diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 7003649e42..6da4f8f9a9 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/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. diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py b/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py index 048be87a53..b3ecef3a0c 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/memory.py +++ b/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: diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py b/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py index fafe906a70..1e4311f195 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py +++ b/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: diff --git a/backend/tests/test_stream_bridge.py b/backend/tests/test_stream_bridge.py index e48edd963d..8e92be5a44 100644 --- a/backend/tests/test_stream_bridge.py +++ b/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