fix(streaming): signal replay history gaps (#4426)

* fix(streaming): signal replay history gaps

* fix(streaming): guard initial Redis replay window

* fix(frontend): align inactive gap recovery

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -275,7 +275,7 @@ Browser login uses `HttpOnly` session cookies. The login page offers a "keep me
 DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser-facing scheme and origin behind a proxy. The bundled nginx sets `X-Forwarded-Proto`, but preserves an upstream HTTPS value and does not overwrite every forwarded header. Configure the outer trusted proxy to replace or strip client-supplied forwarding headers before traffic reaches DeerFlow.
 
 > [!IMPORTANT]
-> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. The bridge shares SSE delivery and `Last-Event-ID` replay across workers, while lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE and `/wait` consumers also refresh durable status on heartbeats as a fallback if terminal publication fails. Malformed reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. IM channel state and other process-local services still need their own multi-worker coordination.
+> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a valid reconnect cursor has been trimmed, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE and `/wait` consumers also refresh durable status on heartbeats as a fallback if terminal publication fails. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. IM channel state and other process-local services still need their own multi-worker coordination.
 >
 > Reconciliation uses an atomic takeover claim that re-checks the lease after candidate selection, so a successful owner renewal wins over orphan recovery and only one reconciler can report a run as recovered.
 

backend/AGENTS.md

@@ -504,6 +504,7 @@ JSONL event stores when `GATEWAY_WORKERS > 1`.
 - Run admission and independent checkpoint writes are first-class thread operations. `runs.operation_kind` distinguishes user-visible `run` rows from internal `checkpoint_write` reservations, while every active kind shares the existing durable active-thread uniqueness constraint. New operation kinds must go through `RunStore.create_thread_operation_atomic()` and `RunManager.reserve_thread_operation()` rather than adding another lock or metadata marker. Live and lease-less reservations are non-interruptible; an expired leased reservation can be reclaimed immediately by interrupt/rollback admission without waiting for orphan reconciliation. Lease-less rows stay fail-closed because the store cannot distinguish a stale row from a live checkpoint writer in another heartbeat-disabled worker; a rare failed delete therefore requires startup reconciliation, and heartbeat-disabled multi-worker deployment remains unsupported. Reservation bodies are attached to their caller task so loss detected by lease renewal cancels the checkpoint writer before it can continue after takeover; the context manager translates that lease-loss cancellation to `ConflictError` after cleanup so Gateway mutation routes return a retryable 409 instead of dropping the HTTP request. The cleanup scope begins immediately after durable admission, including the await that attaches the caller task, so cancellation cannot strand a locally renewed pending reservation. A failed renewal is revalidated under the manager lock before cancellation; if the reservation completed and unregistered while the store update was in flight, its request task must not be cancelled after the checkpoint write. Reservations are excluded from run history/reporting and from run-only helpers such as `list_by_thread()` and `has_inflight()`, release uses the captured owner rather than ambient user context, and local cleanup still runs when the best-effort store delete fails. `RunStore.create_run_atomic()` remains a deprecated compatibility shim for external stores that only admit normal runs; new stores must implement `create_thread_operation_atomic()` to support internal operation kinds.
 - 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.
 - `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).
+- 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 `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching. In `delta` checkpoint mode the worker rewrites that fork into a linear head write before the graph starts (see "A delta-mode run cannot fork" under Checkpoint Channel Modes), because delta state for a fork replays the abandoned sibling's writes.
 - Thread-scoped Gateway runs evaluate an active `ThreadState.goal` after the visible turn completes. `runtime/goal.py` asks a non-thinking evaluator model to judge only visible conversation evidence and return a typed blocker; the evaluator model is created once per run and reused across hidden continuation checks. The evaluator runs after the graph root's tracing scope has already closed, so `create_goal_evaluator_model`/`evaluate_goal_completion` attach their own model-level tracing callbacks (`attach_tracing=True`) and inject Langfuse trace metadata (`thread_id`/`user_id`/`deerflow_trace_id`) directly onto the `ainvoke` call — the same standalone-caller pattern as `oneshot_llm.run_oneshot_llm` and `MemoryUpdater` (see Tracing System below). Satisfied goals are cleared; every non-satisfied evaluation — continuable or stand-down — is persisted with `last_evaluation` (the blocker, reason, and evidence summary; outcomes that stop the loop additionally record a `stand_down_reason` for observability), but only `goal_not_met_yet` evaluations are streamed as hidden `HumanMessage` continuations, and only when a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the thread did not change during evaluation, and the no-progress breaker has not fired. The continuation cap is 8 — a hard maximum in the `0`–`8` range; callers requesting more are clamped (`set_goal`/TUI) or rejected with 422 (`PUT /goal`). The no-progress breaker keys on the latest visible assistant evidence (not the evaluator's free-text reason, which an LLM rewords every turn), so two consecutive continuations that add no new visible assistant output stop the loop after 2 attempts. Model-response cleanup helpers such as think-block stripping and code-fence stripping live in `deerflow.utils.llm_text` so `runtime/goal.py` and Gateway suggestion parsing share the same JSON-prep behavior.

backend/app/gateway/services.py

