diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 04206a407211..cff098f67ff2 100644 --- a/agent/conversation_compression.py +++ b/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 diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index eceee9e0a7da..6d391e86136d 100644 --- a/agent/conversation_loop.py +++ b/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) diff --git a/agent/turn_context.py b/agent/turn_context.py index fdfee87daa52..287885f48667 100644 --- a/agent/turn_context.py +++ b/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 diff --git a/tests/run_agent/test_compression_persistence.py b/tests/run_agent/test_compression_persistence.py index ba05887cc44a..553da93fa59f 100644 --- a/tests/run_agent/test_compression_persistence.py +++ b/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