symfonic.capabilities.memory.background_observation

background_observation

Payload-free terminal facts for background consolidation.

terminal_record

terminal_record(*, cycle: Any, run_id: str, root_run_id: str, state: Any = None, failed: bool = False) -> dict[str, Any]

Build the bounded record a durable sink may retain after a nap ends.

Source code in src/symfonic/capabilities/memory/background_observation.py

def terminal_record(
    *,
    cycle: Any,
    run_id: str,
    root_run_id: str,
    state: Any = None,
    failed: bool = False,
) -> dict[str, Any]:
    """Build the bounded record a durable sink may retain after a nap ends."""
    status = "failed" if failed else str(getattr(state, "status", "unknown"))
    counts = getattr(state, "ledger", {}) if state is not None else {}
    return {
        "kind": "background_consolidation",
        "cycle": str(getattr(cycle, "value", cycle)),
        "run_id": run_id,
        "root_run_id": root_run_id or run_id,
        "outcome": "failed" if failed else "completed",
        "status": status,
        "failure_code": "cycle_error" if failed else "",
        "counts": {str(name): int(value) for name, value in counts.items()},
    }