fix(gateway): live-tail malformed redis reconnect ids (#4012)

* fix(gateway): live-tail malformed redis reconnect ids

* test: avoid redis live-tail heartbeat gaps

* fix: handle terminal malformed redis reconnect ids

README.md

@@ -264,7 +264,7 @@ section, when present, overrides the first two for backward compatibility.
 The unified nginx endpoint is same-origin by default and does not emit browser CORS headers. If you run a split-origin or port-forwarded browser client, set `GATEWAY_CORS_ORIGINS` to comma-separated exact origins such as `http://localhost:3000`; the Gateway then applies the CORS allowlist and matching CSRF origin checks.
 
 > [!IMPORTANT]
-> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). The Redis stream bridge (`stream_bridge.type: redis`) shares SSE delivery and `Last-Event-ID` replay across workers, with a rolling retained-buffer TTL (`stream_ttl_seconds`) as a cleanup safety net. It does not make run cancellation, request de-duplication, or IM channel state fully cross-worker by itself; use single-worker Gateway or explicit sticky routing/ownership before raising `GATEWAY_WORKERS`.
+> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). The Redis stream bridge (`stream_bridge.type: redis`) shares SSE delivery and `Last-Event-ID` replay across workers, with a rolling retained-buffer TTL (`stream_ttl_seconds`) as a cleanup safety net. Malformed reconnect IDs live-tail new events instead of replaying the retained buffer. It does not make run cancellation, request de-duplication, or IM channel state fully cross-worker by itself; use single-worker Gateway or explicit sticky routing/ownership before raising `GATEWAY_WORKERS`.
 
 See [CONTRIBUTING.md](CONTRIBUTING.md) for detailed Docker development guide.
 

backend/AGENTS.md

