diff --git a/gateway/run.py b/gateway/run.py index 2297e264ff2f..eb209efbf99b 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -13846,22 +13846,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _hyg_in_place = bool( getattr(_hyg_agent, "_last_compaction_in_place", False) ) - if _hyg_rotated: - session_entry.session_id = _hyg_new_sid - # The held turn lease follows the - # rotation so an alias key resolving - # the fresh child still serializes - # against this turn (#64934). - self._rebind_turn_lease( - _quick_key, run_generation, _hyg_new_sid - ) - await self.async_session_store._save() - await asyncio.to_thread( - self._sync_telegram_topic_binding, - source, session_entry, - reason="hygiene-compression", - ) - # Only rewrite the transcript when rotation produced # a NEW session id. In-place compaction does NOT # need a rewrite: archive_and_compact() has already @@ -13882,10 +13866,47 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # rewrite_transcript() would DELETE the original # messages and replace them with only the compressed # summary (permanent data loss, #21301). + # + # Write-before-repoint (mirrors manual /compress): + # if we repointed session_entry onto the child SID + # and rewrite_transcript then failed (lock/ENOSPC), + # the live entry would already reference a brand-new + # empty session while the turn continues — the + # conversation silently vanishes. Persist the child + # transcript first; only then rebind the live entry. + if _hyg_rotated: + if not await self.async_session_store.rewrite_transcript( + _hyg_new_sid, _compressed + ): + logger.error( + "Session hygiene: failed to persist " + "compressed transcript for rotated " + "session %s → %s; keeping the live " + "entry on the original session so the " + "conversation is not dropped", + session_entry.session_id, + _hyg_new_sid, + ) + # Fail closed: treat like no rotation. + _hyg_rotated = False + _hyg_in_place = False + else: + session_entry.session_id = _hyg_new_sid + # The held turn lease follows the + # rotation so an alias key resolving + # the fresh child still serializes + # against this turn (#64934). + self._rebind_turn_lease( + _quick_key, run_generation, _hyg_new_sid + ) + await self.async_session_store._save() + await asyncio.to_thread( + self._sync_telegram_topic_binding, + source, session_entry, + reason="hygiene-compression", + ) + if _hyg_rotated: - await self.async_session_store.rewrite_transcript( - session_entry.session_id, _compressed - ) # Reset stored token count — transcript rewritten session_entry.last_prompt_tokens = 0 history = _compressed diff --git a/tests/gateway/test_session_hygiene.py b/tests/gateway/test_session_hygiene.py index 22847aa4f281..379235292e75 100644 --- a/tests/gateway/test_session_hygiene.py +++ b/tests/gateway/test_session_hygiene.py @@ -1636,3 +1636,122 @@ async def test_hygiene_trickle_stream_is_bounded_by_total_ceiling( ] assert len(timeout_warnings) == 1 assert runner._hygiene_compression_failure_cooldowns["sess-progress"] > time.time() + +@pytest.mark.asyncio +async def test_session_hygiene_does_not_repoint_when_rotated_transcript_write_fails( + monkeypatch, tmp_path +): + """Mirrors test_compress_command_does_not_repoint_session_when_transcript_write_fails. + + Hygiene auto-compress used to repoint session_entry.session_id onto the + rotated child SID *before* rewrite_transcript, and ignored a False return. + A failed write (lock/ENOSPC) left the live entry on an empty child while + the turn continued — conversation silently vanished. Write first; only + rebind after a durable persist, matching manual /compress. + """ + fake_dotenv = types.ModuleType("dotenv") + fake_dotenv.load_dotenv = lambda *args, **kwargs: None + monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv) + + class RotatingCompressAgent: + last_instance = None + + def __init__(self, **kwargs): + self.model = kwargs.get("model") + self.session_id = kwargs.get("session_id", "sess-1") + self._last_compaction_in_place = False + self._print_fn = None + self.context_compressor = None + self.shutdown_memory_provider = MagicMock() + self.close = MagicMock() + type(self).last_instance = self + + def _compress_context(self, messages, *_args, **_kwargs): + self.session_id = "sess-2" + return ( + [ + {"role": "user", "content": "kept"}, + {"role": "assistant", "content": "summary"}, + ], + None, + ) + + fake_run_agent = types.ModuleType("run_agent") + fake_run_agent.AIAgent = RotatingCompressAgent + monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) + + gateway_run = importlib.import_module("gateway.run") + GatewayRunner = gateway_run.GatewayRunner + + adapter = HygieneCaptureAdapter() + runner = object.__new__(GatewayRunner) + runner.config = GatewayConfig( + platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="fake-token")} + ) + runner.adapters = {Platform.TELEGRAM: adapter} + runner._voice_mode = {} + runner.hooks = SimpleNamespace(emit=AsyncMock(), loaded_hooks=False) + runner.session_store = MagicMock() + session_entry = SessionEntry( + session_key="agent:main:telegram:dm:user-1", + session_id="sess-1", + created_at=datetime.now(), + updated_at=datetime.now(), + platform=Platform.TELEGRAM, + chat_type="dm", + ) + runner.session_store.get_or_create_session.return_value = session_entry + runner.session_store.load_transcript.return_value = _make_history(6, content_size=400) + runner.session_store.has_any_sessions.return_value = True + # Canonical write fails — must not rebind live entry onto sess-2. + runner.session_store.rewrite_transcript = MagicMock(return_value=False) + runner.session_store.append_to_transcript = MagicMock() + runner.session_store._save = MagicMock() + runner._running_agents = {} + runner._pending_messages = {} + runner._pending_approvals = {} + runner._session_db = object() + runner._is_user_authorized = lambda _source: True + runner._set_session_env = lambda _context: None + runner._rebind_turn_lease = MagicMock() + runner._sync_telegram_topic_binding = MagicMock() + runner._run_agent = AsyncMock( + return_value={ + "final_response": "ok", + "messages": [], + "tools": [], + "history_offset": 0, + "last_prompt_tokens": 0, + } + ) + + monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) + monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "fake"}) + monkeypatch.setattr( + "agent.model_metadata.get_model_context_length", + lambda *_args, **_kwargs: 100, + ) + monkeypatch.setenv("TELEGRAM_HOME_CHANNEL", "795544298") + + event = MessageEvent( + text="hello", + source=SessionSource( + platform=Platform.TELEGRAM, + chat_id="user-1", + chat_type="dm", + user_id="12345", + ), + message_id="1", + ) + + result = await runner._handle_message(event) + + assert result == "ok" + # Rewrite was attempted against the *child* SID (write-first). + runner.session_store.rewrite_transcript.assert_called_once() + assert runner.session_store.rewrite_transcript.call_args[0][0] == "sess-2" + # Live entry must stay on the original session — conversation remains reachable. + assert session_entry.session_id == "sess-1" + runner._rebind_turn_lease.assert_not_called() + runner.session_store._save.assert_not_called() + runner._sync_telegram_topic_binding.assert_not_called()