fix(compression): replace alias task headings during snapshot grounding

Align the iterative update instruction with HISTORICAL_TASK_HEADING and
treat leftover ## Active Task sections as the same snapshot so grounding
replaces them instead of prepending a second live-looking task heading.

Fixes #114479
This commit is contained in:
joaomarcos
2026-09-17 22:12:16 -03:00
committed by Teknium
parent 4ab7dec756
commit d295ef02ee
2 changed files with 81 additions and 3 deletions

View File

@@ -1016,7 +1016,12 @@ _PATH_MENTION_RE = re.compile(r"(?:/|~/?|[A-Za-z]:\\)[^\s`'\")\]}<>]+")
# MEDIA delivery directives must not reach the summarizer — if one leaks into the summary, the downstream
# model may re-emit it as an active directive on the next turn, triggering bogus attachment sends (#14665).
_MEDIA_DIRECTIVE_RE = re.compile(r"MEDIA:\S+")
_HISTORICAL_TASK_SECTION_RE = re.compile(rf"(?ms)^{re.escape(HISTORICAL_TASK_HEADING)}\s*\n.*?(?=^## |\Z)")
# Pre-#44454 alias. A summarizer that still emits it must be replaced, not prepended.
_LEGACY_ACTIVE_TASK_HEADING = "## Active Task"
_TASK_SNAPSHOT_HEADINGS = (HISTORICAL_TASK_HEADING, _LEGACY_ACTIVE_TASK_HEADING)
_HISTORICAL_TASK_SECTION_RE = re.compile(
rf"(?ms)^(?:{'|'.join(re.escape(heading) for heading in _TASK_SNAPSHOT_HEADINGS)})\s*\n.*?(?=^## |\Z)"
)
def _redact_compaction_text(text: Any) -> str:
@@ -3784,8 +3789,8 @@ Write only the summary body. Do not include any preamble or prefix."""
"""Reject user attribution when the source transcript has no user."""
if has_user_turn:
return
match = re.search(rf"(?ms)^{re.escape(HISTORICAL_TASK_HEADING)}\s*\n(.*?)(?=\n##\s|\Z)", summary)
task_snapshot = match.group(1).strip() if match else ""
match = _HISTORICAL_TASK_SECTION_RE.search(summary)
task_snapshot = match.group(0).split("\n", 1)[-1].strip() if match else ""
# The "User asked:" scan can false-positive on quoted tool output; acceptable, since
# the RuntimeError only costs one retry on the existing fallback path.
if task_snapshot != _NO_USER_TASK_SENTINEL or re.search(r"\bUser\s+asked\s*:", summary, re.IGNORECASE):

View File

@@ -0,0 +1,73 @@
"""Task-snapshot heading identity across prompt, template, and grounding.
The iterative summarizer prompt, the emitted template, and deterministic
grounding must share one heading. A leftover ``## Active Task`` section is
not disclaimed by SUMMARY_PREFIX and reads as live work (#114479 / #44454).
"""
from types import SimpleNamespace
from agent.context_compressor import (
ContextCompressor,
HISTORICAL_TASK_HEADING,
)
_LEGACY_ACTIVE_TASK_HEADING = "## Active Task"
def _iterative_update_prompt() -> str:
stub = SimpleNamespace(
tail_mode="lean",
_previous_summary="PREVIOUS SUMMARY BODY",
_bound_summary_input=lambda text: text,
)
stub._summary_template_sections = ContextCompressor._summary_template_sections
stub._build_summary_prompt = ContextCompressor._build_summary_prompt.__get__(stub)
return stub._build_summary_prompt("NEW TURNS", 2000, None, "", True)
def _headings(text: str) -> list[str]:
return [line for line in text.splitlines() if line.startswith("## ")]
def test_iterative_update_instruction_names_canonical_heading():
prompt = _iterative_update_prompt()
assert f'Update "{HISTORICAL_TASK_HEADING}"' in prompt
assert f'Update "{_LEGACY_ACTIVE_TASK_HEADING}"' not in prompt
assert HISTORICAL_TASK_HEADING in prompt
def test_grounding_replaces_legacy_active_task_heading():
body = (
f"{_LEGACY_ACTIVE_TASK_HEADING}\n"
"User asked: 'do X'\n\n"
"## Goal\n"
"thing\n"
)
grounded = ContextCompressor._ground_historical_task_snapshot.__func__(
ContextCompressor,
body,
[{"role": "user", "content": "do X"}],
)
headings = _headings(grounded)
assert headings.count(HISTORICAL_TASK_HEADING) == 1
assert _LEGACY_ACTIVE_TASK_HEADING not in headings
assert "## Goal" in headings
def test_grounding_keeps_single_canonical_heading():
body = (
f"{HISTORICAL_TASK_HEADING}\n"
"User asked: 'stale'\n\n"
"## Goal\n"
"thing\n"
)
grounded = ContextCompressor._ground_historical_task_snapshot.__func__(
ContextCompressor,
body,
[{"role": "user", "content": "fresh ask"}],
)
headings = _headings(grounded)
assert headings.count(HISTORICAL_TASK_HEADING) == 1
assert _LEGACY_ACTIVE_TASK_HEADING not in headings
assert "fresh ask" in grounded