Fireworks Sessions Saver

{
  "jobs": 2,
  "tasks": [
    {
      "id": "context-loss",
      "promptFiles": ["prompts/01-scene-context-loss.md"],
      "image": "01-scene-context-loss.png",
      "provider": "google",
      "model": "gemini-3-pro-image-preview",
      "ar": "16:9",
      "quality": "2k"
    },
    {
      "id": "new-features",
      "promptFiles": ["prompts/02-infographic-new-features.md"],
      "image": "02-infographic-new-features.png",
      "provider": "google",
      "model": "gemini-3-pro-image-preview",
      "ar": "16:9",
      "quality": "2k"
    }
  ]
}

#!/usr/bin/env bash
# fireworks-sessions-saver — one-command installer
# Usage: curl -fsSL https://raw.githubusercontent.com/yizhiyanhua-ai/fireworks-sessions-saver/main/install.sh | bash

set -euo pipefail

REPO="https://raw.githubusercontent.com/yizhiyanhua-ai/fireworks-sessions-saver/main"
SKILL_DIR="$HOME/.claude/skills/fireworks-sessions-saver"
SETTINGS="$HOME/.claude/settings.json"

GREEN='\033[0;32m'; YELLOW='\033[1;33m'; RED='\033[0;31m'; NC='\033[0m'
info()  { echo -e "${GREEN}✓${NC} $*"; }
warn()  { echo -e "${YELLOW}⚠${NC}  $*"; }
error() { echo -e "${RED}✗${NC} $*"; exit 1; }

echo ""
echo "🔥 fireworks-sessions-saver installer"
echo "──────────────────────────────────────"
echo ""

# ── 1. Preflight ───────────────────────────────────────────────────────────────
command -v python3 >/dev/null 2>&1 || error "Python 3 is required but not found."
command -v claude  >/dev/null 2>&1 || error "Claude Code CLI not found. Install from https://claude.ai/code"

PY_MINOR=$(python3 -c "import sys; print(sys.version_info.minor)")
[[ "$PY_MINOR" -lt 9 ]] && error "Python 3.9+ required (found 3.$PY_MINOR)"

info "Python $(python3 --version) found"
info "Claude Code found at $(command -v claude)"

# ── 2. Create directories ──────────────────────────────────────────────────────
mkdir -p "$SKILL_DIR/scripts"
mkdir -p "$SKILL_DIR/references"
mkdir -p "$HOME/.claude/sessions"
info "Directories ready"

# ── 3. Download skill files ────────────────────────────────────────────────────
echo ""
echo "📥 Downloading skill files..."

for f in SKILL.md; do
  curl -fsSL "$REPO/skill/$f" -o "$SKILL_DIR/$f"
  info "$f"
done

for f in session-format.md log-discovery.md; do
  curl -fsSL "$REPO/skill/references/$f" -o "$SKILL_DIR/references/$f"
  info "references/$f"
done

for f in save_session.py list_sessions.py restore_session.py heartbeat.py diff_session.py dashboard.py; do
  curl -fsSL "$REPO/skill/scripts/$f" -o "$SKILL_DIR/scripts/$f"
  info "scripts/$f"
done

# ── 4. Syntax check ────────────────────────────────────────────────────────────
for f in save_session.py list_sessions.py restore_session.py heartbeat.py diff_session.py dashboard.py; do
  python3 -m py_compile "$SKILL_DIR/scripts/$f" || error "Syntax error in $f"
done
info "All scripts verified (syntax OK)"

# ── 5. Patch settings.json ─────────────────────────────────────────────────────
echo ""
echo "⚙️  Configuring heartbeat hook in $SETTINGS ..."

[[ ! -f "$SETTINGS" ]] && echo '{}' > "$SETTINGS" && warn "Created new $SETTINGS"

python3 - "$SETTINGS" "$SKILL_DIR" <<'PYEOF'
import json, sys
from pathlib import Path

settings_path = Path(sys.argv[1])
skill_dir     = sys.argv[2]

settings = json.loads(settings_path.read_text())
hooks = settings.setdefault("hooks", {})

