#!/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: (S153 1-char fix: the marker here USED to be a U+1F534 emoji = the ONLY byte that broke this file's OWN declared "ASCII-only source" invariant at the bottom of this docstring. Docstring only -- 0 behaviour delta for machine-block / secrets-sweep.) 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] # CANONICAL SINGLE-TRUTH for the `ts:` KEY-LINE probe (S153, ctx soft-memory 0df10df4). # [!] ANY future reader that needs "the most recent session marker timestamp" (the C10-LATENESS # gap in scripts/governance-detectors.ps1 openly declares it has "no reliable source" of ts) # MUST import THIS constant -- do NOT hand-roll a second ts regex anywhere. Two regexes = two # truths, which is precisely what rao-2 of the _mind rule-block forbids. # WHY the probe exists at all: SE markers `_pause-.md` / `_tiep-.md` / `_snapshot-.md` # carry `ts:` as their FIRST field BY DESIGN, so a future latest-marker reader will grep `_*` # without filtering by filename => a `ts:` key line inside `_mind-s-.md` would be read as a # marker timestamp. Hence `_mind` content must never contain one. # Matches a KEY line only (line-start, optional indent, optional space before colon); a ts that # lives INSIDE a heading ("## MIND-1 -- 2026-07-26T...") is NOT a key line and must not match. _TS_KEY_RE = re.compile(r'^[ \t]*ts[ \t]*:', re.I | re.M) # --------------------------------------------------------------------------- # 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 # --------------------------------------------------------------------------- # (c) mind-check -- gate for the SOFT-memory file _mind-s-.md # S153, adopt of broadcast 0df10df4 (ctx soft-memory). 7 standard checks + 2 aux. # 4 levels per check: dat / TRUOT / co / bo-qua-co-khai. # exit 0 = 0 TRUOT | 1 = >=1 TRUOT | 3 = config-or-state (fail-loud, never a default). # --------------------------------------------------------------------------- _MIND_RULES_END = '' _MIND_TOP_MARKER = '' # Real blocks only: `^## MIND-`. The rule-block schema writes `## MIND-` with a # literal `` placeholder precisely so this parser cannot false-match it (same trick as # .claude/templates/session-context-template.md ENTRY-SCHEMA). _MIND_BLOCK_RE = re.compile(r'^##[ \t]+MIND-(\d+)\b', re.M) # rao-1: anh's VERBATIM words live in ONE place only (_context FLOW `> anh:`); the soft file # may only POINT at `PAUSE-`. Copying them here would fork the single-source. _ANH_LINE_RE = re.compile(r'^[ \t]*>[ \t]*anh[ \t]*:', re.I | re.M) # rao-3 exception: the ONLY accepted masked-secret label, spelled with diacritics exactly as the # rule-block emits it. Non-ASCII expressed as \uXXXX so the SOURCE stays ASCII (module header). # `[da che <8-hex>]` (ASCII fallback) is DELIBERATELY NOT accepted: on a secret gate an # over-strict miss costs an edit, a lenient miss costs a permanent leak (blocks are immutable). _MASK_LABEL_RE = re.compile(r'\[[ \t]*\u0111\u00e3[ \t]+che[ \t]+[0-9a-fA-F]{8}[ \t]*\]') # aux-9 structure: A-E section labels. Form FAMILY (not one hardcoded shape) because the template # is authored in a PARALLEL lane -- this code binds to the SPEC markers (MIND-RULES-*/MIND-TOP/ # `## MIND-`), and stays permissive about how A..E are decorated. Accepted: # `### A -- x` | `**A.** x` | `- **A** -- x` | `A) x` | `#### A: x` _SECTION_LETTER_RE = re.compile( r'^[ \t]{0,3}(?:#{2,6}[ \t]*)?(?:[-*][ \t]+)?(?:\*\*[ \t]*)?([A-E])(?:[ \t]*\*\*)?' r'[ \t]*[.):\u2014-]', re.M) # aux-9: every idea in section D must carry ONE of the 4 stance labels (moi-neu / dang-cai / # gan-chot / treo-cho-anh) -- \uXXXX-escaped, ASCII source. _D_LABELS = ('m\u1edbi-n\u00eau', '\u0111ang-c\u00e3i', 'g\u1ea7n-ch\u1ed1t', 'treo-ch\u1edd-anh') # An empty section must be DECLARED empty -- `(trong ...` with the diacritic o-circumflex-acute. _EMPTY_DECL_RE = re.compile(r'\([ \t]*tr\u1ed1ng') _BULLET_RE = re.compile(r'^[ \t]*(?:[-*+][ \t]+|\d+[.)][ \t]+)') _BACKTICK_RE = re.compile(r'`([^`\n]+)`') # Self-DECLARED counters (session-end writes them into `_end`). CROSS-CHECK ONLY -- never a # counting source; disk wins, a mismatch is an INFO flag (hub measured real under-declaration). _SELF_DECL_RES = (('mind-blocks', re.compile(r'mind-blocks[ \t]*[=:][ \t]*(\d+)')), ('_pause', re.compile(r'_pause[ \t]*[=:][ \t]*(\d+)'))) DAT = 'dat' TRUOT = 'TRUOT' CO = 'co' SKIP = 'bo-qua-co-khai' class MindConfigError(Exception): """Config / state error -> exit 3. Raised INSTEAD of assuming a default value.""" def _ascii_safe(s, limit=110): """ASCII-only rendering of an excerpt. A VN-diacritic line printed to a cp1252 console raises UnicodeEncodeError = the gate would DIE mid-check; backslash-escape instead.""" s = s.strip().replace('\t', ' ') if len(s) > limit: s = s[:limit] + '...' return s.encode('ascii', 'backslashreplace').decode('ascii') def _is_int(v): return isinstance(v, int) and not isinstance(v, bool) def _require_budget_kb(repo_root, keys): """{key: cap_in_BYTES} read LIVE from .claude/agent-memory/memory-budget.json (value is KB, reader multiplies by 1024). Missing file / missing key / non-int -> MindConfigError. NO DEFAULT, ever (house rule, h24_cadence._note): a hardcoded ceiling would silently re-create the drift that single-source config exists to prevent. The message NAMES the missing key -- a nameless 'config error' sends the reader hunting.""" p = Path(repo_root) / '.claude' / 'agent-memory' / 'memory-budget.json' if not p.is_file(): raise MindConfigError('config absent: %s' % p) try: data = json.loads(p.read_text(encoding='utf-8')) except (OSError, ValueError) as e: raise MindConfigError('config unreadable: %s (%s)' % (p, e)) missing = [k for k in keys if not _is_int(data.get(k))] if missing: raise MindConfigError('missing key(s) %s in %s -- CAM default (khuon H24-2 fail-loud)' % (', '.join(missing), p)) return {k: int(data[k]) * 1024 for k in keys} def _split_content_region(text): """(region, line_offset, rules_end_found) -- content = everything BELOW . enclosure use-vs-mention: the rule block ITSELF spells out the forbidden strings in order to explain them, so a flat grep over the whole file would flag the very rules that forbid them (citation-trap class). Marker absent => scan the WHOLE file + raise a flag: a file without the marker is off-template, and the safe failure direction is LOUD, not lenient.""" i = text.find(_MIND_RULES_END) if i < 0: return text, 0, False nl = text.find('\n', i) start = len(text) if nl < 0 else nl + 1 return text[start:], text.count('\n', 0, start), True def _parse_blocks(region, line_offset): """Ordered-by-POSITION list of {num, line, text}. Newest-on-TOP => blocks[0] is the newest. (Opposite direction from _context FLOW, which appends newest-at-BOTTOM.)""" ms = list(_MIND_BLOCK_RE.finditer(region)) out = [] for i, m in enumerate(ms): end = ms[i + 1].start() if i + 1 < len(ms) else len(region) out.append({'num': int(m.group(1)), 'line': line_offset + region.count('\n', 0, m.start()) + 1, 'text': region[m.start():end]}) return out def _block_sections(block_text): """{letter: section_text} for the FIRST occurrence of each A-E label (see form family).""" found = [(m.group(1), m.start()) for m in _SECTION_LETTER_RE.finditer(block_text)] secs = {} for i, (letter, pos) in enumerate(found): end = found[i + 1][1] if i + 1 < len(found) else len(block_text) secs.setdefault(letter, block_text[pos:end]) return secs def _pointer_tokens(sec_text): """Backticked path-ish tokens in section E (`verdict + con-tro` per line).""" toks = [] for m in _BACKTICK_RE.finditer(sec_text): t = m.group(1).strip() if not t or ' ' in t or '\t' in t or '/' not in t: continue if t.startswith('http://') or t.startswith('https://'): continue toks.append(t) return toks def _classify_pointer(token, repo_root): """ok | pending | broken | placeholder (F-1 three-state, + declared 4th for placeholders).""" if any(c in token for c in '<>*{}'): return 'placeholder' raw = token[2:] if token.startswith('./') else token p = Path(raw) if not p.is_absolute(): p = Path(repo_root) / raw if p.exists(): return 'ok' return 'pending' if p.parent.is_dir() else 'broken' def _count_pause_on_disk(session_dir): """p = DISK count, dual-accept both marker forms. `_pause-.md` (hub form, S148+) + `pause-*.md` (legacy <=S147). The glob `pause-*` does NOT match `_pause-1.md` (leading underscore) => adding them cannot double-count; same life-saving property the ORPHAN-L probe relies on (tiep.md:93).""" session_dir = Path(session_dir) if not session_dir.is_dir(): return 0, 0, 0 hub = len([p for p in session_dir.glob('_pause-*.md') if p.is_file()]) legacy = len([p for p in session_dir.glob('pause-*.md') if p.is_file()]) return hub + legacy, hub, legacy def _self_declared(session_dir): """{name: n} parsed from `_end` (e.g. `markers: _pause=1 . mind-blocks=2`). Cross-check ONLY.""" end = Path(session_dir) / '_end' if not end.is_file(): return None text = end.read_text(encoding='utf-8', errors='replace') out = {} for name, rx in _SELF_DECL_RES: m = rx.search(text) if m: out[name] = int(m.group(1)) return out def _chk(checks, cid, name, level, detail=()): checks.append({'id': cid, 'name': name, 'level': level, 'detail': list(detail)}) # --- the three rao (barriers), shared by draft-mode and full-mode ------------------------- def _check_rao(checks, region, line_offset, label): hits = [(line_offset + region.count('\n', 0, m.start()) + 1, region[m.start():region.find('\n', m.start()) if region.find('\n', m.start()) >= 0 else len(region)]) for m in _ANH_LINE_RE.finditer(region)] _chk(checks, '1', 'rao-1 loi-anh-verbatim', TRUOT if hits else DAT, ['%s:%d %s' % (label, ln, _ascii_safe(txt)) for ln, txt in hits] or ['0 hit `> anh:` duoi RULES-END (verbatim chi song o _context FLOW)']) ts_hits = [line_offset + region.count('\n', 0, m.start()) + 1 for m in _TS_KEY_RE.finditer(region)] _chk(checks, '2', 'rao-2 ts-key-line', TRUOT if ts_hits else DAT, ['%s:%d key-line `ts:` (canonical regex session_ctx.py:_TS_KEY_RE)' % (label, ln) for ln in ts_hits] or ['0 key-line `ts:` (heading-embedded ts is NOT a key line -- by design)']) sec_hits = [] for i, line in enumerate(region.splitlines(), 1): if _MASK_LABEL_RE.search(line): continue for pat, rx in _SECRET_RES: if rx.search(line): sec_hits.append((line_offset + i, pat)) break _chk(checks, '3', 'rao-3 secret Category-5', TRUOT if sec_hits else DAT, # snippet DELIBERATELY not printed: this gate must not itself leak the secret into a log. ['%s:%d [%s] (snippet KHONG in -- chong ro-ri log)' % (label, ln, pat) for ln, pat in sec_hits] or ['0 hit / %d pattern; ngoai-le nhan che 8-hex duoc mien' % len(SECRET_PATTERNS)]) def _check_pointer(checks, blocks, repo_root, closed): """(4) F-1 three-state on the pointers of section E of the TOP block.""" if not blocks: _chk(checks, '4', 'con-tro E block-top', SKIP, ['0 block => 0 con-tro de cham']) return top = blocks[0] secs = _block_sections(top['text']) e_text = secs.get('E') if e_text is None: _chk(checks, '4', 'con-tro E block-top', SKIP, ['block MIND-%d: khong tim thay muc E (cau-truc do aux-9 cham)' % top['num']]) return if _EMPTY_DECL_RE.search(e_text) and not _pointer_tokens(e_text): _chk(checks, '4', 'con-tro E block-top', SKIP, ['block MIND-%d: muc E khai `(trong ...)` => 0 con-tro (khai, khong xanh-im)' % top['num']]) return toks = _pointer_tokens(e_text) if not toks: # GAP-#9 (T4 fault-inject @S153): closed-mode MUST fail here. If this stayed CO at close, # deleting a truthful pending pointer (-> this branch, exit 0) would be CHEAPER than keeping # it (pending -> TRUOT at close) => the gradient rewards the WRONG path (family of F-1). lvl = TRUOT if closed else CO _chk(checks, '4', 'con-tro E block-top', lvl, ['block MIND-%d: muc E co noi-dung ma 0 con-tro backtick-path%s' % (top['num'], ' => TRUOT o cua DONG (chan duong-re-nhat-de-xanh = xoa con-tro)' if closed else '')]) return detail, worst = [], DAT for t in toks: state = _classify_pointer(t, repo_root) if state == 'ok': detail.append('dat %s' % _ascii_safe(t, 90)) elif state == 'placeholder': detail.append('%s %s (placeholder <>/* -- khong resolve duoc, khai)' % (SKIP, _ascii_safe(t, 90))) elif state == 'pending': # anti-Goodhart (PHAN B #9): the cheapest way to go green must be the RIGHT way. detail.append('%s %s -- cha CO, file chua sinh => CHO file sinh o cua DONG; ' 'CAM xoa con-tro de lam xanh' % ((TRUOT if closed else CO), _ascii_safe(t, 90))) if closed: worst = TRUOT elif worst != TRUOT: worst = CO else: detail.append('%s %s -- cha KHONG co => hong that (moi mode)' % (TRUOT, _ascii_safe(t, 90))) worst = TRUOT detail.append('KHAI (nguyen van thu SS4): TEN-TEP-SAI trong thu-muc THAT doc thanh ' '"dang-cho" o cua mo, toi cua dong moi siet -- day la KE, khong phai tinh nang') _chk(checks, '4', 'con-tro E block-top', worst, detail) def _check_numbering(checks, blocks): """(5) unique + top >= every number below. NO count formula is enforced: the house may change how numbers are allocated and this check must still work (spec Muc-3 of the broadcast).""" if not blocks: _chk(checks, '5', 'so-hieu block', SKIP, ['0 block']) return nums = [b['num'] for b in blocks] dups = sorted({n for n in nums if nums.count(n) > 1}) top = nums[0] detail = ['day (tren->duoi): %s' % ', '.join(str(n) for n in nums)] level = DAT if dups: level = TRUOT detail.append('%s so-hieu TRUNG: %s' % (TRUOT, ', '.join(str(d) for d in dups))) below_max = max(nums[1:]) if len(nums) > 1 else None if below_max is not None and top < below_max: level = TRUOT detail.append('%s block top MIND-%d < so duoi MIND-%d (moi-nhat-o-TREN bi vi pham)' % (TRUOT, top, below_max)) if level == DAT: detail.append('duy-nhat + top >= moi so duoi (khong neo cong-thuc dem)') _chk(checks, '5', 'so-hieu block', level, detail) def _check_structure(checks, blocks, label): """aux-9: 5 sections A-E present (or declared empty) + every D idea carries a stance label.""" if not blocks: _chk(checks, '9', 'aux cau-truc A-E + nhan D', SKIP, ['0 block']) return level, detail = DAT, [] for b in blocks: secs = _block_sections(b['text']) missing = [c for c in 'ABCDE' if c not in secs] if missing: level = TRUOT detail.append('%s MIND-%d (%s:%d) thieu muc: %s [dang nhan: `### A --` | `**A.**` | ' '`- **A** --` | `A)` | `#### A:`]' % (TRUOT, b['num'], label, b['line'], ', '.join(missing))) continue d_text = secs['D'] d_lines = d_text.splitlines()[1:] # An empty-declaration line is a DECLARATION, not an idea -- exempt it even when it is # written as a bullet (`- (trong -- khai)`), otherwise the rule "muc trong ghi (trong -- # khai)" would itself be un-satisfiable. Caught by the clean fixture, not by reasoning. ideas = [(i, ln) for i, ln in enumerate(d_lines, 1) if _BULLET_RE.match(ln) and not _EMPTY_DECL_RE.search(ln)] if not ideas: if not _EMPTY_DECL_RE.search(d_text): level = TRUOT detail.append('%s MIND-%d muc D: 0 y ma cung KHONG khai `(trong ...)`' % (TRUOT, b['num'])) continue bad = [ln for _, ln in ideas if not any(lb in ln for lb in _D_LABELS)] if bad: level = TRUOT detail.append('%s MIND-%d muc D: %d/%d y thieu nhan {moi-neu|dang-cai|gan-chot|' 'treo-cho-anh} -- vd: %s' % (TRUOT, b['num'], len(bad), len(ideas), _ascii_safe(bad[0], 70))) if level == DAT: detail.append('%d block: 5 muc A-E du + moi y muc D co nhan' % len(blocks)) _chk(checks, '9', 'aux cau-truc A-E + nhan D', level, detail) def _check_context_cap(checks, session_dir, session_n, cap_bytes): """aux-8: the FIRST REAL READER of session_ctx_kb (closes the H18 ghost-wire: the key's own note asked its future reader to (a) read the key (b) x1024 (c) drop the caveat line).""" ctx = Path(session_dir) / ('_context-s-%d.md' % session_n) if not ctx.is_file(): _chk(checks, '8', 'aux tran _context', SKIP, ['%s: khong co tep' % ctx.name]) return size = ctx.stat().st_size over = size > cap_bytes _chk(checks, '8', 'aux tran _context', CO if over else DAT, ['%s = %dB vs session_ctx_kb x1024 = %dB%s' % (ctx.name, size, cap_bytes, ' => VUOT' if over else '')]) def mind_check(session_n, root=None, draft=None, force_closed=False): """Gate _mind-s-.md. Returns (rc, lines). Raises MindConfigError for rc=3 cases. `--session` is PINNED by the caller on purpose (no default-latest): two windows running side by side would race for "the latest" and audit each other's file.""" repo_root = Path(root or ROOT) session_dir = repo_root / '.claude' / 'sessions' / ('session-%d' % session_n) canonical = session_dir / ('_mind-s-%d.md' % session_n) closed = bool(force_closed) or (session_dir / '_end').is_file() mode = 'draft' if draft else ('closed' if closed else 'open') checks = [] out = ['[mind-check] session-%d | mode=%s | root=%s' % (session_n, mode, repo_root)] if draft: dpath = Path(draft) if not dpath.is_file(): alt = repo_root / draft if not alt.is_file(): raise MindConfigError('draft file absent: %s' % dpath) dpath = alt # Draft-mode consumes NO mind_ctx_kb (checks 6/7 are skipped) => requiring it would be a # fail-loud on a key this run never reads. It DOES read session_ctx_kb (aux-8). caps = _require_budget_kb(repo_root, ['session_ctx_kb']) text = dpath.read_text(encoding='utf-8', errors='replace') region, line_offset, _ = _split_content_region(text) out.append(' draft : %s (%dB)' % (dpath, dpath.stat().st_size)) out.append(' canon : %s%s' % (canonical, '' if canonical.is_file() else ' (chua co)')) _check_rao(checks, region, line_offset, dpath.name) dblocks = _parse_blocks(region, line_offset) on_disk = 0 if canonical.is_file(): creg, coff, _ = _split_content_region( canonical.read_text(encoding='utf-8', errors='replace')) on_disk = len(_parse_blocks(creg, coff)) if not dblocks: _chk(checks, '5', 'so-hieu draft vs count-on-disk', TRUOT, ['draft khong co heading `## MIND-` nao']) else: want = on_disk got = dblocks[0]['num'] _chk(checks, '5', 'so-hieu draft vs count-on-disk', DAT if got == want else TRUOT, ['draft MIND-%d vs count-on-disk %d%s' % (got, want, '' if got == want else ' => LECH (so-hieu = DEM block TRUOC chen)')]) _check_pointer(checks, dblocks, repo_root, closed=False) _chk(checks, '6', 'bat-bien |block| vs p', SKIP, ['draft-mode: cham sau khi chen']) _chk(checks, '7', 'tran mind_ctx_kb', SKIP, ['draft-mode: do tren tep THAT sau khi chen (khong doc mind_ctx_kb luot nay)']) _check_context_cap(checks, session_dir, session_n, caps['session_ctx_kb']) _check_structure(checks, dblocks, dpath.name) return _render(out, checks, 'draft (kiem NHAP truoc khi chen -- loi khong bao gio vao tep)') # ---- full mode ----------------------------------------------------------------------- if not canonical.is_file(): cands = sorted(p.name for p in session_dir.glob('*mind*')) if session_dir.is_dir() else [] why = ['thieu: %s' % canonical] if not session_dir.is_dir(): why.append('thu-muc phien KHONG ton tai: %s' % session_dir) # tep-sai-phai-NEU-TEN: "khong co gi de kiem" must never hide a typo'd filename. why.append('glob `*mind*` trong thu-muc: %s' % (', '.join(cands) if cands else '(0 ung vien)')) if cands: why.append('=> co ung vien TEN SAI o tren; canonical dung phai la %s' % canonical.name) level = TRUOT if closed else SKIP if closed: why.append('closed-mode (co `_end`) => TRUOT: phien da dong ma khong co lop MEM') else: why.append('open-mode => bo-qua CO KHAI (chua doc memory-budget.json: khong co gi de do)') _chk(checks, '0', 'ton tai _mind-s-%d.md' % session_n, level, why) return _render(out, checks, 'full/%s (tep canonical vang)' % mode) caps = _require_budget_kb(repo_root, ['mind_ctx_kb', 'session_ctx_kb']) text = canonical.read_text(encoding='utf-8', errors='replace') size = canonical.stat().st_size region, line_offset, rules_end = _split_content_region(text) out.append(' target: %s (%dB)' % (canonical, size)) if not rules_end: _chk(checks, '0', 'enclosure %s' % _MIND_RULES_END, CO, ['marker VANG => quet TRON tep (huong hong NGHIEM); tep dang off-template']) else: _chk(checks, '0', 'enclosure %s' % _MIND_RULES_END, DAT, ['chi quet DUOI marker (use-vs-mention: khoi luat chua chinh chuoi bi cam)']) if _MIND_TOP_MARKER not in text: _chk(checks, '0b', 'marker %s' % _MIND_TOP_MARKER, CO, ['vang => khong biet chen block moi o dau (moi-nhat-o-TREN)']) _check_rao(checks, region, line_offset, canonical.name) blocks = _parse_blocks(region, line_offset) _check_pointer(checks, blocks, repo_root, closed) _check_numbering(checks, blocks) p, hub, legacy = _count_pause_on_disk(session_dir) n = len(blocks) ok = n in (p, p + 1) detail = ['|block| = %d ; p (DEM DIA) = %d [_pause-*.md=%d + pause-*.md=%d dual-accept]' % (n, p, hub, legacy), 'ky vong |block| thuoc {%d, %d} (block-0 + 1 refresh/pause)' % (p, p + 1)] decl = _self_declared(session_dir) info = [] if decl is None: detail.append('cross-check tu-khai: khong co `_end` => bo qua (dia van la nguon duy nhat)') else: for key, want in (('mind-blocks', n), ('_pause', p)): if key in decl and decl[key] != want: info.append('co-INFO lech tu-khai: `_end` %s=%d vs DIA %d (DIA THANG)' % (key, decl[key], want)) detail.append('cross-check tu-khai `_end`: %s' % (', '.join('%s=%d' % (k, v) for k, v in sorted(decl.items())) or '(0 field)')) detail.extend(info) _chk(checks, '6', 'bat-bien |block| vs p', TRUOT if not ok else (CO if info else DAT), detail) cap = caps['mind_ctx_kb'] over = size > cap d7 = ['%dB vs mind_ctx_kb x1024 = %dB (doc LIVE, 0 hardcode)%s' % (size, cap, ' => VUOT' if over else '')] if over: if n < 3: d7.append('TU-CHOI nen: chi %d block (<3) => giuong co vuot tran, KHONG nen' % n) else: d7.append('huong-dan: nen block CU NHAT (tru block-0 va block top); verbatim con o git') _chk(checks, '7', 'tran mind_ctx_kb', CO if over else DAT, d7) _check_context_cap(checks, session_dir, session_n, caps['session_ctx_kb']) _check_structure(checks, blocks, canonical.name) return _render(out, checks, 'full/%s' % mode) def _render(out, checks, mode_note): """Deterministic 4-level report. rc = 1 iff >=1 TRUOT (co / bo-qua-co-khai never gate).""" order = {'0': 0, '0b': 1, '1': 2, '2': 3, '3': 4, '4': 5, '5': 6, '6': 7, '7': 8, '8': 9, '9': 10} checks = sorted(checks, key=lambda c: order.get(c['id'], 99)) tally = {DAT: 0, TRUOT: 0, CO: 0, SKIP: 0} for c in checks: tally[c['level']] = tally.get(c['level'], 0) + 1 tag = '[phu-%s]' % c['id'] if c['id'] in ('8', '9') else '(%s)' % c['id'] out.append(' %-8s %-30s : %s' % (tag, c['name'], c['level'])) for d in c['detail']: out.append(' - %s' % d) rc = 1 if tally[TRUOT] else 0 out.append(' mode-note: %s' % mode_note) out.append(' verdict: dat=%d %s=%d co=%d %s=%d => exit %d' % (tally[DAT], TRUOT, tally[TRUOT], tally[CO], SKIP, tally[SKIP], rc)) return rc, out # --------------------------------------------------------------------------- # 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') # S153 T4 #7 spec-drift fix: seam parity with mind-check (--root was mind-check-only, so the # literal command in PHAN B #7 could not run as written; T4 had to copy the script into the tree). ss.add_argument('--root', default=None, metavar='PATH', help='repo root override = fault-inject seam (mirror mind-check --root)') # S153 ctx soft-memory (0df10df4). --session is REQUIRED with NO default-latest on purpose: # two windows running in parallel would race over "the latest" session. mc = sub.add_parser('mind-check', help='gate the soft-memory file _mind-s-.md') mc.add_argument('--session', type=int, required=True, metavar='N') mc.add_argument('--draft', default=None, metavar='FILE', help='check a DRAFT block before inserting it (checks 1-3 + numbering; ' 'skips 6/7) -- an error must never reach the immutable file') mc.add_argument('--closed', action='store_true', help='force closed-mode (auto-on when `_end` exists); flag = fault-inject seam') mc.add_argument('--root', default=None, metavar='PATH', help='repo root override = fault-inject seam (mirror -RepoRoot of the .ps1 family)') 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': base = (Path(args.root) / '.claude' / 'sessions') if args.root else DEFAULT_SESSIONS_ROOT session_dir = base / 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 if args.cmd == 'mind-check': try: rc, lines = mind_check(args.session, root=args.root, draft=args.draft, force_closed=args.closed) except MindConfigError as e: # exit 3 = config/state. FAIL-LOUD by design: measuring with an assumed ceiling would # be worse than measuring nothing, because it looks like a pass. print(f"[mind-check] CONFIG/STATE ERROR: {e}", file=sys.stderr) return 3 stream = sys.stderr if rc else sys.stdout for ln in lines: print(ln, file=stream) return rc return 2 if __name__ == '__main__': raise SystemExit(main())