TUI resume creating duplicate live sessions fix

NousResearch/hermes-agent · cluster 38315

Data as of 2026-10-23 11:54 UTC

Nothing open on GitHub. Fix merged in #39153. Check it covers your case.

open issues

open PRs

items grouped

Known?: Yes. 4 related items grouped here, 2 issues already closed.
Fix event: A fix was merged to main in PR #39153. Check it covers your case. No release data is tracked.
Tracker: The maintainers' tracker marks this partially resolved, deciding PR #39153. That is separate from the fix event.
Still open: Nothing open on GitHub.

Next step: read #39153, then check it covers your case.

Read #39153 Pick up a Lab task

These issues and pull requests concern a bug where repeatedly resuming in the TUI gateway can create duplicate live sessions. The work focuses on reusing existing live sessions on resume rather than creating new ones.

Open issues0

No open issues.

Open PRs0

No open pull requests.

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

#39153 fix(tui): reuse live session on resume (salvage #38330)

merged to main

files: tests/tui_gateway/test_protocol.py, tui_gateway/server.py, ui-tui/src/app/useSessionLifecycle.ts, ui-tui/src/gatewayTypes.ts

Show the diff

tests/tui_gateway/test_protocol.py

@@ -394,6 +394,225 @@ def get_messages_as_conversation(self, _sid, include_ancestors=False):
     ]
 
 
