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