From d46f0fb2d5d0d68da746d7bb99849e7f130cd2a0 Mon Sep 17 00:00:00 2001 From: 3ASiC Date: Thu, 16 Jul 2026 21:04:17 +0800 Subject: [PATCH] fix(compression): notify context engine after commit --- agent/conversation_compression.py | 101 ++++++++++- cli.py | 19 ++- gateway/slash_commands.py | 23 ++- run_agent.py | 13 +- .../test_compression_boundary_hook.py | 157 ++++++++++++++++++ tests/test_cli_manual_compress.py | 63 ++++++- tests/test_tui_gateway_server.py | 83 +++++++-- tui_gateway/server.py | 67 +++++++- 8 files changed, 485 insertions(+), 41 deletions(-) diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 1135f23f682d..87b2797dae89 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -754,6 +754,73 @@ def _ensure_compressed_has_user_turn(original_messages: list, compressed: list) }) +_PENDING_CONTEXT_ENGINE_NOTIFICATION = ( + "_pending_context_engine_compression_notification" +) + + +def _notify_context_engine_compression_complete( + agent: Any, + *, + new_session_id: str, + old_session_id: str, +) -> bool: + """Notify the active context engine after a durable compression commit.""" + callback = getattr(agent.context_compressor, "on_session_start", None) + if not callable(callback): + return False + try: + callback( + new_session_id, + boundary_reason="compression", + old_session_id=old_session_id, + platform=getattr(agent, "platform", None) or "cli", + conversation_id=getattr(agent, "_gateway_session_key", None), + ) + except Exception: + # Context-engine hooks are observers. A callback failure must not undo + # history that the core or an outer host transaction already committed. + logger.debug( + "context engine on_session_start (compression) failed", + exc_info=True, + ) + return False + return True + + +def _queue_context_engine_compression_notification( + agent: Any, + *, + new_session_id: str, + old_session_id: str, +) -> None: + """Stage exactly one existing hook call for an outer host transaction.""" + if callable(getattr(agent, _PENDING_CONTEXT_ENGINE_NOTIFICATION, None)): + raise RuntimeError("a compression notification is already pending") + + def _notify() -> bool: + return _notify_context_engine_compression_complete( + agent, + new_session_id=new_session_id, + old_session_id=old_session_id, + ) + + setattr(agent, _PENDING_CONTEXT_ENGINE_NOTIFICATION, _notify) + + +def finalize_context_engine_compression_notification( + agent: Any, + *, + committed: bool, +) -> bool: + """Emit or discard a deferred notification; repeated calls are no-ops.""" + pending = getattr(agent, _PENDING_CONTEXT_ENGINE_NOTIFICATION, None) + setattr(agent, _PENDING_CONTEXT_ENGINE_NOTIFICATION, None) + if not committed or not callable(pending): + return False + return bool(pending()) + + def compress_context( agent: Any, messages: list, @@ -763,6 +830,7 @@ def compress_context( task_id: str = "default", focus_topic: Optional[str] = None, force: bool = False, + defer_context_engine_notification: bool = False, ) -> Tuple[list, str]: """Compress conversation context and split the session in SQLite. @@ -780,6 +848,8 @@ def compress_context( by the manual ``/compress`` slash command so users can retry immediately after an auto-compress abort. Auto-compress callers use the default ``False``. + defer_context_engine_notification: Delay the existing context-engine + hook until a manual host commits its outer history transaction. Returns: ``(compressed_messages, new_system_prompt)`` tuple. When @@ -788,6 +858,12 @@ def compress_context( prompt — the session is NOT rotated. Callers should detect the no-op via ``len(returned) == len(input)`` and stop the retry loop. """ + if ( + defer_context_engine_notification + and callable(getattr(agent, _PENDING_CONTEXT_ENGINE_NOTIFICATION, None)) + ): + raise RuntimeError("a compression notification is already pending") + # Codex app-server sessions: the codex agent owns the real thread context; # Hermes' summarizer would only rewrite a local mirror without shrinking # the actual thread (#36801). Route compaction to the app server's own @@ -1260,6 +1336,7 @@ def compress_context( new_system_prompt = agent._build_system_prompt(system_message) agent._cached_system_prompt = new_system_prompt + _session_commit_succeeded = False if agent._session_db: try: # Trigger memory extraction on the current session before the @@ -1440,6 +1517,7 @@ def compress_context( for message in compressed if isinstance(message, dict) } + _session_commit_succeeded = True except Exception as e: # If the rotation rolled back to the parent (orphan-avoidance # above), agent.session_id is the still-indexed parent and @@ -1460,6 +1538,9 @@ def compress_context( # id on rotation, the (unchanged) current id in-place. _old_sid = locals().get("old_session_id") _is_boundary = bool(_old_sid) or in_place + _context_engine_boundary_committed = _session_commit_succeeded and ( + bool(_old_sid) or compacted_in_place + ) _boundary_parent = _old_sid or agent.session_id or "" # Notify the context engine that a compaction boundary occurred. Plugin @@ -1468,17 +1549,19 @@ def compress_context( # re-initializing fresh. See hermes-lcm#68. Built-in ContextCompressor # ignores kwargs. Fires in BOTH modes: rotation passes old→new ids; in-place # passes the SAME id (the boundary is real even though the id didn't move). - try: - if _is_boundary and hasattr(agent.context_compressor, "on_session_start"): - agent.context_compressor.on_session_start( - agent.session_id or "", - boundary_reason="compression", + if _context_engine_boundary_committed: + if defer_context_engine_notification: + _queue_context_engine_compression_notification( + agent, + new_session_id=agent.session_id or "", + old_session_id=_boundary_parent, + ) + else: + _notify_context_engine_compression_complete( + agent, + new_session_id=agent.session_id or "", old_session_id=_boundary_parent, - platform=getattr(agent, "platform", None) or "cli", - conversation_id=getattr(agent, "_gateway_session_key", None), ) - except Exception as _ce_err: - logger.debug("context engine on_session_start (compression): %s", _ce_err) # Notify memory providers of the compaction boundary so provider-cached # per-session state (Hindsight's _document_id, accumulated turn buffers, diff --git a/cli.py b/cli.py index 8850a422831e..b9e6ddbb7141 100644 --- a/cli.py +++ b/cli.py @@ -9928,6 +9928,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): split_history_for_partial_compress, summarize_compress_preview, ) + from agent.conversation_compression import ( + finalize_context_engine_compression_notification, + ) # Args after the command word (e.g. "/compress here 3" -> "here 3"). raw_args = "" @@ -10027,14 +10030,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): approx_tokens=approx_tokens, focus_topic=focus_topic or None, force=True, + defer_context_engine_notification=True, ) - # Re-append the verbatim tail after the compressed head. - # The split guarantees `tail` begins on a user turn, so the - # compressed-head -> tail boundary is normally valid - # (the head's compressed output ends on assistant/tool). - # rejoin_compressed_head_and_tail() additionally guards the - # seam against any illegal user->user / assistant->assistant - # adjacency, defending provider role-alternation rules. if partial and tail: compressed = rejoin_compressed_head_and_tail(compressed, tail) self.conversation_history = compressed @@ -10054,6 +10051,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): # compressed handoff for the child session. Persist it from # offset 0 so resume can recover the continuation after exit. self.agent._flush_messages_to_session_db(self.conversation_history, None) + finalize_context_engine_compression_notification( + self.agent, + committed=True, + ) new_tokens = estimate_request_tokens_rough( self.conversation_history, system_prompt=_sys_prompt, @@ -10078,6 +10079,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): print(f" {summary['note']}") except Exception as e: + finalize_context_engine_compression_notification( + self.agent, + committed=False, + ) print(f" ❌ Compression failed: {e}") diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index d332516137a7..e30bac27498a 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -3423,6 +3423,9 @@ class GatewaySlashCommandsMixin: split_history_for_partial_compress, summarize_compress_preview, ) + from agent.conversation_compression import ( + finalize_context_engine_compression_notification, + ) _raw_args = (event.get_command_args() or "").strip() # Strip --preview/--dry-run/--aggressive before positional parsing # so the flags coexist with 'here [N]' / focus-topic forms. @@ -3551,11 +3554,15 @@ class GatewaySlashCommandsMixin: loop = asyncio.get_running_loop() compressed, _ = await loop.run_in_executor( None, - lambda: tmp_agent._compress_context(head, "", approx_tokens=approx_tokens, focus_topic=focus_topic, force=True) + lambda: tmp_agent._compress_context( + head, + "", + approx_tokens=approx_tokens, + focus_topic=focus_topic, + force=True, + defer_context_engine_notification=True, + ) ) - - # Re-append the verbatim tail after the compressed head, - # guarding the seam against illegal role adjacency. if partial and tail: compressed = rejoin_compressed_head_and_tail(compressed, tail) @@ -3625,6 +3632,10 @@ class GatewaySlashCommandsMixin: await self.async_session_store.update_session( session_entry.session_key, last_prompt_tokens=0 ) + finalize_context_engine_compression_notification( + tmp_agent, + committed=True, + ) new_tokens = estimate_request_tokens_rough( compressed, system_prompt=_sys_prompt, tools=_tools ) @@ -3655,6 +3666,10 @@ class GatewaySlashCommandsMixin: _aux_fail_model = getattr(compressor, "_last_aux_model_failure_model", None) _aux_fail_err = getattr(compressor, "_last_aux_model_failure_error", None) finally: + finalize_context_engine_compression_notification( + tmp_agent, + committed=False, + ) # Evict cached agent so next turn rebuilds system prompt # from current files (SOUL.md, memory, etc.). self._evict_cached_agent(session_key) diff --git a/run_agent.py b/run_agent.py index b92ea16d0a55..7a37b9f3554b 100644 --- a/run_agent.py +++ b/run_agent.py @@ -6162,7 +6162,17 @@ class AIAgent: """ return self.api_mode != "codex_responses" - def _compress_context(self, messages: list, system_message: str, *, approx_tokens: int = None, task_id: str = "default", focus_topic: str = None, force: bool = False) -> tuple: + def _compress_context( + self, + messages: list, + system_message: str, + *, + approx_tokens: int = None, + task_id: str = "default", + focus_topic: str = None, + force: bool = False, + defer_context_engine_notification: bool = False, + ) -> tuple: """Forwarder — see ``agent.conversation_compression.compress_context``. ``force=True`` is passed by the manual ``/compress`` slash command @@ -6175,6 +6185,7 @@ class AIAgent: self, messages, system_message, approx_tokens=approx_tokens, task_id=task_id, focus_topic=focus_topic, force=force, + defer_context_engine_notification=defer_context_engine_notification, ) def _set_tool_guardrail_halt(self, decision: ToolGuardrailDecision) -> None: diff --git a/tests/run_agent/test_compression_boundary_hook.py b/tests/run_agent/test_compression_boundary_hook.py index db9bb7f49620..36e5b97af613 100644 --- a/tests/run_agent/test_compression_boundary_hook.py +++ b/tests/run_agent/test_compression_boundary_hook.py @@ -16,7 +16,11 @@ import tempfile from pathlib import Path from unittest.mock import MagicMock, patch +import pytest +from agent.conversation_compression import ( + finalize_context_engine_compression_notification, +) class TestCompressionBoundaryHook: def _make_agent(self, session_db): @@ -92,6 +96,159 @@ class TestCompressionBoundaryHook: f"Expected new session_id as first positional arg, got {call!r}" assert call.kwargs.get("old_session_id") == original_sid, \ f"Expected old_session_id={original_sid!r}, got {call.kwargs!r}" + assert len(comp_calls) == 1 + + def test_automatic_notification_follows_core_persistence(self): + from hermes_state import SessionDB + + events = [] + with tempfile.TemporaryDirectory() as tmpdir: + db = SessionDB(db_path=Path(tmpdir) / "test.db") + agent = self._make_agent(db) + compressor = MagicMock() + compressor.compress.return_value = [ + {"role": "user", "content": "summary"} + ] + compressor.compression_count = 1 + compressor.last_prompt_tokens = 0 + compressor.last_completion_tokens = 0 + compressor._last_summary_error = None + compressor._last_compress_aborted = False + compressor.on_session_start.side_effect = ( + lambda *_args, **kwargs: events.append( + kwargs.get("boundary_reason") + ) + ) + agent.context_compressor = compressor + original_update = db.update_system_prompt + + def _record_update(*args, **kwargs): + result = original_update(*args, **kwargs) + events.append("persist") + return result + + with patch.object(db, "update_system_prompt", side_effect=_record_update): + agent._compress_context( + [{"role": "user", "content": "request"}], + "sys", + approx_tokens=100, + ) + + assert events == ["persist", "compression"] + + def test_failure_before_persistence_does_not_notify(self): + from hermes_state import SessionDB + + with tempfile.TemporaryDirectory() as tmpdir: + db = SessionDB(db_path=Path(tmpdir) / "test.db") + agent = self._make_agent(db) + compressor = MagicMock() + compressor.compress.side_effect = RuntimeError("synthetic compression failure") + agent.context_compressor = compressor + + with pytest.raises(RuntimeError, match="synthetic compression failure"): + agent._compress_context( + [{"role": "user", "content": "request"}], + "sys", + approx_tokens=100, + ) + + compressor.on_session_start.assert_not_called() + + def test_failure_during_persistence_does_not_notify(self): + from hermes_state import SessionDB + + with tempfile.TemporaryDirectory() as tmpdir: + db = SessionDB(db_path=Path(tmpdir) / "test.db") + agent = self._make_agent(db) + compressor = MagicMock() + compressor.compress.return_value = [ + {"role": "user", "content": "summary"} + ] + compressor.compression_count = 1 + compressor.last_prompt_tokens = 0 + compressor.last_completion_tokens = 0 + compressor._last_summary_error = None + compressor._last_compress_aborted = False + agent.context_compressor = compressor + + with patch.object( + db, + "update_system_prompt", + side_effect=RuntimeError("synthetic commit failure"), + ): + agent._compress_context( + [{"role": "user", "content": "request"}], + "sys", + approx_tokens=100, + ) + + boundary_calls = [ + call + for call in compressor.on_session_start.call_args_list + if call.kwargs.get("boundary_reason") == "compression" + ] + assert boundary_calls == [] + + def test_no_progress_does_not_notify(self): + from hermes_state import SessionDB + + with tempfile.TemporaryDirectory() as tmpdir: + db = SessionDB(db_path=Path(tmpdir) / "test.db") + agent = self._make_agent(db) + compressor = MagicMock() + compressor.compress.side_effect = lambda messages, **_kwargs: messages + compressor._last_compress_aborted = False + agent.context_compressor = compressor + messages = [{"role": "user", "content": "request"}] + + returned, _ = agent._compress_context( + messages, + "sys", + approx_tokens=100, + ) + + assert returned is messages + compressor.on_session_start.assert_not_called() + + @pytest.mark.parametrize("committed", [True, False]) + def test_deferred_notification_finishes_exactly_once(self, committed): + from hermes_state import SessionDB + + events = [] + with tempfile.TemporaryDirectory() as tmpdir: + db = SessionDB(db_path=Path(tmpdir) / "test.db") + agent = self._make_agent(db) + compressor = MagicMock() + compressor.compress.return_value = [ + {"role": "user", "content": "summary"} + ] + compressor.compression_count = 1 + compressor.last_prompt_tokens = 0 + compressor.last_completion_tokens = 0 + compressor._last_summary_error = None + compressor._last_compress_aborted = False + compressor.on_session_start.side_effect = ( + lambda *_args, **_kwargs: events.append("notify") + ) + agent.context_compressor = compressor + + agent._compress_context( + [{"role": "user", "content": "request"}], + "sys", + approx_tokens=100, + force=True, + defer_context_engine_notification=True, + ) + + assert events == [] + assert finalize_context_engine_compression_notification( + agent, committed=committed + ) is committed + assert finalize_context_engine_compression_notification( + agent, committed=True + ) is False + assert events == (["notify"] if committed else []) def test_no_hook_when_no_session_db(self): """Without session_db, session_id does not rotate and the hook is not fired.""" diff --git a/tests/test_cli_manual_compress.py b/tests/test_cli_manual_compress.py index c12bf1a227e5..bcd34a2333fd 100644 --- a/tests/test_cli_manual_compress.py +++ b/tests/test_cli_manual_compress.py @@ -1,5 +1,8 @@ from contextlib import nullcontext +from agent.conversation_compression import ( + _queue_context_engine_compression_notification, +) from cli import HermesCLI @@ -9,8 +12,33 @@ class DummyAgent: self._cached_system_prompt = "FULL CACHED SYSTEM PROMPT SHOULD NOT BE NESTED" self.session_id = "new-session" self.calls = [] + self.flush_calls = [] + self.flush_error = None + self.host_events = [] + self.boundary_calls = [] + self.context_compressor = type("ContextEngineStub", (), {})() + self.context_compressor.on_session_start = self._record_boundary - def _compress_context(self, messages, system_message, *, approx_tokens=None, focus_topic=None, force=False): + def _record_boundary(self, session_id, **kwargs): + self.host_events.append("notify") + self.boundary_calls.append((session_id, kwargs)) + + def _flush_messages_to_session_db(self, messages, _session_id=None): + self.host_events.append("persist") + self.flush_calls.append((list(messages), _session_id)) + if self.flush_error is not None: + raise self.flush_error + + def _compress_context( + self, + messages, + system_message, + *, + approx_tokens=None, + focus_topic=None, + force=False, + defer_context_engine_notification=False, + ): self.calls.append( { "messages": messages, @@ -18,8 +46,17 @@ class DummyAgent: "approx_tokens": approx_tokens, "focus_topic": focus_topic, "force": force, + "defer_context_engine_notification": ( + defer_context_engine_notification + ), } ) + if defer_context_engine_notification: + _queue_context_engine_compression_notification( + self, + new_session_id=self.session_id, + old_session_id="old-session", + ) return ([{"role": "user", "content": "[CONTEXT SUMMARY]: compacted"}], "new system prompt") @@ -56,3 +93,27 @@ def test_manual_compress_does_not_pass_cached_system_prompt(monkeypatch): assert call["focus_topic"] == "database schema" assert cli.session_id == "new-session" assert cli._pending_title is None + assert len(cli.agent.flush_calls) == 1 + assert cli.agent.host_events == ["persist", "notify"] + assert len(cli.agent.boundary_calls) == 1 + + +def test_manual_compress_flush_failure_discards_notification(monkeypatch): + cli = HermesCLI.__new__(HermesCLI) + cli.conversation_history = [ + {"role": "user", "content": "one"}, + {"role": "assistant", "content": "two"}, + {"role": "user", "content": "three"}, + {"role": "assistant", "content": "four"}, + ] + cli.agent = DummyAgent() + cli.agent.flush_error = RuntimeError("synthetic child flush failure") + cli.session_id = "old-session" + cli._pending_title = "old title" + cli._busy_command = lambda _message: nullcontext() + + cli._manual_compress("/compress") + + assert len(cli.agent.flush_calls) == 1 + assert cli.agent.host_events == ["persist"] + assert cli.agent.boundary_calls == [] diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index ba314901e428..125d3eb584e1 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -5926,26 +5926,37 @@ def test_session_compress_reports_aborted_summary_without_success(monkeypatch): def test_session_compress_syncs_session_key_after_rotation(monkeypatch): - """When AIAgent._compress_context rotates session_id (compression split), - the gateway session_key must follow so subsequent approval routing, - DB title/history lookups, and slash worker resume target the new - continuation session — mirrors HermesCLI._manual_compress's - session_id sync (cli.py). - """ - agent = types.SimpleNamespace(session_id="rotated-id") + """LCM notification follows the TUI's final session-key transition.""" + from agent.conversation_compression import ( + _queue_context_engine_compression_notification, + ) + + events = [] + agent = types.SimpleNamespace( + session_id="rotated-id", + context_compressor=types.SimpleNamespace( + on_session_start=lambda *_args, **_kwargs: events.append("notify") + ), + ) server._sessions["sid"] = _session(agent=agent) server._sessions["sid"]["session_key"] = "old-key" server._sessions["sid"]["pending_title"] = "stale title" - monkeypatch.setattr( - server, - "_compress_session_history", - lambda session, focus_topic=None, **_kw: (2, {"total": 42}), - ) + def _compress(session, focus_topic=None, **_kw): + _queue_context_engine_compression_notification( + session["agent"], + new_session_id="rotated-id", + old_session_id="old-key", + ) + return 2, {"total": 42} + + monkeypatch.setattr(server, "_compress_session_history", _compress) monkeypatch.setattr(server, "_session_info", lambda _agent, *a: {"model": "x"}) restart_calls = [] monkeypatch.setattr( - server, "_restart_slash_worker", lambda sid, s: restart_calls.append(s) + server, + "_restart_slash_worker", + lambda sid, s: (restart_calls.append(s), events.append("sync")), ) try: @@ -5961,6 +5972,52 @@ def test_session_compress_syncs_session_key_after_rotation(monkeypatch): assert server._sessions["sid"]["session_key"] == "rotated-id" assert server._sessions["sid"]["pending_title"] is None assert len(restart_calls) == 1 + assert events == ["sync", "notify"] + finally: + server._sessions.pop("sid", None) + + +def test_session_compress_sync_failure_discards_lcm_notification(monkeypatch): + from agent.conversation_compression import ( + _queue_context_engine_compression_notification, + ) + + events = [] + agent = types.SimpleNamespace( + session_id="rotated-id", + context_compressor=types.SimpleNamespace( + on_session_start=lambda *_args, **_kwargs: events.append("notify") + ), + ) + server._sessions["sid"] = _session(agent=agent) + server._sessions["sid"]["session_key"] = "old-key" + + def _compress(session, focus_topic=None, **_kw): + _queue_context_engine_compression_notification( + session["agent"], + new_session_id="rotated-id", + old_session_id="old-key", + ) + return 2, {"total": 42} + + monkeypatch.setattr(server, "_compress_session_history", _compress) + monkeypatch.setattr( + server, + "_session_info", + lambda *_args: (_ for _ in ()).throw(RuntimeError("finalization failed")), + ) + + try: + with patch("tui_gateway.server._emit"): + resp = server.handle_request( + { + "id": "1", + "method": "session.compress", + "params": {"session_id": "sid"}, + } + ) + assert resp["error"]["code"] == 5005 + assert events == [] finally: server._sessions.pop("sid", None) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 455b532c0b0d..1e420add1679 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -3553,6 +3553,9 @@ def _compress_session_history( before_messages: list | None = None, history_version: int | None = None, ) -> tuple[int, dict]: + from agent.conversation_compression import ( + finalize_context_engine_compression_notification, + ) from agent.model_metadata import estimate_request_tokens_rough agent = session["agent"] @@ -3581,16 +3584,28 @@ def _compress_session_history( # cached prompt (which already contains the agent identity block) # makes the rebuild append the identity a second time. Mirrors the # CLI's _manual_compress fix for issue #15281. - compressed, _ = agent._compress_context( - history, - None, - approx_tokens=approx_tokens, - focus_topic=focus_topic or None, - ) + try: + compressed, _ = agent._compress_context( + history, + None, + approx_tokens=approx_tokens, + focus_topic=focus_topic or None, + defer_context_engine_notification=True, + ) + except Exception: + finalize_context_engine_compression_notification( + agent, + committed=False, + ) + raise with session["history_lock"]: if int(session.get("history_version", 0)) != history_version: # External mutation during compaction — drop the compressed # result so we don't clobber concurrent edits. + finalize_context_engine_compression_notification( + agent, + committed=False, + ) usage = _get_usage(agent) return 0, usage session["history"] = compressed @@ -8992,6 +9007,10 @@ def _(rid, params: dict) -> dict: return _err( rid, 4009, "session busy — /interrupt the current turn before /compress" ) + from agent.conversation_compression import ( + finalize_context_engine_compression_notification, + ) + sid = params.get("session_id", "") focus_topic = str(params.get("focus_topic", "") or "").strip() try: @@ -9059,6 +9078,10 @@ def _(rid, params: dict) -> dict: ) info = _session_info(agent, session) _emit("session.info", sid, info) + finalize_context_engine_compression_notification( + agent, + committed=True, + ) return _ok( rid, { @@ -9080,6 +9103,10 @@ def _(rid, params: dict) -> dict: # no-op, or raised. _status_update(sid, "ready") except Exception as e: + finalize_context_engine_compression_notification( + session["agent"], + committed=False, + ) return _err(rid, 5005, str(e)) @@ -13950,6 +13977,10 @@ def _(rid, params: dict) -> dict: return _err( rid, 4009, "session busy — /interrupt the current turn before /compress" ) + from agent.conversation_compression import ( + finalize_context_engine_compression_notification, + ) + sid = params.get("session_id", "") if _session_uses_compute_host(session): command = f"/{name}" + (f" {arg}" if arg else "") @@ -14023,6 +14054,10 @@ def _(rid, params: dict) -> dict: compression_state=getattr(_agent, "context_compressor", None), ) _emit("session.info", sid, _session_info(session.get("agent"), session)) + finalize_context_engine_compression_notification( + _agent, + committed=True, + ) return _ok( rid, { @@ -14033,6 +14068,10 @@ def _(rid, params: dict) -> dict: }, ) except Exception as exc: + finalize_context_engine_compression_notification( + session["agent"], + committed=False, + ) return _err(rid, 5009, f"compress failed: {exc}") return _err(rid, 4018, f"not a quick/plugin/bundle/skill command: {name}") @@ -15058,6 +15097,9 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str: # while CLI and gateway both did. from agent.manual_compression_feedback import summarize_manual_compression from agent.model_metadata import estimate_request_tokens_rough + from agent.conversation_compression import ( + finalize_context_engine_compression_notification, + ) with session["history_lock"]: _before_messages = list(session.get("history", [])) @@ -15097,6 +15139,10 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str: _lines = [_fb["headline"], _fb["token_line"]] if _fb.get("note"): _lines.append(_fb["note"]) + finalize_context_engine_compression_notification( + agent, + committed=True, + ) return "\n".join(_lines) elif name == "fast" and agent: mode = arg.lower() @@ -15112,6 +15158,15 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str: process_registry.kill_all() except Exception as e: + if name == "compress" and agent: + from agent.conversation_compression import ( + finalize_context_engine_compression_notification, + ) + + finalize_context_engine_compression_notification( + agent, + committed=False, + ) return f"live session sync failed: {e}" return ""