From c4bbb14e528c3aa257f0f26c930e339b9afa3bc3 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sat, 15 Aug 2026 16:28:52 -0700 Subject: [PATCH] feat(compression): mechanical anchor index + region-scoping tripwire - _build_anchor_index(): regex-harvests PR/issue numbers, SHAs, branches, file paths, error strings, handles, URLs from the compacted region into a bounded indexed summary section. LLM-free, so needle identifiers cannot be paraphrased away (the GUI-lineage failure class: 10/15 verbatim-or-nothing golds). Doubles as session_search query-anchor map. - evals/compaction/test_region_scoping.py: sentinel tripwire proving the summarizer input carries ONLY the compacted region (head/tail sentinels never reach the serialized turns body) in both legacy and lean modes. --- agent/context_compressor.py | 71 ++++++++++++++++++ evals/compaction/test_region_scoping.py | 98 +++++++++++++++++++++++++ 2 files changed, 169 insertions(+) create mode 100644 evals/compaction/test_region_scoping.py diff --git a/agent/context_compressor.py b/agent/context_compressor.py index e998f7553c..66e7753de0 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -839,6 +839,73 @@ _LOW_SIGNAL_TOOL_RE = re.compile( r"(?:\"exit_code\"\s*:\s*0)?\s*\}?$" ) +# Anchor ledger (#compaction-v2, Pi/Cline file-ops-ledger convergence, adapted): +# mechanically harvest exact identifiers from the compacted region into an +# indexed summary section. No LLM in the loop, so nothing can be paraphrased +# away — this is the defense for needle-facts (SHAs, ids, error strings) that +# honest summarization at 10:1 always loses. Doubles as a query-anchor map +# for session_search recovery. +_LEAN_ANCHOR_HEADING = "## Anchor Index (mechanically extracted, exact)" +_LEAN_ANCHOR_BUDGET_CHARS = 7_000 +_ANCHOR_PATTERNS: "list[tuple[str, re.Pattern[str], int]]" = [ + ("PRs/issues", re.compile(r"#\d{3,6}\b"), 120), + ("commits", re.compile(r"\b[0-9a-f]{9,40}\b"), 40), + ("branches", re.compile(r"\b(?:fix|feat|docs|refactor|chore|salvage|ent)/[A-Za-z0-9._/-]{3,60}"), 40), + ("files", re.compile(r"\b[\w./-]+/[\w.-]+\.(?:py|ts|tsx|js|rs|md|yaml|yml|json|toml|sh)\b"), 80), + ("errors", re.compile(r"\b(?:[A-Z][a-zA-Z]*Error|Exception|ENOSPC|EACCES|SIGKILL|Traceback)\b[^\n]{0,90}"), 40), + ("handles", re.compile(r"@[A-Za-z0-9-]{3,30}\b"), 40), + ("urls", re.compile(r"https?://[^\s)\"']{10,110}"), 30), +] +_ANCHOR_NOISE = frozenset({ + "@teknium", "@teknium1", # session owner, in every transcript +}) + + +def _build_anchor_index(turns: List[Dict[str, Any]]) -> str: + """Regex-harvest exact identifiers from the compacted region. + + Deterministic and LLM-free. Per-category caps keep the section bounded; + within a category, most-frequent first (frequency is a decent proxy for + load-bearing), ties broken by last-seen order (recency). + """ + text_parts: list[str] = [] + for msg in turns: + c = msg.get("content") + if isinstance(c, str) and c: + text_parts.append(c) + text = "\n".join(text_parts) + if not text: + return "" + sections: list[str] = [] + used = 0 + for label, pattern, cap in _ANCHOR_PATTERNS: + counts: dict[str, int] = {} + last_seen: dict[str, int] = {} + for n, m in enumerate(pattern.finditer(text)): + val = m.group(0).strip().rstrip(".,;:") + if val.lower() in _ANCHOR_NOISE: + continue + counts[val] = counts.get(val, 0) + 1 + last_seen[val] = n + if not counts: + continue + ranked = sorted(counts, key=lambda v: (-counts[v], -last_seen[v]))[:cap] + line = f"{label}: " + ", ".join( + f"{v}(x{counts[v]})" if counts[v] > 1 else v for v in ranked + ) + if used + len(line) > _LEAN_ANCHOR_BUDGET_CHARS: + break + sections.append(line) + used += len(line) + if not sections: + return "" + return ( + "\n\n" + _LEAN_ANCHOR_HEADING + "\n" + + "\n".join(sections) + + "\n(Exact identifiers from the compacted region — use these verbatim, " + "and as session_search query anchors to recover their full context.)" + ) + def _digest_worthy(role: str, content: str) -> bool: """Filter no-signal rows out of the digest input. @@ -4146,6 +4213,10 @@ Summary generation was unavailable, so this is a best-effort deterministic fallb """ if getattr(self, "tail_mode", "legacy") != "lean": return summary + if _LEAN_ANCHOR_HEADING not in summary: + summary += _redact_compaction_text( + _build_anchor_index(turns_to_summarize) + ) if _LEAN_DIGESTS_HEADING not in summary: summary += _redact_compaction_text( self._build_chunk_digests(turns_to_summarize) diff --git a/evals/compaction/test_region_scoping.py b/evals/compaction/test_region_scoping.py new file mode 100644 index 0000000000..a6c29c32ba --- /dev/null +++ b/evals/compaction/test_region_scoping.py @@ -0,0 +1,98 @@ +"""Region-scoping tripwire: the summarizer must only see the compacted region. + +Builds a transcript with sentinel strings planted in (a) the protected head, +(b) the middle (to-be-compacted) region, and (c) the tail, mocks call_llm to +capture the prompt, and asserts head/tail sentinels never reach the +summarizer while the middle sentinel does. Runs for both legacy and lean +modes, and asserts the lean deterministic sections (anchors, verbatim users) +also carry only middle-region content. +""" +import json +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT)) + +from agent.context_compressor import ContextCompressor # noqa: E402 + +HEAD_SENTINEL = "HEADSENTINEL_zq81" +MID_SENTINEL = "MIDSENTINEL_kv93" +TAIL_SENTINEL = "TAILSENTINEL_pw27" + + +def _mk_transcript(): + msgs = [ + {"role": "system", "content": "system prompt"}, + {"role": "user", "content": f"first user message {HEAD_SENTINEL}"}, + {"role": "assistant", "content": "ack"}, + ] + for i in range(40): + marker = f" {MID_SENTINEL}-{i}" if i % 5 == 0 else "" + msgs.append({ + "role": "assistant", "content": f"mid step {i}{marker}", + "tool_calls": [{"id": f"m{i}", "function": {"name": "terminal", "arguments": "{}"}}], + }) + msgs.append({"role": "tool", "tool_call_id": f"m{i}", + "content": (f"mid tool output {i} " * 300) + marker}) + for i in range(6): + msgs.append({"role": "assistant", "content": f"tail step {i} {TAIL_SENTINEL}-{i}", + "tool_calls": [{"id": f"t{i}", "function": {"name": "terminal", "arguments": "{}"}}]}) + msgs.append({"role": "tool", "tool_call_id": f"t{i}", "content": f"tail output {i} {TAIL_SENTINEL}-{i}"}) + msgs.append({"role": "user", "content": f"latest user question {TAIL_SENTINEL}-u"}) + msgs.append({"role": "assistant", "content": "final answer in tail"}) + return msgs + + +def run_mode(tail_mode: str): + captured = [] + + def fake_call_llm(messages=None, **kw): + captured.append(messages[0]["content"] if messages else "") + resp = MagicMock() + resp.choices[0].message.content = "## Active Task\nsummarized" + return resp + + comp = ContextCompressor(model="anthropic/claude-fable-5", quiet_mode=True, + tail_mode=tail_mode) + comp.tail_token_budget = 3_000 # force a real middle on the small fixture + comp._session_id = "scope-test" + msgs = _mk_transcript() + with patch("agent.context_compressor.call_llm", side_effect=fake_call_llm), \ + patch("agent.auxiliary_client.call_llm", side_effect=fake_call_llm): + out = comp.compress(msgs, current_tokens=200_000, force=True) + + all_prompts = "\n".join(captured) + assert captured, f"[{tail_mode}] summarizer never called" + assert MID_SENTINEL in all_prompts, f"[{tail_mode}] middle region missing from summarizer input" + # Head/tail user messages MAY appear inside the FOCUS TOPIC steering block + # (intentional: tells the summarizer what the user currently cares about). + # They must NOT appear in the serialized TURNS body being summarized. + for p in captured: + body = p.split("FOCUS TOPIC:")[0] + assert TAIL_SENTINEL not in body, f"[{tail_mode}] TAIL leaked into summarized turns" + assert HEAD_SENTINEL not in body, f"[{tail_mode}] protected HEAD leaked into summarized turns" + + # The tail must survive verbatim; the head user message must survive. + out_text = "\n".join(str(m.get("content")) for m in out) + assert f"{TAIL_SENTINEL}-u" in out_text, f"[{tail_mode}] latest user message lost" + assert HEAD_SENTINEL in out_text, f"[{tail_mode}] head lost" + + if tail_mode == "lean": + summary_msg = next( + (str(m.get("content")) for m in out + if isinstance(m.get("content"), str) and "Anchor Index" in m["content"]), + "", + ) + if summary_msg: + assert TAIL_SENTINEL not in summary_msg.split("END OF CONTEXT SUMMARY")[0], \ + "[lean] tail content leaked into summary sections" + print(f" {tail_mode}: OK ({len(captured)} summarizer call(s), " + f"{len(out)} msgs out)") + + +if __name__ == "__main__": + for mode in ("legacy", "lean"): + run_mode(mode) + print("scoping tripwire: ALL PASS")