new_hook = {
    "type": "command",
    "command": f"python3 {skill_dir}/scripts/heartbeat.py \"$PWD\"",
    "async": True,
}
post_entries = hooks.setdefault("PostToolUse", [])
already = any(
    e.get("matcher") == "Write|Edit|Bash" and any(
        h.get("command", "") == new_hook["command"]
        for h in e.get("hooks", [])
    )
    for e in post_entries
)
if not already:
    post_entries.append({"matcher": "Write|Edit|Bash", "hooks": [new_hook]})

# SessionStart hook — auto-init session on every new window
init_hook = {
    "type": "command",
    "command": f"python3 {skill_dir}/scripts/save_session.py --working-dir \"$PWD\" --tool claude-code --action init",
    "async": True,
}
start_entries = hooks.setdefault("SessionStart", [])
already_start = any(
    any(h.get("command", "") == init_hook["command"] for h in e.get("hooks", []))
    for e in start_entries
)
if not already_start:
    start_entries.append({"hooks": [init_hook]})

settings_path.write_text(json.dumps(settings, indent=2, ensure_ascii=False) + "\n")
print("OK")
PYEOF

info "Heartbeat hook registered"
info "SessionStart auto-init hook registered"

# ── 6. Done ────────────────────────────────────────────────────────────────────
echo ""
echo "──────────────────────────────────────"
echo -e "  ${GREEN}Installation complete!${NC}"
echo "──────────────────────────────────────"
echo ""
echo "  Next step → type  /hooks  in Claude Code to reload config."
echo ""
echo "  How it works:"
echo "  • Sessions are auto-tracked after every file write or bash command."
echo "  • On new session start, Claude checks for recoverable previous sessions."
echo "  • Say '保存进度' or 'save session' to create a rich checkpoint."
echo ""
echo "  Repo: https://github.com/yizhiyanhua-ai/fireworks-sessions-saver"
echo ""

MIT License

Copyright (c) 2026 yizhiyanhua-ai

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

#!/usr/bin/env python3
"""Multi-project session dashboard — show all active sessions across all directories.

Usage: python dashboard.py [--all] [--json]

  --all   Include sessions that are expired (> 7 days) but not yet deleted
  --json  Output raw JSON instead of formatted table
"""

import argparse
import json
import sys
from datetime import datetime, timedelta
from pathlib import Path

SESSIONS_DIR = Path.home() / ".claude" / "sessions"
EXPIRY_DAYS = 7


def time_ago(dt: datetime) -> str:
    delta = datetime.now() - dt
    if delta.days > 0:
        return f"{delta.days}d ago"
    h = delta.seconds // 3600
    if h > 0:
        return f"{h}h ago"
    m = delta.seconds // 60
    return f"{m}m ago" if m > 0 else "just now"


def load_all_sessions(include_expired: bool = False) -> list:
    if not SESSIONS_DIR.exists():
        return []

    sessions = []
    cutoff = datetime.now() - timedelta(days=EXPIRY_DAYS)

    for f in sorted(SESSIONS_DIR.glob("*.json")):
        try:
            data = json.loads(f.read_text())
        except Exception:
            continue

        if data.get("status") == "transferred":
            continue

        last_active_str = data.get("last_active", "")
        try:
            last_active = datetime.fromisoformat(last_active_str)
        except Exception:
            continue

        expired = last_active < cutoff
        if expired and not include_expired:
            continue

        checkpoints = data.get("checkpoints", [])
        latest = checkpoints[-1] if checkpoints else {}

        sessions.append({
            "file": str(f),
            "session_id": data.get("session_id", "?"),
            "tool": data.get("tool", "unknown"),
            "working_dir": data.get("working_dir", "?"),
            "start_time": data.get("start_time", "?"),
            "last_active": last_active,
            "last_active_str": last_active_str,
            "status": "expired" if expired else data.get("status", "active"),
            "checkpoint_count": len(checkpoints),
            "summary": latest.get("summary", ""),
            "current_task": latest.get("current_task", ""),
            "git_branch": latest.get("git_branch", ""),
            "file_count": len(latest.get("files", [])),
        })

    sessions.sort(key=lambda s: s["last_active"], reverse=True)
    return sessions


