fix(tui): a gateway-owned session survives agent.close()

tests/tui_gateway/test_gateway_owned_session_reap.py

@@ -8,9 +8,10 @@ drops the ended-but-routed entry, recovers hours-old parent context, and
 loops.  The TUI is only a viewer of those sessions.
 """
 
+import threading
 from unittest.mock import MagicMock, patch
 
-from tui_gateway.server import _finalize_session, _is_gateway_owned_source
+from tui_gateway.server import _finalize_session, _is_gateway_owned_source, _teardown_session
 
 
 class TestIsGatewayOwnedSource:
@@ -49,6 +50,43 @@ def _make_session(session_id="sess_1"):
     }
 
 
+def _make_real_session(tmp_path, monkeypatch, *, source, session_id):
+    from hermes_state import SessionDB
+    from run_agent import AIAgent
+
+    hermes_home = tmp_path / "hermes_home"
+    hermes_home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+    db_path = hermes_home / "state.db"
+    db = SessionDB(db_path=db_path)
+    db.create_session(session_id, source=source)
+
+    # Exercise the production close path without constructing a model provider.
+    agent = AIAgent.__new__(AIAgent)
+    agent.session_id = session_id
+    agent._session_db = db
+    agent._owns_session_db = True
+    agent._end_session_on_close = True
+    agent._session_messages = []
+    agent._active_children_lock = threading.Lock()
+    agent._active_children = []
+
+    session = _make_session(session_id)
+    session["agent"] = agent
+    session["profile_home"] = str(hermes_home)
+    return db_path, session, agent
+
+
+def _read_real_row(db_path, session_id):
+    from hermes_state import SessionDB
+
+    db = SessionDB(db_path=db_path)
+    try:
+        return db.get_session(session_id)
+    finally:
+        db.close()
+
+
 class TestFinalizeSkipsGatewaySessions:
     @patch("tui_gateway.server._get_db")
     def test_gateway_session_not_ended(self, mock_get_db):
@@ -72,3 +110,18 @@ class TestFinalizeSkipsGatewaySessions:
         _finalize_session(_make_session(), end_reason="tui_close")
 
         db.end_session.assert_called_once_with("sess_1", "tui_close")
+
+
+class TestGatewayOwnedSessionTeardown:
+    def test_viewer_teardown_leaves_gateway_owned_row_open(self, tmp_path, monkeypatch):
+        session_id = "gateway-owned-session"
+        db_path, session, agent = _make_real_session(
+            tmp_path, monkeypatch, source="telegram", session_id=session_id
+        )
+
+        _teardown_session(session, end_reason="ws_orphan_reap")
+
+        row = _read_real_row(db_path, session_id)
+        assert row["ended_at"] is None
+        assert row["end_reason"] is None
+        assert agent._end_session_on_close is False

tui_gateway/session_lifecycle.py

@@ -271,6 +271,8 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No
                     # compression splits back to the reaped child, forever).
                     if _is_gateway_owned_source((db.get_session(session_id) or {}).get("source", "")):
                         _tui_owns_lifecycle = False
+                        if agent is not None:
+                            agent._end_session_on_close = _tui_owns_lifecycle
                     elif _tui_owns_lifecycle and not _desktop_automatic_cleanup:
                         # Automatic Desktop cleanup (ws_orphan_reap, idle_timeout, etc.) reclaims
                         # runtime but must not end the durable row — the conversation stays open