@@ -46,6 +46,7 @@ from deerflow.runtime import (
     RunRecord,
     RunStatus,
     StreamBridge,
+    StreamGap,
     ThreadOperationKind,
     UnsupportedStrategyError,
     build_state_mutation_graph,
@@ -1236,11 +1237,27 @@ async def sse_consumer(
         yield format_sse("end", None)
         return
 
+    gap_emitted = False
     try:
         async for entry in bridge.subscribe(record.run_id, last_event_id=last_event_id):
             if await request.is_disconnected():
                 break
 
+            if isinstance(entry, StreamGap):
+                gap_emitted = True
+                yield format_sse(
+                    "gap",
+                    {
+                        "code": "stream_replay_gap",
+                        "run_id": record.run_id,
+                        "requested_event_id": entry.requested_event_id,
+                        "earliest_available_event_id": entry.earliest_available_event_id,
+                        "latest_available_event_id": entry.latest_available_event_id,
+                        "recovery": "reload_durable_state",
+                    },
+                )
+                return
+
             if entry is HEARTBEAT_SENTINEL:
                 if await _orphan_recovery_observed_after_heartbeat(record, run_mgr):
                     yield format_sse("end", None)
@@ -1259,7 +1276,7 @@ async def sse_consumer(
         # worker holds no in-memory task/abort state for them, so run_mgr.cancel()
         # cannot stop the task (it would 409). Skip on_disconnect cancellation for
         # those and only act on runs this worker actually owns.
-        if not record.store_only and record.status in (RunStatus.pending, RunStatus.running):
+        if not gap_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)
 
@@ -1297,21 +1314,32 @@ async def wait_for_run_completion(
     if await _terminal_record_stream_missing(bridge, record):
         return True
 
+    resume_from_event_id: str | None = None
     try:
-        async for entry in bridge.subscribe(record.run_id):
-            # END_SENTINEL means the run reached a terminal state; honour it
-            # even if the client just disconnected so the caller still serializes
-            # the real final checkpoint.
-            if entry is END_SENTINEL:
-                completed = True
-                return True
-            if entry is HEARTBEAT_SENTINEL and await _orphan_recovery_observed_after_heartbeat(record, run_mgr):
-                completed = True
-                return True
-            if await request.is_disconnected():
-                break
-            # Heartbeats and regular events: keep waiting for END_SENTINEL.
-        return completed
+        while True:
+            gap_seen = False
+            async for entry in bridge.subscribe(record.run_id, last_event_id=resume_from_event_id):
+                # END_SENTINEL means the run reached a terminal state; honour it
+                # even if the client just disconnected so the caller still serializes
+                # the real final checkpoint.
+                if entry is END_SENTINEL:
+                    completed = True
+                    return True
+                if isinstance(entry, StreamGap):
+                    # The wait API only needs terminal completion, not a complete
+                    # event replay. Resume at the retained tail rather than
+                    # treating a bridge gap as a client disconnect.
+                    resume_from_event_id = entry.latest_available_event_id
+                    gap_seen = True
+                    break
+                if entry is HEARTBEAT_SENTINEL and await _orphan_recovery_observed_after_heartbeat(record, run_mgr):
+                    completed = True
+                    return True
+                if await request.is_disconnected():
+                    return False
+                # Heartbeats and regular events: keep waiting for END_SENTINEL.
+            if not gap_seen:
+                return completed
     finally:
         if not completed and record.status in (RunStatus.pending, RunStatus.running):
             if record.on_disconnect == DisconnectMode.cancel:

backend/docs/API.md

@@ -790,6 +790,31 @@ Both endpoints return `Content-Location: /api/threads/{thread_id}/runs/{run_id}`
 The DeerFlow web UI and LangGraph SDK clients rely on this header to discover the
 assigned `thread_id` and `run_id` on the first message of a new chat.
 
+### SSE replay retention and gaps
+
+Clients may reconnect to a run stream with `Last-Event-ID`. Replay history is
+bounded by `stream_bridge.queue_maxsize` (default `256`) and, for Redis, by the
+rolling `stream_ttl_seconds`. A retained cursor resumes after that event with no
+additional control frame.
+
+When a syntactically valid cursor is older than the retained watermark, the
+server sends exactly one `gap` event before any retained data and closes that
+subscription without an `end` event:
+
+```text
+event: gap
+data: {"code":"stream_replay_gap","run_id":"run-123","requested_event_id":"1718000000000-1","earliest_available_event_id":"1718000000100-42","latest_available_event_id":"1718000000200-84","recovery":"reload_durable_state"}
+
+```
+
+The frame deliberately has no SSE `id:`. Consumers must reload durable thread
+state and persisted run events/messages, then may reconnect from
+`latest_available_event_id` to follow newer live events. A gap does not cancel
+the active run. The same signal applies when a no-cursor subscriber has already
+established an empty-stream wait but the first Redis wake-up falls behind before
+delivery; in that case `requested_event_id` is `null`. Malformed cursor handling
+is backend-specific and is not the same as a valid cursor that was evicted.
+
 ---
 
 ## SDK Usage

backend/docs/STREAMING.md

@@ -143,12 +143,28 @@ sequenceDiagram
 关键组件：
 
 - `runtime/runs/worker.py::run_agent` — 在 `asyncio.Task` 里跑 `agent.astream()`，把每个 chunk 通过 `serialize(chunk, mode=mode)` 转成 JSON，再 `bridge.publish()`。
-- `runtime/stream_bridge` — 抽象 Queue。`publish/subscribe` 解耦生产者和消费者，支持 `Last-Event-ID` 重连、心跳、多订阅者 fan-out。Redis backend 会在每次 `publish()` / `publish_end()` 刷新 retained stream key TTL；启动恢复与基于 worker lease 的周期恢复共用 Gateway stream terminalization 路径：`RunManager` 先将 orphan run 持久化为 `error` 并写入显式的 `stop_reason=orphan_recovered`，随后 Gateway 发布 `END_SENTINEL` 并安排 stream cleanup。周期扫描、逐行状态写入和 Gateway callback 作为一个受监督的 single-flight 后台 task 执行；慢任务不会堆积，也不会阻塞唯一的 lease heartbeat。shutdown 优先收敛活跃 run，再处理恢复 task；尚未执行的延迟 stream cleanup 会改为立即删除。只有 runtime `yield` 前、无并发请求的启动恢复会把最新受影响 thread 标记为 error；周期恢复不做非原子的 thread 投影。store-only SSE 与 `/wait` consumer 不能把普通 durable terminal status 当成流已完成，否则可能跳过延迟发布的 error 等尾部事件；只有 `orphan_recovered` 信号能在 heartbeat 时触发 END fallback，因为此时 producer 已被确认失联。TTL 仍是 Redis 内存和故障安全网，不是正常的 subscriber 终止机制。
+- `runtime/stream_bridge` — 抽象 Queue。`publish/subscribe` 解耦生产者和消费者，支持 `Last-Event-ID` 重连、心跳、多订阅者 fan-out。Memory 和 Redis 都只保留 `queue_maxsize` 条数据事件；游标早于保留水位线时返回 `StreamGap`，不会从当前最早事件静默部分重放。Redis backend 会在每次 `publish()` / `publish_end()` 刷新 retained stream key TTL；启动恢复与基于 worker lease 的周期恢复共用 Gateway stream terminalization 路径：`RunManager` 先将 orphan run 持久化为 `error` 并写入显式的 `stop_reason=orphan_recovered`，随后 Gateway 发布 `END_SENTINEL` 并安排 stream cleanup。周期扫描、逐行状态写入和 Gateway callback 作为一个受监督的 single-flight 后台 task 执行；慢任务不会堆积，也不会阻塞唯一的 lease heartbeat。shutdown 优先收敛活跃 run，再处理恢复 task；尚未执行的延迟 stream cleanup 会改为立即删除。只有 runtime `yield` 前、无并发请求的启动恢复会把最新受影响 thread 标记为 error；周期恢复不做非原子的 thread 投影。store-only SSE 与 `/wait` consumer 不能把普通 durable terminal status 当成流已完成，否则可能跳过延迟发布的 error 等尾部事件；只有 `orphan_recovered` 信号能在 heartbeat 时触发 END fallback，因为此时 producer 已被确认失联。TTL 仍是 Redis 内存和故障安全网，不是正常的 subscriber 终止机制。
 - `app/gateway/services.py::sse_consumer` — 从 bridge 订阅，格式化为 SSE wire 帧。
 - `runtime/serialization.py::serialize` — mode-aware 序列化；`messages` mode 下 `serialize_messages_tuple` 把 `(chunk, metadata)` 转成 `[chunk.model_dump(), metadata]`。
 
 **`StreamBridge` 的存在价值**：当生产者（`run_agent` 任务）和消费者（HTTP 连接）在不同的 asyncio task 里运行时，需要一个可以跨 task 传递事件的中介。Queue 同时还承担断连重连的 buffer 和多订阅者的 fan-out。
 
+### 有界历史与 `gap` 恢复
+
+`Last-Event-ID` 只保证在保留窗口内完整重放，不代表无限历史。有效游标仍在窗口内时，bridge 从该 ID 的下一条事件正常恢复；有效游标已经被 `queue_maxsize` 裁剪，或在线消费者慢到落后水位线时，Gateway 会在任何部分重放之前发送：
+
+```text
+event: gap
+data: {"code":"stream_replay_gap","run_id":"...","requested_event_id":"...","earliest_available_event_id":"...","latest_available_event_id":"...","recovery":"reload_durable_state"}
+
+```
+
+`gap` 帧没有 SSE `id:`，后面也没有正常 `end`；当前订阅随即关闭。它是恢复边界而不是客户端断开，因此不会触发 `on_disconnect=cancel`。客户端必须丢弃不再可信的瞬时状态，重新读取 thread checkpoint 和持久化 run-event/message history，再以 `latest_available_event_id` 为游标跟随新事件。DeerFlow Web UI 自动执行此流程并最多连续恢复五次。
+
+Redis 对无游标、空 stream 上已经建立的阻塞等待也遵循相同契约：第一次 `XREAD` 唤醒的数据在交付前仍是 provisional baseline，bridge 会用下一次事务快照确认其尾 ID 仍在保留窗口。若生产者已经裁剪了该基线，订阅直接返回 `requested_event_id: null` 的 `gap`，不会先交付 retained tail。这个检查有明确的性能代价：每轮订阅需要一个包含 `XRANGE`、`XREVRANGE`、非阻塞 `XREAD` 的事务快照；空闲时还需要单独的阻塞 `XREAD` 来唤醒。
+
+“有效但已淘汰”和 malformed cursor 是不同策略：本契约只要求前者产生 `gap`。Redis 的 malformed ID 仍从 live tail 等待；Memory 对 malformed ID 及序号不低于水位线的未知 ID 仍采用既有的最早保留事件策略。对于序号已经低于水位线的数字格式 foreign ID，Memory 无法再校验已淘汰的 timestamp，因此保守返回 `gap`，优先保证客户端不会把不完整重放误认为完整。
+
 ---
 
 ## DeerFlowClient 路径：sync + in-process

backend/packages/harness/deerflow/runtime/__init__.py

@@ -14,7 +14,7 @@ from .store import get_store, make_store, reset_store, store_context
 # NOTE: ``RedisStreamBridge`` is intentionally not re-exported — ``redis`` is an
 # optional extra and importing it here would load ``redis.asyncio`` in every
 # process. Import it from ``deerflow.runtime.stream_bridge.redis`` when needed.
-from .stream_bridge import END_SENTINEL, HEARTBEAT_SENTINEL, MemoryStreamBridge, StreamBridge, StreamEvent, make_stream_bridge
+from .stream_bridge import END_SENTINEL, HEARTBEAT_SENTINEL, MemoryStreamBridge, StreamBridge, StreamEvent, StreamGap, StreamItem, make_stream_bridge
 
 __all__ = [
     # checkpoint state
@@ -56,5 +56,7 @@ __all__ = [
     "MemoryStreamBridge",
     "StreamBridge",
     "StreamEvent",
+    "StreamGap",
+    "StreamItem",
     "make_stream_bridge",
 ]

backend/packages/harness/deerflow/runtime/stream_bridge/__init__.py

@@ -8,7 +8,7 @@ by :mod:`asyncio.Queue`.
 """
 
 from .async_provider import make_stream_bridge
-from .base import END_SENTINEL, HEARTBEAT_SENTINEL, StreamBridge, StreamEvent
+from .base import END_SENTINEL, HEARTBEAT_SENTINEL, StreamBridge, StreamEvent, StreamGap, StreamItem
 from .memory import MemoryStreamBridge
 
 # NOTE: ``RedisStreamBridge`` is intentionally NOT imported here. ``redis`` is an
@@ -25,5 +25,7 @@ __all__ = [
     "MemoryStreamBridge",
     "StreamBridge",
     "StreamEvent",
+    "StreamGap",
+    "StreamItem",
     "make_stream_bridge",
 ]

backend/packages/harness/deerflow/runtime/stream_bridge/base.py

@@ -30,8 +30,24 @@ class StreamEvent:
     data: Any
 
 
+@dataclass(frozen=True)
+class StreamGap:
+    """A subscriber cursor can no longer be replayed completely.
+
+    ``requested_event_id`` is the reconnect cursor, or the most recently
+    delivered event for a live subscriber that fell behind.  The retained
+    bounds let callers reload durable state and resume at the current tail
+    without mistaking a partial replay for a complete one.
+    """
+
+    requested_event_id: str | None
+    earliest_available_event_id: str
+    latest_available_event_id: str
+
+
 HEARTBEAT_SENTINEL = StreamEvent(id="", event="__heartbeat__", data=None)
 END_SENTINEL = StreamEvent(id="", event="__end__", data=None)
+type StreamItem = StreamEvent | StreamGap
 
 
 class StreamBridge(abc.ABC):
@@ -54,12 +70,13 @@ class StreamBridge(abc.ABC):
         *,
         last_event_id: str | None = None,
         heartbeat_interval: float = 15.0,
-    ) -> AsyncIterator[StreamEvent]:
+    ) -> AsyncIterator[StreamItem]:
         """Async iterator that yields events for *run_id* (consumer side).
 
         Yields :data:`HEARTBEAT_SENTINEL` when no event arrives within
         *heartbeat_interval* seconds.  Yields :data:`END_SENTINEL` once
-        the producer calls :meth:`publish_end`.
+        the producer calls :meth:`publish_end`. Yields :class:`StreamGap` and
+        stops when the subscriber has fallen behind retained history.
         """
 
     @abc.abstractmethod

backend/packages/harness/deerflow/runtime/stream_bridge/memory.py

@@ -4,14 +4,16 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import re
 import time
 from collections.abc import AsyncIterator
 from dataclasses import dataclass, field
 from typing import Any
 
-from .base import END_SENTINEL, HEARTBEAT_SENTINEL, StreamBridge, StreamEvent
+from .base import END_SENTINEL, HEARTBEAT_SENTINEL, StreamBridge, StreamEvent, StreamGap, StreamItem
 
 logger = logging.getLogger(__name__)
+_MEMORY_STREAM_ID_RE = re.compile(r"\d+-(\d+)")
 
 
 @dataclass
@@ -56,25 +58,35 @@ class MemoryStreamBridge(StreamBridge):
         event, so it equals the event's absolute offset within the run. Returns
         ``None`` for ids that do not match the expected format.
         """
-        _, sep, seq_text = event_id.rpartition("-")
-        if not sep:
-            return None
-        try:
-            return int(seq_text)
-        except ValueError:
+        match = _MEMORY_STREAM_ID_RE.fullmatch(event_id)
+        if match is None:
             return None
+        return int(match.group(1))
 
-    def _resolve_start_offset(self, stream: _RunStream, last_event_id: str | None) -> int:
+    @staticmethod
+    def _make_gap(stream: _RunStream, requested_event_id: str | None) -> StreamGap:
+        return StreamGap(
+            requested_event_id=requested_event_id,
+            earliest_available_event_id=stream.events[0].id,
+            latest_available_event_id=stream.events[-1].id,
+        )
+
+    def _resolve_start_offset(self, stream: _RunStream, last_event_id: str | None) -> int | StreamGap:
         if last_event_id is None:
             return stream.start_offset
 
         # Event ids embed a per-run, monotonically increasing ``seq`` that equals
         # the event's absolute offset, so locate the event by arithmetic in O(1)
-        # rather than scanning the retained buffer. The id is verified at the
-        # computed index, so a stale/evicted/foreign/malformed id still falls back
-        # to replay-from-earliest — identical to the previous linear scan.
+        # rather than scanning the retained buffer. Retained ids are verified at
+        # the computed index. Once an id is below the retained watermark there is
+        # nothing left to verify its timestamp against, so even a numeric foreign
+        # id takes the conservative gap path; reloading durable state is safer
+        # than silently claiming a complete replay. Unknown ids at or above the
+        # watermark keep the legacy replay-from-earliest behavior.
         seq = self._parse_event_seq(last_event_id)
         if seq is not None:
+            if stream.events and seq < stream.start_offset:
+                return self._make_gap(stream, last_event_id)
             local_index = seq - stream.start_offset
             if 0 <= local_index < len(stream.events) and stream.events[local_index].id == last_event_id:
                 return stream.start_offset + local_index + 1
@@ -115,39 +127,57 @@ class MemoryStreamBridge(StreamBridge):
         *,
         last_event_id: str | None = None,
         heartbeat_interval: float = 15.0,
-    ) -> AsyncIterator[StreamEvent]:
+    ) -> AsyncIterator[StreamItem]:
         stream = self._get_or_create_stream(run_id)
         async with stream.condition:
-            next_offset = self._resolve_start_offset(stream, last_event_id)
+            start = self._resolve_start_offset(stream, last_event_id)
+            if isinstance(start, StreamGap):
+                gap = start
+                next_offset = stream.start_offset
+            else:
+                gap = None
+                next_offset = start
+
+        if gap is not None:
+            yield gap
+            return
+
+        cursor_event_id = last_event_id
 
         while True:
             async with stream.condition:
                 if next_offset < stream.start_offset:
                     logger.warning(
-                        "subscriber for run %s fell behind retained buffer; resuming from offset %s",
+                        "subscriber for run %s fell behind retained buffer at offset %s",
                         run_id,
-                        stream.start_offset,
+                        next_offset,
                     )
-                    next_offset = stream.start_offset
-
-                local_index = next_offset - stream.start_offset
-                if 0 <= local_index < len(stream.events):
-                    entry = stream.events[local_index]
-                    next_offset += 1
-                elif stream.ended:
-                    entry = END_SENTINEL
+                    entry: StreamItem = self._make_gap(stream, cursor_event_id)
+                    should_stop = True
                 else:
-                    try:
-                        await asyncio.wait_for(stream.condition.wait(), timeout=heartbeat_interval)
-                    except TimeoutError:
-                        entry = HEARTBEAT_SENTINEL
+                    should_stop = False
+
+                    local_index = next_offset - stream.start_offset
+                    if 0 <= local_index < len(stream.events):
+                        entry = stream.events[local_index]
+                        next_offset += 1
+                        cursor_event_id = entry.id
+                    elif stream.ended:
+                        entry = END_SENTINEL
                     else:
-                        continue
+                        try:
+                            await asyncio.wait_for(stream.condition.wait(), timeout=heartbeat_interval)
+                        except TimeoutError:
+                            entry = HEARTBEAT_SENTINEL
+                        else:
+                            continue
 
             if entry is END_SENTINEL:
                 yield END_SENTINEL
                 return
             yield entry
+            if should_stop:
+                return
 
     async def cleanup(self, run_id: str, *, delay: float = 0) -> None:
         if delay > 0:

backend/packages/harness/deerflow/runtime/stream_bridge/redis.py

@@ -27,7 +27,7 @@ except ImportError:  # pragma: no cover - only hit when the optional extra is mi
         "Or switch to stream_bridge.type: memory in config.yaml for single-process deployment."
     ) from None
 
-from .base import END_SENTINEL, HEARTBEAT_SENTINEL, StreamBridge, StreamEvent
+from .base import END_SENTINEL, HEARTBEAT_SENTINEL, StreamBridge, StreamEvent, StreamGap, StreamItem
 
 logger = logging.getLogger(__name__)
 
@@ -139,6 +139,30 @@ class RedisStreamBridge(StreamBridge):
             data=self._decode_data(payload.get("data")),
         )
 