def print_dashboard(sessions: list) -> None:
    if not sessions:
        print("No active sessions found.")
        print(f"Sessions are stored in: {SESSIONS_DIR}")
        return

    active   = [s for s in sessions if s["status"] == "active"]
    expired  = [s for s in sessions if s["status"] == "expired"]

    print("=" * 70)
    print(f"  SESSION DASHBOARD  —  {len(sessions)} session(s) found")
    print("=" * 70)

    if active:
        print(f"\n  ACTIVE  ({len(active)})\n")
        for i, s in enumerate(active, 1):
            ago = time_ago(s["last_active"])
            branch = f"  [{s['git_branch']}]" if s["git_branch"] else ""
            print(f"  [{i}] {s['last_active'].strftime('%Y-%m-%d %H:%M')} ({ago})  —  {s['tool']}{branch}")
            print(f"      Dir:   {s['working_dir']}")
            if s["summary"]:
                print(f"      Last:  {s['summary'][:60]}")
            if s["current_task"]:
                print(f"      Task:  {s['current_task'][:60]}")
            print(f"      CPs:   {s['checkpoint_count']}  ·  Files: {s['file_count']}")
            print(f"      File:  {s['file']}")
            print()

    if expired:
        print(f"  EXPIRED  ({len(expired)})  —  older than {EXPIRY_DAYS} days\n")
        for s in expired:
            ago = time_ago(s["last_active"])
            print(f"  · {s['last_active'].strftime('%Y-%m-%d')} ({ago})  {s['working_dir']}")
        print()

    print("=" * 70)
    print(f"  Storage: {SESSIONS_DIR}")
    print("=" * 70)


def main() -> int:
    parser = argparse.ArgumentParser(description="Multi-project session dashboard")
    parser.add_argument("--all", action="store_true", help="Include expired sessions")
    parser.add_argument("--json", action="store_true", help="Output raw JSON")
    args = parser.parse_args()

    sessions = load_all_sessions(include_expired=args.all)

    if args.json:
        output = [
            {k: v for k, v in s.items() if k != "last_active"}
            for s in sessions
        ]
        print(json.dumps(output, indent=2, ensure_ascii=False))
        return 0

    print_dashboard(sessions)
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Show a diff between two checkpoints in a session file.

Usage:
  python diff_session.py --session-file <path> [--from N] [--to M]

  N and M are 1-based checkpoint indices (default: last two checkpoints).
  Use --list to show all available checkpoints.
