fix(gateway): persist compressed transcript before repointing /compress session

When /compress rotates the session, the handler repointed the live
session entry onto the new (empty) continuation session_id and _save()d
that BEFORE writing the compressed transcript — and rewrite_transcript
swallowed DB write failures at DEBUG. A transient write failure (SQLite
lock under concurrent writes, ENOSPC, disk/IO error) left the session
pointing at an empty id while the handler still reported a cheerful
'Compressed: N → M' success. The active conversation vanished from view.

- gateway/session.py: rewrite_transcript now returns bool (True on write
  success or no-DB, False on canonical write failure). /retry, /undo, and
  yuanbao recall ignore the result, so their behavior is unchanged.
- gateway/slash_commands.py: _handle_compress_command persists the
  compressed transcript FIRST and treats a write failure as fatal (raises
  into the outer handler's 'compress failed' banner). Only repoints +
  _save()s the session on a successful write. Widened beyond the original
  rotation case to also cover in-place compaction (#38763): a failed
  in-place write would otherwise leave the DB untouched while still
  reporting success.
- tests: regression tests for both the rotation and in-place write-failure
  paths — assert a failure banner, unchanged session_id, and no _save().

Co-authored-by: Hermes Agent <agent@nousresearch.com>

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.

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 "

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()