fix(compression): restore the transcript when a compression candidate is refused

Context engines may mutate the message list they receive, and the
context-engine plugin guide says that mutation only becomes visible if
the pass commits. Callers that pass their own commit fence (gateway
session hygiene) or set compression.context_timeout_seconds to 0 hand
the live list to the engine without the facade's deep snapshot.

compress_context restored the pre-compression snapshot on the
no-progress and superseded refusals, but the compressor-abort,
empty-transcript and would-grow refusals returned the live list as the
engine left it. An engine that cleared its input and returned [] made
compress_context return an empty transcript while warning that the
conversation continues unchanged.

Restore the snapshot in compress_context whenever _candidate_rejected
refuses a candidate or the commit refuses a grown one, so every refusal
returns the original transcript. The no-progress and superseded
branches no longer restore on their own because the caller now does it
for every refusal. Warnings, telemetry and the commit path are
unchanged.

Signed-off-by: mnajafian-nv <mnajafian@nvidia.com>
(cherry picked from commit d14fc7c96031d15f85acb00d433bcfdd39804554)

agent/conversation_compression.py

@@ -3545,8 +3545,6 @@ def _candidate_rejected(
     if compressed == messages_before_compression or (
         _strip_marker_for_comparison(compressed) == _strip_marker_for_comparison(messages_before_compression)
     ):
-        if messages != messages_before_compression:
-            messages[:] = copy.deepcopy(messages_before_compression)
         logger.info(
             "Compression made no progress (session=%s) — skipping boundary rewrite.", agent.session_id or "none"
         )
@@ -3584,7 +3582,6 @@ def _candidate_rejected(
             _working_gen,
             agent.session_id or "none",
         )
-        _restore_messages_snapshot(messages, messages_before_compression)
         agent._last_compaction_in_place = False
         _emit_aborted_attempt_telemetry(agent, attempt_started_at, "attempt_superseded")
         return True
@@ -4209,6 +4206,8 @@ def compress_context(
             agent, compressed, messages, messages_before_compression, attempt_generation=attempt.generation,
             attempt_started_at=attempt.started_at,
         ):
+            # Engines may mutate their input in place; a refused candidate must not leak that mutation.
+            _restore_messages_snapshot(messages, messages_before_compression)
             return messages, _existing_system_prompt(agent, system_message)
         if commit_fence is not None:
             _commit_fence_entered = commit_fence.begin_commit(_hard_cancel_event)
@@ -4264,6 +4263,7 @@ def compress_context(
             carried_messages=[reinserted_reply] if reinserted_reply is not None else None,
         )
         if commit.refused_prompt is not None:
+            _restore_messages_snapshot(messages, messages_before_compression)
             return messages, commit.refused_prompt
         compressed = commit.compressed
         split_status = commit.split_status

tests/agent/test_compression_refusal_restores_transcript.py

@@ -0,0 +1,102 @@
+"""Regression: a refused compression candidate hands back the original transcript.
+
+Context engines may mutate the message list they receive (the documented legacy
+contract), and that mutation must only become visible if the pass commits. Callers
+that pass their own commit fence (gateway session hygiene) or disable the
+compression timeout hand the live list to the engine without the facade's deep
+snapshot, so every refusal in ``compress_context`` has to put the pre-compression
+snapshot back before returning. The compressor-abort, empty-transcript and
+would-grow refusals returned the mutated list instead.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from agent.conversation_compression import CompressionCommitFence
+from hermes_state import SessionDB
+
+
+def _build_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,
+        )
+    # Skip the one-time aux-model feasibility probe (may hit the network).
+    agent._compression_feasibility_checked = True
+    return agent
+
+
+def _aborting_engine(compressor):
+    def _compress(msgs, **_kw):
+        msgs.clear()
+        compressor._last_compress_aborted = True
+        compressor._last_summary_error = "summary provider unavailable"
+        return msgs
+
+    return _compress
+
+
+def _empty_transcript_engine(_compressor):
+    def _compress(msgs, **_kw):
+        msgs.clear()
+        return []
+
+    return _compress
+
+
+def _growing_engine(_compressor):
+    def _compress(msgs, **_kw):
+        msgs.pop()
+        return list(msgs) + [{"role": "assistant", "content": "GROWN " * 20_000}]
+
+    return _compress
+
+
+@pytest.mark.parametrize(
+    "make_engine",
+    [_aborting_engine, _empty_transcript_engine, _growing_engine],
+    ids=["compressor_aborted", "empty_transcript", "would_grow"],
+)
+def test_refused_candidate_from_mutating_engine_returns_original_transcript(
+    tmp_path: Path, make_engine
+) -> None:
+    db = SessionDB(db_path=tmp_path / "state.db")
+    session_id = "REFUSED_CANDIDATE"
+    db.create_session(session_id, source="test")
+    agent = _build_agent(db, session_id)
+    compressor = agent.context_compressor
+
+    messages = [{"role": "user", "content": f"m{i} " + "x" * 200} for i in range(21)]
+    original = [dict(m) for m in messages]
+
+    with patch.object(type(compressor), "compress", side_effect=make_engine(compressor)), patch.object(
+        db, "archive_and_compact", wraps=db.archive_and_compact
+    ) as archive_and_compact:
+        # A caller-owned fence selects the direct path that gateway session hygiene uses.
+        returned, _sp = agent._compress_context(
+            messages, "sys", approx_tokens=120_000, commit_fence=CompressionCommitFence()
+        )
+
+    assert returned == original
+    assert messages == original
+    assert agent.session_id == session_id
+    assert db._conn.execute(
+        "SELECT COUNT(*) FROM sessions WHERE parent_session_id = ?", (session_id,)
+    ).fetchone()[0] == 0
+    assert db.get_session(session_id)["end_reason"] is None
+    archive_and_compact.assert_not_called()
+    assert db.get_compression_lock_holder(session_id) is None