Checkpoints

akms.orchestrator.checkpoint

checkpoint.py — File-Based Checkpoint Interface (§3 of system design).

Checkpoints are blocking pause points where the developer reviews and approves before the next stage begins. At each checkpoint, the orchestrator presents:

1. Stage output — the primary artifact (plan, task JSONs, PCD, review report)
2. AKMS status — new/modified nodes, confidence changes, pitfalls added
3. Action menu — stage-specific options (approve, reject, edit, abort)
4. Warnings — any graph health issues

The developer's response is one of: - approve → proceed to next stage - reject [reason] → re-run the stage with feedback - edit → developer modifies artifacts directly, then approves - abort → halt, preserve all state for resumption

File-based protocol: - Orchestrator writes: knowledge/checkpoints/{stage}_{timestamp}.yaml - Developer writes: knowledge/checkpoints/{stage}_{timestamp}_response.yaml

CheckpointData

CheckpointData(
    stage: Stage,
    status: str = "awaiting_review",
    stage_output: str = "",
    akms_status: dict | None = None,
    actions: list[str] | None = None,
    warnings: list[str] | None = None,
    phase: int | None = None,
    timestamp: str | None = None,
)

Data structure for a checkpoint file.

Source code in packages/akms/src/akms/orchestrator/checkpoint.py

def __init__(
    self,
    stage: Stage,
    status: str = "awaiting_review",
    stage_output: str = "",
    akms_status: dict | None = None,
    actions: list[str] | None = None,
    warnings: list[str] | None = None,
    phase: int | None = None,
    timestamp: str | None = None,
):
    self.stage = stage
    self.status = status
    self.stage_output = stage_output
    self.akms_status = akms_status or {}
    self.actions = actions or list(CHECKPOINT_ACTIONS)
    self.warnings = warnings or []
    self.phase = phase
    self.timestamp = timestamp or datetime.now().isoformat()

to_dict

to_dict() -> dict

Serialize to a dict for YAML output.

Source code in packages/akms/src/akms/orchestrator/checkpoint.py

def to_dict(self) -> dict:
    """Serialize to a dict for YAML output."""
    data: dict[str, Any] = {
        "stage": self.stage.name.lower(),
        "status": self.status,
        "timestamp": self.timestamp,
        "stage_output": self.stage_output,
        "akms_status": self.akms_status,
        "actions": self.actions,
        "warnings": self.warnings,
    }
    if self.phase is not None:
        data["phase"] = self.phase
    return data

CheckpointResponse

CheckpointResponse(
    action: str | CheckpointAction,
    reason: str = "",
    edits: dict | None = None,
)

Response from the developer.

Source code in packages/akms/src/akms/orchestrator/checkpoint.py

def __init__(
    self,
    action: str | CheckpointAction,
    reason: str = "",
    edits: dict | None = None,
):
    if isinstance(action, CheckpointAction):
        self.action = action.value
    else:
        self.action = str(action)
    self.reason = reason
    self.edits = edits or {}

CheckpointHandler

Bases: ABC

Abstract base class for checkpoint presentation strategies.

Implementations decide how to present a checkpoint to the developer (file-based polling, terminal prompt, auto-approve, etc.) and return a CheckpointAction indicating the developer's decision.

present abstractmethod

present(
    state: "PipelineState",
    stage_output: str,
    akms_status: str,
    warnings: list[str],
    repo_root: Path,
) -> CheckpointAction

Present a checkpoint and return the developer's action.

Parameters:

Name	Type	Description	Default
state	'PipelineState'	Current pipeline state (stage, phase).	required
stage_output	str	Primary artifact text for review.	required
akms_status	str	AKMS status summary string.	required
warnings	list[str]	List of warning messages to display.	required
repo_root	Path	Repository root path.	required

Returns:

Type	Description
CheckpointAction	CheckpointAction indicating developer decision.

Source code in packages/akms/src/akms/orchestrator/checkpoint.py

@abstractmethod
def present(
    self,
    state: "PipelineState",
    stage_output: str,
    akms_status: str,
    warnings: list[str],
    repo_root: Path,
) -> CheckpointAction:
    """Present a checkpoint and return the developer's action.

    Args:
        state: Current pipeline state (stage, phase).
        stage_output: Primary artifact text for review.
        akms_status: AKMS status summary string.
        warnings: List of warning messages to display.
        repo_root: Repository root path.

    Returns:
        CheckpointAction indicating developer decision.
    """
    ...

FileCheckpointHandler

FileCheckpointHandler(
    timeout: float | None = None, poll_interval: float = 1.0
)

Bases: CheckpointHandler

Checkpoint handler that writes a YAML file and polls for a response file.

Follows the write_checkpoint / read_checkpoint_response protocol. Returns ABORT on timeout, REJECT on unrecognised action string.

Timeout policy: pass an explicit timeout for asynchronous file-based approval (a human answering from another terminal — the design use case for this handler). When timeout is left at its default (None), the wait adapts to the environment: an interactive run (stdin is a TTY) waits :data:INTERACTIVE_TIMEOUT seconds, while a non-interactive run (CI, cron, piped stdin) waits only :data:HEADLESS_TIMEOUT seconds — an unattended process must fail visibly within a minute, not block for an hour on a gate nothing will ever answer. Either way, the wait is announced on stderr with the exact response path and accepted actions.

Source code in packages/akms/src/akms/orchestrator/checkpoint.py

def __init__(self, timeout: float | None = None, poll_interval: float = 1.0):
    self.timeout = timeout
    self.poll_interval = poll_interval

TerminalCheckpointHandler

Bases: CheckpointHandler

Checkpoint handler that uses print() / input() for interactive review.