"""

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path


def time_ago(iso: str) -> str:
    try:
        delta = datetime.now() - datetime.fromisoformat(iso)
        if delta.days > 0:
            return f"{delta.days}d ago"
        h = delta.seconds // 3600
        if h > 0:
            return f"{h}h ago"
        m = delta.seconds // 60
        return f"{m}m ago" if m > 0 else "just now"
    except Exception:
        return iso


def diff_list(old: list, new: list) -> list[str]:
    old_set, new_set = set(old), set(new)
    lines = []
    for item in old_set - new_set:
        lines.append(f"  - {item}")
    for item in new_set - old_set:
        lines.append(f"  + {item}")
    return lines


def diff_files(old: list, new: list) -> list[str]:
    old_map = {f["path"]: f for f in old}
    new_map = {f["path"]: f for f in new}
    lines = []
    for path in set(old_map) - set(new_map):
        lines.append(f"  - {path}  [{old_map[path].get('role', '')}]")
    for path in set(new_map) - set(old_map):
        lines.append(f"  + {path}  [{new_map[path].get('role', '')}]")
    for path in set(old_map) & set(new_map):
        o, n = old_map[path], new_map[path]
        if o.get("role") != n.get("role"):
            lines.append(f"  ~ {path}  {o.get('role','')} → {n.get('role','')}")
        elif o.get("note") != n.get("note") and n.get("note"):
            lines.append(f"  ~ {path}  note: {n.get('note','')}")
    return lines


def main() -> int:
    parser = argparse.ArgumentParser(description="Diff two session checkpoints")
    parser.add_argument("--session-file", required=True)
    parser.add_argument("--from", dest="from_idx", type=int, default=None,
                        help="1-based index of the older checkpoint")
    parser.add_argument("--to", dest="to_idx", type=int, default=None,
                        help="1-based index of the newer checkpoint")
    parser.add_argument("--list", action="store_true", help="List all checkpoints")
    args = parser.parse_args()

    path = Path(args.session_file)
    if not path.exists():
        print(f"File not found: {path}", file=sys.stderr)
        return 1

    try:
        session = json.loads(path.read_text())
    except Exception as e:
        print(f"Failed to parse session file: {e}", file=sys.stderr)
        return 1

    checkpoints = session.get("checkpoints", [])
    if not checkpoints:
        print("No checkpoints found in this session.")
        return 0

    if args.list:
        print(f"Checkpoints in {path.name}:")
        for i, cp in enumerate(checkpoints, 1):
            print(f"  [{i}] {cp.get('timestamp', '?')}  —  {cp.get('summary', '(no summary)')[:60]}")
        return 0

    n = len(checkpoints)
    from_idx = args.from_idx if args.from_idx is not None else max(1, n - 1)
    to_idx   = args.to_idx   if args.to_idx   is not None else n

    if not (1 <= from_idx <= n) or not (1 <= to_idx <= n):
        print(f"Index out of range. Session has {n} checkpoint(s).", file=sys.stderr)
        return 1
    if from_idx == to_idx:
        print("--from and --to must be different checkpoints.", file=sys.stderr)
        return 1

    old_cp = checkpoints[from_idx - 1]
    new_cp = checkpoints[to_idx - 1]

    print("=" * 60)
    print(f"CHECKPOINT DIFF  [{from_idx}] → [{to_idx}]")
    print("=" * 60)
    print(f"From:  {old_cp.get('timestamp', '?')}  ({time_ago(old_cp.get('timestamp',''))})")
    print(f"To:    {new_cp.get('timestamp', '?')}  ({time_ago(new_cp.get('timestamp',''))})")
    print()

    changed = False

    # Task
    old_task = old_cp.get("current_task") or ""
    new_task = new_cp.get("current_task") or ""
    if old_task != new_task:
        changed = True
        print("── Task ──────────────────────────────────────────────")
        if old_task:
            print(f"  - {old_task}")
        if new_task:
            print(f"  + {new_task}")
        print()

    # Summary
    old_sum = old_cp.get("summary") or ""
    new_sum = new_cp.get("summary") or ""
    if old_sum != new_sum:
        changed = True
        print("── Summary ───────────────────────────────────────────")
        if old_sum:
            print(f"  - {old_sum}")
        if new_sum:
            print(f"  + {new_sum}")
        print()

    # Files
    file_lines = diff_files(old_cp.get("files", []), new_cp.get("files", []))
    if file_lines:
        changed = True
        print("── Files ─────────────────────────────────────────────")
        for line in file_lines:
            print(line)
        print()

    # Decisions
    dec_lines = diff_list(
        old_cp.get("key_decisions", []),
        new_cp.get("key_decisions", [])
    )
    if dec_lines:
        changed = True
        print("── Key Decisions ─────────────────────────────────────")
        for line in dec_lines:
            print(line)
        print()

    # Open questions
    q_lines = diff_list(
        old_cp.get("open_questions", []),
        new_cp.get("open_questions", [])
    )
    if q_lines:
        changed = True
        print("── Open Questions ────────────────────────────────────")
        for line in q_lines:
            print(line)
        print()

    # Git branch
    old_branch = old_cp.get("git_branch")
    new_branch = new_cp.get("git_branch")
    if old_branch != new_branch:
        changed = True
        print("── Git Branch ────────────────────────────────────────")
        print(f"  - {old_branch or '(none)'}")
        print(f"  + {new_branch or '(none)'}")
        print()

    if not changed:
        print("No differences found between the two checkpoints.")

    print("=" * 60)
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Lightweight heartbeat — updates last_active and records recently touched files.

Called automatically via PostToolUse hook. Designed to be fast and silent.
Does NOT create a full checkpoint; that's the AI's responsibility.

Usage: python heartbeat.py [working_dir]
"""

