fix(streaming): validate numeric reconnect cursors before subscribing (#6690)

* fix(streaming): validate numeric reconnect cursors before subscribing

* fix(streaming): address reconnect cursor review feedback

---------

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

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

@@ -11,6 +11,8 @@ Observed Redis streams removed during replay yield `StreamGap` with empty bounds
 
 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.
 
+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.
+
 ### 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

@@ -13,7 +13,7 @@ from typing import Any
 from .base import DEFAULT_HEARTBEAT_INTERVAL_SECONDS, END_SENTINEL, HEARTBEAT_SENTINEL, StreamBridge, StreamEvent, StreamGap, StreamItem
 
 logger = logging.getLogger(__name__)
-_MEMORY_STREAM_ID_RE = re.compile(r"\d+-(\d+)")
+_MEMORY_STREAM_ID_RE = re.compile(r"[0-9]+-([0-9]+)")
 
 
 @dataclass
@@ -62,7 +62,11 @@ class MemoryStreamBridge(StreamBridge):
         match = _MEMORY_STREAM_ID_RE.fullmatch(event_id)
         if match is None:
             return None
-        return int(match.group(1))
+        try:
+            return int(match.group(1))
+        except ValueError:
+            # Numeric-looking reconnect ids can exceed Python's digit limit.
+            return None
 
     @staticmethod
     def _make_gap(stream: _RunStream, requested_event_id: str | None) -> StreamGap:

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

@@ -33,7 +33,8 @@ logger = logging.getLogger(__name__)
 
 _KIND_EVENT = "event"
 _KIND_END = "end"
-_REDIS_STREAM_ID_RE = re.compile(r"\d+(-\d+)?")
+_REDIS_STREAM_ID_RE = re.compile(r"[0-9]+(-[0-9]+)?")
+_MAX_STREAM_ID_COMPONENT = 2**64 - 1
 
 # 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
@@ -150,7 +151,12 @@ class RedisStreamBridge(StreamBridge):
         if _REDIS_STREAM_ID_RE.fullmatch(event_id) is None:
             return None
         milliseconds, separator, sequence = event_id.partition("-")
-        return int(milliseconds), int(sequence) if separator else 0
+        try:
+            parts = int(milliseconds), int(sequence) if separator else 0
+        except ValueError:
+            return None
+        # Redis stream id components are unsigned 64-bit integers.
+        return parts if max(parts) <= _MAX_STREAM_ID_COMPONENT else None
 
     @classmethod
     def _stream_id_lt(cls, left: str, right: str) -> bool:
@@ -207,7 +213,7 @@ class RedisStreamBridge(StreamBridge):
     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):
+        if self._parse_stream_id(last_event_id) is not None:
             return last_event_id
         entries = await self._redis.xrevrange(key, count=1)
         if not entries:

backend/tests/test_stream_bridge.py

@@ -23,6 +23,14 @@ from deerflow.runtime import END_SENTINEL, HEARTBEAT_SENTINEL, MemoryStreamBridg
 # directly from the submodule.
 from deerflow.runtime.stream_bridge.redis import RedisStreamBridge
 
+_INVALID_NUMERIC_REDIS_CURSORS = [
+    pytest.param("9" * 5000 + "-0", id="oversized-milliseconds"),
+    pytest.param("1-" + "9" * 5000, id="oversized-sequence"),
+    pytest.param("18446744073709551616-0", id="milliseconds-overflow"),
+    pytest.param("1-18446744073709551616", id="sequence-overflow"),
+    pytest.param("١-٠", id="non-ascii-digits"),
+]
+
 
 def _stream_id_gt(left: str, right: str) -> bool:
     left_ms, left_seq = left.split("-", 1)
