From 7de7a073c65024429d075f0b703fc90e2ecb762b Mon Sep 17 00:00:00 2001 From: embwl0x Date: Thu, 16 Jul 2026 03:10:35 -0500 Subject: [PATCH] fix(compression): preserve synthetic user provenance --- agent/context_compressor.py | 53 +++++- agent/conversation_compression.py | 12 +- ...context_compressor_zero_user_provenance.py | 157 ++++++++++++++++++ tools/todo_tool.py | 7 +- 4 files changed, 213 insertions(+), 16 deletions(-) diff --git a/agent/context_compressor.py b/agent/context_compressor.py index e5d96c594522..cb49a131818e 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -34,6 +34,7 @@ from agent.model_metadata import ( ) from agent.redact import redact_sensitive_text from agent.turn_context import drop_stale_api_content +from tools.todo_tool import TODO_INJECTION_HEADER logger = logging.getLogger(__name__) @@ -134,6 +135,15 @@ COMPRESSED_SUMMARY_HAS_USER_TURN_KEY = "_compressed_summary_has_user_turn" _DB_PERSISTED_MARKER = "_db_persisted" _NO_USER_TASK_SENTINEL = "None. This session contains no user-authored turns." +COMPRESSION_CONTINUATION_USER_CONTENT = ( + "Continue from the compressed conversation context above. " + "This marker exists because no human user turn was available." +) +_LEGACY_COMPRESSION_CONTINUATION_USER_CONTENT = ( + "Continue from the compressed conversation context above. " + "This marker exists because the compacted transcript contained " + "no preserved user turn." +) def _fresh_compaction_message_copy(msg: Dict[str, Any]) -> Dict[str, Any]: @@ -1999,6 +2009,9 @@ class ContextCompressor(ContextEngine): role = msg.get("role", "unknown") text = _compact_fallback_turn(msg.get("content")) _collect_path_mentions(text, relevant_files) + synthetic_user = ( + role == "user" and self._is_synthetic_compression_user_turn(msg) + ) turn_text = text turn_tool_names: list[str] = [] @@ -2009,12 +2022,13 @@ class ContextCompressor(ContextEngine): if turn_tool_names: prefix = "tool calls: " + ", ".join(turn_tool_names[:6]) turn_text = f"{prefix}; {turn_text}" if turn_text else prefix - _remember_dropped_turn(str(role).upper(), turn_text) + turn_label = "INTERNAL CONTEXT" if synthetic_user else str(role).upper() + _remember_dropped_turn(turn_label, turn_text) if len(text) > 600: text = text[:420].rstrip() + " ... " + text[-160:].lstrip() - if role == "user" and text: + if role == "user" and text and not synthetic_user: user_asks.append(text) elif role == "assistant": tool_names: list[str] = [] @@ -2745,13 +2759,33 @@ This compaction should PRIORITISE preserving all information related to the focu for message in messages: if not isinstance(message, dict) or message.get("role") != "user": continue - if cls._has_compressed_summary_metadata(message): - continue - if cls._is_context_summary_content(message.get("content")): + if cls._is_synthetic_compression_user_turn(message): continue return True return False + @classmethod + def _is_synthetic_compression_user_turn(cls, message: Any) -> bool: + """Recognize internal user-role rows after SessionDB projection. + + SessionDB preserves role/content but not underscore-prefixed metadata, + so the stable todo and continuation content markers are authoritative. + """ + if not isinstance(message, dict) or message.get("role") != "user": + return False + if cls._has_compressed_summary_metadata(message): + return True + content = message.get("content") + if cls._is_context_summary_content(content): + return True + text = _content_text_for_contains(content).strip() + return text in { + COMPRESSION_CONTINUATION_USER_CONTENT, + _LEGACY_COMPRESSION_CONTINUATION_USER_CONTENT, + } or text.startswith( + TODO_INJECTION_HEADER + "\n" + ) + @staticmethod def _validate_summary_user_provenance(summary: str, has_user_turn: bool) -> None: """Reject user attribution when the source transcript has no user.""" @@ -2782,9 +2816,9 @@ This compaction should PRIORITISE preserving all information related to the focu msg = messages[idx] if msg.get("role") != "user": continue - content = msg.get("content") - if cls._is_context_summary_content(content): + if cls._is_synthetic_compression_user_turn(msg): continue + content = msg.get("content") text = redact_sensitive_text(_content_text_for_contains(content).strip()) if not text: continue @@ -3065,8 +3099,9 @@ This compaction should PRIORITISE preserving all information related to the focu """ for i in range(len(messages) - 1, head_end - 1, -1): msg = messages[i] - if msg.get("role") == "user" and not self._is_context_summary_content( - msg.get("content") + if ( + msg.get("role") == "user" + and not self._is_synthetic_compression_user_turn(msg) ): return i return -1 diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 01f882be2ee7..cd6d93bc7cd2 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -613,7 +613,7 @@ def _is_real_user_message(message: Any) -> bool: return False from agent.context_compressor import ContextCompressor - return not ContextCompressor._is_context_summary_content(text) + return not ContextCompressor._is_synthetic_compression_user_turn(message) def _merge_anchor_into_user_message(target: dict, anchor: dict) -> None: @@ -690,7 +690,10 @@ def _ensure_compressed_has_user_turn(original_messages: list, compressed: list) """Preserve human intent, not merely a synthetic user-role placeholder.""" if any(_is_real_user_message(message) for message in compressed): return - from agent.context_compressor import _fresh_compaction_message_copy + from agent.context_compressor import ( + COMPRESSION_CONTINUATION_USER_CONTENT, + _fresh_compaction_message_copy, + ) for message in reversed(original_messages): if _is_real_user_message(message): @@ -701,10 +704,7 @@ def _ensure_compressed_has_user_turn(original_messages: list, compressed: list) return compressed.append({ "role": "user", - "content": ( - "Continue from the compressed conversation context above. " - "This marker exists because no human user turn was available." - ), + "content": COMPRESSION_CONTINUATION_USER_CONTENT, }) diff --git a/tests/agent/test_context_compressor_zero_user_provenance.py b/tests/agent/test_context_compressor_zero_user_provenance.py index 47974a0cd0b3..eeb044a1ffbd 100644 --- a/tests/agent/test_context_compressor_zero_user_provenance.py +++ b/tests/agent/test_context_compressor_zero_user_provenance.py @@ -1,11 +1,13 @@ """Regression coverage for zero-user compaction integrity (#64539).""" +import os from types import SimpleNamespace from unittest.mock import patch import pytest from agent.context_compressor import ( + COMPRESSION_CONTINUATION_USER_CONTENT, COMPRESSED_SUMMARY_HAS_USER_TURN_KEY, COMPRESSED_SUMMARY_METADATA_KEY, HISTORICAL_TASK_HEADING, @@ -13,6 +15,12 @@ from agent.context_compressor import ( ContextCompressor, _NO_USER_TASK_SENTINEL, ) +from agent.conversation_compression import ( + _ensure_compressed_has_user_turn, + compress_context, +) +from hermes_state import SessionDB +from tools.todo_tool import TODO_INJECTION_HEADER def _response(content: str) -> SimpleNamespace: @@ -67,6 +75,40 @@ def _assistant_tool_turns(start: int, count: int) -> list[dict]: return turns +def _assistant_turns(start: int, count: int) -> list[dict]: + return [ + { + "role": "assistant", + "content": f"Scheduled step {idx} completed. " + ("x" * 500), + } + for idx in range(start, start + count) + ] + + +def _lifecycle_agent(db: SessionDB, session_id: str): + with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}): + from run_agent import AIAgent + + agent = AIAgent( + api_key="test-key", + base_url="https://openrouter.ai/api/v1", + model="test/model", + quiet_mode=True, + session_db=db, + session_id=session_id, + skip_context_files=True, + skip_memory=True, + ) + agent.compression_in_place = True + agent.context_compressor.protect_first_n = 0 + agent.context_compressor.protect_last_n = 2 + agent.context_compressor.tail_token_budget = 80 + agent._todo_store.write( + [{"id": "inspect", "content": "Inspect artifacts", "status": "pending"}] + ) + return agent + + @pytest.fixture() def compressor() -> ContextCompressor: with patch( @@ -175,6 +217,121 @@ def test_zero_user_provenance_survives_iterative_compaction(compressor): assert second_handoffs[0][COMPRESSED_SUMMARY_HAS_USER_TURN_KEY] is False +def test_compress_context_todo_snapshot_stays_synthetic_across_two_boundaries( + tmp_path, monkeypatch +): + hermes_home = tmp_path / "hermes-home" + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + db = SessionDB(db_path=tmp_path / "state.db") + session_id = "zero-user-todo-lifecycle" + db.create_session(session_id, source="cron", model="test/model") + + first_agent = _lifecycle_agent(db, session_id) + with patch( + "agent.context_compressor.call_llm", + return_value=_response(_valid_zero_user_summary("First boundary")), + ): + first, _ = compress_context( + first_agent, + _assistant_turns(0, 24), + "system", + approx_tokens=90_000, + force=True, + ) + + first_handoff = next( + message + for message in first + if message.get(COMPRESSED_SUMMARY_METADATA_KEY) + ) + assert first_handoff[COMPRESSED_SUMMARY_HAS_USER_TURN_KEY] is False + assert "First boundary" in first_handoff["content"] + assert any( + message.get("role") == "user" + and str(message.get("content") or "").startswith(TODO_INJECTION_HEADER) + for message in first + ) + projected = db.get_messages_as_conversation(session_id) + assert projected + assert all( + COMPRESSED_SUMMARY_METADATA_KEY not in message + and COMPRESSED_SUMMARY_HAS_USER_TURN_KEY not in message + for message in projected + ) + + second_agent = _lifecycle_agent(db, session_id) + with patch( + "agent.context_compressor.call_llm", + return_value=_response(_valid_zero_user_summary("Second boundary")), + ): + second, _ = compress_context( + second_agent, + [*projected, *_assistant_turns(30, 24)], + "system", + approx_tokens=90_000, + force=True, + ) + + handoff = next( + message + for message in second + if message.get(COMPRESSED_SUMMARY_METADATA_KEY) + ) + assert handoff[COMPRESSED_SUMMARY_HAS_USER_TURN_KEY] is False + assert "Second boundary" in handoff["content"] + assert "User asked:" not in handoff["content"] + db.close() + + +def test_continuation_user_marker_is_not_reused_as_real_provenance(): + todo_snapshot = f"{TODO_INJECTION_HEADER}\n- [ ] inspect. Inspect artifacts (pending)" + compressed = [{"role": "assistant", "content": "Scheduled work completed."}] + + _ensure_compressed_has_user_turn( + [{"role": "user", "content": todo_snapshot}], + compressed, + ) + + assert compressed[-1] == { + "role": "user", + "content": COMPRESSION_CONTINUATION_USER_CONTENT, + } + projected = [{"role": row["role"], "content": row["content"]} for row in compressed] + assert ContextCompressor._transcript_has_real_user_turn(projected) is False + + +def test_continuation_markers_are_not_human_anchors(): + from agent.conversation_compression import _is_real_user_message + + legacy = ( + "Continue from the compressed conversation context above. " + "This marker exists because the compacted transcript contained " + "no preserved user turn." + ) + assert not _is_real_user_message( + {"role": "user", "content": COMPRESSION_CONTINUATION_USER_CONTENT} + ) + assert not _is_real_user_message({"role": "user", "content": legacy}) + + +def test_static_fallback_does_not_attribute_synthetic_rows_to_user(compressor): + todo_snapshot = f"{TODO_INJECTION_HEADER}\n- [ ] inspect. Inspect artifacts (pending)" + fallback = compressor._build_static_fallback_summary( + [ + {"role": "user", "content": todo_snapshot}, + { + "role": "user", + "content": COMPRESSION_CONTINUATION_USER_CONTENT, + }, + *_assistant_tool_turns(0, 2), + ] + ) + + assert _NO_USER_TASK_SENTINEL in fallback + assert "User asked:" not in fallback + assert "INTERNAL CONTEXT:" in fallback + + def test_zero_user_deterministic_fallback_uses_same_provenance(compressor): messages = _assistant_tool_turns(0, 12) diff --git a/tools/todo_tool.py b/tools/todo_tool.py index 3c657c034d6d..13b5fd4aad93 100644 --- a/tools/todo_tool.py +++ b/tools/todo_tool.py @@ -36,6 +36,11 @@ MAX_TODO_ITEMS = 256 # before it is parsed and re-injected (see AIAgent._hydrate_todo_store). MAX_TODO_RESULT_CHARS = 512_000 _TRUNCATION_MARKER = "… [truncated]" +# Persisted as ordinary message content. ContextCompressor uses this stable +# header to distinguish the synthetic post-compaction row from a real user. +TODO_INJECTION_HEADER = ( + "[Your active task list was preserved across context compression]" +) class TodoStore: @@ -135,7 +140,7 @@ class TodoStore: if not active_items: return None - lines = ["[Your active task list was preserved across context compression]"] + lines = [TODO_INJECTION_HEADER] for item in active_items: marker = markers.get(item["status"], "[?]") lines.append(f"- {marker} {item['id']}. {item['content']} ({item['status']})")