+    @classmethod
+    def _is_end_entry(cls, fields: Mapping[Any, Any]) -> bool:
+        return cls._normalise_fields(fields).get("kind") == _KIND_END
+
+    @staticmethod
+    def _parse_stream_id(event_id: str) -> tuple[int, int] | None:
+        if _REDIS_STREAM_ID_RE.fullmatch(event_id) is None:
+            return None
+        milliseconds, separator, sequence = event_id.partition("-")
+        return int(milliseconds), int(sequence) if separator else 0
+
+    @classmethod
+    def _stream_id_lt(cls, left: str, right: str) -> bool:
+        left_parts = cls._parse_stream_id(left)
+        right_parts = cls._parse_stream_id(right)
+        return left_parts is not None and right_parts is not None and left_parts < right_parts
+
+    @classmethod
+    def _response_tail_id(cls, response: list[Any]) -> str | None:
+        for _stream_name, entries in reversed(response):
+            if entries:
+                return cls._decode(entries[-1][0])
+        return None
+
     async def publish(self, run_id: str, event: str, data: Any) -> None:
         key = self._stream_key(run_id)
         await self._xadd_retained(
@@ -178,28 +202,63 @@ class RedisStreamBridge(StreamBridge):
             return "0-0"
         return self._decode(event_id)
 
+    async def _read_retained_snapshot(
+        self,
+        key: str,
+        stream_id: str,
+    ) -> tuple[list[Any], list[Any], list[Any]]:
+        """Atomically read retained bounds and entries after ``stream_id``.
+
+        A blocking ``XREAD`` cannot participate in a Redis transaction. Live
+        subscribers therefore use this non-blocking atomic snapshot for
+        correctness, and a separate blocking read only as a wake-up signal.
+        This deliberately adds a three-command pipeline per poll (and a second
+        round trip while idle) so retained-bound checks cannot race with reads.
+        """
+        async with self._redis.pipeline(transaction=True) as pipe:
+            pipe.xrange(key, count=1)
+            pipe.xrevrange(key, count=1)
+            pipe.xread({key: stream_id}, count=_XREAD_COUNT)
+            earliest, latest, response = await pipe.execute()
+        return earliest, latest, response
+
     async def subscribe(
         self,
         run_id: str,
         *,
         last_event_id: str | None = None,
         heartbeat_interval: float = 15.0,
-    ) -> AsyncIterator[StreamEvent]:
+    ) -> AsyncIterator[StreamItem]:
         key = self._stream_key(run_id)
         stream_id = await self._resolve_start_stream_id(key, last_event_id)
+        gap_detection_enabled = last_event_id is not None and self._parse_stream_id(last_event_id) is not None
+        pending_initial_response: list[Any] | None = None
         block_ms = max(1, int(heartbeat_interval * 1000)) if heartbeat_interval > 0 else 1
         consecutive_errors = 0
 
         while True:
+            snapshot_stream_id = stream_id
+            pending_initial_tail_id = None
+            if pending_initial_response is not None:
+                pending_initial_tail_id = self._response_tail_id(pending_initial_response)
+                if pending_initial_tail_id is None:
+                    pending_initial_response = None
+                else:
+                    # The first blocking XREAD began against a proven-empty
+                    # stream. Its response is a provisional live baseline, not
+                    # yet client-visible data. Validate that baseline against
+                    # the retained watermark before yielding any of it.
+                    snapshot_stream_id = pending_initial_tail_id
+
             try:
-                response = await self._redis.xread({key: stream_id}, count=_XREAD_COUNT, block=block_ms)
+                earliest_entries, latest_entries, response = await self._read_retained_snapshot(key, snapshot_stream_id)
             except ResponseError:
                 # Last-Event-ID is client-controlled and validated before XREAD.
                 # If Redis still rejects the id, fail instead of resetting to
                 # 0-0, which would replay the whole retained buffer on reconnect.
                 logger.warning(
                     "Redis rejected stream id %r for stream bridge subscription",
-                    stream_id,
+                    snapshot_stream_id,
                     exc_info=True,
                 )
                 raise
@@ -218,21 +277,93 @@ class RedisStreamBridge(StreamBridge):
                 await asyncio.sleep(delay)
                 continue
             else:
-                consecutive_errors = 0
+                # A non-empty snapshot is forward progress. For an empty
+                # snapshot, keep any preceding wake-up failure count until the
+                # blocking XREAD itself succeeds; otherwise a permanently
+                # failing blocking read could retry forever because the
+                # non-blocking transaction succeeds between attempts.
+                if response:
+                    consecutive_errors = 0
 
-            if not response:
-                yield HEARTBEAT_SENTINEL
+            if earliest_entries and (gap_detection_enabled or pending_initial_tail_id is not None):
+                earliest_id = self._decode(earliest_entries[0][0])
+                if self._stream_id_lt(snapshot_stream_id, earliest_id):
+                    latest_id = self._decode(latest_entries[0][0])
+                    logger.warning(
+                        "subscriber for Redis stream %s fell behind retained history at %s",
+                        key,
+                        snapshot_stream_id,
+                    )
+                    yield StreamGap(
+                        requested_event_id=None if pending_initial_tail_id is not None else stream_id,
+                        earliest_available_event_id=earliest_id,
+                        latest_available_event_id=latest_id,
+                    )
+                    return
+
+            responses_to_process = []
+            if pending_initial_response is not None:
+                responses_to_process.append(pending_initial_response)
+                pending_initial_response = None
+            if response:
+                responses_to_process.append(response)
+
+            if not responses_to_process:
+                if latest_entries and self._decode(latest_entries[0][0]) == stream_id and self._is_end_entry(latest_entries[0][1]):
+                    yield END_SENTINEL
+                    return
+
+                try:
+                    wake_response = await self._redis.xread(
+                        {key: stream_id},
+                        count=_XREAD_COUNT,
+                        block=block_ms,
+                    )
+                except ResponseError:
+                    logger.warning(
+                        "Redis rejected stream id %r for stream bridge subscription",
+                        stream_id,
+                        exc_info=True,
+                    )
+                    raise
+                except RedisError:
+                    consecutive_errors += 1
+                    if consecutive_errors > _MAX_SUBSCRIBE_RETRIES:
+                        raise
+                    delay = min(2**consecutive_errors, heartbeat_interval)
+                    logger.warning(
+                        "Transient Redis error in stream bridge subscriber (retry %d/%d); backing off %.1fs",
+                        consecutive_errors,
+                        _MAX_SUBSCRIBE_RETRIES,
+                        delay,
+                        exc_info=True,
+                    )
+                    await asyncio.sleep(delay)
+                    continue
+                else:
+                    consecutive_errors = 0
+
+                if not wake_response:
+                    yield HEARTBEAT_SENTINEL
+                elif last_event_id is None and not gap_detection_enabled and not earliest_entries:
+                    # Do not discard the first wake-up from a proven-empty
+                    # stream. Holding it until the next atomic bounds check
+                    # gives no-cursor subscribers the same fell-behind signal
+                    # as Memory without changing malformed-cursor live tailing.
+                    pending_initial_response = wake_response
                 continue
 
-            for _stream_name, entries in response:
-                for event_id, fields in entries:
-                    event_id = self._decode(event_id)
-                    stream_id = event_id
-                    entry = self._entry_from_redis(event_id, fields)
-                    if entry is END_SENTINEL:
-                        yield END_SENTINEL
-                        return
-                    yield entry
+            for retained_response in responses_to_process:
+                for _stream_name, entries in retained_response:
+                    for event_id, fields in entries:
+                        event_id = self._decode(event_id)
+                        stream_id = event_id
+                        gap_detection_enabled = True
+                        entry = self._entry_from_redis(event_id, fields)
+                        if entry is END_SENTINEL:
+                            yield END_SENTINEL
+                            return
+                        yield entry
 
     async def cleanup(self, run_id: str, *, delay: float = 0) -> None:
         if delay > 0:

