feat: gpt-5 harmony errata fixes
- Replaced `===== ... =====` eval cell headers with `*** Begin ` / `*** End ` markers; legacy format remains renderable in HTML exports. - Replaced hashline patch grammar with `*** Begin Patch` / `*** End Patch` envelope; old inputs without the envelope are still accepted. - Extracted `sniffEvalLanguage` into a shared `sniff.ts` module reused by the parser and tool. - Added `docs/ERRATA-GPT5-HARMONY.md` and `scripts/session-stats/harmony_backtest.py` documenting and backtesting the GPT-5 Harmony-header leak defect.
This commit is contained in:
can1357
Executable
+991
@@ -0,0 +1,991 @@
|
||||
#!/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_RE = re.compile(r"^(?:@(?P<at>\S.*)|\*\*\* Update File:\s+(?P<upd>\S.*))\s*$")
|
||||
BEGIN_PATCH_RE = re.compile(r"^\*\*\* Begin Patch\s*$")
|
||||
END_PATCH_RE = re.compile(r"^\*\*\* End Patch\s*$")
|
||||
INSERT_RE = re.compile(r"^[+<]\s*(?P<anchor>BOF|EOF|[1-9][0-9]*[A-Za-z]{2})(?:\s*~(?P<inline>.*))?\s*$")
|
||||
RANGE_RE = re.compile(r"(?P<a>[1-9][0-9]*[A-Za-z]{2})(?:\.\.(?P<b>[1-9][0-9]*[A-Za-z]{2}))?")
|
||||
DELETE_RE = re.compile(r"^-\s*(?P<range>[1-9][0-9]*[A-Za-z]{2}(?:\.\.[1-9][0-9]*[A-Za-z]{2})?)\s*$")
|
||||
REPLACE_RE = re.compile(r"^=\s*(?P<range>[1-9][0-9]*[A-Za-z]{2}(?:\.\.[1-9][0-9]*[A-Za-z]{2})?)\s*$")
|
||||
|
||||
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<range>[1-9][0-9]*[A-Za-z]{2}(?:\.\.[1-9][0-9]*[A-Za-z]{2})?)(?P<op>[=+<])(?P<tail>.*)$"
|
||||
)
|
||||
old_delete = re.compile(
|
||||
r"^\s*-(?P<range>[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 range_deleted_lines(raw_range: str) -> int:
|
||||
m = 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 parse_edit_boundary(text: str, *, legacy_loose_tail: bool = False) -> EditBoundary:
|
||||
sections: list[EditSection] = []
|
||||
cur: EditSection | None = None
|
||||
parsed_end = 0
|
||||
line_no = 0
|
||||
needs_payload = False
|
||||
payload_allowed = False
|
||||
saw_required_payload = False
|
||||
seen_content = False
|
||||
|
||||
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
|
||||
target = (header.group("at") or header.group("upd") or "").strip()
|
||||
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 line.startswith("~"):
|
||||
if not payload_allowed:
|
||||
break
|
||||
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
|
||||
|
||||
trimmed = line.lstrip()
|
||||
ins = INSERT_RE.match(trimmed)
|
||||
if ins:
|
||||
cur.op_count += 1
|
||||
inline = ins.group("inline")
|
||||
if inline is None:
|
||||
needs_payload = True
|
||||
payload_allowed = True
|
||||
saw_required_payload = False
|
||||
# Not complete until at least one payload line appears.
|
||||
else:
|
||||
cur.payload_lines += 1
|
||||
parsed_end = end
|
||||
needs_payload = False
|
||||
payload_allowed = False
|
||||
saw_required_payload = False
|
||||
continue
|
||||
|
||||
dele = DELETE_RE.match(trimmed)
|
||||
if dele:
|
||||
cur.op_count += 1
|
||||
cur.deleted_lines += range_deleted_lines(dele.group("range"))
|
||||
parsed_end = end
|
||||
needs_payload = False
|
||||
payload_allowed = False
|
||||
saw_required_payload = False
|
||||
continue
|
||||
|
||||
repl = REPLACE_RE.match(trimmed)
|
||||
if repl:
|
||||
cur.op_count += 1
|
||||
cur.deleted_lines += 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 "<unknown>"
|
||||
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 '<unknown>'}:{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 '<unknown>'}:{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 '<none>'}")
|
||||
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 '<none>'}"
|
||||
)
|
||||
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())
|
||||
Reference in New Issue
Block a user