fix(tui-gateway): retry and log unflushed-message persist failures before teardown discards them

tests/tui_gateway/test_finalize_session_persist.py

@@ -9,6 +9,9 @@
   2. Force-quit mid-tool (double Ctrl+C) — session["history"] has previous turns
   3. Empty session — no-op, no crash
   4. Agent with _persist_session missing — graceful no-op
+  5. Persist *failure* — must be logged, not silently swallowed, and must
+     give _teardown_session a chance to retry before agent.close() destroys
+     the unflushed messages (agent.close() clears _session_messages).
 """
 
 import threading
@@ -148,6 +151,117 @@ def test_db_end_session_still_called(self, mock_get_db):
         mock_db.end_session.assert_called_once_with("sess_123", "test")
 
 
+class TestFinalizeSessionPersistFailureIsNotSilent:
+    """Regression: a persist failure in _finalize_session must never be a
+    silent no-op. Before this fix, ``except Exception: pass`` swallowed the
+    error, ``_finalized`` was already latched True (so the failure could
+    never be retried through _finalize_session again), and the *next* call
+    into _teardown_session unconditionally ran agent.close() — which clears
+    agent._session_messages (run_agent.py) — destroying the unflushed
+    messages with no trace anywhere.
+    """
+
+    def test_persist_failure_is_logged_and_flagged(self, caplog):
+        """A persist exception must be logged (not swallowed) and recorded on
+        the session so _teardown_session knows to retry before agent.close()
+        would otherwise discard the data."""
+        import logging
+        from tui_gateway.server import _finalize_session
+
+        agent = _make_agent()
+        agent._session_messages = [{"role": "user", "content": "unflushed"}]
+        agent._persist_session.side_effect = RuntimeError("db is locked")
+        session = _make_session(agent=agent, history=[{"role": "user", "content": "x"}])
+
+        with caplog.at_level(logging.ERROR, logger="tui_gateway.server"):
+            _finalize_session(session)
+
+        assert session.get("_persist_failed") is True
+        assert "persist" in caplog.text.lower(), caplog.text
+
+    def test_persist_success_does_not_flag_failure(self):
+        """Sanity check: the happy path must not set _persist_failed."""
+        from tui_gateway.server import _finalize_session
+
+        agent = _make_agent()
+        agent._session_messages = [{"role": "user", "content": "flushed fine"}]
+        session = _make_session(agent=agent, history=[{"role": "user", "content": "x"}])
+
+        _finalize_session(session)
+
+        assert not session.get("_persist_failed")
+
+
+class TestTeardownRetriesFailedPersistBeforeClose:
+    """Regression: _teardown_session must not let agent.close() destroy
+    messages that _finalize_session failed to persist without at least one
+    retry, and must fail loud if the retry also fails."""
+
+    def test_teardown_retries_and_succeeds_clears_flag(self, monkeypatch):
+        from tui_gateway.server import _teardown_session
+
+        agent = _make_agent()
+        agent.close = MagicMock()
+        agent._session_messages = [{"role": "user", "content": "unflushed"}]
+        session = _make_session(agent=agent)
+        session["_persist_failed"] = True  # as _finalize_session would leave it
+        session["_finalized"] = True  # skip the finalize body for this test
+        monkeypatch.setattr(
+            "tui_gateway.server._announce_session_reclaimed", lambda *a, **k: None
+        )
+
+        _teardown_session(session, end_reason="test")
+
+        # Retried once via _persist_session, succeeded, so agent.close() ran
+        # (which is what would normally wipe the message list) but the
+        # session is no longer flagged as having lost anything.
+        agent._persist_session.assert_called_once_with(
+            [{"role": "user", "content": "unflushed"}]
+        )
+        assert session.get("_persist_failed") is False
+        agent.close.assert_called_once()
+
+    def test_teardown_logs_loud_when_retry_also_fails(self, monkeypatch, caplog):
+        import logging
+        from tui_gateway.server import _teardown_session
+
+        agent = _make_agent()
+        agent.close = MagicMock()
+        agent._session_messages = [{"role": "user", "content": "still unflushed"}]
+        agent._persist_session.side_effect = RuntimeError("db is locked")
+        session = _make_session(agent=agent)
+        session["_persist_failed"] = True
+        session["_finalized"] = True
+        monkeypatch.setattr(
+            "tui_gateway.server._announce_session_reclaimed", lambda *a, **k: None
+        )
+
+        with caplog.at_level(logging.ERROR, logger="tui_gateway.server"):
+            _teardown_session(session, end_reason="test")
+
+        # agent.close() still runs (teardown must not hang forever), but the
+        # discard is now logged loudly instead of silent, and the failure
+        # flag is left set (nothing pretends the data survived).
+        agent.close.assert_called_once()
+        assert session.get("_persist_failed") is True
+        assert "retry" in caplog.text.lower(), caplog.text
+
+    def test_teardown_skips_retry_when_no_persist_failure_flagged(self):
+        """No _persist_failed flag → no retry attempted; existing behaviour
+        for the common (successful) case is unchanged."""
+        from tui_gateway.server import _teardown_session
+
+        agent = _make_agent()
+        agent.close = MagicMock()
+        session = _make_session(agent=agent)
+        session["_finalized"] = True
+
+        _teardown_session(session, end_reason="test")
+
+        agent._persist_session.assert_not_called()
+        agent.close.assert_called_once()
+
+
 class TestFinalizeSessionPersistE2E:
     """End-to-end: _finalize_session must actually land unflushed turns in
     state.db on disconnect/restart.

tui_gateway/server.py

@@ -694,7 +694,22 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No
             try:
                 agent._persist_session(snapshot)
             except Exception:
-                pass
+                # Do NOT let this fail silently: below, ``_teardown_session``
+                # unconditionally calls ``agent.close()``, which discards
+                # ``agent._session_messages`` (run_agent.py) whether or not
+                # they made it to disk. Flag the failure so the caller gets
+                # one more chance to persist before that happens, and log
+                # loudly so an operator can see the data-loss risk instead of
+                # it vanishing without a trace.
+                session["_persist_failed"] = True
+                logger.error(
+                    "_finalize_session: failed to persist %d unflushed "
+                    "message(s) for session %s; they will be lost if the "
+                    "retry in _teardown_session also fails",
+                    len(snapshot),
+                    session.get("session_key"),
+                    exc_info=True,
+                )
 
     # ── Plugin hook: on_session_end ────────────────────────────────────
     # Signals every plugin that the session is closing, with
@@ -855,6 +870,26 @@ def _teardown_session(session: dict | None, *, end_reason: str = "tui_close") ->
     try:
         agent = session.get("agent")
         if agent is not None and hasattr(agent, "close"):
+            if session.get("_persist_failed") and hasattr(agent, "_persist_session"):
+                # _finalize_session's persist attempt failed and flagged it.
+                # agent.close() below unconditionally clears
+                # agent._session_messages (run_agent.py), so this is the last
+                # chance to save them — retry once, and fail loud either way
+                # so the outcome is never silent.
+                snapshot = getattr(agent, "_session_messages", None)
+                try:
+                    if snapshot:
+                        agent._persist_session(snapshot)
+                    session["_persist_failed"] = False
+                except Exception:
+                    logger.error(
+                        "_teardown_session: retry persist failed for "
+                        "session %s; agent.close() will now discard %d "
+                        "unflushed message(s)",
+                        session.get("session_key"),
+                        len(snapshot or []),
+                        exc_info=True,
+                    )
             agent.close()
     except Exception:
         pass