#!/usr/bin/env python3 """SessionStart checkout-health guard (checkout-health-hardening-2026-06-08 §1). Keeps the shared PRIMARY checkout (~/CLAUDE_PROJECTS) fresh against origin/main so the next /dispatch never runs stale code ("merged != running"). It AUTO fast-forwards ONLY in the narrow safe case: clean, on main, stash-free, behind-only, origin/main fetched, and NOT held by a *different live* primary-checkout session. Everything else is warn-loud-only. It NEVER blocks a session and NEVER discards uncommitted work; the only mutation is the clean+behind-only ff-merge. Canonical source lives in the repo for PR review, but the deployed SessionStart copy runs from ~/.claude/hooks. It must not import repo scripts. Machine-agnostic; all paths via Path.home(). Fail-open top-level try/except -> sys.exit(0), mirroring session-collision-warn.py. Liveness of a different lease holder reuses checkout-lease-gate.py's semantics (transcript-fresh-within-window OR heartbeat-TTL-valid), NOT tmux (interactive Claude sessions are not in tmux; see checkout-lease-gate.py:9-21). """ import json import os import subprocess import sys import time from pathlib import Path # Repo + lease locations follow the SAME env knobs as the rest of the system, so they are # configurable per machine rather than hard-coded: CANONICAL_REPO (session_workspace.sh) and # CHECKOUT_LEASE_STATE_DIR (checkout-lease-gate.py). Defaults are the canonical paths. REPO = Path(os.environ.get("CANONICAL_REPO") or (Path.home() / "CLAUDE_PROJECTS")).expanduser() _STATE_DIR = Path(os.environ.get("CHECKOUT_LEASE_STATE_DIR") or (Path.home() / ".claude" / "state")).expanduser() LEASE_FILE = _STATE_DIR / "checkout-lease.json" LIVE_WINDOW_SEC = 600 # mirror checkout-lease-gate.py LIVE_WINDOW_SEC def _run(args, timeout=8): try: return subprocess.run(args, capture_output=True, text=True, timeout=timeout) except Exception: return None def _read_stdin(): try: data = json.load(sys.stdin) return data if isinstance(data, dict) else {} except Exception: return {} def _git(*args, timeout=8): return _run(["git", "-C", str(REPO), *args], timeout=timeout) def _is_git_checkout(): r = _git("rev-parse", "--show-toplevel") if not (r and r.returncode == 0): return False try: return Path(r.stdout.strip()).resolve() == REPO.resolve() except Exception: return r.stdout.strip() == str(REPO) def _current_session_id(payload): # Claude Code hands every hook a stable session_id; used to recognize "the lease is mine". return payload.get("session_id") if isinstance(payload, dict) else None def _holder_is_live(lease, now): """Mirror checkout-lease-gate.py._holder_is_live: transcript-fresh OR heartbeat-TTL-valid.""" transcript = lease.get("transcript") if transcript: try: if (now - os.path.getmtime(transcript)) <= LIVE_WINDOW_SEC: return True except Exception: pass exp = lease.get("expires_epoch") if isinstance(exp, (int, float)): return now < exp return False def _different_live_holder(my_session_id, now): """True if a DIFFERENT live session holds the primary-checkout lease. Corrupt/unreadable lease -> treat as 'cannot confirm safe' -> True (warn-only, no auto-ff). """ try: if not LEASE_FILE.exists(): return False except Exception: return True try: with open(LEASE_FILE) as f: lease = json.load(f) if not isinstance(lease, dict): return True except Exception: return True # corrupt lease -> do not auto-sync holder = lease.get("session_id") if holder and my_session_id and holder == my_session_id: return False # it's mine return _holder_is_live(lease, now) def _warn(lines): print("\n".join(["⚠️ CHECKOUT-HEALTH (primary ~/CLAUDE_PROJECTS)"] + [" " + l for l in lines])) def main(): payload = _read_stdin() if not _is_git_checkout(): # Not the primary git checkout (or git unavailable) -> nothing to do. return 0 # Fetch ONLY origin/main, explicit refspec, quiet. Slow/unreachable -> freshness unknown. fetch = _git("fetch", "origin", "refs/heads/main:refs/remotes/origin/main", "--quiet", timeout=8) fetched_ok = bool(fetch and fetch.returncode == 0) br = _git("symbolic-ref", "--quiet", "--short", "HEAD") branch = br.stdout.strip() if (br and br.returncode == 0) else "DETACHED" dr = _git("status", "--porcelain", "--untracked-files=all") dirty = dr.stdout.strip() if (dr and dr.returncode == 0) else "UNKNOWN" sr = _git("stash", "list") stash_unknown = not (sr and sr.returncode == 0) stash = sr.stdout.strip() if not stash_unknown else "" # ahead/behind vs origin/main ahead, behind = 0, 0 counts_ok = False cr = _git("rev-list", "--left-right", "--count", "HEAD...origin/main") if cr and cr.returncode == 0: parts = cr.stdout.split() if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit(): ahead, behind = int(parts[0]), int(parts[1]) counts_ok = True if not fetched_ok: _warn(["origin/main fetch failed (freshness UNKNOWN) — not auto-syncing. Check connectivity."]) return 0 if branch != "main": _warn([f"on branch '{branch}', not main — NOT auto-syncing (preserving branch work).", "If this is the primary checkout, switch to main and re-open a session to refresh."]) return 0 if dirty == "UNKNOWN": _warn(["could not read working-tree status — not auto-syncing."]) return 0 if dirty: n_dirty = len([ln for ln in dirty.splitlines() if ln.strip()]) lines = [f"working tree has {n_dirty} uncommitted/untracked change(s) — NOT auto-syncing (preserving work)."] if counts_ok and behind > 0: # The behind-count is the load-bearing signal: a dirty tree silently BLOCKS the # 5-min launchd autopull, so local code grounds against stale code that compounds # every time the peer machine merges. Surfacing the count here is what would have # caught the 54-behind incident (2026-06-21) at session start instead of hours later. bang = "🚨 " if behind >= 10 else "⛔ " lines.append(f"{bang}AND you are {behind} commit(s) BEHIND origin/main — autopull is BLOCKED by these changes.") lines.append(" Local code/self-gates ground STALE until you clear them. Fix: commit or stash, then it self-heals.") _warn(lines) return 0 if stash_unknown: _warn(["could not read stash list — not auto-syncing (stash state UNKNOWN)."]) return 0 if stash: _warn(["stash present — NOT auto-syncing (checkout is not operator-clean)."]) return 0 if not counts_ok: _warn(["could not compute ahead/behind vs origin/main — not auto-syncing."]) return 0 if ahead > 0 and behind > 0: _warn([f"DIVERGENT from origin/main (ahead {ahead}, behind {behind}) — NOT auto-syncing."]) return 0 if ahead > 0: _warn([f"local has {ahead} unpushed commit(s) ahead of origin/main — NOT auto-syncing."]) return 0 if behind == 0: return 0 # already up to date, no action, no noise # behind-only & clean & on main & stash-free & fetched. Last gate: no different live holder. now = int(time.time()) if _different_live_holder(_current_session_id(payload), now): _warn([f"behind origin/main by {behind} but another LIVE session holds the primary checkout —", "not auto-syncing while it is in use."]) return 0 ff = _git("merge", "--ff-only", "origin/main") if ff and ff.returncode == 0: print(f"✅ checkout-health: fast-forwarded ~/CLAUDE_PROJECTS to origin/main (+{behind} commit(s)).") else: _warn([f"behind by {behind} but `git merge --ff-only origin/main` failed — not forcing. " "Resolve manually."]) return 0 if __name__ == "__main__": try: sys.exit(main()) except SystemExit: raise except Exception: # Never let the guard block or error a session start. sys.exit(0)