#!/usr/bin/env python3 """session_ctx.py -- session-model derive helpers (SE port, S148 2026-07-24). SE PORT of AI_INFRA scripts/session_ctx.py -- 🔴 TRIMMED ON PURPOSE, KHAI RO: hub's original (~26 KB) also carries jsonl transcript parsing, anh-message containment verification, token-overhead accounting and a context-cap getter. SE ports only the TWO capabilities its rituals actually call: machine-block -> derive {anchor, changed_files, run_id} for /snapshot + /pause secrets-sweep -> pre-commit Category-5 gate over a session dir NOT ported (do NOT claim these exist here): verify-containment / overhead / cap-getter / window-jsonl resolution. Porting them without a caller would be ghost-wire -- the exact class SE keeps getting bitten by. Add one ONLY when a ritual actually needs it. Fidelity pin: field semantics + CLI shape follow hub so a future hub change can be re-pulled by diff. Behaviour differences from hub are marked `SE-DELTA`. stdlib-only . Python 3.11+ . UTF-8 no-BOM . LF . ASCII-only source. """ from __future__ import annotations import argparse import datetime import json import re import subprocess import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent DEFAULT_SESSIONS_ROOT = ROOT / ".claude" / "sessions" # Commits authored by the Stop-hook (wal-flush.ps1) -- never a valid anchor. _WAL_SUBJECT_RE = re.compile(r'^wal:') # runId as written into WAL by the hmw run-trace convention. _RUNID_RE = re.compile(r'\b(wf_[A-Za-z0-9_-]+)') # Category-5 secret prefixes. ACCEPTED-GAP: pattern-bounded, token-anchored -- this is a # conservative pre-commit BLOCK, NOT a proof of cleanliness (a DB password in an anh-msg # verbatim would pass). Say so; do not sell it as complete. SECRET_PATTERNS = ('voyage', 'sk-', 'gitea_pat', 'cfut_') _SECRET_RES = [(p, re.compile(r'\b' + re.escape(p), re.I)) for p in SECRET_PATTERNS] # --------------------------------------------------------------------------- # git helpers # --------------------------------------------------------------------------- def _git(args, repo_root): """Run `git -C `; return stdout. Fail-loud: a bad range is a real error for a derive that must be reproducible.""" try: r = subprocess.run(['git', '-C', str(repo_root)] + list(args), capture_output=True, text=True, encoding='utf-8', errors='replace') except (OSError, subprocess.SubprocessError) as e: raise SystemExit(f"[session_ctx] git failed to launch: {e}") if r.returncode != 0: raise SystemExit(f"[session_ctx] git {' '.join(args)} -> rc={r.returncode}: {r.stderr.strip()}") return r.stdout def _parse_log_line(line): parts = line.split('\x1f') return {'sha': parts[0], 'subject': parts[1], 'ts': parts[2]} if len(parts) == 3 else None def find_anchor(repo_root, max_scan=100): """Nearest NON-wal named commit walking git log from HEAD -> {sha, subject, ts} or None. Why not HEAD: the Stop-hook auto-commits `wal:` at every turn boundary, and closeout squash rewrites those shas. Anchoring on a named commit survives the rewrite; anchoring on HEAD rots (sha AND subject/ts both move).""" out = _git(['log', f'-n{max_scan}', '--format=%H%x1f%s%x1f%cI'], repo_root) for line in out.splitlines(): rec = _parse_log_line(line) if rec is None or _WAL_SUBJECT_RE.match(rec['subject']): continue return rec return None def _resolve_ref(ref, repo_root): """Resolve an explicit anchor ref (sha/tag/HEAD~n) to {sha, subject, ts}.""" out = _git(['log', '-n1', '--format=%H%x1f%s%x1f%cI', ref], repo_root).strip() return _parse_log_line(out.splitlines()[0]) if out else None def _iso_utc(epoch): return datetime.datetime.fromtimestamp(epoch, datetime.timezone.utc).strftime('%Y-%m-%dT%H:%M:%S') def last_snapshot_mtime(session_dir): """mtime of the most recent _snapshot-* file (fallback --since anchor). None if absent.""" snaps = [s for s in Path(session_dir).glob('_snapshot-*') if s.is_file()] return max(s.stat().st_mtime for s in snaps) if snaps else None def live_run_id(repo_root=None, wal_path=None): """Live runId from .claude/WAL.md (LAST match = most-recently-appended). None if absent.""" wal = Path(wal_path) if wal_path else (Path(repo_root or ROOT) / ".claude" / "WAL.md") if not wal.exists(): return None matches = _RUNID_RE.findall(wal.read_text(encoding='utf-8', errors='replace')) return matches[-1] if matches else None # --------------------------------------------------------------------------- # (a) machine-block # --------------------------------------------------------------------------- def machine_block(session_n, sessions_root=None, repo_root=None, anchor=None): """Derive {session, anchor{subject,ts}, changed_files, changed_count, run_id}. `anchor` = explicit ref override (pin it for reproducibility); default = nearest non-wal. Anchor is identified by {subject, ts}, NOT sha -- sha does not survive closeout squash.""" repo_root = Path(repo_root or ROOT) sessions_root = Path(sessions_root) if sessions_root else DEFAULT_SESSIONS_ROOT session_dir = sessions_root / f"session-{session_n}" a = _resolve_ref(anchor, repo_root) if anchor is not None else find_anchor(repo_root) used_fallback = False if a is not None: files_out = _git(['diff', '--name-only', f'{a["sha"]}..HEAD'], repo_root) anchor_id = {'subject': a['subject'], 'ts': a['ts']} else: used_fallback = True mt = last_snapshot_mtime(session_dir) files_out = (_git(['log', f'--since={_iso_utc(mt)}', '--name-only', '--format='], repo_root) if mt is not None else '') anchor_id = {'subject': None, 'ts': None} changed = sorted({f.strip() for f in files_out.splitlines() if f.strip()}) return { 'session': session_n, 'anchor': anchor_id, 'anchor_from_fallback_since': used_fallback, 'changed_files': changed, 'changed_count': len(changed), 'run_id': live_run_id(repo_root), } def render_machine_block(mb): """Deterministic text rendering for embedding into a _context FLOW entry.""" a = mb['anchor'] lines = [''] lines.append(f'anchor: "{a["subject"]}" @ {a["ts"]}' if a['subject'] is not None else 'anchor: (fallback --since mtime of last _snapshot-*)') lines.append(f'runId: {mb["run_id"] or "(none)"}') lines.append(f'changed-files ({mb["changed_count"]}, git diff --name-only anchor..HEAD):') lines.extend(f' {f}' for f in mb['changed_files']) return '\n'.join(lines) # --------------------------------------------------------------------------- # (b) secrets-sweep # --------------------------------------------------------------------------- def secrets_sweep(session_dir): """Scan the WHOLE write-set of a session dir for Category-5 patterns. Returns list of hits {file, line, pattern, snippet}. See ACCEPTED-GAP at module top.""" session_dir = Path(session_dir) hits = [] if not session_dir.exists(): return hits for p in sorted(session_dir.rglob('*')): if not p.is_file(): continue try: text = p.read_text(encoding='utf-8', errors='replace') except OSError: continue for lineno, line in enumerate(text.splitlines(), 1): for pat, rx in _SECRET_RES: if rx.search(line): hits.append({'file': str(p), 'line': lineno, 'pattern': pat, 'snippet': line.strip()[:120]}) return hits # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main(argv=None): ap = argparse.ArgumentParser(prog='session_ctx.py', description='session-model derive helpers (SE port, trimmed).') sub = ap.add_subparsers(dest='cmd', required=True) mb = sub.add_parser('machine-block', help='derive anchor + changed-files + runId') mb.add_argument('--session', type=int, required=True, metavar='N') mb.add_argument('--anchor', default=None, help='explicit anchor ref (default: nearest non-wal)') mb.add_argument('--json', action='store_true', help='emit JSON (default: rendered text)') ss = sub.add_parser('secrets-sweep', help='Category-5 pre-commit gate over a session dir') ss.add_argument('--session', type=int, required=True, metavar='N') args = ap.parse_args(argv) if args.cmd == 'machine-block': block = machine_block(args.session, anchor=args.anchor) print(json.dumps(block, ensure_ascii=False, indent=2) if args.json else render_machine_block(block)) return 0 if args.cmd == 'secrets-sweep': session_dir = DEFAULT_SESSIONS_ROOT / f"session-{args.session}" hits = secrets_sweep(session_dir) if not hits: print(f"[secrets-sweep] session-{args.session}: 0 hit " f"({len(SECRET_PATTERNS)} pattern, pattern-bounded -- KHONG phai chung-minh sach)") return 0 # SE-DELTA: hub prints hits then exits 1; same contract, message localised. print(f"[secrets-sweep] session-{args.session}: {len(hits)} HIT -- CHAN commit", file=sys.stderr) for h in hits: print(f" {h['file']}:{h['line']} [{h['pattern']}] {h['snippet']}", file=sys.stderr) return 1 return 2 if __name__ == '__main__': raise SystemExit(main())