fix(streaming): report gaps when Redis retention disappears (#6951)

Co-authored-by: ranjianghong526-hash <228174768+ranjianghong526-hash@users.noreply.github.com>

README.md

@@ -567,7 +567,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. 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`); 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 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`); 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 local two-Gateway harness disables IM providers under both `channels` and `channel_connections` by default. `DEERFLOW_MI_KEEP_CHANNELS=1` preserves them; `channels.leader_lease` stays as configured.
 >

backend/app/gateway/services.py

@@ -2776,6 +2776,11 @@ async def wait_for_run_completion(
                     # 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.
+                    if entry.latest_available_event_id is None:
+                        current = await run_mgr.get(record.run_id, user_id=record.user_id) if record.store_only else record
+                        if current is not None and await _terminal_record_stream_missing(bridge, current):
+                            completed = True
+                            return True
                     resume_from_event_id = entry.latest_available_event_id
                     gap_seen = True
                     break

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

@@ -7,6 +7,8 @@ to sibling namespaces. Keep the cursor loop and outage/TTL fallback unchanged.
 
 ### Stream Bridge Heartbeats
 
+Observed Redis streams removed during replay yield `StreamGap` with empty bounds; initially empty streams still wait. After this gap, `/wait` refreshes store-only status and completes only for a terminal run with no retained stream.
+
 Memory/Redis bridges keep startup-only `stream_bridge.heartbeat_interval_seconds`; explicit `subscribe(..., heartbeat_interval=...)` overrides it. Provider contexts drain owned cache/bridge/checkpointer/Store teardown across cancellation. `close_agent_stream()` shields close to completion: preserve counts, defer host cancellation, balance repeats, keep active errors, log attached close failures at callers, map close cancellation to failure, never time out.
 
 ### Checkpoint Channel Modes (`full` / `delta`)

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

@@ -252,6 +252,7 @@ class RedisStreamBridge(StreamBridge):
         pending_initial_response: list[Any] | None = None
         block_ms = max(1, int(heartbeat_interval * 1000)) if heartbeat_interval > 0 else 1
         consecutive_errors = 0
+        stream_observed = False
 
         while True:
             snapshot_stream_id = stream_id
@@ -302,6 +303,19 @@ class RedisStreamBridge(StreamBridge):
                 if response:
                     consecutive_errors = 0
 
+            if earliest_entries:
+                stream_observed = True
+            elif stream_observed or pending_initial_response is not None:
+                # Cleanup/expiry can remove unread events and END while a slow
+                # subscriber drains its fetched batch. An absent observed stream
+                # is lost history, not an idle stream waiting for its producer.
+                yield StreamGap(
+                    requested_event_id=stream_id if pending_initial_response is None else None,
+                    earliest_available_event_id=None,
+                    latest_available_event_id=None,
+                )
+                return
+
             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):

backend/tests/test_redis_stream_cleanup_gap.py

