From dc4078426a00c307bf5c471840d3783ad39bf056 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Tue, 21 Jul 2026 23:11:37 -0400 Subject: [PATCH] fix(#6008): retire failed gateway startup lifecycle --- tests/test_gateway_approval_runs_api.py | 71 +++++++++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/tests/test_gateway_approval_runs_api.py b/tests/test_gateway_approval_runs_api.py index c1ae7341..dcf70e37 100644 --- a/tests/test_gateway_approval_runs_api.py +++ b/tests/test_gateway_approval_runs_api.py @@ -1883,6 +1883,77 @@ def test_chat_cancel_waits_for_worker_published_run_id_before_settlement( _reset_gateway_run_start_state(stream_id) +def test_gateway_worker_prelude_exception_retires_failed_start_after_waiter_consumes(): + import time + + from api import gateway_chat + + stream_id = "stream-start-prelude-exception" + sid = "sess-start-prelude-exception" + wait_for_gateway_run_id = gateway_chat.wait_for_gateway_run_id + mark_starting = gateway_chat._mark_gateway_run_starting + lifecycle = gateway_chat._STREAM_RUN_LIFECYCLE + run_ids = gateway_chat._STREAM_RUN_IDS + observed = {} + waiter_thread = None + session = SimpleNamespace( + profile=None, + workspace="/tmp", + context_messages=[], + messages=[], + active_stream_id=stream_id, + pending_user_source="webui", + process_wakeup_pause={}, + pending_started_at=time.time(), + save=lambda: None, + title="title", + ) + + def waiter(): + observed["result"] = wait_for_gateway_run_id(stream_id, 1.0) + + worker_thread = threading.Thread( + target=gateway_chat._run_gateway_chat_streaming, + kwargs={ + "session_id": sid, + "msg_text": "hi", + "model": "test-model", + "workspace": "/tmp", + "stream_id": stream_id, + }, + daemon=True, + ) + + try: + gateway_chat.STREAMS[stream_id] = SimpleNamespace(put_nowait=lambda *_args, **_kwargs: None) + mark_starting(stream_id) + waiter_thread = threading.Thread(target=waiter, daemon=True) + waiter_thread.start() + deadline = time.time() + 5 + while time.time() < deadline: + if int((lifecycle.get(stream_id) or {}).get("waiters") or 0) == 1: + break + time.sleep(0.01) + assert int((lifecycle.get(stream_id) or {}).get("waiters") or 0) == 1 + with patch("api.gateway_chat.RunJournalWriter", return_value=SimpleNamespace(append_sse_event=lambda *_a, **_k: None)), \ + patch("api.gateway_chat.get_session", return_value=session), \ + patch("api.config.get_config", side_effect=RuntimeError("prelude boom")): + worker_thread.start() + worker_thread.join(timeout=5) + assert not worker_thread.is_alive() + waiter_thread.join(timeout=5) + assert not waiter_thread.is_alive() + assert observed["result"] == (True, None) + assert stream_id not in lifecycle + assert stream_id not in run_ids + finally: + if waiter_thread is not None: + waiter_thread.join(timeout=5) + worker_thread.join(timeout=5) + gateway_chat.STREAMS.pop(stream_id, None) + _reset_gateway_run_start_state(stream_id) + + def test_gateway_failed_start_result_survives_waiter_until_consumed(): import time