save_session_memory.py

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

import argparse
import hashlib
import json
import subprocess
import sys
from pathlib import Path

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


SAVE_PREPARE_STATE = ".save-prepare.json"


def content_fingerprint(path: Path) -> str:
    if not path.exists():
        return ""
    target_prefix = f"- {CURRENT_LAST_UPDATED_FIELD}:"
    lines = [
        line
        for line in path.read_text(encoding="utf-8").splitlines()
        if not line.strip().startswith(target_prefix)
    ]
    normalized = "\n".join(lines).rstrip() + "\n"
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()


def prepare_state_path(target_dir: Path) -> Path:
    return target_dir / SAVE_PREPARE_STATE


def write_prepare_state(target_dir: Path, current_path: Path, history_path: Path) -> None:
    if not current_path.exists() and not history_path.exists():
        return
    target_dir.mkdir(parents=True, exist_ok=True)
    payload = {
        "current_fingerprint": content_fingerprint(current_path),
        "history_fingerprint": content_fingerprint(history_path),
    }
    prepare_state_path(target_dir).write_text(
        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )


def read_prepare_state(target_dir: Path) -> dict[str, str]:
    path = prepare_state_path(target_dir)
    if not path.exists():
        return {}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return {}
    if not isinstance(payload, dict):
        return {}
    return {
        "current_fingerprint": str(payload.get("current_fingerprint") or ""),
        "history_fingerprint": str(payload.get("history_fingerprint") or ""),
    }


def clear_prepare_state(target_dir: Path) -> None:
    path = prepare_state_path(target_dir)
    if path.exists():
        path.unlink()


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Prepare session memory save by running preflight and ensuring files exist."
    )
    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(
        "--init-if-missing",
        action="store_true",
        help="Initialize current/history templates if missing.",
    )
    parser.add_argument(
        "--stage",
        choices=("prepare", "commit"),
        default="prepare",
        help="prepare=ensure files and print paths; commit=register activity after you updated memory files.",
    )
    args = parser.parse_args()

    paths = resolve_memory_paths(Path(args.workspace).resolve(), args.scope)
    current_path = paths["current"]
    history_path = paths["history"]
    target_dir = paths["target_dir"]

    for line in iter_resolution_summary(paths):
        print(line)
    print(f"stage={args.stage}")

    if args.init_if_missing and (not current_path.exists() or not history_path.exists()):
        init_script = Path(__file__).resolve().parent / "init_session_memory.py"
        completed = subprocess.run(
            [
                sys.executable,
                str(init_script),
                "--workspace",
                str(args.workspace),
                "--scope",
                str(args.scope),
            ],
            check=False,
        )
        if completed.returncode != 0:
            return completed.returncode

    if args.stage == "commit":
        prepare_state = read_prepare_state(target_dir)
        if not prepare_state:
            print(f"current={current_path}")
            print(f"history={history_path}")
            print("status=invalid")
            print("reason=missing save prepare state; commit must follow --stage prepare")
            print(
                "next_action=rerun with --stage prepare, edit memory files, then run --stage commit"
            )
            return 1
        current_changed = bool(
            content_fingerprint(current_path) != prepare_state["current_fingerprint"]
        )
        history_changed = bool(
            content_fingerprint(history_path) != prepare_state["history_fingerprint"]
        )
        if current_path.exists() and current_changed:
            current_last_updated = update_file_timestamp(
                current_path, field_name=CURRENT_LAST_UPDATED_FIELD
            )
            print(f"current_last_updated={current_last_updated}")
        else:
            print("current_last_updated=unchanged")
        preflight = run_preflight(
            paths,
            action="save",
            mark_active=current_changed or history_changed,
        )
        print(f"preflight_sleep_status={preflight['sleep_check']['status']}")
        clear_prepare_state(target_dir)
    else:
        write_prepare_state(target_dir, current_path, history_path)

    print(f"current={current_path}")
    print(f"history={history_path}")
    print("status=ready")
    if args.stage == "prepare":
        print(
            "next_action=append past-but-valuable context to history.md, compact current.md to the latest recovery snapshot, then rerun with --stage commit"
        )
    else:
        print("next_action=save committed")
    return 0


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