diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 7504cc67c8..0de94ee662 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -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): diff --git a/tests/agent/test_task_snapshot_heading_contract.py b/tests/agent/test_task_snapshot_heading_contract.py new file mode 100644 index 0000000000..e4967db02c --- /dev/null +++ b/tests/agent/test_task_snapshot_heading_contract.py @@ -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