fix(context-engine): route pre-API and idle compaction status through the quiet-engine resolver

Follow-up for the salvaged #35191: the mid-turn pre-API pressure emit in
conversation_loop.py and the idle-resume emit in turn_context.py were not
routed through automatic_compaction_status_message, so an engine with
emit_automatic_compaction_status=False still leaked those lines. Both now
resolve through the hook (phases "pre_api" and "idle") while keeping the
#69550 template constants as the default wording. Suppression also skips the
#69546 structured 'compacted' terminal edge for compress-phase events that
opened no visible phase; failure warnings (_emit_warning) remain never
suppressible, pinned by test.

agent/conversation_loop.py

@@ -1318,11 +1318,25 @@ def run_conversation(
                 compression_attempts,
                 max_compression_attempts,
             )
-            agent._emit_status(
-                PRE_API_COMPRESSION_STATUS_TEMPLATE.format(
+            _pre_api_status = automatic_compaction_status_message(
+                _compressor,
+                phase="pre_api",
+                default_message=PRE_API_COMPRESSION_STATUS_TEMPLATE.format(
                     tokens=request_pressure_tokens
-                )
+                ),
+                approx_tokens=request_pressure_tokens,
+                threshold_tokens=int(
+                    getattr(_compressor, "threshold_tokens", 0) or 0
+                ),
+                context_length=int(
+                    getattr(_compressor, "context_length", 0) or 0
+                ),
+                model=agent.model,
+                attempt=compression_attempts,
+                max_attempts=max_compression_attempts,
             )
+            if _pre_api_status:
+                agent._emit_status(_pre_api_status)
             _last_preflight_pressure = request_pressure_tokens
             messages, active_system_prompt = agent._compress_context(
                 messages,

agent/turn_context.py

@@ -664,11 +664,18 @@ def build_turn_context(
                     f"{_idle_floor:,}",
                     agent.session_id or "none",
                 )
-                agent._emit_status(
-                    IDLE_COMPACTION_STATUS_TEMPLATE.format(
+                _idle_status = automatic_compaction_status_message(
+                    _compressor,
+                    phase="idle",
+                    default_message=IDLE_COMPACTION_STATUS_TEMPLATE.format(
                         idle_seconds=int(_idle_gap), tokens=_idle_tokens
-                    )
+                    ),
+                    approx_tokens=_idle_tokens,
+                    idle_seconds=int(_idle_gap),
+                    model=agent.model,
                 )
+                if _idle_status:
+                    agent._emit_status(_idle_status)
                 _idle_input = messages
                 messages, active_system_prompt = agent._compress_context(
                     messages, system_message, approx_tokens=_idle_tokens,

tests/agent/test_idle_compaction_lock_and_guards.py

@@ -115,6 +115,53 @@ def test_idle_compaction_runs_through_guarded_path_and_releases_lock(
     assert ctx.messages[ctx.current_turn_user_idx].get("role") == "user"
 
 
+def test_idle_compaction_status_suppressed_when_engine_opts_out(
+    tmp_path: Path,
+) -> None:
+    """Quiet context engines silence the 💤 idle-resume status line too.
+
+    The idle emit routes through ``automatic_compaction_status_message``
+    (phase="idle") the same way the preflight and pre-API emits do — an
+    engine with ``emit_automatic_compaction_status = False`` gets a fully
+    silent idle compaction, while the compaction itself still runs.
+    """
+    db = SessionDB(db_path=tmp_path / "state.db")
+    sid = "IDLE_QUIET"
+    db.create_session(sid, source="cli")
+    agent = _prep_idle_agent(db, sid)
+    agent.context_compressor.emit_automatic_compaction_status = False
+    events = []
+    agent.status_callback = lambda ev, msg: events.append((ev, msg))
+
+    _run_prologue(agent, _history())
+
+    agent.context_compressor.compress.assert_called_once()
+    assert not any(
+        "Resumed after" in str(msg) for _ev, msg in events
+    ), f"idle status leaked despite quiet engine: {events}"
+
+
+def test_idle_compaction_status_emitted_by_default(tmp_path: Path) -> None:
+    """Control: the default engine keeps the 💤 idle-resume status line."""
+    db = SessionDB(db_path=tmp_path / "state.db")
+    sid = "IDLE_LOUD"
+    db.create_session(sid, source="cli")
+    agent = _prep_idle_agent(db, sid)
+    # MagicMock would auto-create the hook attributes as truthy mocks; pin
+    # the default-engine surface explicitly.
+    agent.context_compressor.emit_automatic_compaction_status = True
+    del agent.context_compressor.get_automatic_compaction_status_message
+    events = []
+    agent.status_callback = lambda ev, msg: events.append((ev, msg))
+
+    _run_prologue(agent, _history())
+
+    agent.context_compressor.compress.assert_called_once()
+    assert any(
+        ev == "lifecycle" and "Resumed after" in str(msg) for ev, msg in events
+    ), f"expected idle status line, got: {events}"
+
+
 def test_idle_compaction_defers_to_held_compression_lock(tmp_path: Path) -> None:
     """An idle-triggered compress racing another path must sit the round out.
 

tests/run_agent/test_413_compression.py

@@ -783,7 +783,13 @@ class TestPreflightCompression:
         agent.status_callback = lambda ev, msg: events.append((ev, msg))
         agent.context_compressor.emit_automatic_compaction_status = False
 
-        def _fake_compress(messages, current_tokens=None, focus_topic=None):
+        def _fake_compress(
+            messages,
+            current_tokens=None,
+            focus_topic=None,
+            force=False,
+            memory_context="",
+        ):
             events.append(("compress", "started"))
             return [{"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"}]
 
@@ -798,8 +804,16 @@ class TestPreflightCompression:
                 approx_tokens=1234,
             )
 
-        assert compressed == [{"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"}]
-        assert new_system_prompt == "new system prompt"
+        # The compressor returned only the user-role summary; the human-anchor
+        # repair restores the real user turn after it (same contract as
+        # test_compress_context_emits_lifecycle_status_before_work above).
+        assert compressed == [
+            {"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"},
+            {"role": "user", "content": "hello"},
+        ]
+        # Memory snapshot unchanged → the cached prompt is retained (same
+        # contract as test_compress_context_emits_lifecycle_status_before_work).
+        assert new_system_prompt == "You are helpful."
         assert events == [("compress", "started")]
 
     def test_compress_context_force_keeps_manual_status_when_engine_opts_out(self, agent):
@@ -831,6 +845,117 @@ class TestPreflightCompression:
         assert "Compacting context" in events[0][1]
         assert events[1] == ("compress", "started")
 
+    def test_compress_context_abort_warning_is_never_suppressed(self, agent):
+        """Failure warnings stay visible even when a quiet engine suppresses
+        routine automatic status — only ROUTINE lifecycle lines are quiet."""
+        agent.compression_enabled = False
+        agent.context_compressor.emit_automatic_compaction_status = False
+        events = []
+        agent.status_callback = lambda event, message: events.append((event, message))
+        messages = [{"role": "user", "content": "hello"}]
+
+        def _abort_compression(current_messages, **_kwargs):
+            agent.context_compressor._last_compress_aborted = True
+            agent.context_compressor._last_summary_error = "auxiliary model unavailable"
+            return current_messages
+
+        with patch.object(agent.context_compressor, "compress", side_effect=_abort_compression):
+            compressed, prompt = agent._compress_context(messages, "system prompt")
+
+        assert compressed is messages
+        assert prompt == "You are helpful."
+        # Routine lifecycle + structured compacted edges are suppressed (the
+        # quiet engine opened no visible phase), but the abort warning fires.
+        assert [event for event, _ in events] == ["warn"]
+        assert "Compression aborted" in events[0][1]
+
+    def test_pre_api_compression_status_suppressed_when_engine_opts_out(self, agent):
+        """The mid-turn pre-API pressure emit routes through the resolver too.
+
+        Regression guard for the #35191 review gap: with
+        ``emit_automatic_compaction_status = False`` the 📦 Pre-API line must
+        not leak even though compaction itself still runs.
+        """
+        agent.compression_enabled = True
+        agent.context_compressor.context_length = 200_000
+        agent.context_compressor.threshold_tokens = 130_000
+        agent.context_compressor.emit_automatic_compaction_status = False
+
+        history = [
+            {"role": "user", "content": "earlier question"},
+            {"role": "assistant", "content": "earlier answer"},
+        ]
+        ok_resp = _mock_response(content="After quiet pre-API", finish_reason="stop")
+        agent.client.chat.completions.create.side_effect = [ok_resp]
+        status_messages = []
+        agent.status_callback = lambda ev, msg: status_messages.append((ev, msg))
+
+        with (
+            # Keep the turn-prologue preflight quiet-by-size so only the
+            # in-loop pre-API pressure gate fires.
+            patch("agent.turn_context.estimate_request_tokens_rough", return_value=10_000),
+            patch("agent.conversation_loop.estimate_request_tokens_rough", return_value=144_669),
+            patch(
+                "agent.conversation_loop.estimate_messages_tokens_rough",
+                return_value=144_669,
+            ),
+            patch.object(
+                agent,
+                "_compress_context",
+                side_effect=lambda msgs, *a, **k: (msgs, agent._cached_system_prompt),
+            ) as mock_compress,
+            patch.object(agent, "_persist_session"),
+            patch.object(agent, "_save_trajectory"),
+            patch.object(agent, "_cleanup_task_resources"),
+        ):
+            result = agent.run_conversation("hello", conversation_history=history)
+
+        assert result["completed"] is True
+        assert mock_compress.call_count >= 1, "pre-API compression never ran"
+        assert not any(
+            "Pre-API compression" in msg for _ev, msg in status_messages
+        )
+
+    def test_pre_api_compression_status_emitted_by_default(self, agent):
+        """Control: the default (non-quiet) engine keeps the pre-API line."""
+        agent.compression_enabled = True
+        agent.context_compressor.context_length = 200_000
+        agent.context_compressor.threshold_tokens = 130_000
+
+        history = [
+            {"role": "user", "content": "earlier question"},
+            {"role": "assistant", "content": "earlier answer"},
+        ]
+        ok_resp = _mock_response(content="After pre-API", finish_reason="stop")
+        agent.client.chat.completions.create.side_effect = [ok_resp]
+        status_messages = []
+        agent.status_callback = lambda ev, msg: status_messages.append((ev, msg))
+
+        with (
+            patch("agent.turn_context.estimate_request_tokens_rough", return_value=10_000),
+            patch("agent.conversation_loop.estimate_request_tokens_rough", return_value=144_669),
+            patch(
+                "agent.conversation_loop.estimate_messages_tokens_rough",
+                return_value=144_669,
+            ),
+            patch.object(
+                agent,
+                "_compress_context",
+                side_effect=lambda msgs, *a, **k: (msgs, agent._cached_system_prompt),
+            ) as mock_compress,
+            patch.object(agent, "_persist_session"),
+            patch.object(agent, "_save_trajectory"),
+            patch.object(agent, "_cleanup_task_resources"),
+        ):
+            result = agent.run_conversation("hello", conversation_history=history)
+
+        assert result["completed"] is True
+        assert mock_compress.call_count >= 1
+        assert any(
+            ev == "lifecycle" and "Pre-API compression" in msg
+            for ev, msg in status_messages
+        )
+
     def test_preflight_compresses_oversized_history(self, agent):
         """When loaded history exceeds the model's context threshold, compress before API call."""
         agent.compression_enabled = True