import hashlib
import json
import os
import subprocess
import sys
from datetime import datetime
from pathlib import Path

SESSIONS_DIR = Path.home() / ".claude" / "sessions"


def now_iso() -> str:
    return datetime.now().strftime("%Y-%m-%dT%H:%M:%S")


def workdir_hash(working_dir: str) -> str:
    return hashlib.md5(working_dir.encode()).hexdigest()[:8]


def get_modified_files(working_dir: str) -> "list[str]":
    """Return absolute paths of files modified in git working tree."""
    try:
        out = subprocess.check_output(
            ["git", "diff", "--name-only", "HEAD"],
            cwd=working_dir, stderr=subprocess.DEVNULL
        ).decode().strip()
        files = [os.path.join(working_dir, f) for f in out.split("\n") if f]
        # Also include untracked files
        untracked = subprocess.check_output(
            ["git", "ls-files", "--others", "--exclude-standard"],
            cwd=working_dir, stderr=subprocess.DEVNULL
        ).decode().strip()
        files += [os.path.join(working_dir, f) for f in untracked.split("\n") if f]
        return files[:10]  # cap to avoid bloat
    except Exception:
        return []


def main():
    working_dir = os.path.abspath(sys.argv[1] if len(sys.argv) > 1 else os.getcwd())
    ap = SESSIONS_DIR / f"active_{workdir_hash(working_dir)}.json"

    if not ap.exists():
        return  # no active session, nothing to do

    try:
        session = json.loads(ap.read_text())
    except Exception:
        return

    session["last_active"] = now_iso()

    # Merge recently modified files into the latest checkpoint's file list
    modified = get_modified_files(working_dir)
    if modified and session.get("checkpoints"):
        latest = session["checkpoints"][-1]
        existing_paths = {f["path"] for f in latest.get("files", [])}
        for fpath in modified:
            if fpath not in existing_paths:
                p = Path(fpath)
                mtime = None
                try:
                    mtime = datetime.fromtimestamp(p.stat().st_mtime).strftime("%Y-%m-%dT%H:%M:%S")
                except Exception:
                    pass
                latest.setdefault("files", []).append({
                    "path": fpath,
                    "last_modified": mtime,
                    "role": "editing",
                    "note": ""
                })
        # Keep file list bounded
        latest["files"] = latest["files"][-20:]

    try:
        ap.write_text(json.dumps(session, indent=2, ensure_ascii=False))
    except Exception:
        pass  # silent failure — heartbeat must never break the user's workflow


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""List recoverable sessions for a working directory.

