checkpoint_session_memory.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
from pathlib import Path

from session_memory_common import (
    CURRENT_LAST_UPDATED_FIELD,
    get_path_freshness,
    iter_resolution_summary,
    resolve_memory_paths,
    run_preflight,
)

SAVE_EVENTS = {
    "goal-changed",
    "approach-changed",
    "research-confirmed",
    "blocker-cleared",
    "next-step-changed",
    "task-completed",
    "session-ending",
}


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Decide whether session memory should be updated now."
    )
    parser.add_argument(
        "--workspace",
        default=".",
        help="Starting directory used to resolve the target location.",
    )
    parser.add_argument(
        "--scope",
        choices=("auto", "workspace", "global"),
        default="auto",
        help="auto=shared project memory, optionally routed by config.toml; workspace=current path only; global=${SESSION_MEMORY_HOME:-$HOME/.session-memory}/global",
    )
    parser.add_argument(
        "--event",
        action="append",
        default=[],
        choices=sorted(SAVE_EVENTS),
        help="Signal that a high-value state transition happened. May be passed multiple times.",
    )
    parser.add_argument(
        "--stale-minutes",
        type=int,
        default=180,
        help="Recommend saving if current.md is older than this threshold.",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Print path resolution, freshness details, and preflight status.",
    )
    args = parser.parse_args()

    paths = resolve_memory_paths(Path(args.workspace).resolve(), args.scope)
    preflight = run_preflight(paths, action="checkpoint")
    current_path = paths["current"]
    history_path = paths["history"]

    if args.verbose:
        for line in iter_resolution_summary(paths):
            print(line)
        print(f"preflight_sleep_status={preflight['sleep_check']['status']}")

    if not current_path.exists() or not history_path.exists():
        print("decision=INIT_FIRST")
        print("reason=memory files are missing")
        print("next_action=run init_session_memory.py")
        return 1

    current_freshness = get_path_freshness(
        current_path, field_names=(CURRENT_LAST_UPDATED_FIELD,)
    )
    current_age = current_freshness.age_minutes
    events = sorted(set(args.event))

    if args.verbose:
        print(f"current_last_updated={current_freshness.raw_value or '-'}")
        print(f"current_freshness={current_freshness.status}")
        print(f"current_freshness_source={current_freshness.source}")
        print(f"current_age_minutes={current_age}")
        print(f"events={','.join(events) if events else '-'}")

    reasons: list[str] = []
    if events:
        reasons.append("high-value event detected")
    if current_age is not None and current_age >= args.stale_minutes:
        reasons.append("current.md needs refresh")
    if current_freshness.status == "stale":
        reasons.append("current.md may no longer reflect the real project state")

    if reasons:
        print("decision=SAVE_NOW")
        print(f"reason={'; '.join(reasons)}")
        if events:
            print(
                "next_action=append past-but-valuable context to history.md, then compact current.md to the latest recovery snapshot"
            )
        else:
            print(
                "next_action=refresh current.md only if the project state has moved; move old-but-valuable context to history.md first"
            )
        return 0

    print("decision=SKIP")
    print("reason=no high-value trigger and current.md is fresh")
    print("next_action=continue current task")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())