diff --git a/gateway/session.py b/gateway/session.py index 110e7827a26..7f98eada2de 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1789,17 +1789,27 @@ class SessionStore: logger.debug("has_platform_message_id lookup failed", exc_info=True) return False - def rewrite_transcript(self, session_id: str, messages: List[Dict[str, Any]]) -> None: + def rewrite_transcript(self, session_id: str, messages: List[Dict[str, Any]]) -> bool: """Replace the entire transcript for a session with new messages. Used by /retry, /undo, and /compress to persist modified conversation history. state.db is the canonical store. + + Returns ``True`` when the write lands (or there is no DB to write to) + and ``False`` when the canonical write fails. Most callers can ignore + the result, but callers that would otherwise commit a destructive state + change on top of a failed write — e.g. /compress repointing the live + session onto a fresh session_id — must check it so they can surface an + error instead of silently dropping the conversation. """ - if self._db: - try: - self._db.replace_messages(session_id, messages) - except Exception as e: - logger.debug("Failed to rewrite transcript in DB: %s", e) + if not self._db: + return True + try: + self._db.replace_messages(session_id, messages) + return True + except Exception as e: + logger.debug("Failed to rewrite transcript in DB: %s", e) + return False def load_transcript(self, session_id: str) -> List[Dict[str, Any]]: """Load all messages from a session's transcript. diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index f678a6fc5b8..223144add84 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -2867,29 +2867,45 @@ class GatewaySlashCommandsMixin: new_session_id = tmp_agent.session_id rotated = new_session_id != session_entry.session_id _in_place = bool(getattr(tmp_agent, "_last_compaction_in_place", False)) - if rotated: - session_entry.session_id = new_session_id - self.session_store._save() - await asyncio.to_thread( - self._sync_telegram_topic_binding, - source, session_entry, reason="compress-command", - ) - # Rewrite the transcript when EITHER rotation produced a new id - # OR in-place compaction succeeded. The danger this guards - # against is the THIRD case: _compress_context could NOT rotate - # AND was not in-place (e.g. legacy mode but _session_db - # unavailable / the DB split raised) — there session_id is - # unchanged for a FAILURE reason, and rewrite_transcript() would - # DELETE the original messages and replace them with only the - # compressed summary (permanent data loss #44794, #39704). In - # in-place mode the unchanged id is SUCCESS, so the rewrite is - # exactly right (and is the durable write when the throwaway - # /compress agent has no _session_db of its own). + # Persist the compressed transcript BEFORE repointing the live + # session onto the new session_id. Order matters: if we + # repointed first and the canonical DB write then failed (lock + # contention under concurrent writes, ENOSPC, a disk/IO error), + # the session entry would already reference a brand-new, empty + # session_id while the handler still reported success — the + # user's active conversation would silently vanish from view. + # Writing first, and treating a write failure as fatal, keeps + # the old history reachable (on rotation the entry still points + # at it; in place the original transcript is untouched) and lets + # the outer handler surface a "compress failed" banner instead. + # + # The rewrite runs when EITHER rotation produced a new id OR + # in-place compaction succeeded. It is skipped in the THIRD + # case: _compress_context could NOT rotate AND was not in-place + # (e.g. legacy mode but _session_db unavailable / the DB split + # raised) — there session_id is unchanged for a FAILURE reason, + # and rewrite_transcript() would DELETE the original messages and + # replace them with only the compressed summary (permanent data + # loss #44794, #39704). In in-place mode the unchanged id is + # SUCCESS, so the rewrite is exactly right (and is the durable + # write when the throwaway /compress agent has no _session_db of + # its own). if rotated or _in_place: - self.session_store.rewrite_transcript( + if not self.session_store.rewrite_transcript( new_session_id, compressed - ) + ): + raise RuntimeError( + f"failed to persist compressed transcript for " + f"session {new_session_id}" + ) + if rotated: + session_entry.session_id = new_session_id + self.session_store._save() + await asyncio.to_thread( + self._sync_telegram_topic_binding, + source, session_entry, reason="compress-command", + ) else: logger.warning( "Manual /compress: session rotation did not occur " diff --git a/tests/gateway/test_compress_command.py b/tests/gateway/test_compress_command.py index 9c47e76db23..21029c7b775 100644 --- a/tests/gateway/test_compress_command.py +++ b/tests/gateway/test_compress_command.py @@ -305,3 +305,111 @@ async def test_compress_command_passes_session_db_and_persists_rotated_session() ) agent_instance.shutdown_memory_provider.assert_called_once() agent_instance.close.assert_called_once() + + +@pytest.mark.asyncio +async def test_compress_command_does_not_repoint_session_when_transcript_write_fails(): + """If the canonical transcript write fails after compression produces a new + continuation session_id, /compress must NOT repoint the live session onto + that empty session_id, and must report the failure instead of a success + banner. Otherwise a transient DB/IO error during compression would silently + drop the user's active conversation while still claiming success.""" + history = _make_history() + compressed = [ + history[0], + {"role": "assistant", "content": "summary"}, + history[-1], + ] + runner = _make_runner(history) + runner._session_db = object() + session_entry = runner.session_store.get_or_create_session.return_value + # Simulate the canonical DB write failing (lock contention, ENOSPC, ...). + runner.session_store.rewrite_transcript = MagicMock(return_value=False) + # Telegram topic re-binding must never run on the failure path. + runner._sync_telegram_topic_binding = MagicMock() + + agent_instance = MagicMock() + agent_instance.shutdown_memory_provider = MagicMock() + agent_instance.close = MagicMock() + agent_instance._cached_system_prompt = "" + agent_instance.tools = None + agent_instance.context_compressor.has_content_to_compress.return_value = True + agent_instance._last_compaction_in_place = False + agent_instance.session_id = "sess-1" + + def _compress(messages, *_args, **_kwargs): + # Compression rotated the session: the agent now holds a NEW session_id. + agent_instance.session_id = "sess-2" + return compressed, "" + + agent_instance._compress_context.side_effect = _compress + + def _estimate(messages, **_kwargs): + return 100 + + with ( + patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "***"}), + patch("gateway.run._resolve_gateway_model", return_value="test-model"), + patch("run_agent.AIAgent", return_value=agent_instance), + patch("agent.model_metadata.estimate_request_tokens_rough", side_effect=_estimate), + ): + result = await runner._handle_compress_command(_make_event()) + + # The user sees a failure banner, not a success banner. + assert "failed" in result.lower() + assert "Compressed:" not in result + # The live session was NOT repointed onto the empty new session_id, so the + # original conversation stays reachable. + assert session_entry.session_id == "sess-1" + runner.session_store._save.assert_not_called() + runner._sync_telegram_topic_binding.assert_not_called() + # Resources are still cleaned up even though the command errored. + agent_instance.shutdown_memory_provider.assert_called_once() + agent_instance.close.assert_called_once() + + +@pytest.mark.asyncio +async def test_compress_command_in_place_write_failure_reports_error(): + """In-place compaction (compression.in_place / #38763) does not rotate the + session_id, so a failed rewrite_transcript would leave the DB untouched + while the handler reported success. The write failure must surface as a + failure banner, not a false "Compressed" success.""" + history = _make_history() + compressed = [ + history[0], + {"role": "assistant", "content": "compacted summary"}, + history[-1], + ] + runner = _make_runner(history) + runner._session_db = object() + session_entry = runner.session_store.get_or_create_session.return_value + runner.session_store.rewrite_transcript = MagicMock(return_value=False) + + agent_instance = MagicMock() + agent_instance.shutdown_memory_provider = MagicMock() + agent_instance.close = MagicMock() + agent_instance._cached_system_prompt = "" + agent_instance.tools = None + agent_instance.context_compressor.has_content_to_compress.return_value = True + # In-place compaction: session_id is UNCHANGED but marked as a success. + agent_instance._last_compaction_in_place = True + agent_instance.session_id = "sess-1" + agent_instance._compress_context.return_value = (compressed, "") + + def _estimate(messages, **_kwargs): + return 100 + + with ( + patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "***"}), + patch("gateway.run._resolve_gateway_model", return_value="test-model"), + patch("run_agent.AIAgent", return_value=agent_instance), + patch("agent.model_metadata.estimate_request_tokens_rough", side_effect=_estimate), + ): + result = await runner._handle_compress_command(_make_event()) + + assert "failed" in result.lower() + assert "Compressed:" not in result + assert session_entry.session_id == "sess-1" + runner.session_store._save.assert_not_called() + agent_instance.shutdown_memory_provider.assert_called_once() + agent_instance.close.assert_called_once()