From 9f6224033cde9c699bcf7e35a1dc607da7a86177 Mon Sep 17 00:00:00 2001 From: Black-Kylin <97433668+justinbao19@users.noreply.github.com> Date: Sat, 13 Jun 2026 04:53:27 -0700 Subject: [PATCH] fix(gateway): sync compression split on failed turns Sync gateway session pointers immediately after context compression rotates the agent session, even when the follow-up model call fails before a final response. Co-authored-from: https://github.com/NousResearch/hermes-agent/pull/25747 --- gateway/run.py | 56 ++--- .../test_compression_failure_session_sync.py | 196 ++++++++++++++++++ 2 files changed, 225 insertions(+), 27 deletions(-) create mode 100644 tests/gateway/test_compression_failure_session_sync.py diff --git a/gateway/run.py b/gateway/run.py index 7ee9582321ab..a00dcae1011a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -14565,6 +14565,32 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _context_length = getattr(_agent.context_compressor, "context_length", 0) or 0 _resolved_model = getattr(_agent, "model", None) if _agent else None + # Compression can rotate the agent to a new session before the + # follow-up model call succeeds. Sync that rotation before any + # early return so the next gateway turn reloads the compressed + # child transcript instead of the stale pre-compression parent. + agent = agent_holder[0] + _session_was_split = False + effective_session_id = getattr(agent, 'session_id', session_id) if agent else session_id + if agent and session_key and effective_session_id != session_id: + _session_was_split = True + logger.info( + "Session split detected: %s → %s (compression)", + session_id, effective_session_id, + ) + entry = self.session_store._entries.get(session_key) + if entry: + entry.session_id = effective_session_id + self.session_store._save() + self._sync_telegram_topic_binding( + source, entry, reason="agent-result-compression", + ) + + # When compression created a new session, the messages list was + # shortened. Using the original history offset would make gateway + # persistence slice away the compressed handoff/tail on failure. + _effective_history_offset = 0 if _session_was_split else len(agent_history) + if not final_response: error_msg = f"⚠️ {result['error']}" if result.get("error") else "" return { @@ -14579,7 +14605,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "error": result.get("error"), "compression_exhausted": result.get("compression_exhausted", False), "tools": tools_holder[0] or [], - "history_offset": len(agent_history), + "history_offset": _effective_history_offset, + "session_id": effective_session_id, "last_prompt_tokens": _last_prompt_toks, "input_tokens": _input_toks, "output_tokens": _output_toks, @@ -14625,23 +14652,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew unique_tags.insert(0, "[[audio_as_voice]]") final_response = final_response + "\n" + "\n".join(unique_tags) - # Sync session_id: the agent may have created a new session during - # mid-run context compression (_compress_context splits sessions). - # If so, update the session store entry so the NEXT message loads - # the compressed transcript, not the stale pre-compression one. - agent = agent_holder[0] - _session_was_split = False - if agent and session_key and hasattr(agent, 'session_id') and agent.session_id != session_id: - _session_was_split = True - logger.info( - "Session split detected: %s → %s (compression)", - session_id, agent.session_id, - ) - entry = self.session_store._entries.get(session_key) - if entry: - entry.session_id = agent.session_id - self.session_store._save() - + if _session_was_split: # If this is a Telegram DM and source.thread_id was lost during # the session split (synthetic / recovered event), restore it # from the binding so _thread_metadata_for_source produces the @@ -14673,15 +14684,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew exc_info=True, ) - effective_session_id = getattr(agent, 'session_id', session_id) if agent else session_id - - # When compression created a new session, the messages list was - # shortened. Using the original history offset would produce an - # empty new_messages slice, causing the gateway to write only a - # user/assistant pair — losing the compressed summary and tail. - # Reset to 0 so the gateway writes ALL compressed messages. - _effective_history_offset = 0 if _session_was_split else len(agent_history) - # Auto-generate session title after first exchange (non-blocking) if final_response and self._session_db: try: diff --git a/tests/gateway/test_compression_failure_session_sync.py b/tests/gateway/test_compression_failure_session_sync.py new file mode 100644 index 000000000000..fc4392b03831 --- /dev/null +++ b/tests/gateway/test_compression_failure_session_sync.py @@ -0,0 +1,196 @@ +import asyncio +import sys +import threading +import types +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import gateway.run as gateway_run +from gateway.config import Platform +from gateway.session import SessionSource + + +SESSION_KEY = "agent:main:telegram:dm:12345" + + +class _SaveTrackingSessionStore: + def __init__(self): + self.entry = SimpleNamespace(session_id="session-before-compression") + self._entries = {SESSION_KEY: self.entry} + self.save_calls = 0 + self.topic_sync_calls = [] + + def _save(self): + self.save_calls += 1 + + +class _CompressionThenFailureAgent: + def __init__(self, **kwargs): + self.session_id = kwargs["session_id"] + self.model = kwargs["model"] + self.tools = [] + self.context_compressor = SimpleNamespace( + last_prompt_tokens=4321, + context_length=200000, + ) + self.session_prompt_tokens = 4321 + self.session_completion_tokens = 0 + + def run_conversation(self, user_message, conversation_history=None, task_id=None): + self.session_id = "session-after-compression" + return { + "failed": True, + "error": ( + "APIConnectionError: Codex auxiliary Responses stream exceeded " + "120.0s total timeout" + ), + "messages": [ + { + "role": "user", + "content": ( + "[Context compressed: previous long transcript was " + "summarized before retry]" + ), + }, + {"role": "user", "content": user_message}, + ], + "api_calls": 1, + } + + def interrupt(self, *_args, **_kwargs): + pass + + +class _ImmediateStreamConsumer: + final_response_sent = False + + def __init__(self, *_args, **_kwargs): + pass + + async def run(self): + return None + + def finish(self): + pass + + +class _QuietAdapter: + SUPPORTS_MESSAGE_EDITING = True + _pending_messages = {} + + def get_pending_message(self, _session_key): + return None + + +def _install_fake_agent(monkeypatch): + fake_run_agent = types.ModuleType("run_agent") + fake_run_agent.AIAgent = _CompressionThenFailureAgent + monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) + + +def _make_runner(session_store): + runner = object.__new__(gateway_run.GatewayRunner) + runner.adapters = { + Platform.TELEGRAM: _QuietAdapter(), + } + runner._ephemeral_system_prompt = "" + runner._prefill_messages = [] + runner._reasoning_config = None + runner._provider_routing = {} + runner._fallback_model = None + runner._running_agents = {} + runner._pending_model_notes = {} + runner._pending_skills_reload_notes = {} + runner._session_db = None + runner._agent_cache = {} + runner._agent_cache_lock = threading.Lock() + runner._session_model_overrides = {} + runner._draining = False + runner.config = SimpleNamespace(streaming=None) + runner.hooks = SimpleNamespace(loaded_hooks=False, emit=AsyncMock()) + runner.session_store = session_store + runner._get_proxy_url = lambda: None + runner._resolve_session_agent_runtime = lambda **_kwargs: ( + "gpt-5.4", + { + "provider": "openai-codex", + "api_mode": "codex_responses", + "base_url": "https://chatgpt.com/backend-api/codex", + "api_key": "token", + }, + ) + runner._resolve_session_reasoning_config = lambda **_kwargs: None + runner._resolve_turn_agent_config = lambda message, model, runtime: { + "model": model, + "runtime": runtime, + } + runner._load_service_tier = lambda: None + runner._agent_config_signature = lambda *_args, **_kwargs: ("sig",) + runner._extract_cache_busting_config = lambda _config: () + runner._thread_metadata_for_source = lambda *_args, **_kwargs: None + runner._is_telegram_topic_lane = lambda _source: False + runner._sync_telegram_topic_binding = lambda source, entry, *, reason: session_store.topic_sync_calls.append( + (source, entry.session_id, reason) + ) + runner._release_running_agent_state = MagicMock() + return runner + + +def _source(): + return SessionSource( + platform=Platform.TELEGRAM, + chat_id="12345", + chat_type="dm", + user_id="user-1", + ) + + +def test_failed_turn_still_syncs_compression_session_split(monkeypatch): + """A post-compression API failure must not leave the session store on the + stale pre-compression transcript. + """ + _install_fake_agent(monkeypatch) + monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "off") + monkeypatch.setenv("HERMES_AGENT_TIMEOUT", "0") + monkeypatch.setattr(gateway_run, "_load_gateway_config", lambda: {}) + monkeypatch.setattr( + "gateway.stream_consumer.GatewayStreamConsumer", + _ImmediateStreamConsumer, + ) + + import hermes_cli.tools_config as tools_config + + monkeypatch.setattr( + tools_config, + "_get_platform_tools", + lambda *_args, **_kwargs: {"core"}, + ) + + session_store = _SaveTrackingSessionStore() + runner = _make_runner(session_store) + + result = asyncio.run( + asyncio.wait_for( + runner._run_agent( + message="continue", + context_prompt="", + history=[ + {"role": "user", "content": "old question"}, + {"role": "assistant", "content": "old answer"}, + ], + source=_source(), + session_id="session-before-compression", + session_key=SESSION_KEY, + ), + timeout=2, + ), + ) + + assert result["failed"] is True + assert result["session_id"] == "session-after-compression" + assert result["history_offset"] == 0 + assert session_store.entry.session_id == "session-after-compression" + assert session_store.save_calls == 1 + assert session_store.topic_sync_calls == [ + (_source(), "session-after-compression", "agent-result-compression") + ]