diff --git a/README.md b/README.md index 708230039..b081c2102 100644 --- a/README.md +++ b/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. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 641f7a0ba..55d6f66da 100644 --- a/backend/AGENTS.md +++ b/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. diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index a6d7dab2b..58e767212 100644 --- a/backend/app/gateway/services.py +++ b/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: diff --git a/backend/docs/API.md b/backend/docs/API.md index b23ddeb9d..58295ddc4 100644 --- a/backend/docs/API.md +++ b/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 diff --git a/backend/docs/STREAMING.md b/backend/docs/STREAMING.md index 5b748ce78..809aee166 100644 --- a/backend/docs/STREAMING.md +++ b/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 diff --git a/backend/packages/harness/deerflow/runtime/__init__.py b/backend/packages/harness/deerflow/runtime/__init__.py index c3c4dfb6c..8a65f38df 100644 --- a/backend/packages/harness/deerflow/runtime/__init__.py +++ b/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", ] diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/__init__.py b/backend/packages/harness/deerflow/runtime/stream_bridge/__init__.py index a9ec58f63..eafbef950 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/__init__.py +++ b/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", ] diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/base.py b/backend/packages/harness/deerflow/runtime/stream_bridge/base.py index bcaf65a66..b303a1d9b 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/base.py +++ b/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 diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py b/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py index 2690c8f4c..f9f3cf304 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py +++ b/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: diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py b/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py index c1a488a9d..1db1a3193 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py +++ b/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: diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index 457bb5059..a9bd9b44f 100644 --- a/backend/tests/test_gateway_services.py +++ b/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 diff --git a/backend/tests/test_stream_bridge.py b/backend/tests/test_stream_bridge.py index b5c413e53..ebf037891 100644 --- a/backend/tests/test_stream_bridge.py +++ b/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 diff --git a/backend/tests/test_wait_disconnect_handling.py b/backend/tests/test_wait_disconnect_handling.py index dcfec4a03..011744de7 100644 --- a/backend/tests/test_wait_disconnect_handling.py +++ b/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 diff --git a/config.example.yaml b/config.example.yaml index 476e82123..1282788e1 100644 --- a/config.example.yaml +++ b/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. diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 1040ceed8..9906415b1 100644 --- a/frontend/AGENTS.md +++ b/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. diff --git a/frontend/src/core/api/api-client.ts b/frontend/src/core/api/api-client.ts index 5a4370591..f557edcee 100644 --- a/frontend/src/core/api/api-client.ts +++ b/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; + 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; + resume: (runId: string, lastEventId: string) => AsyncIterable; +}): AsyncGenerator { + 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; +}): AsyncGenerator { + 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; diff --git a/frontend/src/core/i18n/locales/en-US.ts b/frontend/src/core/i18n/locales/en-US.ts index 42de3ac12..dd00518d6 100644 --- a/frontend/src/core/i18n/locales/en-US.ts +++ b/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 diff --git a/frontend/src/core/i18n/locales/types.ts b/frontend/src/core/i18n/locales/types.ts index 7d5e3ece1..b037f348e 100644 --- a/frontend/src/core/i18n/locales/types.ts +++ b/frontend/src/core/i18n/locales/types.ts @@ -405,6 +405,7 @@ export interface Translations { startConversation: string; branchCreated: string; branchFailed: string; + streamReplayGap: string; }; // Chats diff --git a/frontend/src/core/i18n/locales/zh-CN.ts b/frontend/src/core/i18n/locales/zh-CN.ts index 4544b9429..48fcd7713 100644 --- a/frontend/src/core/i18n/locales/zh-CN.ts +++ b/frontend/src/core/i18n/locales/zh-CN.ts @@ -483,6 +483,7 @@ export const zhCN: Translations = { startConversation: "开始新的对话以查看消息", branchCreated: "已创建分叉对话", branchFailed: "创建分叉对话失败。", + streamReplayGap: "部分实时更新已过期，已从持久化状态恢复对话。", }, // Chats diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index 39225040e..92825368a 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/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({ @@ -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(); + 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); diff --git a/frontend/tests/unit/core/api/api-client.test.ts b/frontend/tests/unit/core/api/api-client.test.ts index 0d6f39afd..e494569c6 100644 --- a/frontend/tests/unit/core/api/api-client.test.ts +++ b/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 diff --git a/frontend/tests/unit/core/threads/stream-options.test.ts b/frontend/tests/unit/core/threads/stream-options.test.ts index da9f8adfa..cc089cb19 100644 --- a/frontend/tests/unit/core/threads/stream-options.test.ts +++ b/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(), }));