+def test_session_resume_reuses_existing_live_session(server, monkeypatch):
+    """Repeated resume must not allocate duplicate live agents."""
+
+    target = "20260409_010101_abc123"
+    created_sids: list[str] = []
+    closed_sids: list[str] = []
+    first_agent_started = threading.Event()
+    agent_can_finish = threading.Event()
+
+    class _DB:
+        def get_session(self, _sid):
+            return {"id": target}
+
+        def get_session_by_title(self, _title):
+            return None
+
+        def reopen_session(self, _sid):
+            return None
+
+        def get_messages_as_conversation(self, _sid, include_ancestors=False):
+            return [
+                {"role": "user", "content": "hello"},
+                {"role": "assistant", "content": "yo"},
+            ]
+
+    class _Worker:
+        def close(self):
+            pass
+
+    class _Agent:
+        def __init__(self, sid, session_id):
+            self.sid = sid
+            self.model = "test/model"
+            self.session_id = session_id
+
+        def close(self):
+            closed_sids.append(self.sid)
+
+    def make_agent(sid, key, session_id=None):
+        created_sids.append(sid)
+        first_agent_started.set()
+        assert agent_can_finish.wait(timeout=1)
+        return _Agent(sid, session_id or key)
+
+    monkeypatch.setattr(server, "_get_db", lambda: _DB())
+    monkeypatch.setattr(server, "_make_agent", make_agent)
+    monkeypatch.setattr(server, "_SlashWorker", lambda _key, _model: _Worker())
+    monkeypatch.setattr(
+        server,
+        "_start_notification_poller",
+        lambda _sid, _session: threading.Event(),
+    )
+    monkeypatch.setattr(server, "_notify_session_boundary", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(server, "_wire_callbacks", lambda _sid: None)
+    monkeypatch.setattr(server, "_emit", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        server,
+        "_session_info",
+        lambda _agent, _session=None: {"model": "test/model"},
+    )
+
+    fake_approval = types.SimpleNamespace(
+        load_permanent_allowlist=lambda: None,
+        register_gateway_notify=lambda *_args, **_kwargs: None,
+    )
+
+    with patch.dict(sys.modules, {"tools.approval": fake_approval}):
+        first_holder = {}
+
+        def resume_first():
+            first_holder["resp"] = server.handle_request(
+                {
+                    "id": "first",
+                    "method": "session.resume",
+                    "params": {"session_id": target, "cols": 100},
+                }
+            )
+
+        first_thread = threading.Thread(target=resume_first)
+        first_thread.start()
+        assert first_agent_started.wait(timeout=1)
+
+        second_holder = {}
+
+        def resume_second():
+            second_holder["resp"] = server.handle_request(
+                {
+                    "id": "second",
+                    "method": "session.resume",
+                    "params": {"session_id": target, "cols": 120},
+                }
+            )
+
+        second_thread = threading.Thread(target=resume_second)
+        second_thread.start()
+        agent_can_finish.set()
+
+        first_thread.join(timeout=1)
+        second_thread.join(timeout=1)
+        assert not first_thread.is_alive()
+        assert not second_thread.is_alive()
+        first = first_holder["resp"]
+        second = second_holder["resp"]
+
+    assert "error" not in first
+    assert "error" not in second
+    # Both resumes resolve to the SAME single live session — the core invariant.
+    assert second["result"]["session_id"] == first["result"]["session_id"]
+    assert len(server._sessions) == 1
+    assert [s.get("session_key") for s in server._sessions.values()].count(target) == 1
+    winner = first["result"]["session_id"]
+    # The agent build happens outside the resume lock, so a racing resume may
+    # build a redundant agent; double-checked locking keeps only one live
+    # session and closes any loser's agent (no worker/poller is wired for it).
+    assert winner in created_sids
+    survivors = [sid for sid in created_sids if sid not in closed_sids]
+    assert survivors == [winner]
+    assert all(sid == winner for sid in server._sessions)
+
+
+def test_session_resume_live_payload_uses_current_history_with_ancestors(server, monkeypatch):
+    """Live resume should not reuse a stale ancestor-inclusive snapshot."""
+
+    target = "20260409_010101_child"
+    ancestor_history = [{"role": "user", "content": "ancestor"}]
+    current_history = [
+        {"role": "user", "content": "current"},
+        {"role": "assistant", "content": "current reply"},
+    ]
+
+    class _DB:
+        def get_session(self, _sid):
+            return {"id": target}
+
+        def get_session_by_title(self, _title):
+            return None
+
+        def reopen_session(self, _sid):
+            return None
+
+        def get_messages_as_conversation(self, _sid, include_ancestors=False):
+            if include_ancestors:
+                return ancestor_history + current_history
+            return list(current_history)
+
+    class _Worker:
+        def close(self):
+            pass
+
+    monkeypatch.setattr(server, "_get_db", lambda: _DB())
+    monkeypatch.setattr(
+        server,
+        "_make_agent",
+        lambda _sid, key, session_id=None: types.SimpleNamespace(
+            model="test/model", session_id=session_id or key
+        ),
+    )
+    monkeypatch.setattr(server, "_SlashWorker", lambda _key, _model: _Worker())
+    monkeypatch.setattr(
+        server,
+        "_start_notification_poller",
+        lambda _sid, _session: threading.Event(),
+    )
+    monkeypatch.setattr(server, "_notify_session_boundary", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(server, "_wire_callbacks", lambda _sid: None)
+    monkeypatch.setattr(server, "_emit", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        server,
+        "_session_info",
+        lambda _agent, _session=None: {"model": "test/model"},
+    )
+
+    fake_approval = types.SimpleNamespace(
+        load_permanent_allowlist=lambda: None,
+        register_gateway_notify=lambda *_args, **_kwargs: None,
+    )
+
+    with patch.dict(sys.modules, {"tools.approval": fake_approval}):
+        first = server.handle_request(
+            {
+                "id": "first",
+                "method": "session.resume",
+                "params": {"session_id": target, "cols": 100},
+            }
+        )
+
+        assert "error" not in first
+        sid = first["result"]["session_id"]
+        assert first["result"]["messages"] == [
+            {"role": "user", "text": "ancestor"},
+            {"role": "user", "text": "current"},
+            {"role": "assistant", "text": "current reply"},
+        ]
+
+        with server._sessions[sid]["history_lock"]:
+            server._sessions[sid]["history"] = current_history + [
+                {"role": "user", "content": "new live turn"},
+                {"role": "assistant", "content": "new live reply"},
+            ]
+
+        second = server.handle_request(
+            {
+                "id": "second",
+                "method": "session.resume",
+                "params": {"session_id": target, "cols": 120},
+            }
+        )
+
+    assert "error" not in second
+    assert second["result"]["session_id"] == sid
+    assert second["result"]["messages"] == [
+        {"role": "user", "text": "ancestor"},
+        {"role": "user", "text": "current"},
+        {"role": "assistant", "text": "current reply"},
+        {"role": "user", "text": "new live turn"},
+        {"role": "assistant", "text": "new live reply"},
+    ]
+
+
 def test_make_agent_accepts_list_system_prompt(server, monkeypatch):
     captured = {}

tui_gateway/server.py

@@ -128,6 +128,7 @@ def _thread_panic_hook(args):
 _cfg_cache: dict | None = None
 _cfg_mtime: float | None = None
 _cfg_path = None
+_session_resume_lock = threading.Lock()
 try:
     _slash_timeout = float(os.environ.get("HERMES_TUI_SLASH_TIMEOUT_S") or "45")
 except (ValueError, TypeError):
@@ -2979,6 +2980,10 @@ def _(rid, params: dict) -> dict:
     target = params.get("session_id", "")
     if not target:
         return _err(rid, 4006, "session_id required")
+    try:
+        cols = int(params.get("cols", 80))
+    except (TypeError, ValueError):
+        cols = 80
     db = _get_db()
     if db is None:
         return _db_unavailable_error(rid, code=5000)
@@ -2989,6 +2994,25 @@ def _(rid, params: dict) -> dict:
             target = found["id"]
         else:
             return _err(rid, 4007, "session not found")
+    # Fast path: if the session is already live, reuse it under the lock.
+    with _session_resume_lock:
+        live = _find_live_session_by_key(target)
+        if live is not None:
+            sid, session = live
+            payload = _live_session_payload(
+                sid,
+                session,
+                cols=cols,
+                touch=True,
+                transport=current_transport() or _stdio_transport,
+            )
+            payload["resumed"] = target
+            return _ok(rid, payload)
+
+    # Build the agent OUTSIDE the lock — _make_agent can block for seconds
+    # (MCP discovery, prompt/skill build, AIAgent construction). Holding
+    # _session_resume_lock across it would stall session.close on the main
+    # dispatch thread (it's not a _LONG_HANDLER), blocking fast-path RPCs.
     sid = uuid.uuid4().hex[:8]
     _enable_gateway_prompts()
     try:
@@ -2997,15 +3021,46 @@ def _(rid, params: dict) -> dict:
         display_history = db.get_messages_as_conversation(
             target, include_ancestors=True
         )
+        display_history_prefix = display_history[
+            : max(0, len(display_history) - len(history))
+        ]
         messages = _history_to_messages(display_history)
         tokens = [redacted:secret-assignment]
         try:
             agent = _make_agent(sid, target, session_id=target)
         finally:
             _clear_session_context(tokens)
-        _init_session(sid, target, agent, history, cols=int(params.get("cols", 80)))
     except Exception as e:
         return _err(rid, 5000, f"resume failed: {e}")
+
+    # Double-checked locking: another concurrent resume may have created the
+    # live session while we were building. Re-check under the lock; if it won,
+    # discard our just-built agent and reuse theirs (no worker/poller wired yet).
+    with _session_resume_lock:
+        live = _find_live_session_by_key(target)
+        if live is not None:
+            try:
+                if hasattr(agent, "close"):
+                    agent.close()
+            except Exception:
+                pass
+            other_sid, other_session = live
+            payload = _live_session_payload(
+                other_sid,
+                other_session,
+                cols=cols,
+                touch=True,
+                transport=current_transport() or _stdio_transport,
+            )
+            payload["resumed"] = target
+            return _ok(rid, payload)
+        try:
+            _init_session(sid, target, agent, history, cols=cols)
+            if sid in _sessions:
+                _sessions[sid]["display_history_prefix"] = display_history_prefix
+        except Exception as e:
+            return _err(rid, 5000, f"resume failed: {e}")
+        session = _sessions.get(sid) or {}
     return _ok(
         rid,
         {
@@ -3013,7 +3068,12 @@ def _(rid, params: dict) -> dict:
             "resumed": target,
             "message_count": len(messages),
             "messages": messages,
-            "info": _session_info(agent, _sessions.get(sid)),
+            "info": _session_info(agent, session),
+            "inflight": None,
+            "running": False,
+            "session_key": target,
+            "started_at": float(session.get("created_at") or time.time()),
+            "status": "idle",
         },
     )
 
@@ -3106,6 +3166,15 @@ def _session_live_item(sid: str, session: dict, current_sid: str = "") -> dict:
     }
 
 
+def _find_live_session_by_key(session_key: str) -> tuple[str, dict] | None:
+    for sid, session in list(_sessions.items()):
+        if session.get("_finalized"):
+            continue
+        if str(session.get("session_key") or "") == session_key:
+            return sid, session
+    return None
+
+
 def _fallback_session_info(session: dict) -> dict:
     agent = session.get("agent")
     if agent is not None:
@@ -3119,6 +3188,41 @@ def _fallback_session_info(session: dict) -> dict:
     }
 
 
