fix(streaming): avoid false replay gaps after cursor eviction (#6812)

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

README.md

@@ -591,7 +591,7 @@ Administrators can suspend and restore accounts from **Settings → Users** (adm
 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. Kubernetes replicas run one worker per Pod, which the worker count cannot see: declare them with `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`, which deploy tooling such as a Helm chart can set from its replica count) so the startup gate enforces the same prerequisites instead of staying inert. Declared instances that store credentials (`channel_connections`) must also share one at-rest `DEER_FLOW_CREDENTIALS_KEY` (the Helm chart and `make up` generate it; back it up, see [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)). 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. An already-observed Redis stream removed during replay also signals a gap, so slow subscribers reload durable state instead of heartbeating forever after cleanup. 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, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. 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. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); IM bot credentials entered in the web UI live in the shared `channel_runtime_configs` table, encrypted with `DEER_FLOW_CREDENTIALS_KEY` (an existing `channels/runtime-config.json` is imported once and then deleted), so a save on any instance reaches the provider's connection within one lease renewal; each provider's platform connection is held by one instance through the shared `channel_leases` table — `channels.leader_lease` (default `auto`, on wherever the multi-process signal is) hands it to another instance when the holder dies, and the startup gate refuses `enabled: false` while an IM channel other than GitHub is enabled (see [IM channel leader lease](backend/docs/CONFIGURATION.md#im-channel-leader-lease-channelsleader_lease)); follow-up buffers still need their own multi-worker coordination. To try this topology on one machine, `make dev-multi` starts two Gateways on shared Postgres, Redis and `DEER_FLOW_HOME`, and `make dev-multi-check` runs the cross-instance checks; see the local two-Gateway harness in [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#local-two-gateway-harness).
+> 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. Kubernetes replicas run one worker per Pod, which the worker count cannot see: declare them with `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`, which deploy tooling such as a Helm chart can set from its replica count) so the startup gate enforces the same prerequisites instead of staying inert. Declared instances that store credentials (`channel_connections`) must also share one at-rest `DEER_FLOW_CREDENTIALS_KEY` (the Helm chart and `make up` generate it; back it up, see [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)). The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a trimmed reconnect cursor cannot be verified for complete replay, 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. The Memory bridge can replay without recovery when the cursor is its exact last evicted event: every successor is still retained. Older or unverifiable evicted cursors still trigger recovery. An already-observed Redis stream removed during replay also signals a gap, so slow subscribers reload durable state instead of heartbeating forever after cleanup. 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, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. 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. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); IM bot credentials entered in the web UI live in the shared `channel_runtime_configs` table, encrypted with `DEER_FLOW_CREDENTIALS_KEY` (an existing `channels/runtime-config.json` is imported once and then deleted), so a save on any instance reaches the provider's connection within one lease renewal; each provider's platform connection is held by one instance through the shared `channel_leases` table — `channels.leader_lease` (default `auto`, on wherever the multi-process signal is) hands it to another instance when the holder dies, and the startup gate refuses `enabled: false` while an IM channel other than GitHub is enabled (see [IM channel leader lease](backend/docs/CONFIGURATION.md#im-channel-leader-lease-channelsleader_lease)); follow-up buffers still need their own multi-worker coordination. To try this topology on one machine, `make dev-multi` starts two Gateways on shared Postgres, Redis and `DEER_FLOW_HOME`, and `make dev-multi-check` runs the cross-instance checks; see the local two-Gateway harness in [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#local-two-gateway-harness).
 >
 > With run ownership heartbeats enabled, terminal success requires acknowledgment
 > from that worker's own durable write. A matching database row alone cannot

backend/docs/API.md

@@ -1956,7 +1956,13 @@ 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
+The Memory bridge also remembers its exact last evicted event ID. Reconnecting
+from that known predecessor replays every retained successor without a `gap`,
+because none of the events after the cursor has been lost. This uses one ID per
+run and does not increase the retained event count. Redis keeps its conservative
+trimmed-cursor policy.
+
+For other syntactically valid cursors older than the retained watermark, the
 server sends exactly one `gap` event before any retained data and closes that
 subscription without an `end` event:
 

backend/packages/harness/deerflow/runtime/AGENTS.md

@@ -29,6 +29,15 @@ Memory/Redis bridges keep startup-only `stream_bridge.heartbeat_interval_seconds
 
 Reconnect cursor parsing requires ASCII digits and must treat integer-conversion failures as malformed input. Redis cursors additionally require unsigned 64-bit components; use the same parser before sending a cursor to `XREAD` and when detecting replay gaps. Preserve malformed-cursor policy: memory replays retained events, Redis tails active streams and replays ended streams through `end`. `tests/test_stream_bridge.py` directly tests both parsers offline, including oversized sequences, Redis component overflow, non-ASCII digits, and valid uint64 boundaries; integration tests also exercise malformed cursors against real Redis.
 
+### Memory Stream Replay
+
+Memory keeps the exact last evicted event ID as bounded per-run metadata. A
+reconnect from that known predecessor replays all retained successors and END,
+matching a live subscriber at the same position. Older or unverifiable evicted
+cursors still emit `StreamGap`; Redis replay policy is unchanged. Coverage in
+`tests/test_memory_stream_reconnect_boundary.py` includes the Gateway SSE consumer
+with real ASGI requests and LangGraph SDK decoding.
+
 ### Checkpoint Channel Modes (`full` / `delta`)
 
 Checkpointer storage runs in one of two channel modes, selected by `checkpoint_channel_mode` in `config.yaml` (default `full`). `delta` mode adopts LangGraph 1.2's `DeltaChannel` for `messages`: checkpoints store a sentinel + per-step writes instead of the full message list, so storage/serde grows O(N) instead of O(N²) in turns. All checkpointer backends (memory/sqlite/postgres) serve both modes unchanged — the semantics live in the compiled graph's channel table, not in the saver.

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

@@ -22,6 +22,7 @@ class _RunStream:
     condition: asyncio.Condition = field(default_factory=asyncio.Condition)
     ended: bool = False
     start_offset: int = 0
+    last_evicted_event_id: str | None = None
     retired: bool = False
 
 
@@ -83,14 +84,15 @@ class MemoryStreamBridge(StreamBridge):
             return StreamGap(last_event_id, None, None)
         if last_event_id is None:
             return stream.start_offset
+        if last_event_id == stream.last_evicted_event_id:
+            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. 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
+        # the computed index. The exact last evicted id is a known predecessor
+        # whose successors are all retained. Other ids below the watermark cannot
+        # be verified and take the conservative gap path. 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:
@@ -120,6 +122,7 @@ class MemoryStreamBridge(StreamBridge):
             stream.events.append(entry)
             if len(stream.events) > self._maxsize:
                 overflow = len(stream.events) - self._maxsize
+                stream.last_evicted_event_id = stream.events[overflow - 1].id
                 del stream.events[:overflow]
                 stream.start_offset += overflow
             stream.condition.notify_all()

backend/tests/test_gateway_services.py

@@ -106,6 +106,7 @@ async def test_sse_consumer_emits_gap_without_cancelling_run():
     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})
+    await bridge.publish(record.run_id, "event-4", {"step": 4})
     retained = bridge._streams[record.run_id].events
 
     worker_started = asyncio.Event()

backend/tests/test_memory_stream_reconnect_boundary.py

@@ -0,0 +1,135 @@
+import asyncio
+
+import httpx
+import pytest
+from fastapi import FastAPI, Request
+from langgraph_sdk.sse import SSEDecoder, aiter_lines_raw
+from starlette.responses import StreamingResponse
+
+from deerflow.runtime.stream_bridge import END_SENTINEL, MemoryStreamBridge, StreamEvent, StreamGap
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("capacity", [1, 2, 256])
+async def test_reconnect_replays_all_successors_of_last_evicted_event(capacity):
+    bridge = MemoryStreamBridge(queue_maxsize=capacity)
+    run_id = "complete-reconnect"
+    await bridge.publish(run_id, "messages", {"index": 0})
+    live = bridge.subscribe(run_id)
+    try:
+        cursor = await anext(live)
+        assert isinstance(cursor, StreamEvent)
+        for index in range(1, capacity + 1):
+            await bridge.publish(run_id, "messages", {"index": index})
+        await bridge.publish_end(run_id)
+
+        live_events = [entry async for entry in live]
+        reconnect = bridge.subscribe(run_id, last_event_id=cursor.id)
+        try:
+            replayed = await asyncio.wait_for(anext(reconnect), timeout=1)
+            assert replayed == live_events[0]
+            remaining = [entry async for entry in reconnect]
+            assert [replayed, *remaining] == live_events
+        finally:
+            await reconnect.aclose()
+
+        assert [entry.data for entry in live_events[:-1]] == [{"index": index} for index in range(1, capacity + 1)]
+        assert live_events[-1] is END_SENTINEL
+    finally:
+        await live.aclose()
+        await bridge.close()
+
+
+@pytest.mark.asyncio
+async def test_reconnect_reports_gap_when_a_successor_was_evicted():
+    bridge = MemoryStreamBridge(queue_maxsize=2)
+    run_id = "incomplete-reconnect"
+    await bridge.publish(run_id, "messages", {"index": 0})
+    live = bridge.subscribe(run_id)
+    try:
+        cursor = await anext(live)
+        for index in range(1, 4):
+            await bridge.publish(run_id, "messages", {"index": index})
+        await bridge.publish_end(run_id)
+        live_events = [entry async for entry in live]
+        replayed = [entry async for entry in bridge.subscribe(run_id, last_event_id=cursor.id)]
+        assert len(replayed) == 1
+        assert isinstance(replayed[0], StreamGap)
+        assert replayed == live_events
+        assert replayed[0].requested_event_id == cursor.id
+    finally:
+        await live.aclose()
+        await bridge.close()
+
+
+@pytest.mark.asyncio
+async def test_foreign_cursor_with_same_evicted_sequence_still_reports_gap():
+    bridge = MemoryStreamBridge(queue_maxsize=1)
+    run_id = "foreign-reconnect"
+    try:
+        await bridge.publish(run_id, "messages", {"index": 0})
+        stream = bridge.subscribe(run_id)
+        try:
+            cursor = await anext(stream)
+        finally:
+            await stream.aclose()
+        timestamp, sequence = cursor.id.split("-")
+        foreign_cursor = f"{int(timestamp) + 1}-{sequence}"
+        await bridge.publish(run_id, "messages", {"index": 1})
+        await bridge.publish_end(run_id)
+        replayed = [entry async for entry in bridge.subscribe(run_id, last_event_id=foreign_cursor)]
+        assert len(replayed) == 1
+        assert isinstance(replayed[0], StreamGap)
+        assert replayed[0].requested_event_id == foreign_cursor
+    finally:
+        await bridge.close()
+
+
+@pytest.mark.asyncio
+async def test_gateway_sse_reconnect_preserves_complete_retained_output():
+    from langchain_core.messages import AIMessageChunk
+
+    from app.gateway.services import sse_consumer
+    from deerflow.runtime import RunManager, RunStatus
+    from deerflow.runtime.serialization import serialize
+
+    bridge = MemoryStreamBridge(queue_maxsize=2)
+    manager = RunManager()
+    record = await manager.create("complete-sse-reconnect")
+    await manager.set_status(record.run_id, RunStatus.success)
+    app = FastAPI()
+
+    @app.get("/stream")
+    async def stream(request: Request):
+        return StreamingResponse(sse_consumer(bridge, record, request, manager, apply_on_disconnect=False), media_type="text/event-stream")
+
+    try:
+        payloads = [serialize((AIMessageChunk(content=f"part-{index}", id="answer"), {"langgraph_node": "model"}), mode="messages") for index in range(3)]
+        await bridge.publish(record.run_id, "messages", payloads[0])
+        initial = bridge.subscribe(record.run_id)
+        try:
+            cursor = await anext(initial)
+        finally:
+            await initial.aclose()
+        for index in (1, 2):
+            await bridge.publish(record.run_id, "messages", payloads[index])
+        await bridge.publish_end(record.run_id)
+
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            response = await client.get("/stream", headers={"Last-Event-ID": cursor.id})
+        assert response.status_code == 200
+        decoder = SSEDecoder()
+        frames = []
+        async for line in aiter_lines_raw(response):
+            decoded = decoder.decode(line)
+            if decoded is not None:
+                frames.append(decoded)
+        assert [(frame.event, frame.data) for frame in frames] == [
+            ("messages", payloads[1]),
+            ("messages", payloads[2]),
+            ("end", None),
+        ]
+        assert [frame.id for frame in frames[:-1]] == [entry.id for entry in bridge._streams[record.run_id].events]
+    finally:
+        await manager.shutdown()
+        await bridge.close()

backend/tests/test_memory_stream_retirement.py

@@ -57,11 +57,13 @@ async def _attach_behind_retirement(bridge, run_id, stream, operation, monkeypat
 @pytest.mark.parametrize("operation", ["cleanup", "close"])
 @pytest.mark.parametrize("ended", [False, True])
 @pytest.mark.parametrize("with_cursor", [False, True])
-async def test_attachment_queued_behind_retirement_reports_gap(monkeypatch, operation, ended, with_cursor):
-    bridge = MemoryStreamBridge()
+@pytest.mark.parametrize("capacity", [1, 256])
+async def test_attachment_queued_behind_retirement_reports_gap(monkeypatch, operation, ended, with_cursor, capacity):
+    bridge = MemoryStreamBridge(queue_maxsize=capacity)
     await bridge.publish("attach-window", "messages", "earlier")
-    await bridge.publish("attach-window", "messages", "unread")
     cursor = bridge._streams["attach-window"].events[0].id if with_cursor else None
+    # With capacity one, this makes the cursor the exact last evicted event.
+    await bridge.publish("attach-window", "messages", "unread")
     if ended:
         await bridge.publish_end("attach-window")
     stream = bridge.subscribe("attach-window", last_event_id=cursor)

backend/tests/test_stream_bridge.py

@@ -384,6 +384,7 @@ async def test_evicted_last_event_id_yields_gap_before_partial_replay():
     stream = bridge._streams[run_id]
     e1_id = stream.events[0].id
     await bridge.publish(run_id, "e3", {"step": 3})  # trims e1
+    await bridge.publish(run_id, "e4", {"step": 4})  # trims the unread e2
     await bridge.publish_end(run_id)
 
     received = []
@@ -1199,11 +1200,11 @@ 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[6:]]
+    candidates = [None, "garbage", "1718000000000-x", "999999-999999", foreign_id, *ids[5:]]
     for eid in candidates:
         assert bridge._resolve_start_offset(stream, eid) == _linear_resolve(stream, eid), eid
 
-    for evicted_id in ids[:6]:
+    for evicted_id in ids[:5]:
         assert bridge._resolve_start_offset(stream, evicted_id) == StreamGap(
             requested_event_id=evicted_id,
             earliest_available_event_id=stream.events[0].id,