fix(streaming): report replay gap for future Redis Last-Event-ID (#6605)

* fix(stream): report replay gap for future Redis stream cursors

* test(stream): future reconnect cursors report gap on live and ended runs

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

@@ -318,6 +318,26 @@ class RedisStreamBridge(StreamBridge):
                     )
                     return
 
+            # A syntactically valid but future Last-Event-ID cannot replay from
+            # this retained stream. In particular, an already-ended run would
+            # otherwise heartbeat forever because XREAD never sees its end marker.
+            if latest_entries and gap_detection_enabled:
+                latest_id = self._decode(latest_entries[0][0])
+                if self._stream_id_lt(latest_id, stream_id):
+                    earliest_id = self._decode(earliest_entries[0][0])
+                    logger.warning(
+                        "subscriber for Redis stream %s requested future cursor %s (latest %s)",
+                        key,
+                        stream_id,
+                        latest_id,
+                    )
+                    yield StreamGap(
+                        requested_event_id=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)

backend/tests/test_redis_stream_future_cursor.py

@@ -0,0 +1,60 @@
+"""Regression tests for reconnect cursors beyond retained Redis stream history."""
+
+import pytest
+
+from deerflow.runtime import StreamGap
+from deerflow.runtime.stream_bridge.redis import RedisStreamBridge
+
+
+class _SnapshotPipeline:
+    def __init__(self, entries):
+        self.entries = entries
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_args):
+        return None
+
+    def xrange(self, *_args, **_kwargs):
+        return self
+
+    def xrevrange(self, *_args, **_kwargs):
+        return self
+
+    def xread(self, *_args, **_kwargs):
+        return self
+
+    async def execute(self):
+        return self.entries[:1], self.entries[-1:], []
+
+
+class _RetainedStream:
+    def __init__(self, entries):
+        self.entries = entries
+
+    def pipeline(self, *, transaction=True):
+        assert transaction
+        return _SnapshotPipeline(self.entries)
+
+    async def xread(self, *_args, **_kwargs):
+        raise AssertionError("a future cursor must not enter blocking XREAD")
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("ended", [False, True])
+async def test_future_last_event_id_reports_gap_without_blocking(ended):
+    entries = [("1-0", {"kind": "event", "event": "message", "data": "{}"})]
+    if ended:
+        entries.append(("2-0", {"kind": "end"}))
+    bridge = RedisStreamBridge(redis_url="redis://unused", client=_RetainedStream(entries))
+    cursor = "9999999999999-0"
+    subscriber = bridge.subscribe("run", last_event_id=cursor)
+
+    assert await anext(subscriber) == StreamGap(
+        requested_event_id=cursor,
+        earliest_available_event_id=entries[0][0],
+        latest_available_event_id=entries[-1][0],
+    )
+    with pytest.raises(StopAsyncIteration):
+        await anext(subscriber)