@@ -337,7 +337,7 @@ metadata only.
 - `cancel()` and `create_or_reject(..., multitask_strategy="interrupt"|"rollback")` persist interrupted status through `RunStore.update_status()`, matching normal `set_status()` transitions.
 - Store-only hydrated runs are readable history. If the current worker has no in-memory task/control state for that run, cancellation APIs can return 409 because this worker cannot stop the task.
 - `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).
-- 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 orphan recovery publishes `END_SENTINEL` and schedules stream cleanup for recovered runs; do not broaden this into a shared-database multi-pod reaper without adding worker ownership/liveness first.
+- 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 orphan recovery publishes `END_SENTINEL` and schedules stream cleanup for recovered runs; malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Do not broaden this into a shared-database multi-pod reaper without adding worker ownership/liveness first.
 - 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.
 - 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. 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.
 

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

@@ -6,6 +6,7 @@ import asyncio
 import inspect
 import json
 import logging
+import re
 from collections.abc import AsyncIterator, Mapping
 from typing import Any
 
@@ -32,6 +33,7 @@ logger = logging.getLogger(__name__)
 
 _KIND_EVENT = "event"
 _KIND_END = "end"
+_REDIS_STREAM_ID_RE = re.compile(r"\d+(-\d+)?")
 
 # Batch size for ``XREAD``. Reading more than one entry per round-trip collapses
 # a large ``Last-Event-ID`` replay into far fewer calls; live tailing still
@@ -162,6 +164,20 @@ class RedisStreamBridge(StreamBridge):
         """Return whether Redis still has retained stream data for *run_id*."""
         return bool(await self._redis.exists(self._stream_key(run_id)))
 
+    async def _resolve_start_stream_id(self, key: str, last_event_id: str | None) -> str:
+        if last_event_id is None:
+            return "0-0"
+        if _REDIS_STREAM_ID_RE.fullmatch(last_event_id):
+            return last_event_id
+        entries = await self._redis.xrevrange(key, count=1)
+        if not entries:
+            return "0-0"
+        event_id, fields = entries[0]
+        payload = self._normalise_fields(fields)
+        if payload.get("kind") == _KIND_END:
+            return "0-0"
+        return self._decode(event_id)
+
     async def subscribe(
         self,
         run_id: str,
@@ -170,7 +186,7 @@ class RedisStreamBridge(StreamBridge):
         heartbeat_interval: float = 15.0,
     ) -> AsyncIterator[StreamEvent]:
         key = self._stream_key(run_id)
-        stream_id = last_event_id or "0-0"
+        stream_id = await self._resolve_start_stream_id(key, last_event_id)
         block_ms = max(1, int(heartbeat_interval * 1000)) if heartbeat_interval > 0 else 1
         consecutive_errors = 0
 
@@ -178,22 +194,15 @@ class RedisStreamBridge(StreamBridge):
             try:
                 response = await self._redis.xread({key: stream_id}, count=_XREAD_COUNT, block=block_ms)
             except ResponseError:
-                # The only client-controllable stream ID is the Last-Event-ID
-                # header, so a rejected ID means a malformed reconnect token:
-                # fall back to replaying from the earliest retained event. We key
-                # off the control flow rather than the error wording, which is the
-                # server's text (Redis/Valkey/Dragonfly) and not a stable API. If
-                # the reset read from "0-0" also fails, the stream/connection is
-                # genuinely broken, so re-raise.
-                if stream_id == "0-0":
-                    raise
+                # 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 Last-Event-ID %r for stream bridge; replaying from earliest retained event",
+                    "Redis rejected stream id %r for stream bridge subscription",
                     stream_id,
                     exc_info=True,
                 )
-                stream_id = "0-0"
-                continue
+                raise
             except RedisError:
                 consecutive_errors += 1
                 if consecutive_errors > _MAX_SUBSCRIBE_RETRIES:

backend/tests/test_stream_bridge.py

@@ -58,6 +58,10 @@ class _FakeRedis:
                 except TimeoutError:
                     return []
 
+    async def xrevrange(self, name, max="+", min="-", count=None):
+        entries = list(reversed(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)
@@ -479,6 +483,76 @@ 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_invalid_last_event_id_tails_live_events(redis_bridge: RedisStreamBridge):
+    """Malformed reconnect ids should not replay retained Redis events."""
+    run_id = "redis-run-invalid-last-event-id"
+
+    await redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
+    received = []
+
+    async def publish_later() -> None:
+        await anyio.sleep(0.05)
+        await redis_bridge.publish(run_id, "values", {"step": 1})
+        await redis_bridge.publish_end(run_id)
+
+    with anyio.fail_after(2):
+        async with anyio.create_task_group() as task_group:
+            task_group.start_soon(publish_later)
+            async for entry in redis_bridge.subscribe(run_id, last_event_id="-1", heartbeat_interval=0.01):
+                if entry is HEARTBEAT_SENTINEL:
+                    continue
+                received.append(entry)
+                if entry is END_SENTINEL:
+                    break
+
+    assert [entry.event for entry in received[:-1]] == ["values"]
+    assert received[-1] is END_SENTINEL
+
+
+@pytest.mark.anyio
+async def test_redis_invalid_last_event_id_tails_empty_stream(redis_bridge: RedisStreamBridge):
+    """Malformed reconnect ids should still wait for the first Redis event."""
+    run_id = "redis-run-invalid-empty"
+    received = []
+
+    async def publish_later() -> None:
+        await anyio.sleep(0.05)
+        await redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
+        await redis_bridge.publish_end(run_id)
+
+    with anyio.fail_after(2):
+        async with anyio.create_task_group() as task_group:
+            task_group.start_soon(publish_later)
+            async for entry in redis_bridge.subscribe(run_id, last_event_id="-1", heartbeat_interval=0.01):
+                if entry is HEARTBEAT_SENTINEL:
+                    continue
+                received.append(entry)
+                if entry is END_SENTINEL:
+                    break
+
+    assert [entry.event for entry in received[:-1]] == ["metadata"]
+    assert received[-1] is END_SENTINEL
+
+
+@pytest.mark.anyio
+async def test_redis_invalid_last_event_id_on_terminal_run_replays_end(redis_bridge: RedisStreamBridge):
+    """Malformed reconnect ids on terminal streams should drain END instead of hanging."""
+    run_id = "redis-run-invalid-terminal"
+
+    await redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
+    await redis_bridge.publish_end(run_id)
+
+    received = []
+    async for entry in redis_bridge.subscribe(run_id, last_event_id="-1", heartbeat_interval=1.0):
+        received.append(entry)
+        if entry is END_SENTINEL:
+            break
+
+    assert [entry.event for entry in received[:-1]] == ["metadata"]
+    assert received[-1] is END_SENTINEL
+
+
 @pytest.mark.anyio
 async def test_redis_heartbeat(redis_bridge: RedisStreamBridge):
     """Redis bridge should yield heartbeats when XREAD times out on an existing stream."""
@@ -787,9 +861,9 @@ async def test_make_stream_bridge_passes_redis_options(monkeypatch):
 # `make test` stays green without Redis. Point at a server with
 # DEER_FLOW_TEST_REDIS_URL (defaults to redis://localhost:6379/15 — DB 15 to
 # avoid clobbering real data) and select with `pytest -m integration`. They
-# cover what _FakeRedis cannot: the real ResponseError fallback for a malformed
-# Last-Event-ID, real XADD/XREAD semantics, the server <ms>-<seq> ID format,
-# and MAXLEN trimming.
+# cover what _FakeRedis only approximates: real XADD/XREAD semantics, live-tail
+# reconnects for malformed Last-Event-ID values, the server <ms>-<seq> ID
+# format, and MAXLEN trimming.
 
 REDIS_TEST_URL = os.environ.get("DEER_FLOW_TEST_REDIS_URL", "redis://localhost:6379/15")
 
@@ -878,25 +952,28 @@ async def test_redis_integration_replays_after_last_event_id(real_redis_bridge):
 @pytest.mark.integration
 @requires_redis
 @pytest.mark.anyio
-async def test_redis_integration_invalid_last_event_id_falls_back(real_redis_bridge):
-    """A malformed Last-Event-ID must trigger the ResponseError fallback.
-
-    Real Redis raises ``ResponseError`` for a syntactically-invalid stream ID;
-    ``_FakeRedis`` cannot reproduce this path, so the ``except ResponseError``
-    branch in ``subscribe`` is only exercised here.
-    """
+async def test_redis_integration_invalid_last_event_id_tails_live_events(real_redis_bridge):
+    """A malformed Last-Event-ID should wait at the live tail."""
     run_id = "integ-bad-leid"
     await real_redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
-    await real_redis_bridge.publish_end(run_id)
-
     received = []
-    async for entry in real_redis_bridge.subscribe(run_id, last_event_id="not-a-valid-id", heartbeat_interval=1.0):
-        received.append(entry)
-        if entry is END_SENTINEL:
-            break
 
-    # Falls back to replaying from the earliest retained event instead of raising.
-    assert [e.event for e in received[:-1]] == ["metadata"]
+    async def publish_later() -> None:
+        await anyio.sleep(0.05)
+        await real_redis_bridge.publish(run_id, "values", {"step": 1})
+        await real_redis_bridge.publish_end(run_id)
+
+    with anyio.fail_after(2):
+        async with anyio.create_task_group() as task_group:
+            task_group.start_soon(publish_later)
+            async for entry in real_redis_bridge.subscribe(run_id, last_event_id="not-a-valid-id", heartbeat_interval=0.01):
+                if entry is HEARTBEAT_SENTINEL:
+                    continue
+                received.append(entry)
+                if entry is END_SENTINEL:
+                    break
+
+    assert [e.event for e in received[:-1]] == ["values"]
     assert received[-1] is END_SENTINEL