From c748049bf7291737caa1d3e9fe1fc555cc9b2f1b Mon Sep 17 00:00:00 2001 From: James Drake Date: Tue, 11 Aug 2026 23:30:08 -0500 Subject: [PATCH 1/5] fix: clean up failed chat worker launches --- api/routes.py | 25 +++++ .../test_issue6869_gateway_launch_failure.py | 93 +++++++++++++++++++ 2 files changed, 118 insertions(+) create mode 100644 tests/test_issue6869_gateway_launch_failure.py diff --git a/api/routes.py b/api/routes.py index 7d30ab90..b73ec9af 100644 --- a/api/routes.py +++ b/api/routes.py @@ -22810,6 +22810,30 @@ def _prepare_chat_start_session_for_stream( s.save() +def _cleanup_chat_start_launch_failure(session, stream_id: str) -> None: + """Release state registered before a worker thread successfully starts.""" + clear_session_writeback_owner_if_owned(session.session_id, stream_id) + unregister_stream_owner(stream_id) + with STREAMS_LOCK: + STREAMS.pop(stream_id, None) + STREAM_GOAL_RELATED.pop(stream_id, None) + if getattr(session, "active_stream_id", None) != stream_id: + return + session.active_stream_id = None + session.pending_user_message = None + session.pending_attachments = [] + session.pending_started_at = None + session.pending_user_source = None + try: + session.save() + except Exception: + logger.debug( + "Failed to persist chat-start cleanup after worker launch failure for %s", + stream_id, + exc_info=True, + ) + + def _is_hidden_empty_session(s) -> bool: return ( getattr(s, "title", "Untitled") == "Untitled" @@ -23393,6 +23417,7 @@ def _start_chat_stream_for_session( _clear_gateway_run_starting(stream_id) except Exception: logger.debug("Failed to record gateway run-start failure for stream %s", stream_id, exc_info=True) + _cleanup_chat_start_launch_failure(s, stream_id) raise response = { "stream_id": stream_id, diff --git a/tests/test_issue6869_gateway_launch_failure.py b/tests/test_issue6869_gateway_launch_failure.py new file mode 100644 index 00000000..7d8c6fe0 --- /dev/null +++ b/tests/test_issue6869_gateway_launch_failure.py @@ -0,0 +1,93 @@ +"""Regression tests for #6869: failed Gateway worker launch cleanup.""" + +from types import SimpleNamespace + +import api.config as config +import api.routes as routes +from api import turn_journal + + +def test_gateway_thread_start_failure_releases_writeback_owner_and_stream_state(monkeypatch): + """A thread-start exception must not leave one owner per failed Gateway launch.""" + class FailingThread: + def __init__(self, *args, **kwargs): + pass + + def start(self): + raise RuntimeError("thread launch failed") + + def fake_prepare(session, *, stream_id, **kwargs): + session.active_stream_id = stream_id + session.pending_user_message = kwargs["msg"] + session.pending_started_at = 1.0 + config.register_session_writeback_owner(session.session_id, stream_id) + + monkeypatch.setattr(routes.threading, "Thread", FailingThread) + monkeypatch.setattr(routes, "_prepare_chat_start_session_for_stream", fake_prepare) + monkeypatch.setattr(routes, "set_last_workspace", lambda _workspace: None) + monkeypatch.setattr(routes, "_is_hidden_empty_session", lambda _session: False) + monkeypatch.setattr( + turn_journal, + "append_turn_journal_event", + lambda *_args, **_kwargs: {}, + ) + + for index in range(3): + session = SimpleNamespace( + session_id=f"session-launch-failure-{index}", + title="Existing", + active_stream_id=None, + pending_user_message=None, + pending_attachments=[], + pending_started_at=None, + pending_user_source=None, + messages=[{"role": "user", "content": "old"}], + workspace="/tmp", + model="old-model", + model_provider=None, + worktree_path=None, + profile=None, + save=lambda: None, + ) + + try: + routes._start_chat_stream_for_session( + session, + msg="start gateway turn", + attachments=[], + workspace="/tmp", + model="test-model", + external_runtime_owned=True, + ) + except RuntimeError as exc: + assert str(exc) == "thread launch failed" + else: + raise AssertionError("thread-start failure must propagate") + + assert config.SESSION_WRITEBACK_OWNERS == {} + assert config.STREAM_SESSION_OWNERS == {} + assert config.STREAMS == {} + + +def test_gateway_launch_failure_cleanup_does_not_clear_successor_owner(): + """Cleanup must be compare-and-clear when a successor already took over.""" + session = SimpleNamespace( + session_id="session-launch-successor", + active_stream_id="new-stream", + pending_user_message="new prompt", + pending_attachments=["new.txt"], + pending_started_at=2.0, + pending_user_source="webui", + save=lambda: None, + ) + config.register_session_writeback_owner(session.session_id, "old-stream") + config.register_session_writeback_owner(session.session_id, "new-stream") + config.register_stream_owner("old-stream", session.session_id) + config.STREAMS["old-stream"] = object() + + routes._cleanup_chat_start_launch_failure(session, "old-stream") + + assert config.session_writeback_owner(session.session_id) == "new-stream" + assert session.active_stream_id == "new-stream" + assert session.pending_user_message == "new prompt" + assert "old-stream" not in config.STREAMS From b5c0136545a2ed65aeb98163a954f2275721d452 Mon Sep 17 00:00:00 2001 From: James Drake Date: Thu, 27 Aug 2026 08:34:52 -0500 Subject: [PATCH 2/5] fix: run launch-failure session reset under session lock The session-field half of _cleanup_chat_start_launch_failure mutated and saved the passed-in session object without holding the per-session lock and without re-resolving the canonical session. A concurrent successor turn (same session_id, new stream_id) or a delete could be clobbered: pending fields wiped or a deleted session resurrected via save(). Mirror the established #1533 pattern (routes.py:3077) and anchor-scene write guard (routes.py:5140): hold _get_session_agent_lock, re-resolve with get_session(), early-return on KeyError (session deleted) or when the canonical no longer points at the failed stream, then reset pending fields on the canonical object only. Add deterministic race coverage: successor admitted during the check-then-clear window keeps its pending fields and is not saved away, and a deleted session is not resurrected by save(). Both tests fail against the previous implementation. --- api/routes.py | 42 +++-- .../test_issue6869_gateway_launch_failure.py | 150 ++++++++++++++---- 2 files changed, 144 insertions(+), 48 deletions(-) diff --git a/api/routes.py b/api/routes.py index b73ec9af..da9537b3 100644 --- a/api/routes.py +++ b/api/routes.py @@ -22817,21 +22817,33 @@ def _cleanup_chat_start_launch_failure(session, stream_id: str) -> None: with STREAMS_LOCK: STREAMS.pop(stream_id, None) STREAM_GOAL_RELATED.pop(stream_id, None) - if getattr(session, "active_stream_id", None) != stream_id: - return - session.active_stream_id = None - session.pending_user_message = None - session.pending_attachments = [] - session.pending_started_at = None - session.pending_user_source = None - try: - session.save() - except Exception: - logger.debug( - "Failed to persist chat-start cleanup after worker launch failure for %s", - stream_id, - exc_info=True, - ) + # The session-field reset needs the same concurrency discipline as the + # registry half: hold the per-session lock and re-resolve the canonical + # session before clearing anything. Mutating the passed-in stale object + # could wipe a concurrent successor turn's pending fields, and saving it + # could resurrect a session deleted while the launch was failing. Same + # pattern as the #1533 race fix (routes.py:3077) and the anchor-scene + # write guard (routes.py:5140). + with _get_session_agent_lock(session.session_id): + try: + canonical = get_session(session.session_id) + except KeyError: + return # session deleted while the thread launch was failing + if getattr(canonical, "active_stream_id", None) != stream_id: + return # a successor turn already owns the session + canonical.active_stream_id = None + canonical.pending_user_message = None + canonical.pending_attachments = [] + canonical.pending_started_at = None + canonical.pending_user_source = None + try: + canonical.save() + except Exception: + logger.debug( + "Failed to persist chat-start cleanup after worker launch failure for %s", + stream_id, + exc_info=True, + ) def _is_hidden_empty_session(s) -> bool: diff --git a/tests/test_issue6869_gateway_launch_failure.py b/tests/test_issue6869_gateway_launch_failure.py index 7d8c6fe0..703209b9 100644 --- a/tests/test_issue6869_gateway_launch_failure.py +++ b/tests/test_issue6869_gateway_launch_failure.py @@ -7,6 +7,32 @@ import api.routes as routes from api import turn_journal +def _make_session(session_id, *, active_stream_id=None, pending_user_message=None, save=None): + return SimpleNamespace( + session_id=session_id, + title="Existing", + active_stream_id=active_stream_id, + pending_user_message=pending_user_message, + pending_attachments=[], + pending_started_at=None, + pending_user_source=None, + messages=[{"role": "user", "content": "old"}], + workspace="/tmp", + model="old-model", + model_provider=None, + worktree_path=None, + profile=None, + save=save or (lambda: None), + ) + + +def _register_failed_stream(session_id, stream_id): + """Mirror the registry state _prepare_chat_start_session_for_stream installs.""" + config.register_session_writeback_owner(session_id, stream_id) + config.register_stream_owner(stream_id, session_id) + config.STREAMS[stream_id] = object() + + def test_gateway_thread_start_failure_releases_writeback_owner_and_stream_state(monkeypatch): """A thread-start exception must not leave one owner per failed Gateway launch.""" class FailingThread: @@ -16,16 +42,24 @@ def test_gateway_thread_start_failure_releases_writeback_owner_and_stream_state( def start(self): raise RuntimeError("thread launch failed") + canonical_by_id = {} + def fake_prepare(session, *, stream_id, **kwargs): session.active_stream_id = stream_id session.pending_user_message = kwargs["msg"] session.pending_started_at = 1.0 - config.register_session_writeback_owner(session.session_id, stream_id) + _register_failed_stream(session.session_id, stream_id) + canonical_by_id[session.session_id] = session monkeypatch.setattr(routes.threading, "Thread", FailingThread) monkeypatch.setattr(routes, "_prepare_chat_start_session_for_stream", fake_prepare) monkeypatch.setattr(routes, "set_last_workspace", lambda _workspace: None) monkeypatch.setattr(routes, "_is_hidden_empty_session", lambda _session: False) + monkeypatch.setattr( + routes, + "get_session", + lambda sid, metadata_only=False: canonical_by_id[sid], + ) monkeypatch.setattr( turn_journal, "append_turn_journal_event", @@ -33,23 +67,7 @@ def test_gateway_thread_start_failure_releases_writeback_owner_and_stream_state( ) for index in range(3): - session = SimpleNamespace( - session_id=f"session-launch-failure-{index}", - title="Existing", - active_stream_id=None, - pending_user_message=None, - pending_attachments=[], - pending_started_at=None, - pending_user_source=None, - messages=[{"role": "user", "content": "old"}], - workspace="/tmp", - model="old-model", - model_provider=None, - worktree_path=None, - profile=None, - save=lambda: None, - ) - + session = _make_session(f"session-launch-failure-{index}") try: routes._start_chat_stream_for_session( session, @@ -63,31 +81,97 @@ def test_gateway_thread_start_failure_releases_writeback_owner_and_stream_state( assert str(exc) == "thread launch failed" else: raise AssertionError("thread-start failure must propagate") + assert session.active_stream_id is None + assert session.pending_user_message is None + assert session.pending_started_at is None assert config.SESSION_WRITEBACK_OWNERS == {} assert config.STREAM_SESSION_OWNERS == {} assert config.STREAMS == {} -def test_gateway_launch_failure_cleanup_does_not_clear_successor_owner(): - """Cleanup must be compare-and-clear when a successor already took over.""" - session = SimpleNamespace( - session_id="session-launch-successor", +def test_gateway_launch_failure_cleanup_does_not_clear_successor_owner(monkeypatch): + """A successor fully installed before cleanup keeps its registry ownership.""" + canonical = _make_session( + "session-launch-successor", active_stream_id="new-stream", pending_user_message="new prompt", - pending_attachments=["new.txt"], - pending_started_at=2.0, - pending_user_source="webui", save=lambda: None, ) - config.register_session_writeback_owner(session.session_id, "old-stream") - config.register_session_writeback_owner(session.session_id, "new-stream") - config.register_stream_owner("old-stream", session.session_id) - config.STREAMS["old-stream"] = object() + monkeypatch.setattr( + routes, + "get_session", + lambda sid, metadata_only=False: canonical, + ) + stale = _make_session("session-launch-successor") + _register_failed_stream(stale.session_id, "old-stream") + config.register_session_writeback_owner(stale.session_id, "new-stream") - routes._cleanup_chat_start_launch_failure(session, "old-stream") + routes._cleanup_chat_start_launch_failure(stale, "old-stream") - assert config.session_writeback_owner(session.session_id) == "new-stream" - assert session.active_stream_id == "new-stream" - assert session.pending_user_message == "new prompt" + assert config.session_writeback_owner(stale.session_id) == "new-stream" + assert canonical.active_stream_id == "new-stream" + assert canonical.pending_user_message == "new prompt" assert "old-stream" not in config.STREAMS + + +def test_gateway_launch_failure_cleanup_does_not_wipe_successor_admitted_during_cleanup(monkeypatch): + """A successor admitted mid-cleanup survives: the guard runs on the re-resolved + canonical session, not on the stale passed-in object that still shows the + failed stream.""" + saved = [] + + canonical = _make_session( + "session-launch-race", + active_stream_id="new-stream", + pending_user_message="successor prompt", + save=lambda: saved.append("canonical"), + ) + monkeypatch.setattr( + routes, + "get_session", + lambda sid, metadata_only=False: canonical, + ) + stale = _make_session( + "session-launch-race", + active_stream_id="old-stream", + pending_user_message="failed-stream prompt", + save=lambda: saved.append("stale"), + ) + _register_failed_stream(stale.session_id, "old-stream") + config.register_session_writeback_owner(stale.session_id, "new-stream") + + routes._cleanup_chat_start_launch_failure(stale, "old-stream") + + assert config.session_writeback_owner(stale.session_id) == "new-stream" + assert canonical.active_stream_id == "new-stream" + assert canonical.pending_user_message == "successor prompt" + assert "old-stream" not in config.STREAMS + assert saved == [], "cleanup must not save a session it early-returns on" + + +def test_gateway_launch_failure_cleanup_does_not_resurrect_deleted_session(monkeypatch): + """A session deleted while the launch was failing must not be recreated: + the resolver raises KeyError and cleanup returns without saving anything.""" + saved = [] + + def deleted_resolver(sid, metadata_only=False): + raise KeyError(sid) + + monkeypatch.setattr(routes, "get_session", deleted_resolver) + stale = _make_session( + "session-launch-deleted", + active_stream_id="old-stream", + pending_user_message="pending prompt", + save=lambda: saved.append("stale"), + ) + _register_failed_stream(stale.session_id, "old-stream") + + routes._cleanup_chat_start_launch_failure(stale, "old-stream") + + assert config.session_writeback_owner(stale.session_id) is None + assert "old-stream" not in config.STREAM_SESSION_OWNERS + assert "old-stream" not in config.STREAMS + assert stale.active_stream_id == "old-stream" + assert stale.pending_user_message == "pending prompt" + assert saved == [], "cleanup must not save a deleted session" \ No newline at end of file From f444cd1d2e0eeb8f9d5a516a4a308f34d583f9d5 Mon Sep 17 00:00:00 2001 From: James Drake Date: Fri, 18 Sep 2026 08:23:29 -0500 Subject: [PATCH 3/5] test: accept the profile kwarg in the launch-failure stub Master added a profile= kwarg to set_last_workspace; the monkeypatched stub in this test took only the workspace arg and broke on the rebase. --- tests/test_issue6869_gateway_launch_failure.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_issue6869_gateway_launch_failure.py b/tests/test_issue6869_gateway_launch_failure.py index 703209b9..fdb731bb 100644 --- a/tests/test_issue6869_gateway_launch_failure.py +++ b/tests/test_issue6869_gateway_launch_failure.py @@ -53,7 +53,7 @@ def test_gateway_thread_start_failure_releases_writeback_owner_and_stream_state( monkeypatch.setattr(routes.threading, "Thread", FailingThread) monkeypatch.setattr(routes, "_prepare_chat_start_session_for_stream", fake_prepare) - monkeypatch.setattr(routes, "set_last_workspace", lambda _workspace: None) + monkeypatch.setattr(routes, "set_last_workspace", lambda _workspace, **_kw: None) monkeypatch.setattr(routes, "_is_hidden_empty_session", lambda _session: False) monkeypatch.setattr( routes, From 218164f6c889fa3e1cf53b2bad41492933e5361d Mon Sep 17 00:00:00 2001 From: James Drake Date: Fri, 18 Sep 2026 08:28:03 -0500 Subject: [PATCH 4/5] test: scope the launch-failure registry asserts to this test's own streams scripts/test.sh runs the whole selection in one process (no per-file subprocess isolation), so asserting the global registries are empty made this test fail on unrelated files that leave entries behind. --- tests/test_issue6869_gateway_launch_failure.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/tests/test_issue6869_gateway_launch_failure.py b/tests/test_issue6869_gateway_launch_failure.py index fdb731bb..d4ea0ea0 100644 --- a/tests/test_issue6869_gateway_launch_failure.py +++ b/tests/test_issue6869_gateway_launch_failure.py @@ -43,6 +43,7 @@ def test_gateway_thread_start_failure_releases_writeback_owner_and_stream_state( raise RuntimeError("thread launch failed") canonical_by_id = {} + registered_stream_ids = [] def fake_prepare(session, *, stream_id, **kwargs): session.active_stream_id = stream_id @@ -50,6 +51,7 @@ def test_gateway_thread_start_failure_releases_writeback_owner_and_stream_state( session.pending_started_at = 1.0 _register_failed_stream(session.session_id, stream_id) canonical_by_id[session.session_id] = session + registered_stream_ids.append(stream_id) monkeypatch.setattr(routes.threading, "Thread", FailingThread) monkeypatch.setattr(routes, "_prepare_chat_start_session_for_stream", fake_prepare) @@ -85,9 +87,15 @@ def test_gateway_thread_start_failure_releases_writeback_owner_and_stream_state( assert session.pending_user_message is None assert session.pending_started_at is None - assert config.SESSION_WRITEBACK_OWNERS == {} - assert config.STREAM_SESSION_OWNERS == {} - assert config.STREAMS == {} + # Scope these to the streams this test registered. scripts/test.sh runs the + # whole selection in one process, so asserting the global registries are + # empty makes this test fail on unrelated files that leave entries behind. + assert registered_stream_ids, "the launch path must register state before failing" + for stream_id in registered_stream_ids: + assert stream_id not in config.STREAM_SESSION_OWNERS + assert stream_id not in config.STREAMS + for session_id in canonical_by_id: + assert config.session_writeback_owner(session_id) is None def test_gateway_launch_failure_cleanup_does_not_clear_successor_owner(monkeypatch): From 047dfa40a54ad5d3957ab76cdaf013edd2b4b50e Mon Sep 17 00:00:00 2001 From: James Drake Date: Sat, 19 Sep 2026 00:29:07 -0500 Subject: [PATCH 5/5] fix(routes): never mask the launch error during cleanup _cleanup_chat_start_launch_failure runs inside the launch-failure handler, but the session re-resolve only caught KeyError. An unreadable or undeserializable session store raises something else, and that escaped the helper, replacing the worker-launch error the caller is about to report and aborting the reset after the registries were already cleared. Contain the resolution-and-reset so the helper cannot raise. The KeyError branch keeps its meaning (deleted session, return without saving); anything else is logged at debug like the existing save() failure path. --- api/routes.py | 52 ++++++++++++------- .../test_issue6869_gateway_launch_failure.py | 34 +++++++++++- 2 files changed, 65 insertions(+), 21 deletions(-) diff --git a/api/routes.py b/api/routes.py index da9537b3..e8054150 100644 --- a/api/routes.py +++ b/api/routes.py @@ -22824,26 +22824,38 @@ def _cleanup_chat_start_launch_failure(session, stream_id: str) -> None: # could resurrect a session deleted while the launch was failing. Same # pattern as the #1533 race fix (routes.py:3077) and the anchor-scene # write guard (routes.py:5140). - with _get_session_agent_lock(session.session_id): - try: - canonical = get_session(session.session_id) - except KeyError: - return # session deleted while the thread launch was failing - if getattr(canonical, "active_stream_id", None) != stream_id: - return # a successor turn already owns the session - canonical.active_stream_id = None - canonical.pending_user_message = None - canonical.pending_attachments = [] - canonical.pending_started_at = None - canonical.pending_user_source = None - try: - canonical.save() - except Exception: - logger.debug( - "Failed to persist chat-start cleanup after worker launch failure for %s", - stream_id, - exc_info=True, - ) + # + # This runs while the original launch failure is being handled, so it must + # never raise. Lock acquisition and session resolution can fail on their own + # (I/O, deserialization), and an escaping error here would mask the launch + # failure the caller is about to report while leaving the reset half done. + try: + with _get_session_agent_lock(session.session_id): + try: + canonical = get_session(session.session_id) + except KeyError: + return # session deleted while the thread launch was failing + if getattr(canonical, "active_stream_id", None) != stream_id: + return # a successor turn already owns the session + canonical.active_stream_id = None + canonical.pending_user_message = None + canonical.pending_attachments = [] + canonical.pending_started_at = None + canonical.pending_user_source = None + try: + canonical.save() + except Exception: + logger.debug( + "Failed to persist chat-start cleanup after worker launch failure for %s", + stream_id, + exc_info=True, + ) + except Exception: + logger.debug( + "Failed to reset session state after worker launch failure for %s", + stream_id, + exc_info=True, + ) def _is_hidden_empty_session(s) -> bool: diff --git a/tests/test_issue6869_gateway_launch_failure.py b/tests/test_issue6869_gateway_launch_failure.py index d4ea0ea0..01563067 100644 --- a/tests/test_issue6869_gateway_launch_failure.py +++ b/tests/test_issue6869_gateway_launch_failure.py @@ -182,4 +182,36 @@ def test_gateway_launch_failure_cleanup_does_not_resurrect_deleted_session(monke assert "old-stream" not in config.STREAMS assert stale.active_stream_id == "old-stream" assert stale.pending_user_message == "pending prompt" - assert saved == [], "cleanup must not save a deleted session" \ No newline at end of file + assert saved == [], "cleanup must not save a deleted session" + + +def test_gateway_launch_failure_cleanup_never_masks_the_launch_error(monkeypatch): + """A resolution failure during cleanup must not replace the launch error. + + Cleanup runs inside the launch-failure handler, so an error it fails to + contain is reported instead of the real launch failure. Resolution can fail + with something other than KeyError when the session store is unreadable or + cannot be deserialized. + """ + saved = [] + + def failing_resolver(sid, metadata_only=False): + raise OSError("session store unreadable") + + monkeypatch.setattr(routes, "get_session", failing_resolver) + stale = _make_session( + "session-launch-resolution-error", + active_stream_id="old-stream", + pending_user_message="pending prompt", + save=lambda: saved.append("stale"), + ) + _register_failed_stream(stale.session_id, "old-stream") + + # Must not raise: the caller is already handling the launch failure. + routes._cleanup_chat_start_launch_failure(stale, "old-stream") + + # The registry half still completed, and nothing was saved. + assert config.session_writeback_owner(stale.session_id) is None + assert "old-stream" not in config.STREAM_SESSION_OWNERS + assert "old-stream" not in config.STREAMS + assert saved == [] \ No newline at end of file