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 <CorgiBoyG@users.noreply.github.com>

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.

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))

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)."""

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())