@@ -0,0 +1,256 @@
+"""An established subscriber must not idle forever after retained data is deleted."""
+
+import asyncio
+import json
+
+import anyio
+import pytest
+from test_stream_bridge import REDIS_TEST_URL, _DelayedBlockingReadRedis, _FakeRedis, requires_redis
+
+from deerflow.runtime import RunManager, RunStatus, StreamGap
+from deerflow.runtime.stream_bridge.base import END_SENTINEL, HEARTBEAT_SENTINEL
+from deerflow.runtime.stream_bridge.redis import RedisStreamBridge
+
+
+class _ConnectedRequest:
+    headers = {}
+
+    async def is_disconnected(self):
+        return False
+
+
+@pytest.fixture
+async def cleanup_bridge(request):
+    real = request.param == "redis"
+    bridge = RedisStreamBridge(
+        redis_url=REDIS_TEST_URL if real else "redis://fake",
+        key_prefix="test-cleanup-gap",
+        heartbeat_interval=0.01,
+        client=None if real else _FakeRedis(),
+    )
+    try:
+        yield bridge
+    finally:
+        await bridge.cleanup("cleanup-gap")
+        await bridge.close()
+
+
+BACKENDS = ["fake", pytest.param("redis", marks=[pytest.mark.integration, requires_redis])]
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("cleanup_bridge", BACKENDS, indirect=True)
+@pytest.mark.parametrize("consumed", [1, 64, 65])
+async def test_deleted_observed_stream_reports_gap(cleanup_bridge, consumed):
+    bridge = cleanup_bridge
+    for index in range(130):
+        await bridge.publish("cleanup-gap", "messages", {"index": index})
+    await bridge.publish_end("cleanup-gap")
+    stream = bridge.subscribe("cleanup-gap")
+    try:
+        cursor = None
+        for _ in range(consumed):
+            cursor = await anext(stream)
+        await bridge.cleanup("cleanup-gap")
+        # The already-fetched batch may drain, but the absent next batch is a gap.
+        with anyio.fail_after(2):
+            while True:
+                entry = await anext(stream)
+                assert entry is not HEARTBEAT_SENTINEL
+                assert entry is not END_SENTINEL
+                if isinstance(entry, StreamGap):
+                    break
+                cursor = entry
+        assert entry == StreamGap(cursor.id, None, None)
+        with pytest.raises(StopAsyncIteration):
+            await anext(stream)
+    finally:
+        await stream.aclose()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("cleanup_bridge", BACKENDS, indirect=True)
+async def test_unpublished_stream_still_waits_for_its_producer(cleanup_bridge):
+    bridge = cleanup_bridge
+    stream = bridge.subscribe("cleanup-gap")
+    try:
+        with anyio.fail_after(2):
+            assert await anext(stream) is HEARTBEAT_SENTINEL
+            await bridge.publish("cleanup-gap", "messages", {"index": 0})
+            await bridge.publish_end("cleanup-gap")
+            entries = [entry async for entry in stream]
+        assert entries[0].data == {"index": 0}
+        assert entries[-1] is END_SENTINEL
+    finally:
+        await stream.aclose()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("cleanup_bridge", BACKENDS, indirect=True)
+async def test_deleted_stream_after_initial_wakeup_reports_gap(cleanup_bridge):
+    bridge = cleanup_bridge
+    delayed = _DelayedBlockingReadRedis(bridge._redis)
+    bridge._redis = delayed
+    stream = bridge.subscribe("cleanup-gap", heartbeat_interval=1)
+    pending = asyncio.create_task(anext(stream))
+    try:
+        with anyio.fail_after(2):
+            await delayed.blocking_read_started.wait()
+            await bridge.publish("cleanup-gap", "messages", {"index": 0})
+            await delayed.wake_response_captured.wait()
+            await bridge.cleanup("cleanup-gap")
+            delayed.release_wake_response.set()
+            assert await pending == StreamGap(None, None, None)
+        with pytest.raises(StopAsyncIteration):
+            await anext(stream)
+    finally:
+        pending.cancel()
+        await asyncio.gather(pending, return_exceptions=True)
+        await stream.aclose()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("cleanup_bridge", BACKENDS, indirect=True)
+async def test_gateway_emits_recovery_gap_after_stream_cleanup(cleanup_bridge):
+    from app.gateway.services import sse_consumer
+
+    bridge = cleanup_bridge
+    manager = RunManager()
+    record = await manager.create("cleanup-gap-thread")
+    run_id = record.run_id
+    try:
+        await manager.set_status(run_id, RunStatus.success)
+        for index in range(70):
+            await bridge.publish(run_id, "messages", {"index": index})
+        await bridge.publish_end(run_id)
+        stream = sse_consumer(bridge, record, _ConnectedRequest(), manager, apply_on_disconnect=False)
+        try:
+            assert (await anext(stream)).startswith("event: messages\n")
+            await bridge.cleanup(run_id)
+            with anyio.fail_after(2):
+                frames = [frame async for frame in stream]
+            assert frames[-1].startswith("event: gap\n")
+            data = json.loads(frames[-1].split("data: ", 1)[1])
+            assert data["recovery"] == "reload_durable_state"
+            assert data["latest_available_event_id"] is None
+            assert not any(frame.startswith("event: end\n") for frame in frames)
+            assert record.status == RunStatus.success
+        finally:
+            await stream.aclose()
+    finally:
+        await bridge.cleanup(run_id)
+        await manager.shutdown()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("store_only", [False, True])
+async def test_wait_completes_when_gap_has_no_retained_tail(store_only):
+    from app.gateway.services import wait_for_run_completion
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    store = MemoryRunStore() if store_only else None
+    manager = RunManager(store=store)
+    if store_only:
+        await store.put("cleanup-gap", thread_id="cleanup-thread", status="running")
+        record = await manager.get("cleanup-gap")
+    else:
+        record = await manager.create("cleanup-thread")
+        await manager.set_status(record.run_id, RunStatus.running)
+
+    class DeletedStreamBridge:
+        async def stream_exists(self, run_id):
+            return False
+
+        async def subscribe(self, run_id, *, last_event_id=None):
+            await asyncio.sleep(0)
+            if store_only:
+                await store.update_status(run_id, "success")
+            else:
+                await manager.set_status(run_id, RunStatus.success)
+            yield StreamGap("123-0", None, None)
+            # A new subscription must not wait on the already-deleted stream.
+            await asyncio.Event().wait()
+
+    try:
+        with anyio.fail_after(2):
+            assert await wait_for_run_completion(DeletedStreamBridge(), record, _ConnectedRequest(), manager)
+    finally:
+        await manager.shutdown()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("cleanup_bridge", BACKENDS, indirect=True)
+@pytest.mark.parametrize("store_only", [False, True])
+async def test_wait_survives_cleanup_with_real_bridge(cleanup_bridge, store_only):
+    from app.gateway.services import wait_for_run_completion
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    store = MemoryRunStore() if store_only else None
+    manager = RunManager(store=store)
+    if store_only:
+        await store.put("cleanup-gap", thread_id="cleanup-thread", status="running")
+        record = await manager.get("cleanup-gap")
+    else:
+        record = await manager.create("cleanup-thread")
+        await manager.set_status(record.run_id, RunStatus.running)
+    bridge = cleanup_bridge
+    run_id = record.run_id
+
+    class CleanupAfterFirstMessage(_ConnectedRequest):
+        async def is_disconnected(self):
+            if store_only:
+                await store.update_status(run_id, "success")
+            else:
+                await manager.set_status(run_id, RunStatus.success)
+            await bridge.cleanup(run_id)
+            return False
+
+    try:
+        for index in range(70):
+            await bridge.publish(run_id, "messages", {"index": index})
+        await bridge.publish_end(run_id)
+        with anyio.fail_after(2):
+            assert await wait_for_run_completion(bridge, record, CleanupAfterFirstMessage(), manager)
+    finally:
+        await bridge.cleanup(run_id)
+        await manager.shutdown()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("terminal_with_retained_tail", [False, True])
+async def test_empty_gap_does_not_complete_running_or_retained_run(terminal_with_retained_tail):
+    from app.gateway.services import wait_for_run_completion
+
+    manager = RunManager()
+    record = await manager.create("cleanup-control")
+    await manager.set_status(record.run_id, RunStatus.success if terminal_with_retained_tail else RunStatus.running)
+    resumed = asyncio.Event()
+    allow_end = asyncio.Event()
+
+    class GapThenEndBridge:
+        subscriptions = 0
+
+        async def stream_exists(self, run_id):
+            # Initial probe must not skip the subscription.
+            return True if self.subscriptions == 0 else terminal_with_retained_tail
+
+        async def subscribe(self, run_id, *, last_event_id=None):
+            self.subscriptions += 1
+            if self.subscriptions == 1:
+                yield StreamGap("123-0", None, None)
+                return
+            resumed.set()
+            await allow_end.wait()
+            yield END_SENTINEL
+
+    task = asyncio.create_task(wait_for_run_completion(GapThenEndBridge(), record, _ConnectedRequest(), manager))
+    try:
+        with anyio.fail_after(2):
+            await resumed.wait()
+            assert not task.done()
+            allow_end.set()
+            assert await task
+    finally:
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+        await manager.shutdown()