trading_agent.workspace.proposal_store

Proposed changes to the user's files, waiting for the user to accept or reject them.

The agent cannot edit Agents.md, User.md, TradingRules.md or a Persona.md. It writes a proposal instead: the complete new text of one file and the reason. Accepting a proposal keeps a copy of the file it replaces.

Typical usage example:

store = ProposalStore(layout) proposal = store.create('main', 'User.md', new_text, 'Filled in from the interview') store.accept(proposal.proposal_id)

ProposalError

Bases: ValueError

A proposal does not exist, or is not pending.

Proposal dataclass

Proposal(proposal_id, desk, file_name, content, reason, created_at, status)

One proposed change.

Attributes:

Name	Type	Description
proposal_id	str	The identifier the user types to accept or reject it.
desk	str	The desk that proposed it.
file_name	str	The user file it would replace.
content	str	The complete new text.
reason	str	Why the desk proposes it.
created_at	str	When it was proposed, as ISO 8601 text in UTC.
status	str	"pending", "accepted" or "rejected".

from_json_dictionary classmethod

from_json_dictionary(dictionary)

Reads a proposal saved by to_json_dictionary.

Parameters:

Name	Type	Description	Default
dictionary	dict[str, Any]	The saved fields.	required

Returns:

Type	Description
Proposal	The proposal.

Source code in src/trading_agent/workspace/proposal_store.py

@classmethod
def from_json_dictionary(cls, dictionary: dict[str, Any]) -> Self:
    """Reads a proposal saved by to_json_dictionary.

    Args:
        dictionary (dict[str, Any]): The saved fields.

    Returns:
        Proposal: The proposal.
    """
    return cls(**dictionary)

to_json_dictionary

to_json_dictionary()

Converts the proposal to a dictionary that json.dumps can write.

Returns:

Type	Description
dict[str, Any]	Every field.

Source code in src/trading_agent/workspace/proposal_store.py

def to_json_dictionary(self) -> dict[str, Any]:
    """Converts the proposal to a dictionary that json.dumps can write.

    Returns:
        dict[str, Any]: Every field.
    """
    return dataclasses.asdict(self)

ProposalStore

ProposalStore(layout)

Saves, lists, accepts and rejects proposals.

Creates the store.

Parameters:

Name	Type	Description	Default
layout	WorkspaceLayout	The workspace whose desks' proposals folders hold the proposals.	required

Source code in src/trading_agent/workspace/proposal_store.py

def __init__(self, layout: workspace_layout.WorkspaceLayout):
    """Creates the store.

    Args:
        layout (workspace_layout.WorkspaceLayout): The workspace whose desks' proposals folders hold the proposals.
    """
    self._layout = layout

create

create(desk_name, file_name, content, reason)

Saves a new pending proposal.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk proposing the change.	required
file_name	str	One of the user's files.	required
content	str	The complete new text.	required
reason	str	Why.	required

Returns:

Type	Description
Proposal	The saved proposal.

Raises:

Type	Description
ValueError	The file is not one of the user's files.

Source code in src/trading_agent/workspace/proposal_store.py

def create(
    self,
    desk_name: str,
    file_name: str,
    content: str,
    reason: str,
) -> Proposal:
    """Saves a new pending proposal.

    Args:
        desk_name (str): The desk proposing the change.
        file_name (str): One of the user's files.
        content (str): The complete new text.
        reason (str): Why.

    Returns:
        Proposal: The saved proposal.

    Raises:
        ValueError: The file is not one of the user's files.
    """
    self._layout.user_owned_file(desk_name, file_name)
    now = datetime.datetime.now(datetime.UTC)
    proposal = Proposal(
        proposal_id=f'{now:%Y%m%d-%H%M%S}-{uuid.uuid4().hex[:4]}',
        desk=desk_name,
        file_name=file_name,
        content=content,
        reason=reason,
        created_at=now.isoformat(),
        status=_PENDING,
    )
    self._save(proposal)
    return proposal

all

all()

Every proposal from every desk, oldest first.

Returns:

Type	Description
list[Proposal]	The proposals.

Source code in src/trading_agent/workspace/proposal_store.py

