diff --git a/backend/docs/API.md b/backend/docs/API.md index dcb3e896a9..f5ea99e825 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -846,13 +846,16 @@ data: {"code":"stream_replay_gap","run_id":"run-123","requested_event_id":"17180 ``` -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. +The frame deliberately has no SSE `id:`. Both `earliest_available_event_id` and +`latest_available_event_id` are `string | null` (they are `null` when no events +are retained in the buffer). Consumers must reload durable thread state and +persisted run events/messages, then may reconnect from `latest_available_event_id` +to follow newer live events, or rejoin without a cursor when the buffer is empty +(`latest_available_event_id` is `null`). 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. --- diff --git a/backend/docs/STREAMING.md b/backend/docs/STREAMING.md index 809aee1665..f4c478f053 100644 --- a/backend/docs/STREAMING.md +++ b/backend/docs/STREAMING.md @@ -159,7 +159,7 @@ data: {"code":"stream_replay_gap","run_id":"...","requested_event_id":"...","ear ``` -`gap` 帧没有 SSE `id:`，后面也没有正常 `end`；当前订阅随即关闭。它是恢复边界而不是客户端断开，因此不会触发 `on_disconnect=cancel`。客户端必须丢弃不再可信的瞬时状态，重新读取 thread checkpoint 和持久化 run-event/message history，再以 `latest_available_event_id` 为游标跟随新事件。DeerFlow Web UI 自动执行此流程并最多连续恢复五次。 +`gap` 帧没有 SSE `id:`，后面也没有正常 `end`；当前订阅随即关闭。它是恢复边界而不是客户端断开，因此不会触发 `on_disconnect=cancel`。`earliest_available_event_id` 与 `latest_available_event_id` 在缓冲区无保留事件时为 `null`。客户端必须丢弃不再可信的瞬时状态，重新读取 thread checkpoint 和持久化 run-event/message history，再以 `latest_available_event_id` 为游标跟随新事件（若缓冲区为空即 `latest_available_event_id` 为 `null`，则不带游标重新加入流）。DeerFlow Web UI 自动执行此流程并最多连续恢复五次。 Redis 对无游标、空 stream 上已经建立的阻塞等待也遵循相同契约：第一次 `XREAD` 唤醒的数据在交付前仍是 provisional baseline，bridge 会用下一次事务快照确认其尾 ID 仍在保留窗口。若生产者已经裁剪了该基线，订阅直接返回 `requested_event_id: null` 的 `gap`，不会先交付 retained tail。这个检查有明确的性能代价：每轮订阅需要一个包含 `XRANGE`、`XREVRANGE`、非阻塞 `XREAD` 的事务快照；空闲时还需要单独的阻塞 `XREAD` 来唤醒。 diff --git a/backend/packages/harness/deerflow/config/stream_bridge_config.py b/backend/packages/harness/deerflow/config/stream_bridge_config.py index 6ba9535d66..7e2fe8b991 100644 --- a/backend/packages/harness/deerflow/config/stream_bridge_config.py +++ b/backend/packages/harness/deerflow/config/stream_bridge_config.py @@ -20,6 +20,7 @@ class StreamBridgeConfig(BaseModel): ) queue_maxsize: int = Field( default=256, + ge=1, description="Maximum number of events retained per run (memory bridge queue size / redis stream MAXLEN).", ) max_connections: int | None = Field( diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/base.py b/backend/packages/harness/deerflow/runtime/stream_bridge/base.py index b303a1d9b8..fa16cc1e81 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/base.py +++ b/backend/packages/harness/deerflow/runtime/stream_bridge/base.py @@ -37,12 +37,13 @@ class StreamGap: ``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. + without mistaking a partial replay for a complete one. When nothing is + retained in the buffer, the bounds are ``None``. """ requested_event_id: str | None - earliest_available_event_id: str - latest_available_event_id: str + earliest_available_event_id: str | None + latest_available_event_id: str | None HEARTBEAT_SENTINEL = StreamEvent(id="", event="__heartbeat__", data=None) diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py b/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py index f9f3cf3046..5f2701b580 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py +++ b/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py @@ -32,7 +32,7 @@ class MemoryStreamBridge(StreamBridge): """ def __init__(self, *, queue_maxsize: int = 256) -> None: - self._maxsize = queue_maxsize + self._maxsize = max(1, queue_maxsize) self._streams: dict[str, _RunStream] = {} self._counters: dict[str, int] = {} @@ -67,8 +67,8 @@ class MemoryStreamBridge(StreamBridge): 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, + earliest_available_event_id=stream.events[0].id if stream.events else None, + latest_available_event_id=stream.events[-1].id if stream.events else None, ) def _resolve_start_offset(self, stream: _RunStream, last_event_id: str | None) -> int | StreamGap: @@ -85,7 +85,7 @@ class MemoryStreamBridge(StreamBridge): # 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: + if 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: diff --git a/backend/tests/test_stream_bridge.py b/backend/tests/test_stream_bridge.py index ebf0378914..c05d6bd8c9 100644 --- a/backend/tests/test_stream_bridge.py +++ b/backend/tests/test_stream_bridge.py @@ -8,6 +8,7 @@ from collections import defaultdict import anyio import pytest +from pydantic import ValidationError from deerflow.config.stream_bridge_config import StreamBridgeConfig, set_stream_bridge_config from deerflow.runtime import END_SENTINEL, HEARTBEAT_SENTINEL, MemoryStreamBridge, StreamGap, make_stream_bridge @@ -1072,6 +1073,64 @@ async def test_memory_malformed_last_event_id_is_not_reported_as_gap(): assert all(not isinstance(entry, StreamGap) for entry in received) +@pytest.mark.anyio +async def test_memory_make_gap_handles_empty_events_buffer(): + """_make_gap returns StreamGap with None bounds when events deque is empty.""" + bridge = MemoryStreamBridge(queue_maxsize=2) + run_id = "run-empty-stream-gap" + stream = bridge._get_or_create_stream(run_id) + assert len(stream.events) == 0 + + gap = bridge._make_gap(stream, "100-0") + assert gap == StreamGap( + requested_event_id="100-0", + earliest_available_event_id=None, + latest_available_event_id=None, + ) + + +def test_memory_stream_bridge_clamps_queue_maxsize(): + """queue_maxsize <= 0 is clamped to at least 1.""" + bridge = MemoryStreamBridge(queue_maxsize=0) + assert bridge._maxsize == 1 + bridge_neg = MemoryStreamBridge(queue_maxsize=-5) + assert bridge_neg._maxsize == 1 + + +def test_stream_bridge_config_validates_queue_maxsize(): + """StreamBridgeConfig rejects queue_maxsize < 1.""" + with pytest.raises(ValidationError): + StreamBridgeConfig(queue_maxsize=0) + with pytest.raises(ValidationError): + StreamBridgeConfig(queue_maxsize=-1) + + +@pytest.mark.anyio +async def test_memory_subscribe_replays_gap_when_buffer_empty_and_cursor_behind_watermark(): + """Subscribing with a cursor behind start_offset when buffer is empty yields StreamGap with None bounds.""" + bridge = MemoryStreamBridge(queue_maxsize=2) + run_id = "run-empty-retained-gap" + stream = bridge._get_or_create_stream(run_id) + # Simulate a stream whose retained buffer was cleared / advanced + stream.start_offset = 10 + stream.events.clear() + + received = [ + entry + async for entry in bridge.subscribe( + run_id, + last_event_id="5-0", + heartbeat_interval=0.1, + ) + ] + assert len(received) == 1 + assert received[0] == StreamGap( + requested_event_id="5-0", + earliest_available_event_id=None, + latest_available_event_id=None, + ) + + @pytest.mark.anyio async def test_make_stream_bridge_uses_docker_redis_env(monkeypatch): """Docker can enable Redis bridge without editing config.yaml.""" diff --git a/config.example.yaml b/config.example.yaml index 480ac3d36e..4b8fe52317 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -15,7 +15,7 @@ # ============================================================================ # Bump this number when the config schema changes. # Run `make config-upgrade` to merge new fields into your local config.yaml. -config_version: 36 +config_version: 37 # ============================================================================ # Logging @@ -2330,13 +2330,13 @@ run_ownership: # # stream_bridge: # type: memory # single-process only -# queue_maxsize: 256 # data events retained per run. A reconnect +# queue_maxsize: 256 # data events retained per run (min: 1). 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 # data events retained per run (Redis MAXLEN). +# queue_maxsize: 256 # data events retained per run (Redis MAXLEN, min: 1). # # Older valid cursors receive SSE `gap`. # stream_ttl_seconds: 86400 # rolling TTL for retained stream buffers. # # Refreshed on each publish/publish_end; set 0 diff --git a/deploy/helm/deer-flow/README.md b/deploy/helm/deer-flow/README.md index 684251ae72..3787f1f4a4 100644 --- a/deploy/helm/deer-flow/README.md +++ b/deploy/helm/deer-flow/README.md @@ -124,7 +124,7 @@ they resolve from the `secrets` map): ```yaml config: | - config_version: 36 + config_version: 37 models: - name: gpt-4 use: langchain_openai:ChatOpenAI diff --git a/deploy/helm/deer-flow/values.yaml b/deploy/helm/deer-flow/values.yaml index 3e6d752d68..e137816c2e 100644 --- a/deploy/helm/deer-flow/values.yaml +++ b/deploy/helm/deer-flow/values.yaml @@ -243,7 +243,7 @@ ingress: # -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never # inline literal secret values here. The default enables provisioner sandbox. config: | - config_version: 36 + config_version: 37 log_level: info models: [] diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index cb64249054..fd27ed07a9 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -97,7 +97,7 @@ Edit-and-rerun is deliberately latest-turn-only. `core/messages/utils.ts::getLat - **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 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. +- **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 resumes after the server-provided retained tail when one exists (or rejoins without a cursor if the buffer is empty), 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. - Citation links in message and artifact Markdown must derive their `citation:` label from the full `ReactNode` children tree, since Streamdown may provide element or array children during streaming rather than a plain string. - **Environment validation** uses `@t3-oss/env-nextjs` with Zod schemas (`src/env.js`). Skip with `SKIP_ENV_VALIDATION=1` diff --git a/frontend/src/core/api/api-client.ts b/frontend/src/core/api/api-client.ts index f557edcee2..26a2618372 100644 --- a/frontend/src/core/api/api-client.ts +++ b/frontend/src/core/api/api-client.ts @@ -75,8 +75,8 @@ 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; + earliest_available_event_id: string | null; + latest_available_event_id: string | null; recovery: "reload_durable_state"; }; @@ -93,7 +93,7 @@ export class StreamReplayGapError extends Error { 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})`, + `Unable to recover SSE history after ${recoveryAttempts} attempts (requested ${gap.requested_event_id ?? "initial stream"}, earliest ${gap.earliest_available_event_id ?? "none"})`, ); this.name = "StreamReplayGapError"; } @@ -106,12 +106,16 @@ function parseStreamReplayGap(data: unknown): StreamReplayGapData { const value = data as Record; const requestedEventId = value.requested_event_id; + const earliestAvailableEventId = value.earliest_available_event_id; + const latestAvailableEventId = value.latest_available_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" || + (earliestAvailableEventId !== null && + typeof earliestAvailableEventId !== "string") || + (latestAvailableEventId !== null && + typeof latestAvailableEventId !== "string") || value.recovery !== "reload_durable_state" ) { throw new Error("Invalid stream replay gap payload."); @@ -244,7 +248,7 @@ async function* recoverStreamReplayGaps({ threadId: string | null | undefined; expectedRunId: () => string | undefined; initialStream: AsyncIterable; - resume: (runId: string, lastEventId: string) => AsyncIterable; + resume: (runId: string, lastEventId?: string) => AsyncIterable; }): AsyncGenerator { let stream = initialStream; let recoveryAttempts = 0; @@ -276,8 +280,8 @@ async function* recoverStreamReplayGaps({ // 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. + // durable values, then resume after the retained tail when it exists + // (or rejoin without a cursor if the buffer is empty). clearReconnectRun(threadId, runId); yield { event: "custom", @@ -294,7 +298,7 @@ async function* recoverStreamReplayGaps({ } rememberReconnectRun(threadId, runId); - stream = resume(runId, gap.latest_available_event_id); + stream = resume(runId, gap.latest_available_event_id ?? undefined); } } diff --git a/frontend/tests/unit/core/api/api-client.test.ts b/frontend/tests/unit/core/api/api-client.test.ts index e494569c6f..17e301de29 100644 --- a/frontend/tests/unit/core/api/api-client.test.ts +++ b/frontend/tests/unit/core/api/api-client.test.ts @@ -450,6 +450,70 @@ test("recovers a gap emitted by the initial run stream", async () => { expect(recoveryHeaders[0]?.get("Last-Event-ID")).toBe("5-0"); }); +test("recovers from stream replay gap with null retained bounds", async () => { + const sessionStorage = makeSessionStorage(); + const gap = { + code: "stream_replay_gap", + run_id: "run-null-bounds", + requested_event_id: "1-0", + earliest_available_event_id: null, + latest_available_event_id: null, + recovery: "reload_durable_state", + }; + let recoveryLastEventId: string | null = "sentinel"; + const fetchFn = rs.fn(async (url: string | URL, init?: RequestInit) => { + const path = url.toString(); + if (path.endsWith("/threads/thread-null/runs/stream")) { + return makeSSEResponse(`event: gap\ndata: ${JSON.stringify(gap)}\n\n`, { + "Content-Location": "/threads/thread-null/runs/run-null-bounds", + }); + } + if (path.includes("/threads/thread-null/state")) { + return new Response( + JSON.stringify({ + values: { messages: [{ type: "ai", content: "checkpoint" }] }, + }), + { status: 200 }, + ); + } + if (path.includes("/runs/run-null-bounds/stream")) { + const headers = new Headers(init?.headers); + recoveryLastEventId = headers.get("Last-Event-ID"); + 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-null", + "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(recoveryLastEventId).toBeNull(); +}); + test("clears reconnect metadata when an initial stream gap resume is inactive", async () => { const sessionStorage = makeSessionStorage(); const gap = {