+def _live_session_payload(
+    sid: str,
+    session: dict,
+    *,
+    cols: int | None = None,
+    touch: bool = False,
+    transport: Transport | None = None,
+) -> dict:
+    with session["history_lock"]:
+        if cols is not None:
+            session["cols"] = cols
+        if transport is not None:
+            session["transport"] = transport
+        if touch:
+            session["last_active"] = time.time()
+        history = list(session.get("display_history_prefix") or []) + list(
+            session.get("history") or []
+        )
+        inflight = _inflight_snapshot(session)
+        running = bool(session.get("running"))
+    payload = {
+        "info": _fallback_session_info(session),
+        "message_count": len(history),
+        "messages": _history_to_messages(history),
+        "running": running,
+        "session_id": sid,
+        "session_key": session.get("session_key") or sid,
+        "started_at": float(session.get("created_at") or time.time()),
+        "status": _session_live_status(sid, session),
+    }
+    if inflight:
+        payload["inflight"] = inflight
+    return payload
+
+
 @method("session.active_list")
 def _(rid, params: dict) -> dict:
     """Return live TUI sessions in this gateway process.
@@ -3152,27 +3256,9 @@ def _(rid, params: dict) -> dict:
     if err:
         return err
 
-    with session["history_lock"]:
-        session["last_active"] = time.time()
-        history = list(session.get("display_history") or session.get("history") or [])
-        inflight = _inflight_snapshot(session)
-        running = bool(session.get("running"))
-    status = _session_live_status(sid, session)
-    payload = {
-        "info": _fallback_session_info(session),
-        "message_count": len(history),
-        "messages": _history_to_messages(history),
-        "running": running,
-        "session_id": sid,
-        "session_key": session.get("session_key") or sid,
-        "started_at": float(session.get("created_at") or time.time()),
-        "status": status,
-    }
-    if inflight:
-        payload["inflight"] = inflight
     return _ok(
         rid,
-        payload,
+        _live_session_payload(sid, session, touch=True),
     )
 
 
@@ -3558,28 +3644,32 @@ def _(rid, params: dict) -> dict:
 @method("session.close")
 def _(rid, params: dict) -> dict:
     sid = params.get("session_id", "")
-    session = _sessions.pop(sid, None)
-    if not session:
+    current = _sessions.get(sid)
+    if not current:
         return _ok(rid, {"closed": False})
-    _finalize_session(session)
-    try:
-        from tools.approval import unregister_gateway_notify
+    with _session_resume_lock:
+        session = _sessions.pop(sid, None)
+        if not session:
+            return _ok(rid, {"closed": False})
+        _finalize_session(session)
+        try:
+            from tools.approval import unregister_gateway_notify
 
-        unregister_gateway_notify(session["session_key"])
-    except Exception:
-        pass
-    try:
-        agent = session.get("agent")
-        if agent and hasattr(agent, "close"):
-            agent.close()
-    except Exception:
-        pass
-    try:
-        worker = session.get("slash_worker")
-        if worker:
-            worker.close()
-    except Exception:
-        pass
+            unregister_gateway_notify(session["session_key"])
+        except Exception:
+            pass
+        try:
+            agent = session.get("agent")
+            if agent and hasattr(agent, "close"):
+                agent.close()
+        except Exception:
+            pass
+        try:
+            worker = session.get("slash_worker")
+            if worker:
+                worker.close()
+        except Exception:
+            pass
     return _ok(rid, {"closed": True})

ui-tui/src/app/useSessionLifecycle.ts

@@ -302,38 +302,47 @@ export function useSessionLifecycle(opts: UseSessionLifecycleOptions) {
           return
         }
 
-        closeSession(getUiState().sid === id ? null : getUiState().sid).then(() =>
-          gw
-            .request<SessionResumeResponse>('session.resume', { cols: colsRef.current, session_id: id })
-            .then(raw => {
-              const r = asRpcResult<SessionResumeResponse>(raw)
-
-              if (!r) {
-                sys('error: invalid response: session.resume')
-
-                return patchUiState({ status: 'ready' })
-              }
-
-              resetSession()
-              setSessionStartedAt(Date.now())
-
-              const resumed = toTranscriptMessages(r.messages)
-
-              setHistoryItems(r.info ? [introMsg(r.info), ...resumed] : resumed)
-              writeActiveSessionFile(r.resumed ?? r.session_id)
-              patchUiState({
-                info: r.info ?? null,
-                sid: r.session_id,
-                status: 'ready',
-                usage: usageFrom(r.info ?? null)
-              })
-              setTimeout(() => scrollRef.current?.scrollToBottom(), 0)
-            })
-            .catch((e: Error) => {
-              sys(`error: ${e.message}`)
-              patchUiState({ status: 'ready' })
+        const previousSid = getUiState().sid
+
+        gw.request<SessionResumeResponse>('session.resume', { cols: colsRef.current, session_id: id })
+          .then(raw => {
+            const r = asRpcResult<SessionResumeResponse>(raw)
+
+            if (!r) {
+              sys('error: invalid response: session.resume')
+
+              return patchUiState({ status: 'ready' })
+            }
+
+            const info = r.info ?? null
+            const running = Boolean(r.running || r.status === 'working' || r.status === 'waiting')
+
+            resetSession()
+            setSessionStartedAt(r.started_at ? r.started_at * 1000 : Date.now())
+
+            const resumed = [...toTranscriptMessages(r.messages), ...liveSessionInflightMessages(r.inflight)]
+
+            setHistoryItems(info ? [introMsg(info), ...resumed] : resumed)
+            writeActiveSessionFile(r.resumed ?? r.session_id)
+            patchUiState({
+              busy: running,
+              info,
+              sid: r.session_id,
+              status: statusFromLiveSession(r.status, running),
+              usage: usageFrom(info)
             })
-        )
+            hydrateLiveSessionInflight(r.inflight)
+
+            if (previousSid && previousSid !== r.session_id) {
+              void closeSession(previousSid)
+            }
+
+            setTimeout(() => scrollRef.current?.scrollToBottom(), 0)
+          })
+          .catch((e: Error) => {
+            sys(`error: ${e.message}`)
+            patchUiState({ status: 'ready' })
+          })
       })
     },
     [closeSession, colsRef, gw, panel, resetSession, rpc, scrollRef, setHistoryItems, setSessionStartedAt, sys]

ui-tui/src/gatewayTypes.ts

@@ -122,11 +122,15 @@ export interface SessionCreateResponse {
 }
 
 export interface SessionResumeResponse {
+  inflight?: null | SessionInflightTurn
   info?: SessionInfo
   message_count?: number
   messages: GatewayTranscriptMessage[]
   resumed?: string
+  running?: boolean
   session_id: string
+  started_at?: number
+  status?: LiveSessionStatus
 }
 
 export type LiveSessionStatus = 'idle' | 'starting' | 'waiting' | 'working'

1 PR closed without merging

From the close comments: merged via another change 1

#38330 fix(tui): reuse live session on resume · merged via another change → #39153

Closed issues · 2

Closed#38315Bug: repeated TUI resume can create duplicate live sessions
Closed#38320[Bug]: repeated TUI resume can create duplicate live sessions

How this cluster was grouped

Link evidence (partly structural links): some members are connected by structural evidence (shared changed files, or a closing/superseded reference); the rest are joined by an automated relevance judgement over their text.
Members outside that structural core may belong to a different underlying problem.
Pairwise link density 0.39: the share of item pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. 1.00 means every pair is linked.

Pick up a Lab task

Check whether a fix covers the open issues here, with your own agent. Bring back evidence for maintainers.

Open the Lab

At a glance

Merged PRs: #39153
Closed issues: 2
Attempts: 1 PR closed without merging
Fix link checked: 4 Jun 2026

For your agent

The same facts, typed.

cluster/38315.json
cluster/38315.md