fix(compression): preserve flush baseline after abort

agent/conversation_compression.py

@@ -730,7 +730,11 @@ def replay_compression_warning(agent: Any) -> None:
             pass
 
 
-def conversation_history_after_compression(agent: Any, messages: list) -> Optional[list]:
+def conversation_history_after_compression(
+    agent: Any,
+    messages: list,
+    previous_history: Optional[list] = None,
+) -> Optional[list]:
     """Return the correct flush baseline after a compression boundary.
 
     Legacy compression rotates to a fresh child session. That child has not
@@ -747,7 +751,19 @@ def conversation_history_after_compression(agent: Any, messages: list) -> Option
 
     A shallow copy is intentional: it captures the current compacted dict
     identities as history while allowing later same-turn appends to remain new.
+
+    An aborted or no-op attempt after an earlier in-place compaction must retain
+    the pre-attempt baseline.  Treating all current messages as persisted would
+    drop any later, unflushed turns on restart; clearing the baseline would
+    append the already-persisted compacted rows a second time.
     """
+    if bool(getattr(agent, "_last_compression_attempt_recorded", False)):
+        attempt_in_place = getattr(agent, "_last_compression_attempt_in_place", None)
+        if attempt_in_place is True:
+            return list(messages)
+        if attempt_in_place is False:
+            return None
+        return previous_history
     if bool(getattr(agent, "_last_compaction_in_place", False)):
         return list(messages)
     return None
@@ -1008,6 +1024,13 @@ def compress_context(
     ):
         raise RuntimeError("a compression notification is already pending")
 
+    # ``conversation_history_after_compression()`` needs the latest attempt's
+    # outcome, while ``_last_compaction_in_place`` remains the run-level signal
+    # read by gateway callers. ``None`` means this attempt aborted or made no
+    # boundary, so the previous flush baseline remains authoritative.
+    agent._last_compression_attempt_recorded = True
+    agent._last_compression_attempt_in_place = None
+
     _attempt_started_at = time.monotonic()
     _attempt_id = uuid.uuid4().hex
     _trigger_source = "manual" if force else "auto"