Usage: python list_sessions.py [working_dir]
"""

import hashlib
import json
import os
import sys
from datetime import datetime, timedelta
from pathlib import Path

SESSIONS_DIR = Path.home() / ".claude" / "sessions"
EXPIRY_DAYS = 7


def workdir_hash(working_dir: str) -> str:
    return hashlib.md5(working_dir.encode()).hexdigest()[:8]


def time_ago(dt: datetime) -> str:
    delta = datetime.now() - dt
    if delta.days > 0:
        return f"{delta.days} day(s) ago"
    hours = delta.seconds // 3600
    if hours > 0:
        return f"{hours} hour(s) ago"
    minutes = delta.seconds // 60
    return f"{minutes} minute(s) ago" if minutes > 0 else "just now"


def load_session_summary(path: Path) -> dict | None:
    try:
        data = json.loads(path.read_text())
        if data.get("status") == "transferred":
            return None
        last_active = datetime.fromisoformat(data.get("last_active", data["start_time"]))
        if last_active < datetime.now() - timedelta(days=EXPIRY_DAYS):
            return None
        checkpoints = data.get("checkpoints", [])
        if not checkpoints:
            return None
        latest = checkpoints[-1]
        return {
            "file": str(path),
            "tool": data.get("tool", "unknown"),
            "last_active": last_active,
            "summary": latest.get("summary", "No summary"),
            "current_task": latest.get("current_task", ""),
            "open_questions": latest.get("open_questions", []),
            "files": [f["path"] for f in latest.get("files", [])],
            "checkpoint_count": len(checkpoints),
        }
    except Exception:
        return None


def main():
    working_dir = os.path.abspath(sys.argv[1] if len(sys.argv) > 1 else os.getcwd())
    wh = workdir_hash(working_dir)

    if not SESSIONS_DIR.exists():
        print("No previous sessions found.")
        return

    sessions = []

    # Check archive files for this working dir
    for f in SESSIONS_DIR.glob(f"archive_{wh}_*.json"):
        s = load_session_summary(f)
        if s:
            sessions.append(s)

    # Check active file — may be from a crashed previous session
    active = SESSIONS_DIR / f"active_{wh}.json"
    s = load_session_summary(active)
    if s:
        sessions.append(s)

    sessions.sort(key=lambda x: x["last_active"], reverse=True)

    if not sessions:
        print("No previous sessions found for this project.")
        return

    print(f"Found {len(sessions)} previous session(s) for: {working_dir}\n")
    for i, s in enumerate(sessions, 1):
        ago = time_ago(s["last_active"])
        print(f"[{i}] {s['last_active'].strftime('%Y-%m-%d %H:%M')} ({ago}) — {s['tool']}")
        print(f"    Summary:  {s['summary']}")
        if s["current_task"]:
            print(f"    Task:     {s['current_task']}")
        if s["files"]:
            print(f"    Files:    {', '.join(s['files'][:3])}")
        if s["open_questions"]:
            print(f"    Open:     {s['open_questions'][0]}")
        print(f"    File:     {s['file']}")
        print()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Print a structured context summary from a session archive file.

Usage: python restore_session.py --session-file <path>
"""

import argparse
import json
from datetime import datetime
from pathlib import Path


def time_ago(iso: str) -> str:
    try:
        dt = datetime.fromisoformat(iso)
        delta = datetime.now() - dt
        if delta.days > 0:
            return f"{delta.days} day(s) ago"
        hours = delta.seconds // 3600
        if hours > 0:
            return f"{hours} hour(s) ago"
        minutes = delta.seconds // 60
        return f"{minutes} minute(s) ago" if minutes > 0 else "just now"
    except Exception:
        return iso


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--session-file", required=True)
    args = parser.parse_args()

    path = Path(args.session_file)
    if not path.exists():
        print(f"Session file not found: {path}")
        return 1

    data = json.loads(path.read_text())
    checkpoints = data.get("checkpoints", [])

    print("=" * 60)
    print("SESSION CONTEXT RESTORE")
    print("=" * 60)
    print(f"Tool:         {data.get('tool', 'unknown')}")
    print(f"Working dir:  {data.get('working_dir', '')}")
    print(f"Session ID:   {data.get('session_id', '')}")
    print(f"Started:      {data.get('start_time', '')} ({time_ago(data.get('start_time', ''))})")
    print(f"Last active:  {data.get('last_active', '')} ({time_ago(data.get('last_active', ''))})")
    print(f"Checkpoints:  {len(checkpoints)}")
    print()

    if not checkpoints:
        print("No checkpoints recorded.")
        return 0

    # Most recent checkpoint is the primary restore point
    latest = checkpoints[-1]
    print("─" * 60)
    print("LATEST STATE")
    print("─" * 60)

    if latest.get("summary"):
        print(f"Summary:      {latest['summary']}")
    if latest.get("current_task"):
        print(f"Current task: {latest['current_task']}")
    if latest.get("git_branch"):
        print(f"Git branch:   {latest['git_branch']}")

    commits = latest.get("git_recent_commits", [])
    if commits:
        print("Recent commits:")
        for c in commits:
            print(f"  {c}")

    files = latest.get("files", [])
    if files:
        print("Files in context:")
        for f in files:
            role = f.get("role", "")
            note = f.get("note", "")
            mtime = f.get("last_modified", "")
            exists = Path(f["path"]).exists()
            status = "" if exists else " [MISSING]"
            line = f"  [{role}] {f['path']}{status}"
            if mtime:
                line += f"  (modified {mtime})"
            if note:
                line += f"\n         → {note}"
            print(line)

    decisions = latest.get("key_decisions", [])
    if decisions:
        print("Key decisions:")
        for d in decisions:
            print(f"  • {d}")

    questions = latest.get("open_questions", [])
    if questions:
        print("Open questions / blockers:")
        for q in questions:
            print(f"  ? {q}")

    commands = latest.get("recent_commands", [])
    if commands:
        print("Recent commands:")
        for c in commands:
            print(f"  $ {c}")

    log_refs = latest.get("log_refs", {})
    cc_log = log_refs.get("claude_code")
    codex_log = log_refs.get("codex")
    if cc_log or codex_log:
        print("Log files (for deeper context):")
        if cc_log:
            exists = Path(cc_log).exists()
            print(f"  Claude Code: {cc_log}{'' if exists else ' [NOT FOUND]'}")
        if codex_log:
            exists = Path(codex_log).exists()
            print(f"  Codex:       {codex_log}{'' if exists else ' [NOT FOUND]'}")

    # Show older checkpoints as a timeline if there are multiple
    if len(checkpoints) > 1:
        print()
        print("─" * 60)
        print("CHECKPOINT HISTORY")
        print("─" * 60)
        for i, cp in enumerate(reversed(checkpoints[:-1]), 1):
            ts = cp.get("timestamp", "")
            summary = cp.get("summary", "—")
            task = cp.get("current_task", "")
            print(f"  [{i} back] {ts}  {summary}")
            if task and task != summary:
                print(f"           task: {task}")

    print()
    print("=" * 60)
    print("Restore complete. Use the above context to resume work.")
    print(f"Session file: {path}")
    print("=" * 60)

    return 0


