perf(runtime): resolve SSE resume offset in O(1) via the event-id seq (#3700)

MemoryStreamBridge._resolve_start_offset scanned the retained event buffer
(up to queue_maxsize=256 entries) on every subscribe/reconnect carrying a
Last-Event-ID. Event ids are "{ts}-{seq}" where seq is a per-run monotonic
counter that equals the event's absolute offset, so the offset is computable
arithmetically. Parse seq, index into the buffer, and verify the id matches
exactly -- a stale/evicted/foreign/malformed id falls back to
replay-from-earliest, identical to the previous scan.

Co-authored-by: ly-wang19 <ly-wang19@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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

@@ -48,13 +48,36 @@ class MemoryStreamBridge(StreamBridge):
         seq = self._counters[run_id] - 1
         return f"{ts}-{seq}"
 
+    @staticmethod
+    def _parse_event_seq(event_id: str) -> int | None:
+        """Extract the per-run sequence number from a ``{ts}-{seq}`` event id.
+
+        ``seq`` (assigned by :meth:`_next_id`) increases by one per published
+        event, so it equals the event's absolute offset within the run. Returns
+        ``None`` for ids that do not match the expected format.
+        """
+        _, sep, seq_text = event_id.rpartition("-")
+        if not sep:
+            return None
+        try:
+            return int(seq_text)
+        except ValueError:
+            return None
+
     def _resolve_start_offset(self, stream: _RunStream, last_event_id: str | None) -> int:
         if last_event_id is None:
             return stream.start_offset
 
-        for index, entry in enumerate(stream.events):
-            if entry.id == last_event_id:
-                return stream.start_offset + index + 1
+        # Event ids embed a per-run, monotonically increasing ``seq`` that equals
+        # the event's absolute offset, so locate the event by arithmetic in O(1)
+        # rather than scanning the retained buffer. The id is verified at the
+        # computed index, so a stale/evicted/foreign/malformed id still falls back
+        # to replay-from-earliest — identical to the previous linear scan.
+        seq = self._parse_event_seq(last_event_id)
+        if seq is not None:
+            local_index = seq - stream.start_offset
+            if 0 <= local_index < len(stream.events) and stream.events[local_index].id == last_event_id:
+                return stream.start_offset + local_index + 1
 
         if stream.events:
             logger.warning(

backend/tests/test_stream_bridge.py

@@ -334,3 +334,73 @@ async def test_make_stream_bridge_defaults():
     """make_stream_bridge() with no config yields a MemoryStreamBridge."""
     async with make_stream_bridge() as bridge:
         assert isinstance(bridge, MemoryStreamBridge)
+
+
+# ---------------------------------------------------------------------------
+# _resolve_start_offset: O(1) seq-indexed resolution (parity with linear scan)
+# ---------------------------------------------------------------------------
+
+
+def _linear_resolve(stream, last_event_id):
+    """The original linear-scan resolver, kept as a parity reference."""
+    if last_event_id is None:
+        return stream.start_offset
+    for index, entry in enumerate(stream.events):
+        if entry.id == last_event_id:
+            return stream.start_offset + index + 1
+    return stream.start_offset
+
+
+@pytest.mark.parametrize(
+    "event_id,expected",
+    [
+        ("1718000000000-0", 0),
+        ("1718000000000-42", 42),
+        ("garbage", None),  # no separator
+        ("1718000000000-x", None),  # non-integer seq
+        ("", None),
+    ],
+)
+def test_parse_event_seq(event_id, expected):
+    assert MemoryStreamBridge._parse_event_seq(event_id) == expected
+
+
+@pytest.mark.anyio
+async def test_resolve_start_offset_matches_linear_scan():
+    """The seq-indexed resolver must return exactly what the linear scan returned,
+    across retained, evicted, foreign (same seq / wrong ts), malformed, and None ids."""
+    bridge = MemoryStreamBridge(queue_maxsize=4)
+    run_id = "run-parity"
+    ids = []
+    for i in range(10):
+        await bridge.publish(run_id, f"e{i}", {"i": i})
+        ids.append(bridge._streams[run_id].events[-1].id)  # includes ids that later evict
+    stream = bridge._streams[run_id]
+    assert stream.start_offset == 6  # 10 published, buffer of 4 retains seq 6..9
+
+    # A foreign id: a retained event's seq but a different timestamp -> must NOT match.
+    ts, _, seq_text = stream.events[0].id.rpartition("-")
+    foreign_id = f"{int(ts) + 1}-{seq_text}"
+
+    candidates = [None, "garbage", "1718000000000-x", "999999-999999", foreign_id, *ids]
+    for eid in candidates:
+        assert bridge._resolve_start_offset(stream, eid) == _linear_resolve(stream, eid), eid
+
+
+@pytest.mark.anyio
+async def test_subscribe_with_unknown_last_event_id_replays_from_earliest():
+    """A foreign/garbage Last-Event-ID falls back to replaying retained events."""
+    bridge = MemoryStreamBridge(queue_maxsize=10)
+    run_id = "run-unknown-id"
+    await bridge.publish(run_id, "first", {})
+    await bridge.publish(run_id, "second", {})
+    await bridge.publish_end(run_id)
+
+    received = []
+    async for entry in bridge.subscribe(run_id, last_event_id="not-a-real-id", heartbeat_interval=1.0):
+        received.append(entry)
+        if entry is END_SENTINEL:
+            break
+
+    assert [entry.event for entry in received[:-1]] == ["first", "second"]
+    assert received[-1] is END_SENTINEL