@@ -1823,6 +1846,7 @@ def compress_context(
         # via a rotation-independent flag. The gateway uses this — NOT an
         # id-change diff — to re-baseline transcript handling (history_offset=0 +
         # rewrite on the same id) when compaction happened in place. See #38763.
+        agent._last_compression_attempt_in_place = compacted_in_place
         agent._last_compaction_in_place = compacted_in_place
 
         # Keep the post-compression rough estimate for diagnostics, but do not

agent/conversation_loop.py

@@ -710,6 +710,13 @@ def run_conversation(
         except Exception:
             pass
 
+    # The gateway caches agents across user turns.  Compression state is
+    # per-turn: carrying a prior in-place boundary forward would make a later
+    # uncompressed result look like a compacted transcript to gateway writers.
+    agent._last_compaction_in_place = False
+    agent._last_compression_attempt_recorded = False
+    agent._last_compression_attempt_in_place = None
+
     # ── Per-turn setup (the prologue) ──
     # All once-per-turn setup — stdio guarding, retry-counter resets, user
     # message sanitization, todo/nudge hydration, system-prompt restore-or-
@@ -1340,7 +1347,7 @@ def run_conversation(
             # and preflight compaction sites; see
             # conversation_history_after_compression().
             conversation_history = conversation_history_after_compression(
-                agent, messages
+                agent, messages, conversation_history
             )
             api_call_count -= 1
             agent._api_call_count = api_call_count
@@ -3527,7 +3534,7 @@ def run_conversation(
                             task_id=effective_task_id,
                         )
                         conversation_history = conversation_history_after_compression(
-                            agent, messages
+                            agent, messages, conversation_history
                         )
                         if len(messages) < original_len or old_ctx > _reduced_ctx:
                             agent._buffer_status(
@@ -3782,7 +3789,7 @@ def run_conversation(
                         task_id=effective_task_id,
                     )
                     conversation_history = conversation_history_after_compression(
-                        agent, messages
+                        agent, messages, conversation_history
                     )
 
                     # Re-estimate tokens after compression.  Same-message-count
@@ -4023,7 +4030,7 @@ def run_conversation(
                         task_id=effective_task_id,
                     )
                     conversation_history = conversation_history_after_compression(
-                        agent, messages
+                        agent, messages, conversation_history
                     )
 
                     # Re-estimate tokens after compression.  Same-message-count
@@ -5421,7 +5428,7 @@ def run_conversation(
                         task_id=effective_task_id,
                     )
                     conversation_history = conversation_history_after_compression(
-                        agent, messages
+                        agent, messages, conversation_history
                     )
                 
                 # Save session log incrementally (so progress is visible even if interrupted)

agent/turn_context.py

@@ -806,7 +806,7 @@ def build_turn_context(
                     _preflight_compression_blocked = True
                     break  # Cannot compress further: neither rows nor tokens moved
                 conversation_history = conversation_history_after_compression(
-                    agent, messages
+                    agent, messages, conversation_history
                 )
                 agent._empty_content_retries = 0
                 agent._thinking_prefill_retries = 0

tests/run_agent/test_compression_persistence.py

@@ -19,6 +19,7 @@ Bug scenario (pre-fix):
 import os
 import tempfile
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import patch
 
 
@@ -191,6 +192,125 @@ class TestFlushAfterCompression:
                 "final answer",
             ]
 
+    def test_abort_after_in_place_compaction_preserves_flush_baseline(self):
+        """An aborted retry must retain both compacted rows and new turns."""
+        from agent.conversation_compression import (
+            compress_context,
+            conversation_history_after_compression,
+        )
+        from hermes_state import SessionDB
+
+        class SuccessCompressor:
+            _last_compress_aborted = False
+            _last_summary_error = None
+            compression_count = 1
+            _last_compression_made_progress = True
+            _last_summary_fallback_used = False
+            last_compression_rough_tokens = 0
+            last_prompt_tokens = 0
+            last_completion_tokens = 0
+            awaiting_real_usage_after_compression = False
+
+            def compress(self, _messages, **_kwargs):
+                return [
+                    {"role": "user", "content": "[summary] earlier state"},
+                    {"role": "assistant", "content": "retained tail"},
+                ]
+
+        class AbortCompressor:
+            _last_compress_aborted = False
+            _last_summary_error = "simulated auxiliary timeout"
+            compression_count = 2
+            _last_compression_made_progress = False
+            _last_summary_fallback_used = False
+            last_compression_rough_tokens = 0
+            last_prompt_tokens = 0
+            last_completion_tokens = 0
+            awaiting_real_usage_after_compression = False
+
+            def compress(self, messages, **_kwargs):
+                self._last_compress_aborted = True
+                return messages
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db = SessionDB(db_path=Path(tmpdir) / "test.db")
+            agent = self._make_agent(db)
+            agent.compression_in_place = True
+            original = [
+                {"role": "user", "content": "old question"},
+                {"role": "assistant", "content": "old answer"},
+            ]
+            agent._flush_messages_to_session_db(original, [])
+
+            agent.context_compressor = SuccessCompressor()
+            compacted, _ = compress_context(
+                agent, original, "system", approx_tokens=100_000
+            )
+            history = conversation_history_after_compression(
+                agent, compacted, None
+            )
+
+            messages = compacted + [
+                {"role": "user", "content": "new request"},
+                {"role": "assistant", "content": "new answer"},
+            ]
+            agent.context_compressor = AbortCompressor()
+            returned, _ = compress_context(
+                agent, messages, "system", approx_tokens=100_000
+            )
+            history = conversation_history_after_compression(
+                agent, returned, history
+            )
+            agent._flush_messages_to_session_db(returned, history)
+
+            assert [message["content"] for message in db.get_messages_as_conversation(
+                agent.session_id
+            )] == [
+                "[summary] earlier state",
+                "retained tail",
+                "new request",
+                "new answer",
+            ]
+
+    def test_new_run_clears_stale_in_place_compaction_state(self, monkeypatch):
+        """A cached agent must not report a prior turn's compaction boundary."""
+        import agent.conversation_loop as loop
+
+        observed = {}
+
+        def fake_build_turn_context(agent, *_args, **_kwargs):
+            observed["in_place"] = agent._last_compaction_in_place
+            observed["attempt_recorded"] = agent._last_compression_attempt_recorded
+            observed["attempt_in_place"] = agent._last_compression_attempt_in_place
+            return SimpleNamespace(
+                user_message="hello",
+                original_user_message="hello",
+                messages=[],
+                conversation_history=[],
+                active_system_prompt="system",
+                effective_task_id="default",
+                turn_id="turn-1",
+                current_turn_user_idx=0,
+                should_review_memory=False,
+                plugin_user_context={},
+                ext_prefetch_cache=None,
+            )
+
+        agent = SimpleNamespace(
+            api_mode="codex_app_server",
+            _last_compaction_in_place=True,
+            _last_compression_attempt_recorded=True,
+            _last_compression_attempt_in_place=True,
+        )
+        agent._run_codex_app_server_turn = lambda **_kwargs: dict(observed)
+        monkeypatch.setattr(loop, "build_turn_context", fake_build_turn_context)
+
+        assert loop.run_conversation(agent, "hello") == {
+            "in_place": False,
+            "attempt_recorded": False,
+            "attempt_in_place": None,
+        }
+
     def test_rotation_child_session_flushes_full_compressed_transcript_with_markers(self):
         """Regression for #57491: live cached-agent markers must not block child flush."""
         from agent.conversation_compression import compress_context