Suitable for local developer workflows where the orchestrator runs in a terminal session (per addendum §3.3.1, deviation D3).

write_checkpoint

write_checkpoint(
    repo_root: Path, data: CheckpointData
) -> Path

Write a checkpoint YAML file.

Parameters:

Name	Type	Description	Default
repo_root	Path	Repository root path.	required
data	CheckpointData	Checkpoint data to write.	required

Returns:

Type	Description
Path	Path to the written checkpoint file.

Source code in packages/akms/src/akms/orchestrator/checkpoint.py

def write_checkpoint(
    repo_root: Path,
    data: CheckpointData,
) -> Path:
    """Write a checkpoint YAML file.

    Args:
        repo_root: Repository root path.
        data: Checkpoint data to write.

    Returns:
        Path to the written checkpoint file.
    """
    cp_dir = _checkpoint_dir(repo_root)
    base = _checkpoint_filename(data.stage, data.timestamp, data.phase)
    cp_path = cp_dir / f"{base}.yaml"

    with open(cp_path, "w") as f:
        yaml.dump(data.to_dict(), f, default_flow_style=False, sort_keys=False)

    logger.info("Checkpoint written: %s", cp_path)
    return cp_path

read_checkpoint_response

read_checkpoint_response(
    checkpoint_path: Path,
    timeout: float | None = None,
    poll_interval: float = 1.0,
) -> CheckpointResponse | None

Read a checkpoint response file.

The response file name is derived from the checkpoint path by appending _response. Optionally polls for the file if timeout is set.

Parameters:

Name	Type	Description	Default
checkpoint_path	Path	Path to the checkpoint YAML.	required
timeout	float | None	Seconds to wait for response (None = no wait, just check).	None
poll_interval	float	Seconds between polls.	1.0

Returns:

Type	Description
CheckpointResponse | None	CheckpointResponse if found, None if timeout/not found.

Source code in packages/akms/src/akms/orchestrator/checkpoint.py

def read_checkpoint_response(
    checkpoint_path: Path,
    timeout: float | None = None,
    poll_interval: float = 1.0,
) -> CheckpointResponse | None:
    """Read a checkpoint response file.

    The response file name is derived from the checkpoint path by appending
    ``_response``. Optionally polls for the file if ``timeout`` is set.

    Args:
        checkpoint_path: Path to the checkpoint YAML.
        timeout: Seconds to wait for response (None = no wait, just check).
        poll_interval: Seconds between polls.

    Returns:
        CheckpointResponse if found, None if timeout/not found.
    """
    response_path = checkpoint_path.with_name(checkpoint_path.stem + "_response.yaml")

    if timeout is None:
        if response_path.exists():
            with open(response_path) as f:
                data = yaml.safe_load(f)
            return CheckpointResponse.from_dict(data or {})
        return None

    deadline = time.time() + timeout
    while time.time() < deadline:
        if response_path.exists():
            with open(response_path) as f:
                data = yaml.safe_load(f)
            logger.info("Checkpoint response received: %s", response_path)
            return CheckpointResponse.from_dict(data or {})
        time.sleep(poll_interval)

    logger.warning("Checkpoint response timeout after %.1fs", timeout)
    return None

write_checkpoint_response

write_checkpoint_response(
    checkpoint_path: Path,
    action: str,
    reason: str = "",
    edits: dict | None = None,
) -> Path

Write a checkpoint response file (for testing / CLI).

Parameters:

Name	Type	Description	Default
checkpoint_path	Path	Path to the original checkpoint YAML.	required
action	str	One of: approve, reject, edit, abort.	required
reason	str	Required for reject.	''
edits	dict | None	Optional edits dict.	None

Returns:

Type	Description
Path	Path to the written response file.

Source code in packages/akms/src/akms/orchestrator/checkpoint.py

def write_checkpoint_response(
    checkpoint_path: Path,
    action: str,
    reason: str = "",
    edits: dict | None = None,
) -> Path:
    """Write a checkpoint response file (for testing / CLI).

    Args:
        checkpoint_path: Path to the original checkpoint YAML.
        action: One of: approve, reject, edit, abort.
        reason: Required for reject.
        edits: Optional edits dict.

    Returns:
        Path to the written response file.
    """
    response_path = checkpoint_path.with_name(checkpoint_path.stem + "_response.yaml")
    data: dict[str, Any] = {"action": action}
    if reason:
        data["reason"] = reason
    if edits:
        data["edits"] = edits

    with open(response_path, "w") as f:
        yaml.dump(data, f, default_flow_style=False)

    logger.info("Checkpoint response written: %s → %s", action, response_path)
    return response_path

list_checkpoints

list_checkpoints(repo_root: Path) -> list[dict]

List all checkpoint files in the repository.

Returns:

Type	Description
list[dict]	List of dicts with keys: path, stage, status, timestamp, has_response.

Source code in packages/akms/src/akms/orchestrator/checkpoint.py

def list_checkpoints(repo_root: Path) -> list[dict]:
    """List all checkpoint files in the repository.

    Returns:
        List of dicts with keys: path, stage, status, timestamp, has_response.
    """
    cp_dir = _checkpoint_dir(repo_root)
    results = []

    for cp_file in sorted(cp_dir.glob("*.yaml")):
        if cp_file.stem.endswith("_response"):
            continue

        try:
            with open(cp_file) as f:
                data = yaml.safe_load(f) or {}
        except Exception:
            continue

        response_path = cp_file.with_name(cp_file.stem + "_response.yaml")

        results.append(
            {
                "path": str(cp_file),
                "stage": data.get("stage", "unknown"),
                "status": data.get("status", "unknown"),
                "timestamp": data.get("timestamp", ""),
                "has_response": response_path.exists(),
            }
        )

    return results