From bc71b8bc9520c41ab476db4cfb5978055bf5d76b Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 03:48:34 -0700 Subject: [PATCH] =?UTF-8?q?fix(compression):=20anchor=20on=20the=20LAST=20?= =?UTF-8?q?intent=20row=20=E2=80=94=20newer=20user=20turn=20outranks=20old?= =?UTF-8?q?er=20steer=20(#100053=20follow-up)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the salvaged #100114 commit. Its two-pass anchor selection scanned steers first and real user rows second, so a transcript shaped [user A, tool(steer B), ..., user C] anchored the already-consumed steer B over the newer real request C — the same replay class the PR set out to fix. Replace it with one reversed positional scan that picks whichever intent-bearing row is last (real role=user or steer-bearing role=tool), and make the compressed-transcript steer check count only role=tool rows (the only place the runtime delivers a steer), so a summary quoting the marker cannot masquerade as live intent. Adds S1/S2/S3 regression tests (steer dropped by compaction, steer surviving in tail, newer user turn after steer) plus alternation and use-exactly-once assertions. --- agent/conversation_compression.py | 42 ++--- .../test_compression_busy_steer_anchor.py | 146 ++++++++++++++++++ 2 files changed, 169 insertions(+), 19 deletions(-) create mode 100644 tests/agent/test_compression_busy_steer_anchor.py diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 885e58d1da..71ddcb9b00 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -2927,20 +2927,16 @@ def _extract_steer_text_from_message(message: Any) -> Optional[str]: return extracted if extracted else None -def _find_latest_busy_steer_text(messages: list) -> Optional[str]: - """Return the most recent steer payload in *messages*, if any.""" - for msg in reversed(messages): - if not isinstance(msg, dict): - continue - extracted = _extract_steer_text_from_message(msg) - if extracted: - return extracted - return None - - def _compressed_has_busy_steer(messages: list) -> bool: - """Whether *messages* already carries a steer marker (intent present).""" + """Whether *messages* already carries a steer marker (intent present). + + Only ``role=tool`` rows count: that is the sole place the runtime ever + delivers a steer, so a compaction summary that merely quotes the marker + text must not be mistaken for live intent. + """ for msg in messages: + if not isinstance(msg, dict) or msg.get("role") != "tool": + continue if _message_contains_busy_steer(msg): return True return False @@ -3155,19 +3151,27 @@ def _ensure_compressed_has_user_turn( _fresh_compaction_message_copy, ) - steer_text = _find_latest_busy_steer_text(original_messages) - if steer_text: - return _insert_real_user_anchor( - compressed, - {"role": "user", "content": steer_text}, - ) - + # One reversed positional scan: the anchor is whichever intent-bearing + # row is LAST in the original transcript — a real ``role=user`` turn or + # a steer marker riding inside a ``role=tool`` result. Scanning the two + # kinds separately (steer first, then user) would let an older, already + # consumed steer outrank a newer real user request and replay it + # (#100053 follow-up: ``[user A, tool(steer B), ..., user C]`` must + # anchor C, not B). for message in reversed(original_messages): if _is_real_user_message(message): return _insert_real_user_anchor( compressed, _fresh_compaction_message_copy(message), ) + if not isinstance(message, dict) or message.get("role") != "tool": + continue + steer_text = _extract_steer_text_from_message(message) + if steer_text: + return _insert_real_user_anchor( + compressed, + {"role": "user", "content": steer_text}, + ) from agent.message_metadata import append_message append_message( diff --git a/tests/agent/test_compression_busy_steer_anchor.py b/tests/agent/test_compression_busy_steer_anchor.py new file mode 100644 index 0000000000..3aed06bece --- /dev/null +++ b/tests/agent/test_compression_busy_steer_anchor.py @@ -0,0 +1,146 @@ +"""Regression coverage for busy-steer preservation across compaction (#100053). + +With ``display.busy_input_mode: steer`` the follow-up rides inside the latest +``role=tool`` result (``apply_pending_steer_to_tool_results``), never as a +``role=user`` row. ``_ensure_compressed_has_user_turn`` must treat that marker +as live user intent — and must pick whichever intent-bearing row is LAST in +the original transcript, so an older steer never outranks a newer real user +request. +""" + +import pytest + +from agent.context_compressor import ( + COMPRESSION_CONTINUATION_USER_CONTENT, + SUMMARY_PREFIX, +) +from agent.conversation_compression import ( + _compressed_has_busy_steer, + _ensure_compressed_has_user_turn, +) +from agent.prompt_builder import STEER_MARKER_OPEN, format_steer_marker + +REQUEST_A = "Historical request A: audit the auth module." +STEER_B = "Steer B: stop, switch to fixing the login bug instead." +REQUEST_C = "Newer real user request C: now write the release notes." + + +def _tool_turns(start: int, count: int, *, steer_at: int | None = None) -> list[dict]: + turns: list[dict] = [] + for idx in range(start, start + count): + turns.append( + { + "role": "assistant", + "content": "Working.", + "tool_calls": [ + { + "id": f"call-{idx}", + "function": {"name": "terminal", "arguments": "{}"}, + } + ], + } + ) + content = f"tool output {idx}" + if steer_at == idx: + content += format_steer_marker(STEER_B) + turns.append({"role": "tool", "tool_call_id": f"call-{idx}", "content": content}) + return turns + + +def _summary_row() -> dict: + return {"role": "user", "content": f"{SUMMARY_PREFIX}\n\nEarlier work summarized."} + + +def _assert_alternation(messages: list[dict]) -> None: + roles = [m.get("role") for m in messages] + for left, right in zip(roles, roles[1:]): + assert not (left == right == "user"), f"user/user adjacency in {roles}" + assert not (left == right == "assistant"), f"assistant/assistant adjacency in {roles}" + + +def _user_rows(messages: list[dict]) -> list[str]: + return [str(m.get("content")) for m in messages if m.get("role") == "user"] + + +def test_s1_steer_summarized_away_becomes_anchor_not_historical_request(): + """S1: the steer lived in a tool row that compaction dropped; the only + ``role=user`` row in history is the already-consumed request A. The steer + must be restored as the anchor, and A must not be replayed.""" + original = [{"role": "user", "content": REQUEST_A}] + _tool_turns(0, 6, steer_at=2) + compressed = [_summary_row(), *_tool_turns(5, 1)] + + outcome = _ensure_compressed_has_user_turn(original, compressed) + + assert outcome == "inserted" + _assert_alternation(compressed) + users = _user_rows(compressed) + assert STEER_B in users, users + assert REQUEST_A not in users, "historical request replayed as new input" + assert COMPRESSION_CONTINUATION_USER_CONTENT not in users + # Steer text is used exactly once across the whole compressed transcript. + assert sum(str(m.get("content")).count(STEER_B) for m in compressed) == 1 + + +def test_s2_steer_surviving_in_tail_tool_row_counts_as_present(): + """S2: the steer-bearing tool row survived into the tail. No anchor may be + inserted (the intent is already there) and A must not be cloned.""" + original = [{"role": "user", "content": REQUEST_A}] + _tool_turns(0, 6, steer_at=5) + compressed = [_summary_row(), *_tool_turns(5, 1, steer_at=5)] + before = [dict(m) for m in compressed] + + outcome = _ensure_compressed_has_user_turn(original, compressed) + + assert outcome == "already_present" + assert compressed == before, "transcript mutated despite live steer present" + assert REQUEST_A not in _user_rows(compressed) + assert sum(str(m.get("content")).count(STEER_B) for m in compressed) == 1 + + +def test_s3_newer_real_user_turn_outranks_older_steer(): + """S3: ``[user A, tool(steer B), ..., user C]`` — C is the newest intent. + A steer-first scan would anchor the consumed steer B and replay it.""" + original = ( + [{"role": "user", "content": REQUEST_A}] + + _tool_turns(0, 3, steer_at=1) + + [{"role": "user", "content": REQUEST_C}] + + _tool_turns(3, 4) + ) + compressed = [_summary_row(), *_tool_turns(6, 1)] + + outcome = _ensure_compressed_has_user_turn(original, compressed) + + assert outcome == "inserted" + _assert_alternation(compressed) + users = _user_rows(compressed) + assert REQUEST_C in users, users + assert STEER_B not in users, "older consumed steer replayed over newer user turn" + assert REQUEST_A not in users + assert not any(STEER_B in u for u in users) + + +def test_newer_steer_outranks_older_real_user_turn(): + """Mirror of S3: ``[user A, ..., tool(steer B)]`` — the steer is newest.""" + original = [{"role": "user", "content": REQUEST_A}] + _tool_turns(0, 4, steer_at=3) + compressed = [_summary_row(), *_tool_turns(4, 1)] + + outcome = _ensure_compressed_has_user_turn(original, compressed) + + assert outcome == "inserted" + _assert_alternation(compressed) + users = _user_rows(compressed) + assert STEER_B in users + assert REQUEST_A not in users + + +@pytest.mark.parametrize( + "role", + ["user", "assistant"], +) +def test_compressed_steer_presence_only_counts_tool_rows(role): + """A summary or assistant row that merely quotes the marker text is not a + live steer delivery — only ``role=tool`` rows carry real steers.""" + quoted = {"role": role, "content": f"{SUMMARY_PREFIX}\n{format_steer_marker(STEER_B)}"} + assert _compressed_has_busy_steer([quoted]) is False + assert STEER_MARKER_OPEN in quoted["content"] + live = {"role": "tool", "tool_call_id": "c", "content": f"ok{format_steer_marker(STEER_B)}"} + assert _compressed_has_busy_steer([live]) is True