solution-erp/scripts/session_ctx.py

#!/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 <repo_root> <args>`; 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 = ['<!-- machine-block: scripts/session_ctx.py machine-block (script-derive) -->']
    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())