fix(runtime): honor both cursors when reading thread messages (#6136)

* fix(runtime): honor both cursors when reading thread messages

* docs(runtime): clarify run-scoped combined cursor bounds

backend/README.md

@@ -118,6 +118,14 @@ LLM-powered persistent context retention across conversations:
 | **MCP** | Any Model Context Protocol server (stdio, SSE, HTTP transports) |
 | **Skills** | Domain-specific workflows injected via system prompt |
 
+### Run Event Storage
+
+For direct `RunEventStore.list_messages` callers, `after_seq` and `before_seq`
+bound an exclusive message window. With both cursors, reads return the first
+`limit` messages inside that window in ascending sequence order across memory,
+JSONL, and database backends. Keep `before_seq` fixed and advance `after_seq`
+to the last returned sequence to page forward through a bounded history range.
+
 ### Gateway API
 
 FastAPI application providing REST endpoints for frontend integration:

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

@@ -26,6 +26,8 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c
 
 **Message sequence placement:** Keep backend and frontend message identity rules aligned. Details: `backend/docs/runtime-guidance-details.md`.
 
+**Thread message cursors:** `list_messages` applies both exclusive bounds before `limit`, paging forward whenever `after_seq` is supplied.
+
 **Human-input capture** (`runtime/journal.py`): track capture separately from
 the optional display summary. Image-only input has no text but must still stop
 the batch scan and later model calls from appending another human-input event.

backend/packages/harness/deerflow/runtime/events/store/base.py

@@ -116,6 +116,9 @@ class RunEventStore(abc.ABC):
         - after_seq: return the first ``limit`` records with seq > after_seq (ascending)
         - neither: return the latest ``limit`` records (ascending)
 
+        When both cursors are supplied, return the first ``limit`` records in
+        the exclusive window ``after_seq < seq < before_seq`` (ascending).
+
         ``user_id`` may be passed explicitly by request-independent callers;
         user-scoped backends must apply it according to their isolation model.
         """
@@ -222,6 +225,9 @@ class RunEventStore(abc.ABC):
         - after_seq: return the first ``limit`` records with seq > after_seq (ascending)
         - before_seq: return the last ``limit`` records with seq < before_seq (ascending)
         - neither: return the latest ``limit`` records (ascending)
+
+        When both cursors are supplied, return the first ``limit`` records in
+        the exclusive window ``after_seq < seq < before_seq`` (ascending).
         """
 
     @abc.abstractmethod

backend/packages/harness/deerflow/runtime/events/store/jsonl.py

@@ -341,8 +341,7 @@ class JsonlRunEventStore(RunEventStore):
 
         if before_seq is not None:
             messages = [e for e in messages if e["seq"] < before_seq]
-            return messages[-limit:]
-        elif after_seq is not None:
+        if after_seq is not None:
             messages = [e for e in messages if e["seq"] > after_seq]
             return messages[:limit]
         else:

backend/packages/harness/deerflow/runtime/events/store/memory.py

@@ -128,14 +128,16 @@ class MemoryRunEventStore(RunEventStore):
         # contiguous slice located with bisect (O(log m)) rather than a full scan.
         messages = self._messages.get(thread_id, [])
 
-        if before_seq is not None:
+        if after_seq is not None:
+            # Page forward within both exclusive bounds, without copying the
+            # whole window before applying the limit.
+            lo = bisect.bisect_right(messages, after_seq, key=lambda e: e["seq"])
+            hi = len(messages) if before_seq is None else bisect.bisect_left(messages, before_seq, key=lambda e: e["seq"])
+            return messages[lo : min(hi, lo + limit)]
+        elif before_seq is not None:
             # Records with seq < before_seq, then the last `limit` of them.
             hi = bisect.bisect_left(messages, before_seq, key=lambda e: e["seq"])
             return messages[max(0, hi - limit) : hi]
-        elif after_seq is not None:
-            # Records with seq > after_seq, then the first `limit` of them.
-            lo = bisect.bisect_right(messages, after_seq, key=lambda e: e["seq"])
-            return messages[lo : lo + limit]
         else:
             # Return the latest `limit` records, ascending.
             return messages[-limit:]

backend/tests/test_run_event_store_cursor_window.py

@@ -0,0 +1,73 @@
+"""Thread message cursors bound one exclusive window before applying the limit."""
+
+import pytest
+
+from deerflow.runtime.events.store.memory import MemoryRunEventStore
+
+
+@pytest.fixture(params=["memory", "jsonl", "sqlite"])
+async def window_store(request, tmp_path):
+    if request.param == "memory":
+        yield MemoryRunEventStore()
+    elif request.param == "jsonl":
+        from deerflow.runtime.events.store.jsonl import JsonlRunEventStore
+
+        yield JsonlRunEventStore(base_dir=tmp_path)
+    else:
+        from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+        from deerflow.runtime.events.store.db import DbRunEventStore
+
+        await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'events.db'}", sqlite_dir=str(tmp_path))
+        try:
+            yield DbRunEventStore(get_session_factory())
+        finally:
+            await close_engine()
+
+
+@pytest.fixture
+async def populated_window_store(window_store):
+    # Message seqs are 1, 3, 5, 7, 9 across two runs; trace rows fill the gaps.
+    for i in range(1, 10):
+        await window_store.put(
+            thread_id="t1",
+            run_id="r1" if i < 5 else "r2",
+            event_type="llm.ai.response" if i % 2 else "llm.start",
+            category="message" if i % 2 else "trace",
+            content={"type": "ai", "id": f"m{i}", "content": f"message {i}"},
+        )
+    await window_store.put(thread_id="other", run_id="r1", event_type="llm.ai.response", category="message", content="other thread")
+    return window_store
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("after_seq", "before_seq", "expected"),
+    [(1, 9, [3, 5]), (3, 7, [5]), (4, 8, [5, 7]), (7, 7, []), (9, 3, []), (5, 6, []), (9, 20, [])],
+)
+async def test_thread_message_window_applies_both_exclusive_cursors(populated_window_store, after_seq, before_seq, expected):
+    rows = await populated_window_store.list_messages("t1", after_seq=after_seq, before_seq=before_seq, limit=2)
+
+    assert [row["seq"] for row in rows] == expected
+    assert [row["content"]["id"] for row in rows] == [f"m{seq}" for seq in expected]
+
+
+@pytest.mark.anyio
+async def test_forward_pages_keep_a_fixed_upper_bound(populated_window_store):
+    pages = []
+    after_seq = 1
+    for _ in range(4):
+        rows = await populated_window_store.list_messages("t1", after_seq=after_seq, before_seq=9, limit=2)
+        pages.append([row["seq"] for row in rows])
+        if not rows:
+            break
+        after_seq = rows[-1]["seq"]
+
+    assert pages == [[3, 5], [7], []]
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(("cursors", "expected"), [({}, [7, 9]), ({"before_seq": 7}, [3, 5]), ({"after_seq": 3}, [5, 7])])
+async def test_single_cursor_and_latest_pages_are_unchanged(populated_window_store, cursors, expected):
+    rows = await populated_window_store.list_messages("t1", limit=2, **cursors)
+
+    assert [row["seq"] for row in rows] == expected