backend/tests/test_gateway_services.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
+from contextlib import suppress
 from types import SimpleNamespace
 
 import pytest
@@ -85,6 +86,71 @@ def test_format_sse_no_event_id():
     assert "id:" not in frame
 
 
+@pytest.mark.anyio
+async def test_sse_consumer_emits_gap_without_cancelling_run():
+    """A replay gap is a recovery boundary, not a client disconnect."""
+    from app.gateway.services import sse_consumer
+    from deerflow.runtime import DisconnectMode, MemoryStreamBridge, RunManager, RunStatus
+
+    bridge = MemoryStreamBridge(queue_maxsize=2)
+    run_manager = RunManager()
+    record = await run_manager.create("thread-gap", on_disconnect=DisconnectMode.cancel)
+    await run_manager.set_status(record.run_id, RunStatus.running)
+
+    await bridge.publish(record.run_id, "event-1", {"step": 1})
+    evicted_id = bridge._streams[record.run_id].events[0].id
+    await bridge.publish(record.run_id, "event-2", {"step": 2})
+    await bridge.publish(record.run_id, "event-3", {"step": 3})
+    retained = bridge._streams[record.run_id].events
+
+    worker_started = asyncio.Event()
+
+    async def _pending_worker() -> None:
+        worker_started.set()
+        await asyncio.Event().wait()
+
+    record.task = asyncio.create_task(_pending_worker())
+    await worker_started.wait()
+
+    class _ConnectedRequest:
+        headers = {"Last-Event-ID": evicted_id}
+
+        async def is_disconnected(self) -> bool:
+            return False
+
+    try:
+        frames = [
+            frame
+            async for frame in sse_consumer(
+                bridge,
+                record,
+                _ConnectedRequest(),
+                run_manager,
+            )
+        ]
+
+        assert len(frames) == 1
+        assert frames[0].startswith("event: gap\n")
+        assert "\nid:" not in frames[0]
+        assert "\nevent: end\n" not in frames[0]
+        payload = json.loads(frames[0].split("data: ", 1)[1].splitlines()[0])
+        assert payload == {
+            "code": "stream_replay_gap",
+            "run_id": record.run_id,
+            "requested_event_id": evicted_id,
+            "earliest_available_event_id": retained[0].id,
+            "latest_available_event_id": retained[-1].id,
+            "recovery": "reload_durable_state",
+        }
+        assert record.status == RunStatus.running
+        assert not record.abort_event.is_set()
+        assert not record.task.done()
+    finally:
+        record.task.cancel()
+        with suppress(asyncio.CancelledError):
+            await record.task
+
+
 def test_sanitize_log_param_strips_control_characters():
     from app.gateway.utils import sanitize_log_param
 

backend/tests/test_stream_bridge.py

@@ -10,7 +10,7 @@ import anyio
 import pytest
 
 from deerflow.config.stream_bridge_config import StreamBridgeConfig, set_stream_bridge_config
-from deerflow.runtime import END_SENTINEL, HEARTBEAT_SENTINEL, MemoryStreamBridge, make_stream_bridge
+from deerflow.runtime import END_SENTINEL, HEARTBEAT_SENTINEL, MemoryStreamBridge, StreamGap, make_stream_bridge
 
 # RedisStreamBridge is no longer re-exported from deerflow.runtime (redis is an
 # optional extra; see the NOTE in runtime/stream_bridge/__init__.py). Import it
@@ -62,6 +62,10 @@ class _FakeRedis:
         entries = list(reversed(self.streams.get(name, [])))
         return entries[:count] if count is not None else entries
 
+    async def xrange(self, name, min="-", max="+", count=None):
+        entries = list(self.streams.get(name, []))
+        return entries[:count] if count is not None else entries
+
     async def delete(self, name):
         self.deleted.append(name)
         self.streams.pop(name, None)
@@ -81,6 +85,30 @@ class _FakeRedis:
         self.closed = True
 
 
+class _DelayedBlockingReadRedis:
+    """Hold the first blocking XREAD response while the stream is trimmed."""
+
+    def __init__(self, redis) -> None:
+        self._delegate = redis
+        self.blocking_read_started = asyncio.Event()
+        self.wake_response_captured = asyncio.Event()
+        self.release_wake_response = asyncio.Event()
+
+    def __getattr__(self, name):
+        return getattr(self._delegate, name)
+
+    async def xread(self, streams, count=None, block=None):
+        if block is None:
+            return await self._delegate.xread(streams, count=count, block=block)
+
+        self.blocking_read_started.set()
+        response = await self._delegate.xread(streams, count=count, block=block)
+        if response:
+            self.wake_response_captured.set()
+            await self.release_wake_response.wait()
+        return response
+
+
 class _FakeRedisPipeline:
     def __init__(self, redis: _FakeRedis) -> None:
         self.redis = redis
@@ -100,6 +128,18 @@ class _FakeRedisPipeline:
         self.ops.append(("expire", name, seconds))
         return self
 
+    def xrange(self, name, min="-", max="+", count=None):
+        self.ops.append(("xrange", name, min, max, count))
+        return self
+
+    def xrevrange(self, name, max="+", min="-", count=None):
+        self.ops.append(("xrevrange", name, max, min, count))
+        return self
+
+    def xread(self, streams, count=None, block=None):
+        self.ops.append(("xread", streams, count, block))
+        return self
+
     async def execute(self):
         results = []
         for op in self.ops:
@@ -109,6 +149,15 @@ class _FakeRedisPipeline:
             elif op[0] == "expire":
                 _, name, seconds = op
                 results.append(await self.redis.expire(name, seconds))
+            elif op[0] == "xrange":
+                _, name, min_id, max_id, count = op
+                results.append(await self.redis.xrange(name, min=min_id, max=max_id, count=count))
+            elif op[0] == "xrevrange":
+                _, name, max_id, min_id, count = op
+                results.append(await self.redis.xrevrange(name, max=max_id, min=min_id, count=count))
+            elif op[0] == "xread":
+                _, streams, count, block = op
+                results.append(await self.redis.xread(streams, count=count, block=block))
         return results
 
 
@@ -290,48 +339,58 @@ async def test_subscribe_replays_after_last_event_id(bridge: MemoryStreamBridge)
 
 
 @pytest.mark.anyio
-async def test_slow_subscriber_does_not_skip_after_buffer_trim():
-    """A slow subscriber should continue from the correct absolute offset."""
+async def test_evicted_last_event_id_yields_gap_before_partial_replay():
+    """A valid cursor older than retained history must not silently partial-replay."""
     bridge = MemoryStreamBridge(queue_maxsize=2)
-    run_id = "run-slow-subscriber"
+    run_id = "run-evicted-cursor"
     await bridge.publish(run_id, "e1", {"step": 1})
     await bridge.publish(run_id, "e2", {"step": 2})
 
     stream = bridge._streams[run_id]
     e1_id = stream.events[0].id
-    assert stream.start_offset == 0
-
     await bridge.publish(run_id, "e3", {"step": 3})  # trims e1
-    assert stream.start_offset == 1
-    assert [entry.event for entry in stream.events] == ["e2", "e3"]
-
-    resumed_after_e1 = []
-    async for entry in bridge.subscribe(
-        run_id,
-        last_event_id=e1_id,
-        heartbeat_interval=1.0,
-    ):
-        resumed_after_e1.append(entry)
-        if len(resumed_after_e1) == 2:
-            break
-
-    assert [entry.event for entry in resumed_after_e1] == ["e2", "e3"]
-    e2_id = resumed_after_e1[0].id
-
     await bridge.publish_end(run_id)
 
     received = []
     async for entry in bridge.subscribe(
         run_id,
-        last_event_id=e2_id,
+        last_event_id=e1_id,
         heartbeat_interval=1.0,
     ):
         received.append(entry)
-        if entry is END_SENTINEL:
-            break
 
-    assert [entry.event for entry in received[:-1]] == ["e3"]
-    assert received[-1] is END_SENTINEL
+    assert received == [
+        StreamGap(
+            requested_event_id=e1_id,
+            earliest_available_event_id=stream.events[0].id,
+            latest_available_event_id=stream.events[-1].id,
+        )
+    ]
+
+
+@pytest.mark.anyio
+async def test_slow_subscriber_yields_gap_after_buffer_trim():
+    """A live subscriber that falls behind must not silently jump forward."""
+    bridge = MemoryStreamBridge(queue_maxsize=2)
+    run_id = "run-slow-subscriber"
+    await bridge.publish(run_id, "e1", {"step": 1})
+    await bridge.publish(run_id, "e2", {"step": 2})
+
+    subscriber = bridge.subscribe(run_id, heartbeat_interval=1.0)
+    first = await anext(subscriber)
+    assert first.event == "e1"
+
+    await bridge.publish(run_id, "e3", {"step": 3})
+    await bridge.publish(run_id, "e4", {"step": 4})
+
+    stream = bridge._streams[run_id]
+    assert await anext(subscriber) == StreamGap(
+        requested_event_id=first.id,
+        earliest_available_event_id=stream.events[0].id,
+        latest_available_event_id=stream.events[-1].id,
+    )
+    with pytest.raises(StopAsyncIteration):
+        await anext(subscriber)
 
 
 # ---------------------------------------------------------------------------
@@ -404,11 +463,11 @@ async def test_publish_end_preserves_history_when_space_available():
 
 
 @pytest.mark.anyio
-async def test_concurrent_tasks_end_sentinel():
-    """Multiple concurrent producer/consumer pairs should all terminate properly.
+async def test_concurrent_slow_consumers_receive_gap():
+    """Concurrent consumers must all receive a gap when producers outrun retention.
 