@@ -792,7 +800,8 @@ async def test_redis_invalid_last_event_id_tails_empty_stream(redis_bridge: Redi
 
 
 @pytest.mark.anyio
-async def test_redis_invalid_last_event_id_on_terminal_run_replays_end(redis_bridge: RedisStreamBridge):
+@pytest.mark.parametrize("last_event_id", [pytest.param("-1", id="sentinel"), *_INVALID_NUMERIC_REDIS_CURSORS])
+async def test_redis_invalid_last_event_id_on_terminal_run_replays_end(redis_bridge: RedisStreamBridge, last_event_id: str):
     """Malformed reconnect ids on terminal streams should drain END instead of hanging."""
     run_id = "redis-run-invalid-terminal"
 
@@ -800,10 +809,11 @@ async def test_redis_invalid_last_event_id_on_terminal_run_replays_end(redis_bri
     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
+    with anyio.fail_after(2):
+        async for entry in redis_bridge.subscribe(run_id, last_event_id=last_event_id, 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
@@ -1127,6 +1137,10 @@ def _linear_resolve(stream, last_event_id):
     [
         ("1718000000000-0", 0),
         ("1718000000000-42", 42),
+        ("١-0", None),
+        ("1-٠", None),
+        ("١-٠", None),
+        ("1-" + "9" * 5000, None),
         ("-1", None),  # malformed live-tail sentinel, not a valid event id
         ("garbage", None),  # no separator
         ("1718000000000-x", None),  # non-integer seq
@@ -1137,6 +1151,38 @@ def test_parse_event_seq(event_id, expected):
     assert MemoryStreamBridge._parse_event_seq(event_id) == expected
 
 
+@pytest.mark.parametrize(
+    "event_id,expected",
+    [
+        ("0", (0, 0)),
+        ("0-0", (0, 0)),
+        ("1718000000000", (1718000000000, 0)),
+        ("1718000000000-42", (1718000000000, 42)),
+        ("18446744073709551615", (18446744073709551615, 0)),
+        ("18446744073709551615-0", (18446744073709551615, 0)),
+        ("1-18446744073709551615", (1, 18446744073709551615)),
+        ("18446744073709551615-18446744073709551615", (18446744073709551615, 18446744073709551615)),
+        ("18446744073709551616", None),
+        ("18446744073709551616-0", None),
+        ("1-18446744073709551616", None),
+        ("9" * 5000 + "-0", None),
+        ("1-" + "9" * 5000, None),
+        ("١-0", None),
+        ("1-٠", None),
+        ("١-٠", None),
+        ("", None),
+        ("-1", None),
+        ("garbage", None),
+        ("1-", None),
+        ("1--1", None),
+        ("1-0-0", None),
+        ("1-0\n", None),
+    ],
+)
+def test_parse_stream_id(event_id, expected):
+    assert RedisStreamBridge._parse_stream_id(event_id) == expected
+
+
 @pytest.mark.anyio
 async def test_resolve_start_offset_matches_linear_scan():
     """Retained and unknown cursors preserve the previous linear-scan behavior."""
@@ -1204,10 +1250,21 @@ async def test_subscribe_with_unknown_last_event_id_replays_from_earliest():
 
 
 @pytest.mark.anyio
-async def test_memory_malformed_last_event_id_is_not_reported_as_gap():
+@pytest.mark.parametrize(
+    "last_event_id",
+    [
+        pytest.param("-1", id="sentinel"),
+        pytest.param("1-" + "9" * 5000, id="oversized-sequence"),
+        pytest.param("١-0", id="non-ascii-milliseconds"),
+        pytest.param("1-٠", id="non-ascii-sequence"),
+        pytest.param("١-٠", id="non-ascii-both-components"),
+    ],
+)
+async def test_memory_malformed_last_event_id_is_not_reported_as_gap(last_event_id: str):
     """Malformed cursor policy stays separate from valid evicted cursors."""
     bridge = MemoryStreamBridge(queue_maxsize=2)
     run_id = "run-malformed-id"
+    await bridge.publish(run_id, "evicted", {})
     await bridge.publish(run_id, "first", {})
     await bridge.publish(run_id, "second", {})
     await bridge.publish_end(run_id)
@@ -1216,7 +1273,7 @@ async def test_memory_malformed_last_event_id_is_not_reported_as_gap():
         entry
         async for entry in bridge.subscribe(
             run_id,
-            last_event_id="-1",
+            last_event_id=last_event_id,
             heartbeat_interval=1.0,
         )
     ]
@@ -1429,7 +1486,8 @@ 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_tails_live_events(real_redis_bridge):
+@pytest.mark.parametrize("last_event_id", [pytest.param("not-a-valid-id", id="malformed"), *_INVALID_NUMERIC_REDIS_CURSORS])
+async def test_redis_integration_invalid_last_event_id_tails_live_events(real_redis_bridge, last_event_id: str):
     """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})
@@ -1443,7 +1501,7 @@ async def test_redis_integration_invalid_last_event_id_tails_live_events(real_re
     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):
+            async for entry in real_redis_bridge.subscribe(run_id, last_event_id=last_event_id, heartbeat_interval=0.01):
                 if entry is HEARTBEAT_SENTINEL:
                     continue
                 received.append(entry)
@@ -1454,6 +1512,39 @@ async def test_redis_integration_invalid_last_event_id_tails_live_events(real_re
     assert received[-1] is END_SENTINEL
 
 
+@pytest.mark.integration
+@requires_redis
+@pytest.mark.anyio
+@pytest.mark.parametrize("last_event_id", _INVALID_NUMERIC_REDIS_CURSORS)
+async def test_redis_integration_invalid_numeric_cursor_replays_terminal_stream(real_redis_bridge, last_event_id: str):
+    run_id = "integ-invalid-numeric-terminal"
+    await real_redis_bridge.publish(run_id, "metadata", {"run_id": run_id})
+    await real_redis_bridge.publish_end(run_id)
+
+    with anyio.fail_after(2):
+        received = [entry async for entry in real_redis_bridge.subscribe(run_id, last_event_id=last_event_id, heartbeat_interval=0.01)]
+
+    assert [entry.event for entry in received[:-1]] == ["metadata"]
+    assert received[-1] is END_SENTINEL
+
+
+@pytest.mark.integration
+@requires_redis
+@pytest.mark.anyio
+@pytest.mark.parametrize("last_event_id", ["18446744073709551615-0", "0-18446744073709551615", "18446744073709551615"])
+async def test_redis_integration_uint64_boundary_cursor_reports_gap(real_redis_bridge, last_event_id: str):
+    run_id = "integ-boundary-cursor"
+    await real_redis_bridge.publish(run_id, "metadata", {})
+    await real_redis_bridge.publish_end(run_id)
+
+    with anyio.fail_after(2):
+        received = [entry async for entry in real_redis_bridge.subscribe(run_id, last_event_id=last_event_id, heartbeat_interval=0.01)]
+
+    assert len(received) == 1
+    assert isinstance(received[0], StreamGap)
+    assert received[0].requested_event_id == last_event_id
+
+
 @pytest.mark.integration
 @requires_redis
 @pytest.mark.anyio