#!/usr/bin/env python3 """ Backtest GPT-5 Harmony-header leak handling against ~/.omp/stats.db. This is a dry-run analysis tool. It does not mutate stats.db or session JSONL. It scans stored assistant/tool-call surfaces, applies a selected detection and recovery strategy, and prints which edit inputs would be preserved by a sanitize-tail strategy versus aborted/replayed. """ from __future__ import annotations import argparse import hashlib import json import re import sqlite3 import sys from collections import Counter, defaultdict from dataclasses import dataclass, field from pathlib import Path from typing import Any DB_PATH = Path.home() / ".omp" / "stats.db" MARKER_RE = re.compile(r"\bto=functions\.[A-Za-z_][A-Za-z0-9_]*") HARMONY_RE = re.compile(r"<\|(start|end|channel|message|call|return)\|>") CHANNEL_WORD_RE = re.compile(r"\b(analysis|commentary|assistant|user|system|developer|tool)\s+to=functions\.") GLITCH_RE = re.compile(r"\b(changedFiles|RTLU|Jsii(?:_commentary)?|Japgolly|tRTLUfunctions|Joshi_commentary|Japgolly_commentary|jsii_commentary|Jsii_commentary|Jsii)\b") NULLISH_RE = re.compile(r"\b(undefined|null)\b") BODY_CASCADE_RE = re.compile(r"\bto=functions\.[A-Za-z_][A-Za-z0-9_]*\s+code(?:\s|$)") FAKE_RESULT_RE = re.compile( r"\bto=functions\.[A-Za-z_][A-Za-z0-9_]*(?s:.{0,80}?)code_output\s*\nCell\s+\d+:" ) FENCE_RE = re.compile(r"^\s*(```+|~~~+)") # Python's stdlib re has no Unicode script properties, so keep the exact # ranges local and explicit. SCRIPT_RUN_RE = re.compile( "[" "\u3400-\u4DBF" # CJK Extension A "\u4E00-\u9FFF" # CJK Unified Ideographs "\uF900-\uFAFF" # CJK Compatibility Ideographs "\u0400-\u04FF" # Cyrillic "\u0E00-\u0E7F" # Thai "\u10A0-\u10FF" # Georgian "\u0530-\u058F" # Armenian "\u0C80-\u0CFF" # Kannada "\u0C00-\u0C7F" # Telugu "\u0900-\u097F" # Devanagari "\u0600-\u06FF" # Arabic "\u0D00-\u0D7F" # Malayalam "]{2,}" ) # Header detector — matches any of: # ¶PATH or ¶PATH#hash (current hashline format) # §PATH (legacy hashline format, pre-2026-05) # *** Update File: PATH (Codex apply_patch envelope) HEADER_RE = re.compile( r"^(?:§+(?P.*)|¶+\s*(?P[^\s#¶]+)(?:#[0-9a-f]{4})?|\*\*\* Update File:\s+(?P\S.*))\s*$" ) BEGIN_PATCH_RE = re.compile(r"^\*\*\* Begin Patch\s*$") END_PATCH_RE = re.compile(r"^\*\*\* End Patch\s*$") # Legacy hashline ops (kept for historical session corpus). LEGACY_INSERT_RE = re.compile(r"^(?P[«»])\s*(?PBOF|EOF|[1-9][0-9]*[A-Za-z]{2})\s*$") LEGACY_RANGE_RE = re.compile(r"(?P[1-9][0-9]*[A-Za-z]{2})(?:\.\.(?P[1-9][0-9]*[A-Za-z]{2}))?") LEGACY_REPLACE_RE = re.compile(r"^≔\s*(?P[1-9][0-9]*[A-Za-z]{2}(?:\.\.[1-9][0-9]*[A-Za-z]{2})?)\s*$") # Current hashline ops. NEW_INSERT_RE = re.compile( r"^\s*(?:[>+\-*]+\s*)?(?P[1-9][0-9]*|BOF|EOF)(?P[↑↓])(?P.*)$" ) NEW_RANGE_RE = re.compile( r"^\s*(?:[>+\-*]+\s*)?(?P[1-9][0-9]*)(?:-(?P[1-9][0-9]*))?(?P[:!])(?P.*)$" ) JSON_DECODER = json.JSONDecoder() @dataclass(frozen=True) class Signal: cls: str start: int end: int detail: str @dataclass class MarkerEvidence: start: int end: int classes: set[str] = field(default_factory=lambda: {"M"}) @property def label(self) -> str: return "+".join(sorted(self.classes)) @dataclass class EditSection: target_file: str op_count: int = 0 payload_lines: int = 0 deleted_lines: int = 0 @dataclass class EditBoundary: ok: bool parsed_end: int reason: str sections: list[EditSection] = field(default_factory=list) line_no: int = 0 @property def op_count(self) -> int: return sum(s.op_count for s in self.sections) @property def payload_lines(self) -> int: return sum(s.payload_lines for s in self.sections) @property def deleted_lines(self) -> int: return sum(s.deleted_lines for s in self.sections) @dataclass class ToolBacktest: surface: str row_id: str session_file: str seq: int entry_id: str | None call_id: str | None tool_name: str model: str | None provider: str | None action: str signals: list[str] signal_offsets: list[int] text_len: int parsed_end: int | None = None removed_len: int = 0 removed_sha16: str | None = None removed_preview: str = "" clean_preview: str = "" context_preview: str = "" edit_files: list[str] = field(default_factory=list) edit_ops: int = 0 edit_payload_lines: int = 0 edit_deleted_lines: int = 0 parse_reason: str = "" @dataclass class TextBacktest: surface: str row_id: str session_file: str seq: int entry_id: str | None model: str | None provider: str | None action: str signals: list[str] signal_offsets: list[int] text_len: int context_preview: str def open_ro(path: Path) -> sqlite3.Connection: if not path.exists(): sys.exit(f"db not found: {path}. Run scripts/session-stats/sync.py first.") conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True) conn.row_factory = sqlite3.Row return conn def commas(n: int) -> str: return f"{n:,}" def one_line(text: str, limit: int = 180) -> str: text = text.replace("\r", "\\r").replace("\n", " | ").replace("\t", "\\t") if len(text) <= limit: return text return text[: max(0, limit - 3)] + "..." def snippet(text: str, pos: int, radius: int = 120) -> str: lo = max(0, pos - radius) hi = min(len(text), pos + radius) prefix = "..." if lo > 0 else "" suffix = "..." if hi < len(text) else "" return one_line(prefix + text[lo:hi] + suffix, radius * 2 + 20) def sha16(text: str) -> str: return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()[:16] def is_inside_fenced_block(text: str, pos: int) -> bool: """Best-effort Markdown fence context used to avoid doc/test false positives.""" in_fence = False for line in text[:pos].splitlines(): if FENCE_RE.match(line): in_fence = not in_fence return in_fence def ascii_ratio(text: str) -> float: if not text: return 1.0 ascii_count = sum(1 for ch in text if ord(ch) < 128) return ascii_count / len(text) def script_mismatch_near(text: str, start: int, end: int) -> bool: near = text[max(0, start - 32) : min(len(text), end + 32)] if not SCRIPT_RUN_RE.search(near): return False surrounding = text[max(0, start - 200) : min(len(text), end + 200)] return ascii_ratio(surrounding) >= 0.85 def marker_evidence_for( text: str, marker: re.Match[str], parsed_end: int | None, respect_fences: bool, include_nullish: bool, ) -> MarkerEvidence | None: start, end = marker.span() if respect_fences and is_inside_fenced_block(text, start): return None ev = MarkerEvidence(start=start, end=end) window16 = text[max(0, start - 16) : min(len(text), end + 16)] window200 = text[start : min(len(text), start + 200)] for c in CHANNEL_WORD_RE.finditer(text[max(0, start - 64) : min(len(text), end + 16)]): absolute_start = max(0, start - 64) + c.start() absolute_end = max(0, start - 64) + c.end() if absolute_start <= start < absolute_end: ev.classes.add("C") break if GLITCH_RE.search(window16): ev.classes.add("G") if include_nullish and NULLISH_RE.search(window16): ev.classes.add("N") if script_mismatch_near(text, start, end): ev.classes.add("S") if BODY_CASCADE_RE.match(window200) and MARKER_RE.search(window200[marker.end() - start :]): ev.classes.add("B") if FAKE_RESULT_RE.match(text, start): ev.classes.add("R") if parsed_end is not None and start >= parsed_end: ev.classes.add("T") return ev def detect_signals( text: str, strategy: str, parsed_end: int | None = None, respect_fences: bool = True, include_nullish: bool = False, ) -> tuple[list[Signal], list[MarkerEvidence]]: signals: list[Signal] = [] marker_evidence: list[MarkerEvidence] = [] for h in HARMONY_RE.finditer(text): if respect_fences and is_inside_fenced_block(text, h.start()): continue signals.append(Signal("H", h.start(), h.end(), h.group(0))) for marker in MARKER_RE.finditer(text): ev = marker_evidence_for(text, marker, parsed_end, respect_fences, include_nullish) if ev is None: continue marker_evidence.append(ev) if strategy == "marker": signals.append(Signal(ev.label, ev.start, ev.end, text[ev.start : ev.end])) elif len(ev.classes) > 1: signals.append(Signal(ev.label, ev.start, ev.end, text[ev.start : ev.end])) if strategy == "tail": signals = [s for s in signals if s.cls == "H" or "T" in s.cls.split("+")] signals.sort(key=lambda s: (s.start, s.end, s.cls)) marker_evidence.sort(key=lambda e: (e.start, e.end)) return signals, marker_evidence def complete_json_end(raw: str) -> tuple[bool, int, str]: try: _, end = JSON_DECODER.raw_decode(raw) return True, end, "json-prefix-ok" except json.JSONDecodeError as exc: return False, exc.pos, exc.msg def line_spans(text: str) -> list[tuple[str, int, int]]: out: list[tuple[str, int, int]] = [] pos = 0 for raw in text.splitlines(keepends=True): start = pos pos += len(raw) out.append((raw.rstrip("\r\n"), start, pos)) if text and (text.endswith("\n") or text.endswith("\r")): return out if not text: return [] # splitlines(keepends=True) already includes the final unterminated line. return out def parse_legacy_diff_boundary(text: str, *, loose_tail: bool = False) -> EditBoundary: """Best-effort parser for pre-hashline edit inputs. Older sessions used `---path` followed by compact hash/range operations whose replacement payload was raw text, not `~`-prefixed. That format is not safe enough for production recovery, but this backtest needs to answer where a tail-only cleaner would cut if we supported those historical rows. """ sections: list[EditSection] = [] cur: EditSection | None = None parsed_end = 0 line_no = 0 in_payload = False old_replace = re.compile( r"^\s*(?P[1-9][0-9]*[A-Za-z]{2}(?:\.\.[1-9][0-9]*[A-Za-z]{2})?)(?P[=+<])(?P.*)$" ) old_delete = re.compile( r"^\s*-(?P[1-9][0-9]*[A-Za-z]{2}(?:\.\.[1-9][0-9]*[A-Za-z]{2})?)\s*$" ) for line, _start, end in line_spans(text): line_no += 1 if loose_tail and (MARKER_RE.search(line) or HARMONY_RE.search(line)) and cur is not None and cur.op_count > 0: break if line.startswith("---"): target = line[3:].strip() if not target: break cur = EditSection(target_file=target) sections.append(cur) parsed_end = end in_payload = False continue if cur is None: if not line.strip(): parsed_end = end continue break if line.startswith("+++") or line.startswith("@@"): if line.startswith("@@"): cur.op_count += 1 in_payload = True else: in_payload = False parsed_end = end continue dele = old_delete.match(line) if dele: cur.op_count += 1 cur.deleted_lines += range_deleted_lines(dele.group("range")) parsed_end = end in_payload = False continue repl = old_replace.match(line) if repl: cur.op_count += 1 if repl.group("op") == "=": cur.deleted_lines += range_deleted_lines(repl.group("range")) if repl.group("tail"): cur.payload_lines += 1 parsed_end = end in_payload = True continue if line.startswith("+") and not line.startswith("+++"): cur.payload_lines += 1 parsed_end = end in_payload = True continue if in_payload and (line.startswith("-") or line.startswith(" ") or line.startswith("\\")): if line.startswith("-") and not line.startswith("---"): cur.deleted_lines += 1 parsed_end = end continue if in_payload: cur.payload_lines += 1 parsed_end = end continue if not line.strip(): parsed_end = end continue break return EditBoundary( ok=parsed_end > 0 and bool(sections), parsed_end=parsed_end, reason="legacy-edit-ok" if parsed_end > 0 and sections else "no-complete-edit-prefix", sections=sections, line_no=line_no, ) def anchor_line_no(anchor: str) -> int | None: m = re.match(r"([1-9][0-9]*)", anchor) return int(m.group(1)) if m else None def legacy_range_deleted_lines(raw_range: str) -> int: m = LEGACY_RANGE_RE.fullmatch(raw_range) if not m: return 1 a = anchor_line_no(m.group("a")) or 0 b = anchor_line_no(m.group("b") or m.group("a")) or a return max(1, b - a + 1) def new_range_deleted_lines(a_raw: str, b_raw: str | None) -> int: try: a = int(a_raw) b = int(b_raw) if b_raw else a except ValueError: return 1 return max(1, b - a + 1) def parse_edit_boundary(text: str, *, legacy_loose_tail: bool = False) -> EditBoundary: """Find the longest tool-input prefix that parses as a complete sequence of edit sections. Supports both the current hashline format (``¶PATH#hash`` / ``↑↓:!``) and the legacy format (``§PATH`` / ``«»≔``) since the analyzed corpus spans the format transition. Codex ``*** Update File:`` envelopes are also recognized. """ sections: list[EditSection] = [] cur: EditSection | None = None # Per-section format: "new", "legacy_hl", or "codex". Determines which op # regexes apply and whether `needs_payload` semantics are in effect. cur_format: str | None = None parsed_end = 0 line_no = 0 # Legacy hashline (`«»`) inserts demand at least one payload line before the # boundary advances. New-format inserts are self-completing on their own # line, so these flags only matter when ``cur_format == "legacy_hl"``. needs_payload = False payload_allowed = False saw_required_payload = False seen_content = False legacy_payload_blockers = {"«", "»", "≔", "§"} new_payload_blockers = ("¶",) for line, _start, end in line_spans(text): line_no += 1 stripped = line.strip() if not seen_content and not stripped: parsed_end = end continue seen_content = True if BEGIN_PATCH_RE.match(line): if needs_payload and not saw_required_payload: break parsed_end = end payload_allowed = False continue if END_PATCH_RE.match(line): if needs_payload and not saw_required_payload: break parsed_end = end payload_allowed = False continue header = HEADER_RE.match(line) if header: if needs_payload and not saw_required_payload: break if header.group("hl_new") is not None: target = header.group("hl_new").strip() cur_format = "new" elif header.group("hl_legacy") is not None: target = header.group("hl_legacy").strip() cur_format = "legacy_hl" else: target = (header.group("upd") or "").strip() cur_format = "codex" cur = EditSection(target_file=target) sections.append(cur) parsed_end = end needs_payload = False payload_allowed = False saw_required_payload = False continue if cur is None: break if cur_format == "new": # New format: payload lines are anything that does not start a new # op or header. `↑`/`↓`/`:` ops may be followed by additional # payload lines; `!` ops are self-contained. if payload_allowed and not line.startswith(new_payload_blockers): # Re-check whether the line itself is an op — an op line ends # the current payload run and starts a new op. if not NEW_INSERT_RE.match(line) and not NEW_RANGE_RE.match(line): cur.payload_lines += 1 parsed_end = end continue if not stripped: parsed_end = end payload_allowed = False continue ins = NEW_INSERT_RE.match(line) if ins: cur.op_count += 1 if ins.group("inline"): cur.payload_lines += 1 parsed_end = end payload_allowed = True continue rng = NEW_RANGE_RE.match(line) if rng: sigil = rng.group("sigil") cur.op_count += 1 cur.deleted_lines += new_range_deleted_lines(rng.group("a"), rng.group("b")) if sigil == ":" and rng.group("inline"): cur.payload_lines += 1 parsed_end = end # `!` deletes do not accept payload; `:` replaces may carry # subsequent payload lines. payload_allowed = sigil == ":" continue break # Legacy hashline state machine (preserved verbatim for §«»≔ corpus). if payload_allowed and line[:1] not in legacy_payload_blockers: cur.payload_lines += 1 parsed_end = end saw_required_payload = True needs_payload = False continue if not stripped: if needs_payload and not saw_required_payload: break parsed_end = end payload_allowed = False needs_payload = False saw_required_payload = False continue if needs_payload and not saw_required_payload: break ins = LEGACY_INSERT_RE.match(line) if ins: cur.op_count += 1 needs_payload = True payload_allowed = True saw_required_payload = False # Not complete until at least one payload line appears. continue repl = LEGACY_REPLACE_RE.match(line) if repl: cur.op_count += 1 cur.deleted_lines += legacy_range_deleted_lines(repl.group("range")) parsed_end = end needs_payload = False payload_allowed = True saw_required_payload = False continue break reason = "ok" if parsed_end > 0 and sections else "no-complete-edit-prefix" if needs_payload and not saw_required_payload: reason = "insert-missing-payload" if not sections and text.lstrip().startswith("---"): return parse_legacy_diff_boundary(text, loose_tail=legacy_loose_tail) return EditBoundary( ok=parsed_end > 0 and bool(sections), parsed_end=parsed_end, reason=reason, sections=sections, line_no=line_no, ) def parse_arg_json(raw: str) -> tuple[Any | None, bool, str]: try: return json.loads(raw), True, "ok" except json.JSONDecodeError as exc: return None, False, f"json-error:{exc.pos}:{exc.msg}" def extract_primary_text(tool_name: str, arg_json: str, parsed: Any | None) -> tuple[str, str]: if tool_name == "edit" and isinstance(parsed, dict) and isinstance(parsed.get("input"), str): return "edit.input", parsed["input"] if tool_name == "eval" and isinstance(parsed, dict) and isinstance(parsed.get("input"), str): return "eval.input", parsed["input"] if tool_name == "write" and isinstance(parsed, dict) and isinstance(parsed.get("content"), str): return "write.content", parsed["content"] if tool_name == "bash" and isinstance(parsed, dict) and isinstance(parsed.get("command"), str): return "bash.command", parsed["command"] return "arg_json", arg_json def action_for_tool( tool_name: str, surface: str, text: str, signals: list[Signal], boundary: EditBoundary | None, ) -> str: if not signals: return "allow" if tool_name == "edit" and surface == "edit.input" and boundary is not None and boundary.ok: if all(sig.start >= boundary.parsed_end for sig in signals): return "sanitize_tail" return "abort_replay" return "abort_replay" def evaluate_tool_row( row: sqlite3.Row, strategy: str, respect_fences: bool, include_nullish: bool, legacy_loose_tail: bool = False, ) -> ToolBacktest: arg_json = row["arg_json"] or "" tool_name = row["tool_name"] or "" parsed, json_ok, json_reason = parse_arg_json(arg_json) surface, text = extract_primary_text(tool_name, arg_json, parsed) boundary: EditBoundary | None = None parsed_end: int | None = None parse_reason = json_reason if tool_name == "edit" and surface == "edit.input": boundary = parse_edit_boundary(text, legacy_loose_tail=legacy_loose_tail) parsed_end = boundary.parsed_end if boundary.ok else None parse_reason = boundary.reason else: ok, end, reason = complete_json_end(arg_json) if ok: parsed_end = end parse_reason = reason if json_ok else json_reason signals, _marker_evidence = detect_signals( text, strategy=strategy, parsed_end=parsed_end, respect_fences=respect_fences, include_nullish=include_nullish, ) action = action_for_tool(tool_name, surface, text, signals, boundary) removed = "" clean_preview = "" edit_files: list[str] = [] edit_ops = 0 edit_payload_lines = 0 edit_deleted_lines = 0 if boundary is not None: edit_files = [s.target_file for s in boundary.sections] edit_ops = boundary.op_count edit_payload_lines = boundary.payload_lines edit_deleted_lines = boundary.deleted_lines if action == "sanitize_tail": removed = text[boundary.parsed_end :] cleaned = text[: boundary.parsed_end] clean_preview = tail_preview(cleaned) first_pos = signals[0].start if signals else 0 return ToolBacktest( surface=surface, row_id=str(row["id"]), session_file=row["session_file"], seq=row["seq"], entry_id=row["entry_id"], call_id=row["call_id"], tool_name=tool_name, model=row["model"], provider=row["provider"], action=action, signals=[s.cls for s in signals], signal_offsets=[s.start for s in signals], text_len=len(text), parsed_end=parsed_end, removed_len=len(removed), removed_sha16=sha16(removed) if removed else None, removed_preview=one_line(removed, 200) if removed else "", clean_preview=clean_preview, context_preview=snippet(text, first_pos) if signals else "", edit_files=edit_files, edit_ops=edit_ops, edit_payload_lines=edit_payload_lines, edit_deleted_lines=edit_deleted_lines, parse_reason=parse_reason, ) def tail_preview(text: str, max_lines: int = 8, limit: int = 420) -> str: lines = text.splitlines() tail = "\n".join(lines[-max_lines:]) return one_line(tail, limit) def evaluate_text_row( row: sqlite3.Row, surface: str, text: str, strategy: str, respect_fences: bool, include_nullish: bool, ) -> TextBacktest: signals, _ = detect_signals( text, strategy=strategy, parsed_end=None, respect_fences=respect_fences, include_nullish=include_nullish, ) action = "rewrite_candidate" if signals else "allow" first_pos = signals[0].start if signals else 0 return TextBacktest( surface=surface, row_id=f"{row['session_file']}:{row['seq']}:{surface}", session_file=row["session_file"], seq=row["seq"], entry_id=row["entry_id"], model=row["model"], provider=row["provider"], action=action, signals=[s.cls for s in signals], signal_offsets=[s.start for s in signals], text_len=len(text), context_preview=snippet(text, first_pos) if signals else "", ) def candidate_where(column: str) -> str: return " OR ".join( [ f"{column} LIKE '%to=functions.%'", f"{column} LIKE '%<|start|>%'", f"{column} LIKE '%<|end|>%'", f"{column} LIKE '%<|channel|>%'", f"{column} LIKE '%<|message|>%'", f"{column} LIKE '%<|call|>%'", f"{column} LIKE '%<|return|>%'", ] ) def scan_tools(conn: sqlite3.Connection, args: argparse.Namespace) -> list[ToolBacktest]: where = candidate_where("arg_json") params: list[Any] = [] if args.provider: where = f"({where}) AND provider = ?" params.append(args.provider) if args.model: where = f"({where}) AND model = ?" params.append(args.model) if args.tool: where = f"({where}) AND tool_name = ?" params.append(args.tool) sql = f""" SELECT id, session_file, seq, entry_id, call_id, tool_name, raw_tool_name, timestamp, model, provider, arg_json FROM ss_tool_calls WHERE {where} ORDER BY timestamp, id """ rows = conn.execute(sql, params).fetchall() return [ evaluate_tool_row( row, strategy=args.strategy, respect_fences=not args.no_fence_context, include_nullish=args.include_nullish, legacy_loose_tail=args.legacy_loose_tail, ) for row in rows ] def scan_assistant(conn: sqlite3.Connection, args: argparse.Namespace) -> list[TextBacktest]: if not args.include_assistant: return [] text_where = candidate_where("text_blob") thinking_where = candidate_where("thinking_blob") where = f"({text_where}) OR ({thinking_where})" params: list[Any] = [] if args.provider: where = f"({where}) AND provider = ?" params.append(args.provider) if args.model: where = f"({where}) AND model = ?" params.append(args.model) sql = f""" SELECT session_file, seq, entry_id, timestamp, model, provider, text_blob, thinking_blob FROM ss_assistant_msgs WHERE {where} ORDER BY timestamp, session_file, seq """ out: list[TextBacktest] = [] for row in conn.execute(sql, params): if row["text_blob"]: out.append( evaluate_text_row( row, "assistant_text", row["text_blob"], args.strategy, not args.no_fence_context, args.include_nullish, ) ) if row["thinking_blob"]: out.append( evaluate_text_row( row, "assistant_thinking", row["thinking_blob"], args.strategy, not args.no_fence_context, args.include_nullish, ) ) return out def print_counter(title: str, counter: Counter[str]) -> None: print(title) for key, value in counter.most_common(): print(f" {key:<32} {value:>6}") def print_tool_summary(results: list[ToolBacktest]) -> None: print("=== tool-call scan ===") print(f"candidate rows: {commas(len(results))}") print_counter("\nby action:", Counter(r.action for r in results)) print_counter("\nby tool/action:", Counter(f"{r.tool_name}:{r.action}" for r in results)) print_counter("\nby model/action:", Counter(f"{r.model or ''}:{r.action}" for r in results)) signal_counter: Counter[str] = Counter() for r in results: if r.signals: signal_counter.update(r.signals) else: signal_counter["none"] += 1 print_counter("\nby signal:", signal_counter) edit_results = [r for r in results if r.tool_name == "edit"] if edit_results: sanitized = sum(1 for r in edit_results if r.action == "sanitize_tail") aborted = sum(1 for r in edit_results if r.action == "abort_replay") print("\nedit preservation:") print(f" edit candidates: {commas(len(edit_results))}") print(f" sanitize_tail: {commas(sanitized)}") print(f" abort_replay: {commas(aborted)}") if sanitized: preserved_ops = sum(r.edit_ops for r in edit_results if r.action == "sanitize_tail") preserved_payload = sum(r.edit_payload_lines for r in edit_results if r.action == "sanitize_tail") removed = sum(r.removed_len for r in edit_results if r.action == "sanitize_tail") print(f" ops preserved by sanitize: {commas(preserved_ops)}") print(f" payload lines preserved: {commas(preserved_payload)}") print(f" tail bytes removed: {commas(removed)}") def print_text_summary(results: list[TextBacktest]) -> None: if not results: return print("\n=== assistant message scan ===") print(f"candidate surfaces: {commas(len(results))}") print_counter("\nby action:", Counter(r.action for r in results)) print_counter("\nby surface/action:", Counter(f"{r.surface}:{r.action}" for r in results)) print_counter("\nby model/action:", Counter(f"{r.model or ''}:{r.action}" for r in results)) def signal_summary(labels: list[str], limit: int = 6) -> str: if not labels: return "none" counts = Counter(labels) parts = [f"{label}x{count}" if count > 1 else label for label, count in counts.most_common(limit)] rest = sum(counts.values()) - sum(count for _, count in counts.most_common(limit)) if rest: parts.append(f"+{rest} more") return ",".join(parts) def print_examples(results: list[ToolBacktest], show: int) -> None: if show <= 0: return print(f"\n=== sanitize_tail edit examples (up to {show}) ===") sanitize_examples = [r for r in results if r.tool_name == "edit" and r.action == "sanitize_tail"] for r in sanitize_examples[:show]: print(f"\n[id={r.row_id} seq={r.seq} model={r.model} signals={signal_summary(r.signals)}]") print(f"session: {r.session_file}") print(f"file(s): {', '.join(r.edit_files) if r.edit_files else ''}") print( f"parsed_end={r.parsed_end} text_len={r.text_len} removed={r.removed_len} " f"sha16={r.removed_sha16} ops={r.edit_ops} payload={r.edit_payload_lines}" ) print(f"last clean lines: {r.clean_preview}") print(f"removed preview: {r.removed_preview}") print(f"\n=== abort_replay examples (up to {show}) ===") abort_examples = [r for r in results if r.action == "abort_replay"] for r in abort_examples[:show]: print(f"\n[id={r.row_id} tool={r.tool_name} surface={r.surface} seq={r.seq} model={r.model} signals={signal_summary(r.signals)}]") print(f"session: {r.session_file}") if r.tool_name == "edit": print( f"parse={r.parse_reason} parsed_end={r.parsed_end} text_len={r.text_len} " f"ops={r.edit_ops} payload={r.edit_payload_lines} files={', '.join(r.edit_files) if r.edit_files else ''}" ) print(f"context: {r.context_preview}") def write_json_report(path: Path, tools: list[ToolBacktest], texts: list[TextBacktest]) -> None: def tool_dict(r: ToolBacktest) -> dict[str, Any]: return { "surface": r.surface, "row_id": r.row_id, "session_file": r.session_file, "seq": r.seq, "entry_id": r.entry_id, "call_id": r.call_id, "tool_name": r.tool_name, "model": r.model, "provider": r.provider, "action": r.action, "signals": r.signals, "signal_offsets": r.signal_offsets, "text_len": r.text_len, "parsed_end": r.parsed_end, "removed_len": r.removed_len, "removed_sha16": r.removed_sha16, "removed_preview": r.removed_preview, "clean_preview": r.clean_preview, "context_preview": r.context_preview, "edit_files": r.edit_files, "edit_ops": r.edit_ops, "edit_payload_lines": r.edit_payload_lines, "edit_deleted_lines": r.edit_deleted_lines, "parse_reason": r.parse_reason, } def text_dict(r: TextBacktest) -> dict[str, Any]: return { "surface": r.surface, "row_id": r.row_id, "session_file": r.session_file, "seq": r.seq, "entry_id": r.entry_id, "model": r.model, "provider": r.provider, "action": r.action, "signals": r.signals, "signal_offsets": r.signal_offsets, "text_len": r.text_len, "context_preview": r.context_preview, } payload = { "tool_calls": [tool_dict(r) for r in tools], "assistant_surfaces": [text_dict(r) for r in texts], } path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def main() -> int: ap = argparse.ArgumentParser( description="Backtest Harmony leak detection/recovery against session-stats sqlite tables." ) ap.add_argument("--db", type=Path, default=DB_PATH, help="stats sqlite path") ap.add_argument( "--strategy", choices=("marker", "fusion", "tail"), default="fusion", help=( "marker = trip on bare marker/control token; " "fusion = H or M plus co-signal; " "tail = only H or marker after parsed boundary" ), ) ap.add_argument("--provider", default=None, help="restrict tool/message rows to a provider") ap.add_argument("--model", default=None, help="restrict tool/message rows to a model") ap.add_argument("--tool", default=None, help="restrict tool-call rows to one tool") ap.add_argument("--include-assistant", action="store_true", help="also scan assistant text/thinking surfaces") ap.add_argument("--include-nullish", action="store_true", help="treat adjacent null/undefined as signal N") ap.add_argument("--no-fence-context", action="store_true", help="do not exempt Markdown fenced blocks") ap.add_argument("--legacy-loose-tail", action="store_true", help="model old raw-payload edit inputs as tail-sanitizable at first marker line") ap.add_argument("--show", type=int, default=8, help="examples per action group") ap.add_argument("--json-out", type=Path, default=None, help="write machine-readable report") args = ap.parse_args() conn = open_ro(args.db) print("=== harmony leak backtest ===") print(f"db: {args.db}") print(f"strategy: {args.strategy}") print(f"fences: {'ignored for action' if not args.no_fence_context else 'scanned as active text'}") if args.legacy_loose_tail: print("legacy: loose tail mode") if args.provider: print(f"provider: {args.provider}") if args.model: print(f"model: {args.model}") if args.tool: print(f"tool: {args.tool}") print() tool_results = scan_tools(conn, args) text_results = scan_assistant(conn, args) print_tool_summary(tool_results) print_text_summary(text_results) print_examples(tool_results, args.show) if args.json_out is not None: write_json_report(args.json_out, tool_results, text_results) print(f"\nwrote JSON report: {args.json_out}") return 0 if __name__ == "__main__": sys.exit(main())