From e81be5b66a43bcf3fa65b99240d3d240ed552fea Mon Sep 17 00:00:00 2001 From: ppazosp Date: Fri, 25 Sep 2026 13:51:10 +0200 Subject: [PATCH] fix(compression): preserve the active request after SQLite reload (cherry picked from commit 6ee1d087caf3f158d7379c8e4ca09a67c9c0e684) --- agent/context_compressor.py | 26 ++++++++++++- agent/conversation_compression.py | 4 +- contributors/emails/pablopazosp3@gmail.com | 1 + tests/agent/test_split_turn_compaction.py | 44 ++++++++++++++++++++++ 4 files changed, 71 insertions(+), 4 deletions(-) create mode 100644 contributors/emails/pablopazosp3@gmail.com diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 8ff05ebde8..8968b7f08d 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -4732,6 +4732,28 @@ Write only the summary body. Do not include any preamble or prefix.""" return max(pair_end, head_end + 1) return adjusted + @classmethod + def _has_merged_inflight_replay(cls, message: Any) -> bool: + """Recognize the active request on a handoff, including after DB reload. + + SessionDB and cold-history restore preserve content but not private + in-memory flags. The explicit replay after the summary end marker is + authoritative; a request quoted inside the historical summary is not. + """ + if not isinstance(message, dict): + return False + if message.get(_INFLIGHT_REPLAY_MERGED_KEY): + return True + if not cls._is_context_summary_message(message): + return False + text = _content_text_for_contains(message.get("content")) + _, boundary, remainder = text.partition(_SUMMARY_END_MARKER) + return bool( + boundary + and remainder.lstrip().startswith(_INFLIGHT_TASK_REPLAY_HEADER) + and remainder.lstrip()[len(_INFLIGHT_TASK_REPLAY_HEADER):].strip() + ) + @classmethod def _find_inflight_user_task( cls, messages: List[Dict[str, Any]] @@ -4775,7 +4797,7 @@ Write only the summary body. Do not include any preamble or prefix.""" if cls._is_actionable_user_turn(msg) and _is_real_user_message(msg): last_user_idx = i break - if isinstance(msg, dict) and msg.get(_INFLIGHT_REPLAY_MERGED_KEY): + if cls._has_merged_inflight_replay(msg): # A previous cycle merged the live request onto this summary # carrier; it is the only copy left, so it is still the task. last_user_idx = i @@ -4859,7 +4881,7 @@ Write only the summary body. Do not include any preamble or prefix.""" ) last_visible_role = _last_template_visible_role(compressed) - if inflight.get(_INFLIGHT_REPLAY_MERGED_KEY): + if self._has_merged_inflight_replay(inflight): # Never copy a summary carrier (metadata would mark the replay # synthetic): restate as a plain user row. replay = {"role": "user", "content": task_text} diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index ed7c9ce39b..0fe34c1ca7 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -2471,9 +2471,9 @@ def _ensure_compressed_has_user_turn(original_messages: list, compressed: list) # walk treats the whole compacted transcript as unpersisted and re-INSERTs it — the live set doubles on # every compaction (~58K → ~512K tokens in production). from agent.context_compressor import ( - _INFLIGHT_REPLAY_MERGED_KEY, COMPRESSION_CONTINUATION_USER_CONTENT, _fresh_compaction_message_copy, + ContextCompressor, COMPRESSION_CONTINUATION_USER_CONTENT, _fresh_compaction_message_copy, ) - if any(isinstance(message, dict) and message.get(_INFLIGHT_REPLAY_MERGED_KEY) for message in compressed): + if any(ContextCompressor._has_merged_inflight_replay(message) for message in compressed): # The in-flight request was restated onto the summary carrier (#100818); an anchor would duplicate it. return "already_present" # One reversed scan over BOTH kinds: scanning steer then user would let an older diff --git a/contributors/emails/pablopazosp3@gmail.com b/contributors/emails/pablopazosp3@gmail.com new file mode 100644 index 0000000000..583bfbf986 --- /dev/null +++ b/contributors/emails/pablopazosp3@gmail.com @@ -0,0 +1 @@ +ppazosp diff --git a/tests/agent/test_split_turn_compaction.py b/tests/agent/test_split_turn_compaction.py index c54e836f3f..fbc4579348 100644 --- a/tests/agent/test_split_turn_compaction.py +++ b/tests/agent/test_split_turn_compaction.py @@ -216,3 +216,47 @@ def test_a_tail_that_fits_the_budget_still_anchors_the_active_request() -> None: assert any( m.get("content") == _ACTIVE_REQUEST for m in messages[cut:] ) + + +@pytest.mark.parametrize("reload_from_db", [False, True], ids=["live", "restart"]) +@pytest.mark.parametrize("summary", [None, "Shard checks are continuing."], ids=["fallback", "summary"]) +def test_active_request_survives_repeated_compaction_and_restart( + tmp_path, reload_from_db: bool, summary: str | None, +) -> None: + from agent.context_compressor import _INFLIGHT_TASK_REPLAY_HEADER, _SUMMARY_END_MARKER + from agent.conversation_compression import _ensure_compressed_has_user_turn + from hermes_state import SessionDB + + db_path = tmp_path / "state.db" + db = SessionDB(db_path=db_path) + session_id = "active-turn-restart" + db.create_session(session_id, "test") + messages = _oversized_active_turn() + try: + for cycle in range(3): + if cycle: + for index in range(10 * cycle, 10 * (cycle + 1)): + messages.extend(_tool_group(index)) + original = messages + compressor = _make_compressor() + with patch.object(compressor, "_generate_summary", return_value=summary): + messages = compressor.compress(original, current_tokens=90_000, force=True) + _ensure_compressed_has_user_turn(original, messages) + assert len(messages) < len(original) + _assert_tool_pairs_are_complete(messages) + # Historical summaries may quote the request. Count only actionable + # text after their boundary, not those explicitly historical quotes. + user_content = "\n".join( + str(m.get("content")).rsplit(_SUMMARY_END_MARKER, 1)[-1] + for m in messages if m["role"] == "user" + ) + assert user_content.count(_ACTIVE_REQUEST) == 1 + assert user_content.count(_INFLIGHT_TASK_REPLAY_HEADER) == 1 + assert user_content.rfind(_ACTIVE_REQUEST) > user_content.rfind(_SUMMARY_END_MARKER) + db.archive_and_compact(session_id, messages) + if reload_from_db: + db.close() + db = SessionDB(db_path=db_path) + messages = db.get_messages_as_conversation(session_id) + finally: + db.close()