-    Simulates the production scenario where multiple runs share a single
-    bridge instance — each must receive its own END sentinel.
+    Each producer fills a four-entry bridge without yielding, so subscribers
+    that started at offset zero cannot observe the first six events.
     """
     bridge = MemoryStreamBridge(queue_maxsize=4)
     num_runs = 4
@@ -442,7 +501,9 @@ async def test_concurrent_tasks_end_sentinel():
 
     for run_id in run_ids:
         events = results[run_id]
-        assert events[-1] is END_SENTINEL, f"Run {run_id} did not receive END sentinel"
+        assert len(events) == 1
+        assert isinstance(events[0], StreamGap), f"Run {run_id} did not receive a gap"
+        assert events[0].requested_event_id is None
 
 
 # ---------------------------------------------------------------------------
@@ -500,6 +561,121 @@ async def test_redis_replays_after_last_event_id(redis_bridge: RedisStreamBridge
     assert received[-1] is END_SENTINEL
 
 
+@pytest.mark.anyio
+async def test_redis_evicted_last_event_id_yields_gap_before_partial_replay(
+    redis_bridge: RedisStreamBridge,
+):
+    """Redis must distinguish a trimmed cursor from a complete replay."""
+    run_id = "redis-run-evicted-cursor"
+    await redis_bridge.publish(run_id, "e1", {"step": 1})
+    key = redis_bridge._stream_key(run_id)
+    e1_id = redis_bridge._redis.streams[key][0][0]
+    await redis_bridge.publish(run_id, "e2", {"step": 2})
+    await redis_bridge.publish(run_id, "e3", {"step": 3})
+
+    retained_ids = [event_id for event_id, _fields in redis_bridge._redis.streams[key]]
+    received = [
+        entry
+        async for entry in redis_bridge.subscribe(
+            run_id,
+            last_event_id=e1_id,
+            heartbeat_interval=1.0,
+        )
+    ]
+
+    assert received == [
+        StreamGap(
+            requested_event_id=e1_id,
+            earliest_available_event_id=retained_ids[0],
+            latest_available_event_id=retained_ids[-1],
+        )
+    ]
+
+
+@pytest.mark.anyio
+async def test_redis_slow_subscriber_yields_gap_after_buffer_trim(
+    redis_bridge: RedisStreamBridge,
+):
+    """A live Redis subscriber must detect trimming after its last batch."""
+    run_id = "redis-run-slow-subscriber"
+    await redis_bridge.publish(run_id, "e1", {"step": 1})
+    subscriber = redis_bridge.subscribe(run_id, heartbeat_interval=1.0)
+    first = await anext(subscriber)
+    assert first.event == "e1"
+
+    await redis_bridge.publish(run_id, "e2", {"step": 2})
+    await redis_bridge.publish(run_id, "e3", {"step": 3})
+    await redis_bridge.publish(run_id, "e4", {"step": 4})
+    key = redis_bridge._stream_key(run_id)
+    retained_ids = [event_id for event_id, _fields in redis_bridge._redis.streams[key]]
+
+    assert await anext(subscriber) == StreamGap(
+        requested_event_id=first.id,
+        earliest_available_event_id=retained_ids[0],
+        latest_available_event_id=retained_ids[-1],
+    )
+    with pytest.raises(StopAsyncIteration):
+        await anext(subscriber)
+
+
+@pytest.mark.anyio
+async def test_redis_initial_subscriber_yields_gap_when_first_wake_falls_behind():
+    """An established no-cursor wait must not silently replay a trimmed tail."""
+    fake = _FakeRedis()
+    delayed = _DelayedBlockingReadRedis(fake)
+    bridge = RedisStreamBridge(
+        redis_url="redis://fake",
+        queue_maxsize=2,
+        client=delayed,
+    )
+    run_id = "redis-run-initial-subscriber-gap"
+    subscriber = bridge.subscribe(run_id, heartbeat_interval=1.0)
+    first_item = asyncio.create_task(anext(subscriber))
+
+    with anyio.fail_after(2):
+        await delayed.blocking_read_started.wait()
+        await bridge.publish(run_id, "e1", {"step": 1})
+        await delayed.wake_response_captured.wait()
+        await bridge.publish(run_id, "e2", {"step": 2})
+        await bridge.publish(run_id, "e3", {"step": 3})
+        await bridge.publish(run_id, "e4", {"step": 4})
+        delayed.release_wake_response.set()
+        gap = await first_item
+
+    key = bridge._stream_key(run_id)
+    retained_ids = [event_id for event_id, _fields in fake.streams[key]]
+    assert gap == StreamGap(
+        requested_event_id=None,
+        earliest_available_event_id=retained_ids[0],
+        latest_available_event_id=retained_ids[-1],
+    )
+    with pytest.raises(StopAsyncIteration):
+        await anext(subscriber)
+
+
+@pytest.mark.anyio
+async def test_redis_recovery_cursor_at_end_yields_end_immediately(
+    redis_bridge: RedisStreamBridge,
+):
+    """The latest gap cursor may be the internal end marker."""
+    run_id = "redis-run-end-cursor"
+    await redis_bridge.publish(run_id, "event", {})
+    await redis_bridge.publish_end(run_id)
+    key = redis_bridge._stream_key(run_id)
+    end_id = redis_bridge._redis.streams[key][-1][0]
+
+    received = [
+        entry
+        async for entry in redis_bridge.subscribe(
+            run_id,
+            last_event_id=end_id,
+            heartbeat_interval=0.01,
+        )
+    ]
+
+    assert received == [END_SENTINEL]
+
+
 @pytest.mark.anyio
 async def test_redis_invalid_last_event_id_tails_live_events(redis_bridge: RedisStreamBridge):
     """Malformed reconnect ids should not replay retained Redis events."""
@@ -744,6 +920,27 @@ async def test_redis_transient_error_gives_up_after_max_retries():
             pass
 
 
+@pytest.mark.anyio
+async def test_redis_blocking_wakeup_error_gives_up_after_max_retries():
+    """Successful snapshots must not hide a permanently failing blocking XREAD."""
+    from redis.exceptions import RedisError
+
+    fake = _FakeRedis()
+    original_xread = fake.xread
+
+    async def fail_blocking_xread(streams, count=None, block=None):
+        if block is not None:
+            raise RedisError("Persistent blocking connection error")
+        return await original_xread(streams, count=count, block=block)
+
+    fake.xread = fail_blocking_xread
+    bridge = RedisStreamBridge(redis_url="redis://fake", queue_maxsize=2, client=fake)
+
+    with pytest.raises(RedisError, match="Persistent blocking connection error"):
+        async for _ in bridge.subscribe("redis-run-blocking-fail", heartbeat_interval=0.01):
+            pass
+
+
 # ---------------------------------------------------------------------------
 # Factory tests
 # ---------------------------------------------------------------------------
@@ -757,7 +954,7 @@ async def test_make_stream_bridge_defaults():
 
 
 # ---------------------------------------------------------------------------
-# _resolve_start_offset: O(1) seq-indexed resolution (parity with linear scan)
+# _resolve_start_offset: O(1) seq-indexed resolution
 # ---------------------------------------------------------------------------
 
 
@@ -776,6 +973,7 @@ def _linear_resolve(stream, last_event_id):
     [
         ("1718000000000-0", 0),
         ("1718000000000-42", 42),
+        ("-1", None),  # malformed live-tail sentinel, not a valid event id
         ("garbage", None),  # no separator
         ("1718000000000-x", None),  # non-integer seq
         ("", None),
@@ -787,8 +985,7 @@ def test_parse_event_seq(event_id, expected):
 
 @pytest.mark.anyio
 async def test_resolve_start_offset_matches_linear_scan():
-    """The seq-indexed resolver must return exactly what the linear scan returned,
-    across retained, evicted, foreign (same seq / wrong ts), malformed, and None ids."""
+    """Retained and unknown cursors preserve the previous linear-scan behavior."""
     bridge = MemoryStreamBridge(queue_maxsize=4)
     run_id = "run-parity"
     ids = []
@@ -802,10 +999,36 @@ async def test_resolve_start_offset_matches_linear_scan():
     ts, _, seq_text = stream.events[0].id.rpartition("-")
     foreign_id = f"{int(ts) + 1}-{seq_text}"
 
-    candidates = [None, "garbage", "1718000000000-x", "999999-999999", foreign_id, *ids]
+    candidates = [None, "garbage", "1718000000000-x", "999999-999999", foreign_id, *ids[6:]]
     for eid in candidates:
         assert bridge._resolve_start_offset(stream, eid) == _linear_resolve(stream, eid), eid
 
+    for evicted_id in ids[:6]:
+        assert bridge._resolve_start_offset(stream, evicted_id) == StreamGap(
+            requested_event_id=evicted_id,
+            earliest_available_event_id=stream.events[0].id,
+            latest_available_event_id=stream.events[-1].id,
+        )
+
+
+@pytest.mark.anyio
+async def test_memory_low_numeric_foreign_cursor_conservatively_yields_gap():
+    """An unverifiable numeric cursor below the watermark takes the safe path."""
+    bridge = MemoryStreamBridge(queue_maxsize=2)
+    run_id = "run-low-foreign-cursor"
+    for index in range(3):
+        await bridge.publish(run_id, f"e{index}", {"index": index})
+
+    stream = bridge._streams[run_id]
+    timestamp, _, _sequence = stream.events[0].id.rpartition("-")
+    foreign_evicted_id = f"{int(timestamp) + 1}-0"
+
+    assert bridge._resolve_start_offset(stream, foreign_evicted_id) == StreamGap(
+        requested_event_id=foreign_evicted_id,
+        earliest_available_event_id=stream.events[0].id,
+        latest_available_event_id=stream.events[-1].id,
+    )
+
 
 @pytest.mark.anyio
 async def test_subscribe_with_unknown_last_event_id_replays_from_earliest():
@@ -826,6 +1049,29 @@ async def test_subscribe_with_unknown_last_event_id_replays_from_earliest():
     assert received[-1] is END_SENTINEL
 
 
+@pytest.mark.anyio
+async def test_memory_malformed_last_event_id_is_not_reported_as_gap():
+    """Malformed cursor policy stays separate from valid evicted cursors."""
+    bridge = MemoryStreamBridge(queue_maxsize=2)
+    run_id = "run-malformed-id"
+    await bridge.publish(run_id, "first", {})
+    await bridge.publish(run_id, "second", {})
+    await bridge.publish_end(run_id)
+
+    received = [
+        entry
+        async for entry in bridge.subscribe(
+            run_id,
+            last_event_id="-1",
+            heartbeat_interval=1.0,
+        )
+    ]
+
+    assert [entry.event for entry in received[:-1]] == ["first", "second"]
+    assert received[-1] is END_SENTINEL
+    assert all(not isinstance(entry, StreamGap) for entry in received)
+
+
 @pytest.mark.anyio
 async def test_make_stream_bridge_uses_docker_redis_env(monkeypatch):
     """Docker can enable Redis bridge without editing config.yaml."""
@@ -1009,6 +1255,72 @@ async def test_redis_integration_maxlen_trims_history(real_redis_bridge):
     assert length == 2
 
 
+@pytest.mark.integration
+@requires_redis
+@pytest.mark.anyio
+async def test_redis_integration_evicted_cursor_yields_gap(real_redis_bridge):
+    """Real Redis MAXLEN trimming must produce the same gap contract."""
+    run_id = "integ-gap"
+    await real_redis_bridge.publish(run_id, "event-1", {"i": 1})
+    key = real_redis_bridge._stream_key(run_id)
+    first_id = (await real_redis_bridge._redis.xrange(key, count=1))[0][0]
+    await real_redis_bridge.publish(run_id, "event-2", {"i": 2})
+    await real_redis_bridge.publish(run_id, "event-3", {"i": 3})
+
+    retained = await real_redis_bridge._redis.xrange(key)
+    received = [
+        entry
+        async for entry in real_redis_bridge.subscribe(
+            run_id,
+            last_event_id=first_id,
+            heartbeat_interval=1.0,
+        )
+    ]
+
+    assert received == [
+        StreamGap(
+            requested_event_id=first_id,
+            earliest_available_event_id=retained[0][0],
+            latest_available_event_id=retained[-1][0],
+        )
+    ]
+
+
+@pytest.mark.integration
+@requires_redis
+@pytest.mark.anyio
+async def test_redis_integration_initial_subscriber_yields_gap_when_first_wake_falls_behind(
+    real_redis_bridge,
+):
+    """A real blocking XREAD wake must be validated before first delivery."""
+    raw_redis = real_redis_bridge._redis
+    delayed = _DelayedBlockingReadRedis(raw_redis)
+    real_redis_bridge._redis = delayed
+    run_id = "integ-initial-subscriber-gap"
+    subscriber = real_redis_bridge.subscribe(run_id, heartbeat_interval=1.0)
+    first_item = asyncio.create_task(anext(subscriber))
+
+    with anyio.fail_after(2):
+        await delayed.blocking_read_started.wait()
+        await real_redis_bridge.publish(run_id, "e1", {"step": 1})
+        await delayed.wake_response_captured.wait()
+        await real_redis_bridge.publish(run_id, "e2", {"step": 2})
+        await real_redis_bridge.publish(run_id, "e3", {"step": 3})
+        await real_redis_bridge.publish(run_id, "e4", {"step": 4})
+        delayed.release_wake_response.set()
+        gap = await first_item
+
+    key = real_redis_bridge._stream_key(run_id)
+    retained = await raw_redis.xrange(key)
+    assert gap == StreamGap(
+        requested_event_id=None,
+        earliest_available_event_id=retained[0][0],
+        latest_available_event_id=retained[-1][0],
+    )
+    with pytest.raises(StopAsyncIteration):
+        await anext(subscriber)
+
+
 @pytest.mark.integration
 @requires_redis
 @pytest.mark.anyio

backend/tests/test_wait_disconnect_handling.py

@@ -125,6 +125,36 @@ class TestWaitForRunCompletion:
 
         asyncio.run(run())
 
+    def test_gap_resumes_from_retained_tail_until_run_ends(self) -> None:
+        """The internal wait path may skip payloads but must still observe END."""
+        from app.gateway.services import wait_for_run_completion
+
+        async def run() -> None:
+            mgr = RunManager()
+            bridge = MemoryStreamBridge(queue_maxsize=2)
+            record = await _create_running_record(mgr, on_disconnect=DisconnectMode.cancel)
+            request = _FakeRequest()
+
+            async def overrun_then_finish() -> None:
+                await asyncio.sleep(0)
+                for step in range(4):
+                    await bridge.publish(record.run_id, "values", {"step": step})
+                await asyncio.sleep(0)
+                await mgr.set_status(record.run_id, RunStatus.success)
+                await bridge.publish_end(record.run_id)
+
+            asyncio.create_task(overrun_then_finish())
+            completed = await asyncio.wait_for(
+                wait_for_run_completion(bridge, record, request, mgr),
+                timeout=2.0,
+            )
+
+            assert completed is True
+            assert record.status == RunStatus.success
+            assert not record.abort_event.is_set()
+
+        asyncio.run(run())
+
     def test_cancels_run_on_disconnect_when_cancel_mode(self) -> None:
         """on_disconnect=cancel: real disconnect must call run_mgr.cancel()."""
         from app.gateway.services import wait_for_run_completion

config.example.yaml

@@ -1915,12 +1915,14 @@ run_ownership:
 #
 # stream_bridge:
 #   type: memory                 # single-process only
-#   queue_maxsize: 256
+#   queue_maxsize: 256           # data events retained per run. A reconnect
+#                                # older than this window receives SSE `gap`.
 #
 # stream_bridge:
 #   type: redis                  # recommended for Docker / multi-worker gateway
 #   redis_url: redis://redis:6379/0
-#   queue_maxsize: 256           # events retained per run (redis stream MAXLEN)
+#   queue_maxsize: 256           # data events retained per run (Redis MAXLEN).
+#                                # Older valid cursors receive SSE `gap`.
 #   stream_ttl_seconds: 86400    # rolling TTL for retained stream buffers.
 #                                # Refreshed on each publish/publish_end; set 0
 #                                # to disable. This is not a run timeout.

frontend/AGENTS.md

@@ -93,7 +93,8 @@ Tool-calling AI messages can contain user-visible text as well as `tool_calls`.
 - **Thread hooks** (`useThreadStream`, `useSubmitThread`, `useThreads`) are the primary API interface
 - **Thread routes** — construct Web UI chat paths through `core/threads/utils.ts::pathOfThread()`, which percent-encodes both custom agent names and thread IDs before inserting them into route segments
 - **LangGraph client** is a singleton obtained via `getAPIClient()` in `core/api/`
-- **Run stream options** are sanitized by `core/api/stream-mode.ts`: the Gateway-supported set is `values`, `messages-tuple`, `updates`, `debug`, `tasks`, `checkpoints`, and `custom`; any request containing an unsupported mode throws before HTTP instead of being partially forwarded or silently defaulting to `values`. `streamResumable` is retained by thread hooks only for SDK-side reconnect bookkeeping but stripped before the HTTP request because the Gateway does not implement resumable SSE. Keep this boundary aligned with the backend request schema; `messages` and `events` are not supported and must not be forwarded.
+- **Run stream options** are sanitized by `core/api/stream-mode.ts`: the Gateway-supported set is `values`, `messages-tuple`, `updates`, `debug`, `tasks`, `checkpoints`, and `custom`; any request containing an unsupported mode throws before HTTP instead of being partially forwarded or silently defaulting to `values`. `streamResumable` is retained by thread hooks only for SDK-side reconnect bookkeeping but stripped before the HTTP request because the Gateway does not accept that request option; actual replay uses the SSE `Last-Event-ID` cursor. Keep this boundary aligned with the backend request schema; `messages` and `events` are not supported and must not be forwarded.
+- **SSE replay gaps** are handled in `core/api/api-client.ts`, which wraps both initial and joined run streams because the upstream SDK ignores unknown event names. An id-less backend `gap` control frame clears stale reconnect metadata, emits an internal `stream_replay_gap` custom event, reloads durable thread values, and rejoins after the server-provided retained tail, with up to five recovery rejoins after the original stream (six total stream calls on an all-gap exhaustion path). The wrapper remains a lazy async iterable because the SDK consumes it with `for await`. `core/threads/hooks.ts` clears optimistic/transient/subtask state, invalidates durable history caches, and shows the localized recovery warning; never let a gap fall through as a normal stream finish or cancel the still-running backend run.
 - **Streaming Markdown rendering** is owned by `core/streamdown`: Streamdown's `animated` / `isAnimating` API handles incremental word animation, while the shared `streamdownRenderingPlugins` config registers the named code-highlighting and Mermaid plugins required by Streamdown 2.5. Keep wrappers and derived configs wired to that shared object; do not reintroduce a rehype plugin that wraps every word, because reparsing a growing block remounts old words and replays their animation.
 - **Environment validation** uses `@t3-oss/env-nextjs` with Zod schemas (`src/env.js`). Skip with `SKIP_ENV_VALIDATION=1`
 - **Subtask step history and runtime metadata** (`core/tasks/`) — the subtask card shows a subagent's full step timeline (#3779): its assistant reasoning turns interleaved with the tools it ran. `Subtask.steps[]` is accumulated live from `task_running` events (appended via `mergeSteps`, not overwritten) and backfilled on expand for historical runs by `fetchSubtaskSteps`, which pages the events endpoint scoped to one task (GET `/runs/{runId}/events?event_types=subagent.step&task_id=…&after_seq=…`) until a short page, so the run-wide limit can't truncate the timeline. `task_started` carries the effective `model_name`; `task_running` carries a cumulative usage snapshot after each completed LLM call. `core/tasks/lifecycle.ts` normalizes these additive events, and `computeNextSubtask` keeps the largest cumulative total so replayed or late SSE frames cannot double-count or roll the folded card backward. Terminal ToolMessage metadata (`subagent_model_name` / `subagent_token_usage`) restores the same values from normal history after reload; no per-card event fetch is needed. `core/tasks/steps.ts` is the pure step model: `messageToStep` (live), `eventsToSteps` (reload), `mergeSteps` (dedup by `message_index`), and `stepsForDisplay` (what the card renders — keeps tool steps + AI steps with text, drops the trailing final-answer AI step when completed since it's shown as `result`). `core/tasks/context.tsx`'s `useUpdateSubtask` applies updates against a `tasksRef` mirroring the latest state (not a closure snapshot), so a late-resolving `fetchSubtaskSteps` backfill merges into current state instead of clobbering SSE steps or sibling subtasks that arrived meanwhile. The owning `run_id` is carried onto history content messages in `buildVisibleHistoryMessages` so the card can resolve the events endpoint.

frontend/src/core/api/api-client.ts

@@ -67,6 +67,59 @@ const TERMINAL_RUN_STATUSES = new Set([
   "interrupted",
 ]);
 
+// This is a rejoin budget: the original stream is not counted, so exhausting
+// five recovery attempts can consume six streams in total.
+const MAX_STREAM_GAP_RECOVERIES = 5;
+
+export type StreamReplayGapData = {
+  code: "stream_replay_gap";
+  run_id: string;
+  requested_event_id: string | null;
+  earliest_available_event_id: string;
+  latest_available_event_id: string;
+  recovery: "reload_durable_state";
+};
+
+type StreamPart = {
+  id?: string;
+  event: string;
+  data: unknown;
+};
+
+export class StreamReplayGapError extends Error {
+  constructor(
+    readonly gap: StreamReplayGapData,
+    readonly recoveryAttempts: number,
+    readonly recoveryCause?: unknown,
+  ) {
+    super(
+      `Unable to recover SSE history after ${recoveryAttempts} attempts (requested ${gap.requested_event_id ?? "initial stream"}, earliest ${gap.earliest_available_event_id})`,
+    );
+    this.name = "StreamReplayGapError";
+  }
+}
+
+function parseStreamReplayGap(data: unknown): StreamReplayGapData {
+  if (typeof data !== "object" || data === null) {
+    throw new Error("Invalid stream replay gap payload.");
+  }
+
+  const value = data as Record<string, unknown>;
+  const requestedEventId = value.requested_event_id;
+  if (
+    value.code !== "stream_replay_gap" ||
+    typeof value.run_id !== "string" ||
+    (requestedEventId !== null && typeof requestedEventId !== "string") ||
+    typeof value.earliest_available_event_id !== "string" ||
+    typeof value.latest_available_event_id !== "string" ||
+    value.recovery !== "reload_durable_state"
+  ) {
+    throw new Error("Invalid stream replay gap payload.");
+  }
+
+  return value as StreamReplayGapData;
+}
+
 /**
  * Shared matcher for the gateway's 409 conflict responses. The SDK surfaces
  * non-2xx responses as ``HTTPError { status, message }`` where ``message`` is
@@ -167,6 +220,105 @@ export function clearReconnectRun(
   }
 }
 
+function rememberReconnectRun(
+  threadId: string | null | undefined,
+  runId: string,
+): void {
+  if (typeof window === "undefined" || !threadId) return;
+
+  try {
+    window.sessionStorage.setItem(`lg:stream:${threadId}`, runId);
+  } catch {
+    // Ignore storage access failures so gap recovery remains usable.
+  }
+}
+
+async function* recoverStreamReplayGaps({
+  client,
+  threadId,
+  expectedRunId,
+  initialStream,
+  resume,
+}: {
+  client: LangGraphClient;
+  threadId: string | null | undefined;
+  expectedRunId: () => string | undefined;
+  initialStream: AsyncIterable<StreamPart>;
+  resume: (runId: string, lastEventId: string) => AsyncIterable<StreamPart>;
+}): AsyncGenerator<StreamPart> {
+  let stream = initialStream;
+  let recoveryAttempts = 0;
+
+  while (true) {
+    let gap: StreamReplayGapData | undefined;
+    for await (const entry of stream) {
+      if (entry.event === "gap") {
+        gap = parseStreamReplayGap(entry.data);
+        break;
+      }
+      yield entry;
+    }
+
+    if (!gap) {
+      return;
+    }
+
+    const runId = expectedRunId() ?? gap.run_id;
+    if (!threadId || gap.run_id !== runId) {
+      throw new Error(
+        "Stream replay gap does not match the active thread run.",
+      );
+    }
+    if (recoveryAttempts >= MAX_STREAM_GAP_RECOVERIES) {
+      throw new StreamReplayGapError(gap, recoveryAttempts);
+    }
+    recoveryAttempts += 1;
+
+    // The SDK would otherwise ignore an unknown `gap` event and report a
+    // normal finish. Surface a custom control event to DeerFlow's hook, reload
+    // durable values, then explicitly follow only events newer than the
+    // retained tail captured by the server.
+    clearReconnectRun(threadId, runId);
+    yield {
+      event: "custom",
+      data: { type: "stream_replay_gap", ...gap },
+    };
+
+    const durableState = await client.threads
+      .getState(threadId)
+      .catch((error: unknown) => {
+        throw new StreamReplayGapError(gap, recoveryAttempts, error);
+      });
+    if (durableState.values != null) {
+      yield { event: "values", data: durableState.values };
+    }
+
+    rememberReconnectRun(threadId, runId);
+    stream = resume(runId, gap.latest_available_event_id);
+  }
+}
+
+async function* handleInactiveRunStream({
+  threadId,
+  expectedRunId,
+  stream,
+}: {
+  threadId: string | null | undefined;
+  expectedRunId: () => string | undefined;
+  stream: AsyncIterable<StreamPart>;
+}): AsyncGenerator<StreamPart> {
+  try {
+    yield* stream;
+  } catch (error) {
+    const runId = expectedRunId();
+    if (runId && isInactiveRunStreamError(error)) {
+      clearReconnectRun(threadId, runId);
+      return;
+    }
+    throw error;
+  }
+}
+
 function createCompatibleClient(isMock?: boolean): LangGraphClient {
   if (isStaticWebsiteOnly() && !isMock) {
     return createStaticClient();
@@ -179,12 +331,44 @@ function createCompatibleClient(isMock?: boolean): LangGraphClient {
   });
 
   const originalRunStream = client.runs.stream.bind(client.runs);
-  client.runs.stream = ((threadId, assistantId, payload) =>
-    originalRunStream(
+  const originalJoinStream = client.runs.joinStream.bind(client.runs);
+  // Preserve the SDK's lazy AsyncIterable contract. Its StreamManager consumes
+  // this return value with `for await`, so run creation still starts on first
+  // iteration rather than when `runs.stream()` is called.
+  client.runs.stream = async function* (threadId, assistantId, payload) {
+    const sanitizedPayload = sanitizeRunStreamOptions(payload);
+    const originalOnRunCreated = sanitizedPayload?.onRunCreated;
+    let runId: string | undefined;
+    const initialStream = originalRunStream(threadId, assistantId, {
+      ...sanitizedPayload,
+      onRunCreated(meta) {
+        runId = meta.run_id;
+        originalOnRunCreated?.(meta);
+      },
+    });
+
+    const recoveredStream = recoverStreamReplayGaps({
+      client,
       threadId,
-      assistantId,
-      sanitizeRunStreamOptions(payload),
-    )) as typeof client.runs.stream;
+      expectedRunId: () => runId,
+      initialStream,
+      resume: (resolvedRunId, lastEventId) => {
+        // Keep the recovery run id available to the shared inactive-stream
+        // handler even if the SDK omitted its onRunCreated callback.
+        runId = resolvedRunId;
+        return originalJoinStream(threadId, resolvedRunId, {
+          lastEventId,
+          signal: sanitizedPayload?.signal,
+          streamMode: sanitizedPayload?.streamMode,
+        });
+      },
+    });
+    yield* handleInactiveRunStream({
+      threadId,
+      expectedRunId: () => runId,
+      stream: recoveredStream,
+    });
+  } as typeof client.runs.stream;
 
   const originalCancel = client.runs.cancel.bind(client.runs);
   client.runs.cancel = (async (threadId, runId, wait, action, options) => {
@@ -205,7 +389,6 @@ function createCompatibleClient(isMock?: boolean): LangGraphClient {
     }
   }) as typeof client.runs.cancel;
 
-  const originalJoinStream = client.runs.joinStream.bind(client.runs);
   client.runs.joinStream = async function* (threadId, runId, options) {
     // Short-circuit reconnects to runs that have already finished: otherwise a
     // reload after the backend's stream bridge is reaped blocks forever on a
@@ -215,19 +398,22 @@ function createCompatibleClient(isMock?: boolean): LangGraphClient {
       clearReconnectRun(threadId, runId);
       return;
     }
-    try {
-      yield* originalJoinStream(
+    const sanitizedOptions = sanitizeRunStreamOptions(options);
+    yield* handleInactiveRunStream({
+      threadId,
+      expectedRunId: () => runId,
+      stream: recoverStreamReplayGaps({
+        client,
         threadId,
-        runId,
-        sanitizeRunStreamOptions(options),
-      );
-    } catch (error) {
-      if (isInactiveRunStreamError(error)) {
-        clearReconnectRun(threadId, runId);
-        return;
-      }
-      throw error;
-    }
+        expectedRunId: () => runId,
+        initialStream: originalJoinStream(threadId, runId, sanitizedOptions),
+        resume: (resolvedRunId, lastEventId) =>
+          originalJoinStream(threadId, resolvedRunId, {
+            ...sanitizedOptions,
+            lastEventId,
+          }),
+      }),
+    });
   } as typeof client.runs.joinStream;
 
   return client;

frontend/src/core/i18n/locales/en-US.ts

@@ -506,6 +506,8 @@ export const enUS: Translations = {
     startConversation: "Start a conversation to see messages here",
     branchCreated: "Conversation branch created",
     branchFailed: "Failed to branch conversation.",
+    streamReplayGap:
+      "Some live updates expired. The conversation was restored from saved state.",
   },
 
   // Chats

frontend/src/core/i18n/locales/types.ts

@@ -405,6 +405,7 @@ export interface Translations {
     startConversation: string;
     branchCreated: string;
     branchFailed: string;
+    streamReplayGap: string;
   };
 
   // Chats

frontend/src/core/i18n/locales/zh-CN.ts

@@ -483,6 +483,7 @@ export const zhCN: Translations = {
     startConversation: "开始新的对话以查看消息",
     branchCreated: "已创建分叉对话",
     branchFailed: "创建分叉对话失败。",
+    streamReplayGap: "部分实时更新已过期，已从持久化状态恢复对话。",
   },
 
   // Chats

frontend/src/core/threads/hooks.ts

@@ -23,7 +23,7 @@ import { isHiddenFromUIMessage } from "../messages/utils";
 import type { FileInMessage } from "../messages/utils";
 import type { LocalSettings } from "../settings";
 import { isSidecarThread, SIDECAR_METADATA_KEY } from "../sidecar/thread";
-import { useUpdateSubtask } from "../tasks/context";
+import { useSubtaskContext, useUpdateSubtask } from "../tasks/context";
 import { taskEventToSubtaskUpdate } from "../tasks/lifecycle";
 import { messageToStep } from "../tasks/steps";
 import type { UploadedFileInfo } from "../uploads";
@@ -1208,6 +1208,7 @@ export function useThreadStream({
   }, []);
 
   const queryClient = useQueryClient();
+  const { tasksRef, setTasks } = useSubtaskContext();
   const updateSubtask = useUpdateSubtask();
 
   const thread = useStream<AgentThreadState>({
@@ -1343,6 +1344,25 @@ export function useThreadStream({
           ? (event as { type: unknown }).type
           : undefined;
 
+      if (eventType === "stream_replay_gap") {
+        setOptimisticMessages([]);
+        setOptimisticThreadId(null);
+        setLiveMessagesThreadId(null);
+        setPendingSupersededRunIds(new Set());
+        setPendingSupersededMessageIds(new Set());
+        messagesRef.current = [];
+        transientHistoryBridgeRef.current = [];
+        transientHistoryOrderRef.current = [];
+        transientHistoryThreadIdRef.current = null;
+        summarizedRef.current = new Set<string>();
+        pendingUsageBaselineMessageIdsRef.current = new Set();
+        tasksRef.current = {};
+        setTasks({});
+        invalidateStoppedThreadCaches(queryClient, threadIdRef.current, isMock);
+        toast.warning(t.conversation.streamReplayGap);
+        return;
+      }
+
       const taskUpdate = taskEventToSubtaskUpdate(event);
       if (taskUpdate) {
         updateSubtask(taskUpdate);

frontend/tests/unit/core/api/api-client.test.ts

@@ -5,6 +5,7 @@ import {
   getAPIClient,
   isInactiveRunStreamError,
   isRunNotCancellableError,
+  StreamReplayGapError,
 } from "@/core/api/api-client";
 
 function makeSessionStorage() {
@@ -20,6 +21,16 @@ function makeSessionStorage() {
   };
 }
 
+function makeSSEResponse(body: string, headers?: HeadersInit) {
+  return new Response(body, {
+    status: 200,
+    headers: {
+      "Content-Type": "text/event-stream",
+      ...headers,
+    },
+  });
+}
+
 afterEach(() => {
   rs.unstubAllGlobals();
 });
@@ -301,6 +312,336 @@ test("proceeds to join when the run is still active", async () => {
   expect(sessionStorage.removeItem).toHaveBeenCalledWith("lg:stream:thread-1");
 });
 
+test("recovers a join stream gap from durable state and resumes after the retained tail", async () => {
+  const sessionStorage = makeSessionStorage();
+  sessionStorage.setItem("lg:stream:thread-1", "run-1");
+  const gap = {
+    code: "stream_replay_gap",
+    run_id: "run-1",
+    requested_event_id: "1-0",
+    earliest_available_event_id: "2-0",
+    latest_available_event_id: "3-0",
+    recovery: "reload_durable_state",
+  };
+  const recoveryRequests: RequestInit[] = [];
+  const fetchFn = rs.fn(async (url: string | URL, init?: RequestInit) => {
+    const path = url.toString();
+    if (path.endsWith("/runs/run-1")) {
+      return new Response(JSON.stringify({ status: "running" }), {
+        status: 200,
+      });
+    }
+    if (path.includes("/runs/run-1/stream")) {
+      recoveryRequests.push(init ?? {});
+      if (recoveryRequests.length === 1) {
+        return makeSSEResponse(`event: gap\ndata: ${JSON.stringify(gap)}\n\n`);
+      }
+      return makeSSEResponse("event: end\ndata: null\n\n");
+    }
+    if (path.includes("/threads/thread-1/state")) {
+      return new Response(
+        JSON.stringify({
+          values: { messages: [{ type: "ai", content: "durable" }] },
+          next: [],
+          tasks: [],
+          metadata: {},
+          created_at: null,
+          checkpoint: {},
+          parent_checkpoint: null,
+        }),
+        { status: 200 },
+      );
+    }
+    return new Response(JSON.stringify({ detail: "unexpected request" }), {
+      status: 500,
+    });
+  });
+  rs.stubGlobal("window", {
+    location: { origin: "http://localhost:2026" },
+    sessionStorage,
+  });
+  rs.stubGlobal("fetch", fetchFn);
+
+  const received: Array<{ id?: string; event: string; data: unknown }> = [];
+  for await (const entry of getAPIClient(true).runs.joinStream(
+    "thread-1",
+    "run-1",
+    { lastEventId: "1-0" },
+  )) {
+    received.push(entry);
+  }
+
+  expect(received).toEqual([
+    {
+      event: "custom",
+      data: { type: "stream_replay_gap", ...gap },
+    },
+    {
+      event: "values",
+      data: { messages: [{ type: "ai", content: "durable" }] },
+    },
+    { event: "end", data: null },
+  ]);
+  expect(new Headers(recoveryRequests[1]?.headers).get("Last-Event-ID")).toBe(
+    "3-0",
+  );
+});
+
+test("recovers a gap emitted by the initial run stream", async () => {
+  const sessionStorage = makeSessionStorage();
+  const gap = {
+    code: "stream_replay_gap",
+    run_id: "run-2",
+    requested_event_id: null,
+    earliest_available_event_id: "4-0",
+    latest_available_event_id: "5-0",
+    recovery: "reload_durable_state",
+  };
+  const recoveryHeaders: Headers[] = [];
+  const fetchFn = rs.fn(async (url: string | URL, init?: RequestInit) => {
+    const path = url.toString();
+    if (path.endsWith("/threads/thread-2/runs/stream")) {
+      return makeSSEResponse(`event: gap\ndata: ${JSON.stringify(gap)}\n\n`, {
+        "Content-Location": "/threads/thread-2/runs/run-2",
+      });
+    }
+    if (path.includes("/threads/thread-2/state")) {
+      return new Response(
+        JSON.stringify({
+          values: { messages: [{ type: "ai", content: "checkpoint" }] },
+        }),
+        { status: 200 },
+      );
+    }
+    if (path.includes("/runs/run-2/stream")) {
+      recoveryHeaders.push(new Headers(init?.headers));
+      return makeSSEResponse("event: end\ndata: null\n\n");
+    }
+    return new Response(JSON.stringify({ detail: "unexpected request" }), {
+      status: 500,
+    });
+  });
+  rs.stubGlobal("window", {
+    location: { origin: "http://localhost:2026" },
+    sessionStorage,
+  });
+  rs.stubGlobal("fetch", fetchFn);
+
+  const received: Array<{ id?: string; event: string; data: unknown }> = [];
+  for await (const entry of getAPIClient(true).runs.stream(
+    "thread-2",
+    "lead_agent",
+    { streamResumable: true },
+  )) {
+    received.push(entry);
+  }
+
+  expect(received).toEqual([
+    {
+      event: "custom",
+      data: { type: "stream_replay_gap", ...gap },
+    },
+    {
+      event: "values",
+      data: { messages: [{ type: "ai", content: "checkpoint" }] },
+    },
+    { event: "end", data: null },
+  ]);
+  expect(recoveryHeaders[0]?.get("Last-Event-ID")).toBe("5-0");
+});
+
+test("clears reconnect metadata when an initial stream gap resume is inactive", async () => {
+  const sessionStorage = makeSessionStorage();
+  const gap = {
+    code: "stream_replay_gap",
+    run_id: "run-inactive-resume",
+    requested_event_id: null,
+    earliest_available_event_id: "4-0",
+    latest_available_event_id: "5-0",
+    recovery: "reload_durable_state",
+  };
+  let recoveryRequests = 0;
+  const fetchFn = rs.fn(async (url: string | URL) => {
+    const path = url.toString();
+    if (path.endsWith("/threads/thread-inactive-resume/runs/stream")) {
+      return makeSSEResponse(`event: gap\ndata: ${JSON.stringify(gap)}\n\n`, {
+        "Content-Location":
+          "/threads/thread-inactive-resume/runs/run-inactive-resume",
+      });
+    }
+    if (path.includes("/threads/thread-inactive-resume/state")) {
+      return new Response(
+        JSON.stringify({
+          values: { messages: [{ type: "ai", content: "checkpoint" }] },
+        }),
+        { status: 200 },
+      );
+    }
+    if (path.includes("/runs/run-inactive-resume/stream")) {
+      recoveryRequests += 1;
+      return new Response(
+        JSON.stringify({
+          detail:
+            "Run run-inactive-resume is not active on this worker and cannot be streamed",
+        }),
+        { status: 409 },
+      );
+    }
+    return new Response(JSON.stringify({ detail: "unexpected request" }), {
+      status: 500,
+    });
+  });
+  rs.stubGlobal("window", {
+    location: { origin: "http://localhost:2026" },
+    sessionStorage,
+  });
+  rs.stubGlobal("fetch", fetchFn);
+
+  const stream = getAPIClient(true).runs.stream(
+    "thread-inactive-resume",
+    "lead_agent",
+    { streamResumable: true },
+  );
+
+  await expect(stream.next()).resolves.toMatchObject({
+    done: false,
+    value: {
+      event: "custom",
+      data: { type: "stream_replay_gap", ...gap },
+    },
+  });
+  await expect(stream.next()).resolves.toMatchObject({
+    done: false,
+    value: {
+      event: "values",
+      data: { messages: [{ type: "ai", content: "checkpoint" }] },
+    },
+  });
+  await expect(stream.next()).resolves.toMatchObject({ done: true });
+
+  expect(recoveryRequests).toBe(1);
+  expect(sessionStorage.getItem("lg:stream:thread-inactive-resume")).toBeNull();
+});
+
+test("stops after five consecutive stream gap recoveries", async () => {
+  const sessionStorage = makeSessionStorage();
+  sessionStorage.setItem("lg:stream:thread-gap-loop", "run-gap-loop");
+  let streamCalls = 0;
+  let stateCalls = 0;
+  const fetchFn = rs.fn(async (url: string | URL) => {
+    const path = url.toString();
+    if (path.endsWith("/runs/run-gap-loop")) {
+      return new Response(JSON.stringify({ status: "running" }), {
+        status: 200,
+      });
+    }
+    if (path.includes("/runs/run-gap-loop/stream")) {
+      streamCalls += 1;
+      const gap = {
+        code: "stream_replay_gap",
+        run_id: "run-gap-loop",
+        requested_event_id: `${streamCalls}-0`,
+        earliest_available_event_id: `${streamCalls + 1}-0`,
+        latest_available_event_id: `${streamCalls + 2}-0`,
+        recovery: "reload_durable_state",
+      };
+      return makeSSEResponse(`event: gap\ndata: ${JSON.stringify(gap)}\n\n`);
+    }
+    if (path.includes("/threads/thread-gap-loop/state")) {
+      stateCalls += 1;
+      return new Response(JSON.stringify({ values: { messages: [] } }), {
+        status: 200,
+      });
+    }
+    return new Response(JSON.stringify({ detail: "unexpected request" }), {
+      status: 500,
+    });
+  });
+  rs.stubGlobal("window", {
+    location: { origin: "http://localhost:2026" },
+    sessionStorage,
+  });
+  rs.stubGlobal("fetch", fetchFn);
+
+  const consume = async () => {
+    for await (const entry of getAPIClient(true).runs.joinStream(
+      "thread-gap-loop",
+      "run-gap-loop",
+      { lastEventId: "1-0" },
+    )) {
+      // Drain until the recovery budget is exhausted.
+      void entry;
+    }
+  };
+
+  await expect(consume()).rejects.toBeInstanceOf(StreamReplayGapError);
+  expect(streamCalls).toBe(6);
+  expect(stateCalls).toBe(5);
+});
+
+test("surfaces durable-state recovery failures as a structured gap error", async () => {
+  const sessionStorage = makeSessionStorage();
+  sessionStorage.setItem("lg:stream:thread-gap-state", "run-gap-state");
+  sessionStorage.setItem.mockClear();
+  const gap = {
+    code: "stream_replay_gap",
+    run_id: "run-gap-state",
+    requested_event_id: "1-0",
+    earliest_available_event_id: "2-0",
+    latest_available_event_id: "3-0",
+    recovery: "reload_durable_state",
+  };
+  const fetchFn = rs.fn(async (url: string | URL) => {
+    const path = url.toString();
+    if (path.endsWith("/runs/run-gap-state")) {
+      return new Response(JSON.stringify({ status: "running" }), {
+        status: 200,
+      });
+    }
+    if (path.includes("/runs/run-gap-state/stream")) {
+      return makeSSEResponse(`event: gap\ndata: ${JSON.stringify(gap)}\n\n`);
+    }
+    if (path.includes("/threads/thread-gap-state/state")) {
+      return new Response(JSON.stringify({ detail: "state unavailable" }), {
+        status: 404,
+      });
+    }
+    return new Response(JSON.stringify({ detail: "unexpected request" }), {
+      status: 500,
+    });
+  });
+  rs.stubGlobal("window", {
+    location: { origin: "http://localhost:2026" },
+    sessionStorage,
+  });
+  rs.stubGlobal("fetch", fetchFn);
+
+  const consume = async () => {
+    for await (const entry of getAPIClient(true).runs.joinStream(
+      "thread-gap-state",
+      "run-gap-state",
+      { lastEventId: "1-0" },
+    )) {
+      void entry;
+    }
+  };
+
+  const error = await consume().catch((cause: unknown) => cause);
+  expect(error).toBeInstanceOf(StreamReplayGapError);
+  expect(error).toMatchObject({
+    gap,
+    recoveryAttempts: 1,
+    recoveryCause: expect.any(Error),
+  });
+  expect(sessionStorage.removeItem).toHaveBeenCalledWith(
+    "lg:stream:thread-gap-state",
+  );
+  expect(sessionStorage.setItem).not.toHaveBeenCalledWith(
+    "lg:stream:thread-gap-state",
+    "run-gap-state",
+  );
+});
+
 test("short-circuits reconnect to an interrupted (user-cancelled) run", async () => {
   // Regression: interrupted is a persisted terminal status written by
   // RunManager.cancel(). Reconnecting to it must short-circuit like other

frontend/tests/unit/core/threads/stream-options.test.ts

@@ -57,6 +57,10 @@ async function captureThreadStreamOptions() {
     }),
   }));
   rs.doMock("@/core/tasks/context", () => ({
+    useSubtaskContext: () => ({
+      tasksRef: { current: {} },
+      setTasks: rs.fn(),
+    }),
     useUpdateSubtask: () => rs.fn(),
   }));