def all(self) -> list[Proposal]:
    """Every proposal from every desk, oldest first.

    Returns:
        list[Proposal]: The proposals.
    """
    proposals = []
    for path in sorted(self._layout.root.glob('desks/*/proposals/*.json')):
        dictionary = json.loads(path.read_text(encoding='utf-8'))
        proposals.append(Proposal.from_json_dictionary(dictionary))
    proposals.sort(key=lambda proposal: proposal.created_at)
    return proposals

get

get(proposal_id)

Finds one proposal.

Parameters:

Name	Type	Description	Default
proposal_id	str	Its identifier.	required

Returns:

Type	Description
Proposal	The proposal.

Raises:

Type	Description
ProposalError	No proposal has that identifier.

Source code in src/trading_agent/workspace/proposal_store.py

def get(self, proposal_id: str) -> Proposal:
    """Finds one proposal.

    Args:
        proposal_id (str): Its identifier.

    Returns:
        Proposal: The proposal.

    Raises:
        ProposalError: No proposal has that identifier.
    """
    for proposal in self.all():
        if proposal.proposal_id == proposal_id:
            return proposal
    raise ProposalError(f'No proposal with id {proposal_id!r}')

difference

difference(proposal_id)

Shows what accepting a proposal would change, as a unified diff.

Parameters:

Name	Type	Description	Default
proposal_id	str	Its identifier.	required

Returns:

Type	Description
str	The diff, with "-" lines removed and "+" lines added.

Source code in src/trading_agent/workspace/proposal_store.py

def difference(self, proposal_id: str) -> str:
    """Shows what accepting a proposal would change, as a unified diff.

    Args:
        proposal_id (str): Its identifier.

    Returns:
        str: The diff, with "-" lines removed and "+" lines added.
    """
    proposal = self.get(proposal_id)
    target = self._layout.user_owned_file(proposal.desk, proposal.file_name)
    current = ''
    if target.is_file():
        current = target.read_text(encoding='utf-8')
    lines = difflib.unified_diff(
        current.splitlines(keepends=True),
        proposal.content.splitlines(keepends=True),
        fromfile=f'{proposal.file_name} (now)',
        tofile=f'{proposal.file_name} (proposed)',
    )
    return ''.join(lines)

accept

accept(proposal_id)

Replaces the user file with the proposed text, keeping a copy of the old one.

Parameters:

Name	Type	Description	Default
proposal_id	str	Its identifier.	required

Returns:

Type	Description
Path	The file that was replaced.

Raises:

Type	Description
ProposalError	The proposal does not exist or is not pending.

Source code in src/trading_agent/workspace/proposal_store.py

def accept(self, proposal_id: str) -> Path:
    """Replaces the user file with the proposed text, keeping a copy of the old one.

    Args:
        proposal_id (str): Its identifier.

    Returns:
        Path: The file that was replaced.

    Raises:
        ProposalError: The proposal does not exist or is not pending.
    """
    proposal = self._pending(proposal_id)
    target = self._layout.user_owned_file(proposal.desk, proposal.file_name)
    if target.is_file():
        backup = (
            self._layout.proposals_directory(proposal.desk)
            / 'replaced'
            / f'{proposal.proposal_id}-{proposal.file_name}'
        )
        backup.parent.mkdir(parents=True, exist_ok=True)
        backup.write_text(
            target.read_text(encoding='utf-8'), encoding='utf-8'
        )
    target.write_text(proposal.content, encoding='utf-8')
    self._save(dataclasses.replace(proposal, status=_ACCEPTED))
    return target

reject

reject(proposal_id)

Marks a proposal as rejected without changing any file.

Parameters:

Name	Type	Description	Default
proposal_id	str	Its identifier.	required

Raises:

Type	Description
ProposalError	The proposal does not exist or is not pending.

Source code in src/trading_agent/workspace/proposal_store.py

def reject(self, proposal_id: str) -> None:
    """Marks a proposal as rejected without changing any file.

    Args:
        proposal_id (str): Its identifier.

    Raises:
        ProposalError: The proposal does not exist or is not pending.
    """
    proposal = self._pending(proposal_id)
    self._save(dataclasses.replace(proposal, status=_REJECTED))