if __name__ == "__main__":
    exit(main())

#!/usr/bin/env python3
"""Save or update a session checkpoint.

Actions:
  init        Create a new active session (archives existing one if it has checkpoints)
  checkpoint  Add a checkpoint to the active session
  cleanup     Delete an archive file after successful transfer
"""

import argparse
import hashlib
import json
import os
import subprocess
import uuid
from datetime import datetime
from pathlib import Path

SESSIONS_DIR = Path.home() / ".claude" / "sessions"
MAX_CHECKPOINTS = 10


def workdir_hash(working_dir: str) -> str:
    return hashlib.md5(working_dir.encode()).hexdigest()[:8]


def now_iso() -> str:
    return datetime.now().strftime("%Y-%m-%dT%H:%M:%S")


def now_ts() -> str:
    return datetime.now().strftime("%Y%m%d_%H%M%S")


def active_path(working_dir: str) -> Path:
    return SESSIONS_DIR / f"active_{workdir_hash(working_dir)}.json"


def get_git_info(working_dir: str) -> dict:
    info = {"branch": None, "recent_commits": []}
    try:
        branch = subprocess.check_output(
            ["git", "branch", "--show-current"],
            cwd=working_dir, stderr=subprocess.DEVNULL
        ).decode().strip()
        info["branch"] = branch or None

        commits = subprocess.check_output(
            ["git", "log", "--oneline", "-5"],
            cwd=working_dir, stderr=subprocess.DEVNULL
        ).decode().strip()
        info["recent_commits"] = [c for c in commits.split("\n") if c]
    except Exception:
        pass
    return info


def get_log_refs(tool: str) -> dict:
    refs = {"claude_code": None, "codex": None}  # type: dict

    if tool in ("claude-code", "claude"):
        projects_dir = Path.home() / ".claude" / "projects"
        if projects_dir.exists():
            all_logs = sorted(
                projects_dir.rglob("*.jsonl"),
                key=lambda f: f.stat().st_mtime,
                reverse=True
            )
            if all_logs:
                refs["claude_code"] = str(all_logs[0])

    if tool == "codex":
        codex_dir = Path.home() / ".codex"
        if codex_dir.exists():
            all_logs = sorted(
                list(codex_dir.rglob("*.json")) + list(codex_dir.rglob("*.jsonl")),
                key=lambda f: f.stat().st_mtime,
                reverse=True
            )
            if all_logs:
                refs["codex"] = str(all_logs[0])

    return refs


