fix(compression): preserve the active request after SQLite reload

(cherry picked from commit 6ee1d087caf3f158d7379c8e4ca09a67c9c0e684)

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}

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

contributors/emails/pablopazosp3@gmail.com

@@ -0,0 +1 @@
+ppazosp

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()