From 2ca38e5df43411afe11aedfc34bfd71a2eecd4bb Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 22 Jul 2026 21:38:49 -0700 Subject: [PATCH] test(compression): pin scaffolding-tail standalone append + stale-snapshot refresh Covers the follow-up hardening: continuation-marker and summary-as-user tails keep the flagged standalone snapshot (zero-user provenance #69292 verified via _transcript_has_real_user_turn on the projected rows), stale snapshot rows are refreshed in place, a previously merged snapshot is stripped before re-injection, and an all-completed todo store injects nothing (#26981). --- .../agent/test_compression_rotation_state.py | 188 ++++++++++++++++++ 1 file changed, 188 insertions(+) diff --git a/tests/agent/test_compression_rotation_state.py b/tests/agent/test_compression_rotation_state.py index 86bcc1e60dbb..a5d7b58b962a 100644 --- a/tests/agent/test_compression_rotation_state.py +++ b/tests/agent/test_compression_rotation_state.py @@ -804,3 +804,191 @@ class TestTodoSnapshotMergedNotDuplicated: previous.get("role") == current.get("role") == "user" for previous, current in zip(db_msgs, db_msgs[1:]) ) + + +class TestTodoSnapshotScaffoldingTails: + """Scaffolding tails must never absorb the todo snapshot (#69292).""" + + @staticmethod + def _agent_with_todo(db: SessionDB, session_id: str, tail: dict): + db.create_session(session_id, source="cli") + agent = _build_agent_with_db(db, session_id, platform="cli") + agent.context_compressor.compress.return_value = [ + {"role": "user", "content": "[CONTEXT COMPACTION] summary"}, + {"role": "assistant", "content": "acknowledged"}, + tail, + ] + agent._todo_store.write( + [{"id": "t1", "content": "task A", "status": "pending"}] + ) + return agent + + def test_snapshot_stays_standalone_after_continuation_marker( + self, tmp_path: Path + ): + from agent.context_compressor import ( + COMPRESSION_CONTINUATION_USER_CONTENT, + ContextCompressor, + ) + + db = SessionDB(db_path=tmp_path / "state.db") + agent = self._agent_with_todo( + db, + "PARENT_TODO_MARKER_TAIL", + { + "role": "user", + "content": COMPRESSION_CONTINUATION_USER_CONTENT, + }, + ) + + compressed, _ = agent._compress_context( + _msgs(), "sys", approx_tokens=120_000 + ) + + tail = compressed[-1] + assert tail["role"] == "user" + assert tail.get("_todo_snapshot_synthetic") is True + assert "task A" in tail["content"] + # The continuation marker keeps its exact text so it stays + # recognizable as scaffolding after SessionDB projection. + marker_rows = [ + message + for message in compressed + if message.get("content") == COMPRESSION_CONTINUATION_USER_CONTENT + ] + assert len(marker_rows) == 1 + # Zero-user provenance: neither the marker nor the snapshot may read + # as a real user turn once SessionDB projection strips the flags + # (#69292). The fixture's stub summary text is not a real handoff + # prefix, so assert on the projected scaffolding rows directly. + assert not ContextCompressor._transcript_has_real_user_turn( + [ + {"role": "user", "content": marker_rows[0]["content"]}, + {"role": "user", "content": tail["content"]}, + ] + ) + + def test_snapshot_stays_standalone_after_summary_as_user_tail( + self, tmp_path: Path + ): + from agent.context_compressor import SUMMARY_PREFIX, ContextCompressor + + summary_as_user = f"{SUMMARY_PREFIX}\nzero-user summary body" + db = SessionDB(db_path=tmp_path / "state.db") + agent = self._agent_with_todo( + db, + "PARENT_TODO_SUMMARY_TAIL", + {"role": "user", "content": summary_as_user}, + ) + + compressed, _ = agent._compress_context( + _msgs(), "sys", approx_tokens=120_000 + ) + + tail = compressed[-1] + assert tail.get("_todo_snapshot_synthetic") is True + assert "task A" in tail["content"] + # The summary handoff prefix must stay at the START of its own + # message for downstream summary detection. + summary_rows = [ + message + for message in compressed + if str(message.get("content") or "").startswith(SUMMARY_PREFIX) + ] + assert len(summary_rows) == 1 + # Zero-user provenance (#69292): after SessionDB projection strips + # the flags, both the summary-as-user handoff and the standalone + # snapshot must still classify as synthetic — the merge would have + # buried the header/prefix markers mid-content. + assert not ContextCompressor._transcript_has_real_user_turn( + [ + {"role": "user", "content": summary_rows[0]["content"]}, + {"role": "user", "content": tail["content"]}, + ] + ) + + def test_stale_snapshot_row_is_refreshed_not_stacked(self, tmp_path: Path): + from tools.todo_tool import TODO_INJECTION_HEADER + + stale = f"{TODO_INJECTION_HEADER}\n- [ ] t0. old finished task (pending)" + db = SessionDB(db_path=tmp_path / "state.db") + agent = self._agent_with_todo( + db, + "PARENT_TODO_STALE_ROW", + {"role": "user", "content": stale}, + ) + + compressed, _ = agent._compress_context( + _msgs(), "sys", approx_tokens=120_000 + ) + + tail = compressed[-1] + assert tail.get("_todo_snapshot_synthetic") is True + assert "task A" in tail["content"] + assert "old finished task" not in tail["content"] + snapshot_rows = [ + message + for message in compressed + if str(message.get("content") or "").startswith(TODO_INJECTION_HEADER) + ] + assert len(snapshot_rows) == 1 + + def test_previously_merged_snapshot_is_stripped_before_reinjection( + self, tmp_path: Path + ): + from tools.todo_tool import TODO_INJECTION_HEADER + + previously_merged = ( + "please fix the login bug\n\n" + f"{TODO_INJECTION_HEADER}\n- [ ] t0. old finished task (pending)" + ) + db = SessionDB(db_path=tmp_path / "state.db") + agent = self._agent_with_todo( + db, + "PARENT_TODO_RESTRIP", + {"role": "user", "content": previously_merged}, + ) + + compressed, _ = agent._compress_context( + _msgs(), "sys", approx_tokens=120_000 + ) + + tail = compressed[-1] + assert tail["role"] == "user" + assert "please fix the login bug" in tail["content"] + assert "task A" in tail["content"] + assert "old finished task" not in tail["content"] + assert tail["content"].count(TODO_INJECTION_HEADER) == 1 + assert not any( + previous.get("role") == current.get("role") == "user" + for previous, current in zip(compressed, compressed[1:]) + ) + + def test_empty_todo_store_injects_nothing(self, tmp_path: Path): + from tools.todo_tool import TODO_INJECTION_HEADER + + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "PARENT_TODO_EMPTY" + db.create_session(session_id, source="cli") + agent = _build_agent_with_db(db, session_id, platform="cli") + expected = [ + {"role": "user", "content": "[CONTEXT COMPACTION] summary"}, + {"role": "assistant", "content": "acknowledged"}, + {"role": "user", "content": "tail"}, + ] + agent.context_compressor.compress.return_value = [ + dict(message) for message in expected + ] + agent._todo_store.write( + [{"id": "t1", "content": "done thing", "status": "completed"}] + ) + + compressed, _ = agent._compress_context( + _msgs(), "sys", approx_tokens=120_000 + ) + + assert compressed == expected + assert not any( + TODO_INJECTION_HEADER in str(message.get("content") or "") + for message in compressed + )