def parse_files(files_str: str) -> list:
    """Parse 'path:role,path:role' into file ref objects."""
    result = []
    if not files_str:
        return result
    for entry in files_str.split(","):
        entry = entry.strip()
        if not entry:
            continue
        if ":" in entry:
            path, role = entry.rsplit(":", 1)
        else:
            path, role = entry, "reference"
        p = Path(path.strip())
        mtime = None
        if p.exists():
            mtime = datetime.fromtimestamp(p.stat().st_mtime).strftime("%Y-%m-%dT%H:%M:%S")
        result.append({
            "path": str(p),
            "last_modified": mtime,
            "role": role.strip(),
            "note": ""
        })
    return result


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--working-dir", required=True)
    parser.add_argument("--tool", default="claude-code")
    parser.add_argument("--action", choices=["init", "checkpoint", "cleanup"], required=True)
    parser.add_argument("--summary", default="")
    parser.add_argument("--current-task", default="")
    parser.add_argument("--files", default="",
                        help="Comma-separated 'path:role' pairs, e.g. src/auth.ts:editing,README.md:reference")
    parser.add_argument("--decisions", default="", help="Semicolon-separated key decisions")
    parser.add_argument("--open-questions", default="", help="Semicolon-separated open questions")
    parser.add_argument("--commands", default="", help="Semicolon-separated recent commands")
    parser.add_argument("--archive-file", default="", help="Archive file to delete (for cleanup action)")
    args = parser.parse_args()

    SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
    working_dir = os.path.abspath(args.working_dir)

    if args.action == "init":
        ap = active_path(working_dir)
        if ap.exists():
            existing = json.loads(ap.read_text())
            if existing.get("checkpoints"):
                arch = SESSIONS_DIR / f"archive_{workdir_hash(working_dir)}_{now_ts()}.json"
                ap.rename(arch)
                print(f"Archived previous session: {arch.name}")

        session = {
            "session_id": uuid.uuid4().hex[:8],
            "tool": args.tool,
            "working_dir": working_dir,
            "start_time": now_iso(),
            "last_active": now_iso(),
            "status": "active",
            "checkpoints": []
        }
        ap.write_text(json.dumps(session, indent=2, ensure_ascii=False))
        print(f"Session initialized: {ap}")

    elif args.action == "checkpoint":
        ap = active_path(working_dir)
        if not ap.exists():
            print("No active session found. Run --action init first.")
            return

        session = json.loads(ap.read_text())
        git = get_git_info(working_dir)
        log_refs = get_log_refs(args.tool)

        checkpoint = {
            "timestamp": now_iso(),
            "summary": args.summary,
            "current_task": args.current_task,
            "git_branch": git["branch"],
            "git_recent_commits": git["recent_commits"],
            "files": parse_files(args.files),
            "key_decisions": [d.strip() for d in args.decisions.split(";") if d.strip()],
            "open_questions": [q.strip() for q in args.open_questions.split(";") if q.strip()],
            "recent_commands": [c.strip() for c in args.commands.split(";") if c.strip()],
            "log_refs": log_refs
        }

        checkpoints = session.get("checkpoints", [])
        checkpoints.append(checkpoint)
        session["checkpoints"] = checkpoints[-MAX_CHECKPOINTS:]
        session["last_active"] = now_iso()

        ap.write_text(json.dumps(session, indent=2, ensure_ascii=False))
        print(f"Checkpoint saved ({len(session['checkpoints'])}/{MAX_CHECKPOINTS})")

    elif args.action == "cleanup":
        target = Path(args.archive_file) if args.archive_file else None
        if target and target.exists():
            target.unlink()
            print(f"Deleted transferred session: {target.name}")
        else:
            print("No archive file specified or file not found.")


if __name__ == "__main__":
    main()