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.
This commit is contained in:
Teknium
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user