From a8dc1da2057897299829c3fff4769756b95c79b7 Mon Sep 17 00:00:00 2001 From: Inference_ <68734681+Inference1@users.noreply.github.com> Date: Thu, 1 Oct 2026 20:11:10 +0800 Subject: [PATCH] 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 | 8 ++ .../harness/deerflow/runtime/AGENTS.md | 2 + .../deerflow/runtime/events/store/base.py | 6 ++ .../deerflow/runtime/events/store/jsonl.py | 3 +- .../deerflow/runtime/events/store/memory.py | 12 +-- .../test_run_event_store_cursor_window.py | 73 +++++++++++++++++++ 6 files changed, 97 insertions(+), 7 deletions(-) create mode 100644 backend/tests/test_run_event_store_cursor_window.py diff --git a/backend/README.md b/backend/README.md index b5f69d53f1..6534cf4487 100644 --- a/backend/README.md +++ b/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: diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 42bc4960f0..411f5eb513 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/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. diff --git a/backend/packages/harness/deerflow/runtime/events/store/base.py b/backend/packages/harness/deerflow/runtime/events/store/base.py index e59a17a5d5..afd54be0a1 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/base.py +++ b/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 diff --git a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py b/backend/packages/harness/deerflow/runtime/events/store/jsonl.py index 647410f689..4a79a0bad1 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/jsonl.py +++ b/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: diff --git a/backend/packages/harness/deerflow/runtime/events/store/memory.py b/backend/packages/harness/deerflow/runtime/events/store/memory.py index 1796eeb980..8b8475e821 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/memory.py +++ b/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:] diff --git a/backend/tests/test_run_event_store_cursor_window.py b/backend/tests/test_run_event_store_cursor_window.py new file mode 100644 index 0000000000..cb1794c176 --- /dev/null +++ b/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