trading_agent.memory.checkpoints

Snapshots of a desk's own files, so any change the desk makes can be undone.

Each desk gets its own git repository in workspace/.checkpoints/<desk>.git, kept apart from the project's git. Its working tree is the desk folder, but only Memory.md, memory/, notes/ and skills/agent/ are tracked; everything else in the folder is ignored. A snapshot is taken before every write the desk makes, so each checkpoint is the state just before one change.

Typical usage example:

checkpoints = Checkpoints(Path('workspace/.checkpoints/main.git'), Path('workspace/desks/main')) checkpoints.snapshot('before write_file Memory.md') checkpoints.restore('3f2a9c1')

TRACKED_PATHS module-attribute

TRACKED_PATHS = ('Memory.md', 'memory', 'notes', 'skills/agent')

CheckpointError

Bases: RuntimeError

git could not take, list or restore a checkpoint.

Checkpoint dataclass

Checkpoint(checkpoint_id, created_at, reason)

One snapshot.

Attributes:

Name	Type	Description
checkpoint_id	str	The snapshot's short git commit id.
created_at	str	When it was taken, as an ISO 8601 string.
reason	str	What was about to happen, such as "before write_file Memory.md".

Checkpoints

Checkpoints(repository_directory, desk_directory)

The checkpoint repository of one desk.

Attributes:

Name	Type	Description
repository_directory		The git directory that holds the snapshots.
desk_directory		The desk folder whose files are snapshotted.

Describes the repository without touching the disk.

Parameters:

Name	Type	Description	Default
repository_directory	Path	The git directory, such as workspace/.checkpoints/main.git.	required
desk_directory	Path	The desk folder.	required

Source code in src/trading_agent/memory/checkpoints.py

def __init__(self, repository_directory: Path, desk_directory: Path):
    """Describes the repository without touching the disk.

    Args:
        repository_directory (Path): The git directory, such as workspace/.checkpoints/main.git.
        desk_directory (Path): The desk folder.
    """
    self.repository_directory = repository_directory
    self.desk_directory = desk_directory
    self._lock = threading.Lock()

snapshot

snapshot(reason)

Records the desk's files as they are now, if they changed since the last snapshot.

Parameters:

Name	Type	Description	Default
reason	str	What is about to happen, kept as the snapshot's message.	required

Returns:

Type	Description
str | None	The new checkpoint's id, or None when nothing had changed.

Raises:

Type	Description
CheckpointError	git failed.

Source code in src/trading_agent/memory/checkpoints.py

def snapshot(self, reason: str) -> str | None:
    """Records the desk's files as they are now, if they changed since the last snapshot.

    Args:
        reason (str): What is about to happen, kept as the snapshot's message.

    Returns:
        str | None: The new checkpoint's id, or None when nothing had changed.

    Raises:
        CheckpointError: git failed.
    """
    with self._lock:
        return self._snapshot(reason)

list

list(limit=20)

Lists the most recent checkpoints, newest first.

Parameters:

Name	Type	Description	Default
limit	int	The most checkpoints to list.	20

Returns:

Type	Description
list[Checkpoint]	The checkpoints, or an empty list when none has been taken.

Raises:

Type	Description
CheckpointError	git failed.

Source code in src/trading_agent/memory/checkpoints.py

def list(self, limit: int = 20) -> list[Checkpoint]:
    """Lists the most recent checkpoints, newest first.

    Args:
        limit (int): The most checkpoints to list.

    Returns:
        list[Checkpoint]: The checkpoints, or an empty list when none has been taken.

    Raises:
        CheckpointError: git failed.
    """
    with self._lock:
        if not self._has_commits():
            return []
        output = self._git(
            'log',
            f'-n{limit}',
            f'--format=%h{_FIELD_SEPARATOR}%cI{_FIELD_SEPARATOR}%s',
        )
    checkpoints = []
    for line in output.splitlines():
        parts = line.split(_FIELD_SEPARATOR)
        if len(parts) != 3:
            continue
        checkpoints.append(
            Checkpoint(
                checkpoint_id=parts[0],
                created_at=parts[1],
                reason=parts[2],
            )
        )
    return checkpoints

restore

restore(checkpoint_id)

Puts the desk's files back as they were at a checkpoint.

The current files are snapshotted first, so a restore can itself be undone. Files created after the checkpoint are removed; files outside Memory.md, memory/, notes/ and skills/agent/ are never touched.

Parameters:

Name	Type	Description	Default
checkpoint_id	str	The checkpoint to return to.	required

Returns:

Type	Description
str	The id of the snapshot taken just before restoring, or of the current state when nothing had changed.

Raises:

Type	Description
CheckpointError	The checkpoint does not exist, or git failed.

Source code in src/trading_agent/memory/checkpoints.py

def restore(self, checkpoint_id: str) -> str:
    """Puts the desk's files back as they were at a checkpoint.

    The current files are snapshotted first, so a restore can itself be undone. Files created after the checkpoint are removed; files outside Memory.md, memory/, notes/ and skills/agent/ are never touched.

    Args:
        checkpoint_id (str): The checkpoint to return to.

    Returns:
        str: The id of the snapshot taken just before restoring, or of the current state when nothing had changed.

    Raises:
        CheckpointError: The checkpoint does not exist, or git failed.
    """
    with self._lock:
        self._snapshot(f'before restoring {checkpoint_id}')
        try:
            self._git(
                'rev-parse', '--verify', f'{checkpoint_id}^{{commit}}'
            )
        except CheckpointError as error:
            raise CheckpointError(
                f'No checkpoint {checkpoint_id!r} in {self.repository_directory}'
            ) from error
        before = self._git('rev-parse', '--short', 'HEAD').strip()
        self._git(
            'restore',
            f'--source={checkpoint_id}',
            '--staged',
            '--worktree',
            '--',
            '.',
        )
        self._commit_if_changed(f'restored {checkpoint_id}')
        return before