From a3d75e843f5252a0961dbe143629006a651002ad Mon Sep 17 00:00:00 2001 From: RongJie G <111257566+CorgiBoyG@users.noreply.github.com> Date: Mon, 28 Sep 2026 16:00:52 +0800 Subject: [PATCH] fix(gateway): distinguish bridge failures from disconnects (#5623) * fix(gateway): distinguish bridge failures from disconnects * fix(gateway): harden SSE terminal disconnect fencing --------- Co-authored-by: CorgiBoyG --- backend/app/gateway/AGENTS.md | 2 +- backend/app/gateway/services.py | 35 ++- backend/tests/test_sse_observer_disconnect.py | 198 ++++++++++++++- .../tests/test_wait_disconnect_handling.py | 233 +++++++++++++++++- 4 files changed, 460 insertions(+), 8 deletions(-) diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 822b1b64fb..95df8dd1e4 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -196,7 +196,7 @@ startup gate rejects process-local memory and JSONL event stores when - Gateway checkpoint mutations outside run execution must use `services.reserve_checkpoint_write()`, which composes the process-local thread lock with the durable `checkpoint_write` reservation. Manual compaction, `POST /threads/{id}/state`, and both goal mutation routes (`PUT` / `DELETE /threads/{id}/goal`, including creation of a missing goal checkpoint) use this boundary, so an existing run blocks the write and the reservation blocks new reject/interrupt/rollback runs across workers. - Branch/state-update checkpoints copy only the source checkpoint's persisted `deerflow_agent_name`, never request metadata; missing or malformed bindings stay unbound so compaction fails closed. - `DELETE /api/threads/{id}` holds a durable `delete` reservation for the whole cleanup and removes thread filesystem data, checkpoints, owner-scoped historical runs, run events, feedback and thread metadata best-effort. Historical-run cleanup removes only `operation_kind="run"` rows, so internal thread-operation rows — including the reservation protecting that request — stay durable until `RunManager.reserve_thread_operation()` exits. This cleans persisted rows; preventing an already-admitted write from re-creating state for a deleted thread is a separate thread-incarnation contract, not part of this cleanup. -- `POST /wait` (both thread-scoped and `/api/runs/wait`) drains the stream bridge via `wait_for_run_completion()` instead of bare `await record.task`, so it honours the run's `on_disconnect` setting and cancels the background run on real client disconnect rather than returning a stale checkpoint (issue #3265). +- Creator SSE and `/wait` apply `on_disconnect` only to confirmed request disconnect or cancellation; bridge failures propagate without cancelling the run. `/wait` consumes the bridge instead of `record.task`, preventing stale checkpoint responses (#3265). - Memory and Redis `StreamBridge` implementations retain only `stream_bridge.queue_maxsize` data events. A syntactically valid `Last-Event-ID` older than the retained watermark, or a live subscriber that falls behind it, yields `StreamGap` before any partial replay. `sse_consumer` maps that control item to an id-less SSE `gap` payload (`stream_replay_gap`) and intentionally leaves the run active; internal `/wait` consumers resume from its latest retained ID because they only need terminal completion. Redis checks bounds plus the non-blocking read in one transaction, using blocking `XREAD` only as a wake-up before repeating the atomic snapshot. For a no-cursor subscriber that established a wait on an empty stream, the first wake response remains provisional until that next snapshot verifies its tail is still retained; this closes the pre-first-delivery trimming window without changing malformed-cursor live tailing. The correctness tradeoff is one three-command snapshot pipeline per poll plus the blocking wake round trip while idle. Malformed cursor behavior remains backend-specific. Memory treats a syntactically numeric cursor below its watermark conservatively as a gap even when the evicted timestamp can no longer be verified; unknown ids at or above the watermark retain the legacy replay-from-earliest policy. - Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup and lease-driven periodic orphan recovery share one Gateway stream-terminalization path: after `RunManager` durably marks a run `error` with `stop_reason=orphan_recovered`, Gateway publishes `END_SENTINEL` and schedules stream cleanup. The periodic store scan, per-row status writes, and Gateway callback run as one supervised single-flight task, so a slow pass is skipped at the next interval instead of piling up or pausing the sole lease-renewal loop. Store retries have bounded attempts/backoff; an individual operation still relies on the database driver/pool timeout. `RunManager.shutdown()` gives active user runs priority within its shared deadline, then drains or cancels orphan recovery. Gateway tracks delayed recovered-stream cleanups and converts unfinished delays to immediate deletes before closing the bridge; the Redis TTL remains the outage safety net. Only startup recovery, before the runtime yields to requests, projects the latest affected thread to `error`; periodic recovery deliberately avoids that non-atomic projection because `ThreadMetaStore` has no `latest_run_id` conditional-update contract. Store-only SSE and `/wait` consumers wait for the bridge's real END marker after an ordinary durable terminal status, because status persistence can precede tail events. The explicit `orphan_recovered` signal is the only heartbeat fallback: its publisher is known to be gone, so it supplies the liveness boundary if END publication fails or the retained key expires. Malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Keep cross-component recovery orchestration in Gateway through the generic `RunManager.on_orphans_recovered` callback; do not introduce a harness-to-app dependency. Callback failure warnings include every recovered `run_id` so operators can identify rows whose Gateway-side terminalization needs inspection. - Thread-scoped run creation accepts an optional `Idempotency-Key` header on create, stream, and wait. Gateway hashes the caller key with the authenticated owner and `thread_id` before passing it to `RunManager`, whose persistence index is process-wide; never pass an unscoped external key to that index. The same scoped key is shared across `/runs`, `/runs/stream`, and `/runs/wait`; a reused admission whose stored `input` or `assistant_id` differs from the retry returns 409. `/wait` must not treat `task is None` as completion: `store_only` records without a cross-process bridge return durable `status`/`error` instead of serializing the current checkpoint; otherwise wait on the bridge. An idempotent reuse must not serialize the latest thread checkpoint as this run's result — a later run on the same thread may have advanced the head — so reused `/wait` returns durable `status`/`error`. Capture that reuse decision before awaiting completion; `idempotency_reused` is sticky on the shared cached record and an overlapping retry must not suppress the original creating request's checkpoint. After observing completion, refresh store-backed `status`/`error` before returning them — a hydrated peer record still holds admission-time fields. A creating-endpoint retry of a terminal record whose stream is gone emits SSE `gap`/`stream_replay_gap` with `recovery: reload_durable_state` rather than a bare `end`; observer joins of that same record still emit `end`. That gap is opt-in via `sse_consumer(..., emit_gap_on_missing_stream=True)` from thread-scoped `/runs/stream` on this request's reuse — do not key it off `apply_on_disconnect` or the sticky `idempotency_reused` flag. Default `sse_consumer` callers, including stateless `/api/runs/stream`, still emit `end`. A reused still-running `store_only` record on a process-local bridge returns 409 from `/stream` with no `Retry-After`, matching `join`. Missing headers preserve ordinary non-idempotent admission. Stateless `/api/runs/*` stays outside this contract because a request without an explicit thread creates a fresh temporary thread before admission. diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 4fbe5663b3..3f8c94e3cf 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -97,6 +97,7 @@ from deerflow.sandbox.lease import SANDBOX_SERVER_OWNED_CONTEXT_KEYS from deerflow.subagents.status_contract import SUBAGENT_ACCEPTANCE_VERDICT_KEY, SUBAGENT_RECEIPT_VERDICT_KEY, SUBAGENT_TOOL_RECEIPTS_KEY from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, ensure_trace_context, ensure_trace_id from deerflow.utils.assembly_io import run_assembly +from deerflow.utils.file_io import await_drained from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, UNTRUSTED_INPUT_KEY from deerflow.utils.thread_id import validate_thread_id @@ -2344,9 +2345,12 @@ async def sse_consumer( return gap_emitted = False + terminal_emitted = False + disconnect_observed = False try: async for entry in bridge.subscribe(record.run_id, last_event_id=last_event_id): if await request.is_disconnected(): + disconnect_observed = True break if isinstance(entry, StreamGap): @@ -2366,26 +2370,36 @@ async def sse_consumer( if entry is HEARTBEAT_SENTINEL: if await _orphan_recovery_observed_after_heartbeat(record, run_mgr): + terminal_emitted = True yield format_sse("end", None) return yield ": heartbeat\n\n" continue if entry is END_SENTINEL: + terminal_emitted = True yield format_sse("end", None, event_id=entry.id or None) return yield format_sse(entry.event, entry.data, event_id=entry.id or None) + if not disconnect_observed: + raise RuntimeError("stream bridge subscription ended before a terminal event") + except (GeneratorExit, asyncio.CancelledError): + # Starlette closes the response generator or cancels its request task + # when the creator connection disappears. Ordinary bridge exceptions + # must not be mistaken for that client-owned lifecycle signal. + disconnect_observed = True + raise finally: # store_only records are cross-worker observation handles. An explicit # cancel-then-stream action has already persisted its request before # subscribing; a plain join disconnect must not invent a new # cancellation request. Only apply on_disconnect to locally-owned runs, # and only on the creator's own stream — never on an observer join. - if apply_on_disconnect and not gap_emitted and not record.store_only and record.status in (RunStatus.pending, RunStatus.running): + if disconnect_observed and apply_on_disconnect and not gap_emitted and not terminal_emitted and not record.store_only and record.status in (RunStatus.pending, RunStatus.running): if record.on_disconnect == DisconnectMode.cancel: - await run_mgr.cancel(record.run_id) + await await_drained(run_mgr.cancel(record.run_id)) async def wait_for_run_completion( @@ -2422,8 +2436,13 @@ async def wait_for_run_completion( disconnected. Callers must skip checkpoint serialization on ``False`` so a partial checkpoint is not returned as a normal response. + + Raises: + RuntimeError: The bridge subscription ended without a terminal event. + Other bridge failures propagate unchanged. """ completed = False + disconnect_observed = False if await _terminal_record_stream_missing(bridge, record): return True @@ -2449,11 +2468,17 @@ async def wait_for_run_completion( completed = True return True if await request.is_disconnected(): + disconnect_observed = True return False # Heartbeats and regular events: keep waiting for END_SENTINEL. if not gap_seen: - return completed + raise RuntimeError("stream bridge subscription ended before a terminal event") + except asyncio.CancelledError: + # Request-task cancellation is the non-streaming equivalent of + # Starlette closing an SSE response generator. + disconnect_observed = True + raise finally: - if not completed and record.status in (RunStatus.pending, RunStatus.running): + if disconnect_observed and not completed and record.status in (RunStatus.pending, RunStatus.running): if record.on_disconnect == DisconnectMode.cancel: - await run_mgr.cancel(record.run_id) + await await_drained(run_mgr.cancel(record.run_id)) diff --git a/backend/tests/test_sse_observer_disconnect.py b/backend/tests/test_sse_observer_disconnect.py index dd4e8694e1..f4afd1bba8 100644 --- a/backend/tests/test_sse_observer_disconnect.py +++ b/backend/tests/test_sse_observer_disconnect.py @@ -14,8 +14,11 @@ import asyncio import inspect from types import SimpleNamespace +import anyio +import pytest + from app.gateway.services import sse_consumer -from deerflow.runtime import DisconnectMode, RunRecord, RunStatus +from deerflow.runtime import END_SENTINEL, HEARTBEAT_SENTINEL, DisconnectMode, RunRecord, RunStatus def _running_record() -> RunRecord: @@ -39,6 +42,66 @@ class _StubBridge: return _gen() +class _TwoEventBridge: + """Yields twice so the request can disconnect on the second poll.""" + + def subscribe(self, run_id, last_event_id=None): + async def _gen(): + yield SimpleNamespace(event="message", data='{"step": 1}', id="1") + yield SimpleNamespace(event="message", data='{"step": 2}', id="2") + + return _gen() + + +class _HeartbeatBridge: + def subscribe(self, run_id, last_event_id=None): + async def _gen(): + yield HEARTBEAT_SENTINEL + + return _gen() + + +class _FailingBridge: + """Fails before yielding: the client remains connected.""" + + def subscribe(self, run_id, last_event_id=None): + async def _gen(): + raise RuntimeError("stream bridge unavailable") + yield + + return _gen() + + +class _BlockingBridge: + def __init__(self): + self.subscribed = asyncio.Event() + + def subscribe(self, run_id, last_event_id=None): + async def _gen(): + self.subscribed.set() + await asyncio.Event().wait() + yield + + return _gen() + + +class _ExhaustedBridge: + def subscribe(self, run_id, last_event_id=None): + async def _gen(): + if False: + yield + + return _gen() + + +class _EndBridge: + def subscribe(self, run_id, last_event_id=None): + async def _gen(): + yield END_SENTINEL + + return _gen() + + class _CancelRecorder: """Stands in for the RunManager: records cancel calls, mutates nothing.""" @@ -49,6 +112,12 @@ class _CancelRecorder: self.cancelled.append(run_id) +class _DrainingCancelRecorder(_CancelRecorder): + async def cancel(self, run_id, action="interrupt"): + await asyncio.sleep(0) + self.cancelled.append(run_id) + + class _StubRequest: """Minimal request: headers for Last-Event-ID, never-disconnected client (the disconnect under test happens between events, via generator close).""" @@ -60,6 +129,17 @@ class _StubRequest: return False +class _PollingDisconnectRequest(_StubRequest): + def __init__(self, *, disconnect_after: int): + super().__init__() + self._disconnect_after = disconnect_after + self._polls = 0 + + async def is_disconnected(self) -> bool: + self._polls += 1 + return self._polls > self._disconnect_after + + def _request() -> _StubRequest: return _StubRequest() @@ -84,6 +164,23 @@ def test_creator_stream_disconnect_applies_cancel_policy(): assert asyncio.run(scenario()) == ["run-1"] +def test_creator_stream_poll_disconnect_applies_cancel_policy(): + """The primary request-poll arm must mark the disconnect before breaking.""" + + async def scenario(): + recorder = _CancelRecorder() + request = _PollingDisconnectRequest(disconnect_after=1) + consumer = sse_consumer(_TwoEventBridge(), _running_record(), request, recorder) + first_frame = await anext(consumer) + assert "event: message\n" in first_frame + assert "id: 1\n" in first_frame + with pytest.raises(StopAsyncIteration): + await anext(consumer) + return recorder.cancelled + + assert asyncio.run(scenario()) == ["run-1"] + + def test_observer_join_disconnect_does_not_cancel(): """A join/observer stream closing must not cancel the run — including for a read-only credential that never held runs:cancel.""" @@ -97,6 +194,105 @@ def test_observer_join_disconnect_does_not_cancel(): assert asyncio.run(scenario()) == [] +def test_creator_stream_bridge_failure_does_not_apply_disconnect_policy(): + """A server-side subscription failure is not a client disconnect.""" + + async def scenario(): + recorder = _CancelRecorder() + consumer = sse_consumer(_FailingBridge(), _running_record(), _request(), recorder) + with pytest.raises(RuntimeError, match="stream bridge unavailable"): + await anext(consumer) + return recorder.cancelled + + assert asyncio.run(scenario()) == [] + + +def test_creator_stream_request_cancellation_applies_disconnect_policy(): + """Cancelling the hosting request task retains creator disconnect semantics.""" + + async def scenario(): + bridge = _BlockingBridge() + recorder = _CancelRecorder() + consumer = sse_consumer(bridge, _running_record(), _request(), recorder) + next_frame = asyncio.create_task(anext(consumer)) + await bridge.subscribed.wait() + + next_frame.cancel() + try: + await next_frame + except asyncio.CancelledError: + pass + else: + raise AssertionError("cancelled request task must propagate cancellation") + return recorder.cancelled + + assert asyncio.run(scenario()) == ["run-1"] + + +def test_creator_stream_anyio_cancel_scope_drains_disconnect_policy(): + """Level cancellation cannot interrupt the owned run-cancel operation.""" + + async def scenario(): + bridge = _BlockingBridge() + recorder = _DrainingCancelRecorder() + consumer = sse_consumer(bridge, _running_record(), _request(), recorder) + + with anyio.CancelScope() as scope: + scope.cancel() + await anext(consumer) + + return recorder.cancelled + + assert anyio.run(scenario) == ["run-1"] + + +def test_creator_stream_unexpected_exhaustion_does_not_cancel(): + """A subscription ending without END is a server error, not disconnect.""" + + async def scenario(): + recorder = _CancelRecorder() + consumer = sse_consumer(_ExhaustedBridge(), _running_record(), _request(), recorder) + with pytest.raises(RuntimeError, match="ended before a terminal event"): + await anext(consumer) + return recorder.cancelled + + assert asyncio.run(scenario()) == [] + + +def test_creator_stream_close_after_end_does_not_cancel(): + """Closing after a terminal frame cannot reinterpret END as disconnect.""" + + async def scenario(): + recorder = _CancelRecorder() + consumer = sse_consumer(_EndBridge(), _running_record(), _request(), recorder) + assert await anext(consumer) == "event: end\ndata: null\n\n" + await consumer.aclose() + return recorder.cancelled + + assert asyncio.run(scenario()) == [] + + +def test_creator_stream_close_after_orphan_recovery_end_does_not_cancel(monkeypatch): + """A synthetic orphan terminal frame owns the same local invariant as END.""" + + async def scenario(): + recorder = _CancelRecorder() + + async def orphan_recovered(record, run_mgr): + return True + + monkeypatch.setattr( + "app.gateway.services._orphan_recovery_observed_after_heartbeat", + orphan_recovered, + ) + consumer = sse_consumer(_HeartbeatBridge(), _running_record(), _request(), recorder) + assert await anext(consumer) == "event: end\ndata: null\n\n" + await consumer.aclose() + return recorder.cancelled + + assert asyncio.run(scenario()) == [] + + def test_join_routes_wire_sse_consumer_as_observers(): """Both join surfaces must be wired as observers, and the creator's create-and-stream endpoints must keep the creator policy (default).""" diff --git a/backend/tests/test_wait_disconnect_handling.py b/backend/tests/test_wait_disconnect_handling.py index 011744de7c..566c5fa275 100644 --- a/backend/tests/test_wait_disconnect_handling.py +++ b/backend/tests/test_wait_disconnect_handling.py @@ -20,9 +20,15 @@ from __future__ import annotations import asyncio from dataclasses import dataclass, field +from types import SimpleNamespace from typing import Any +from unittest.mock import AsyncMock, patch -from deerflow.runtime import ORPHAN_RECOVERY_STOP_REASON, RunManager, RunRecord, RunStatus +import anyio +import pytest +from _router_auth_helpers import call_unwrapped + +from deerflow.runtime import ORPHAN_RECOVERY_STOP_REASON, CancelOutcome, RunManager, RunRecord, RunStatus from deerflow.runtime.runs.schemas import DisconnectMode from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge @@ -83,6 +89,54 @@ class _FastHeartbeatBridge(MemoryStreamBridge): ) +class _FailingBridge: + """Fails before yielding: the client remains connected.""" + + def subscribe(self, run_id, *, last_event_id=None, heartbeat_interval=15.0): + async def _gen(): + raise RuntimeError("stream bridge unavailable") + yield + + return _gen() + + +class _BlockingBridge: + def __init__(self) -> None: + self.subscribed = asyncio.Event() + + def subscribe(self, run_id, *, last_event_id=None, heartbeat_interval=15.0): + async def _gen(): + self.subscribed.set() + await asyncio.Event().wait() + yield + + return _gen() + + +class _ExhaustedBridge: + def subscribe(self, run_id, *, last_event_id=None, heartbeat_interval=15.0): + async def _gen(): + if False: + yield + + return _gen() + + +@dataclass +class _CancelRecorder: + cancelled: list[str] = field(default_factory=list) + + async def cancel(self, run_id: str, action: str = "interrupt") -> None: + self.cancelled.append(run_id) + + +@dataclass +class _DrainingCancelRecorder(_CancelRecorder): + async def cancel(self, run_id: str, action: str = "interrupt") -> None: + await asyncio.sleep(0) + self.cancelled.append(run_id) + + async def _create_running_record(mgr: RunManager, *, on_disconnect: DisconnectMode) -> Any: record = await mgr.create_or_reject( THREAD_ID, @@ -221,6 +275,108 @@ class TestWaitForRunCompletion: asyncio.run(run()) + def test_bridge_failure_does_not_apply_disconnect_policy(self) -> None: + """A server-side subscription failure is not a client disconnect.""" + from app.gateway.services import wait_for_run_completion + + async def run() -> None: + record = RunRecord( + run_id="run-bridge-failure", + thread_id=THREAD_ID, + assistant_id=None, + status=RunStatus.running, + on_disconnect=DisconnectMode.cancel, + ) + recorder = _CancelRecorder() + + try: + await wait_for_run_completion(_FailingBridge(), record, _FakeRequest(), recorder) + except RuntimeError as exc: + assert str(exc) == "stream bridge unavailable" + else: + raise AssertionError("bridge failure must propagate") + + assert recorder.cancelled == [] + + asyncio.run(run()) + + def test_request_task_cancellation_applies_disconnect_policy(self) -> None: + """Cancelling the hosting request task retains disconnect semantics.""" + from app.gateway.services import wait_for_run_completion + + async def run() -> None: + record = RunRecord( + run_id="run-request-cancelled", + thread_id=THREAD_ID, + assistant_id=None, + status=RunStatus.running, + on_disconnect=DisconnectMode.cancel, + ) + bridge = _BlockingBridge() + recorder = _CancelRecorder() + wait_task = asyncio.create_task(wait_for_run_completion(bridge, record, _FakeRequest(), recorder)) + await bridge.subscribed.wait() + + wait_task.cancel() + try: + await wait_task + except asyncio.CancelledError: + pass + else: + raise AssertionError("cancelled request task must propagate cancellation") + + assert recorder.cancelled == ["run-request-cancelled"] + + asyncio.run(run()) + + def test_anyio_cancel_scope_drains_disconnect_policy(self) -> None: + """Level cancellation cannot interrupt the owned run-cancel operation.""" + from app.gateway.services import wait_for_run_completion + + async def run() -> None: + record = RunRecord( + run_id="run-request-cancelled", + thread_id=THREAD_ID, + assistant_id=None, + status=RunStatus.running, + on_disconnect=DisconnectMode.cancel, + ) + bridge = _BlockingBridge() + recorder = _DrainingCancelRecorder() + + with anyio.CancelScope() as scope: + scope.cancel() + await wait_for_run_completion(bridge, record, _FakeRequest(), recorder) + + assert recorder.cancelled == ["run-request-cancelled"] + + anyio.run(run) + + def test_unexpected_subscription_exhaustion_does_not_cancel(self) -> None: + """A subscription ending without END is a server error, not disconnect.""" + from app.gateway.services import wait_for_run_completion + + async def run() -> None: + record = RunRecord( + run_id="run-bridge-exhausted", + thread_id=THREAD_ID, + assistant_id=None, + status=RunStatus.running, + on_disconnect=DisconnectMode.cancel, + ) + recorder = _CancelRecorder() + + try: + await wait_for_run_completion(_ExhaustedBridge(), record, _FakeRequest(), recorder) + except RuntimeError as exc: + assert str(exc) == "stream bridge subscription ended before a terminal event" + else: + raise AssertionError("unexpected bridge exhaustion must fail") + + assert recorder.cancelled == [] + + asyncio.run(run()) + def test_no_cancel_when_run_already_finished(self) -> None: """If the run ended (END_SENTINEL) before disconnect is observed, the finally block must not call cancel — the run is already terminal.""" @@ -438,3 +594,78 @@ class TestWaitForRunCompletion: assert completed is True asyncio.run(run()) + + +@pytest.mark.parametrize("route_name", ["stateless_wait", "thread_wait", "cancel_wait", "cancel_stream_wait"]) +def test_wait_routes_propagate_bridge_failures(route_name: str) -> None: + """Every wait caller must preserve infrastructure failures as errors.""" + from app.gateway.routers import runs, thread_runs + + error = RuntimeError("stream bridge unavailable") + request = SimpleNamespace() + body = thread_runs.RunCreateRequest(config={"configurable": {"thread_id": THREAD_ID}}) + record = SimpleNamespace( + run_id="run-route-bridge-failure", + thread_id=THREAD_ID, + task=object(), + store_only=False, + status=RunStatus.running, + error=None, + idempotency_reused=False, + ) + bridge = SimpleNamespace(supports_cross_process=True) + manager = SimpleNamespace( + get=AsyncMock(return_value=record), + cancel=AsyncMock(return_value=CancelOutcome.requested), + ) + + async def run() -> None: + if route_name == "stateless_wait": + with ( + patch.object(runs, "get_stream_bridge", return_value=bridge), + patch.object(runs, "get_run_manager", return_value=manager), + patch.object(runs, "start_run", AsyncMock(return_value=record)), + patch.object(runs, "wait_for_run_completion", AsyncMock(side_effect=error)), + ): + await call_unwrapped(runs.stateless_wait, body, request) + return + + if route_name == "thread_wait": + with ( + patch.object(thread_runs, "get_stream_bridge", return_value=bridge), + patch.object(thread_runs, "get_run_manager", return_value=manager), + patch.object(thread_runs, "start_run", AsyncMock(return_value=record)), + patch.object(thread_runs, "wait_for_run_completion", AsyncMock(side_effect=error)), + ): + await call_unwrapped(thread_runs.wait_run, THREAD_ID, body, request) + return + + record.task = None + record.store_only = True + with ( + patch.object(thread_runs, "_require_run_visible_to_scope", AsyncMock()), + patch.object(thread_runs, "get_stream_bridge", return_value=bridge), + patch.object(thread_runs, "get_run_manager", return_value=manager), + patch.object(thread_runs, "require_cancel_permission_when_action"), + patch.object(thread_runs, "wait_for_run_completion", AsyncMock(side_effect=error)), + ): + if route_name == "cancel_wait": + await call_unwrapped( + thread_runs.cancel_run, + THREAD_ID, + record.run_id, + request, + wait=True, + action="interrupt", + ) + else: + await thread_runs._stream_existing_run( + THREAD_ID, + record.run_id, + request, + action="interrupt", + wait=1, + ) + + with pytest.raises(RuntimeError, match="stream bridge unavailable"): + asyncio.run(run())