From f82f9b2829145bf4e0d309e683c79d39c21d72f4 Mon Sep 17 00:00:00 2001 From: NanPan <111261006+poijygfdyy@users.noreply.github.com> Date: Wed, 28 Oct 2026 20:59:10 +0800 Subject: [PATCH] fix(events): rebase RunEvent snapshot isolation onto current main (#7423) --- .../harness/deerflow/runtime/AGENTS.md | 9 ++ .../deerflow/runtime/events/store/base.py | 24 +++- .../deerflow/runtime/events/store/memory.py | 57 ++++----- .../test_conversation_access_edge_cases.py | 5 +- .../test_conversation_message_continuation.py | 2 +- backend/tests/test_history_batch_queries.py | 9 +- ...est_memory_run_event_snapshot_isolation.py | 108 ++++++++++++++++++ 7 files changed, 174 insertions(+), 40 deletions(-) create mode 100644 backend/tests/test_memory_run_event_snapshot_isolation.py diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index dd3ec5df9a..773760528c 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -210,6 +210,15 @@ owner, but `AUTO` resolves to the internal user's `make_safe_user_id` form, so Gateway thread/run reads (including run-row lookups) must pass `_run_scope_user_id()` explicitly. +**In-memory RunEvent snapshot ownership** (`runtime/events/store/memory.py`): +The thread- and run-keyed indexes deliberately share their internal event dicts, +but caller-owned `content` and `metadata` must be copied at admission, and +public put/read results must be detached snapshots. Otherwise mutating a write +receipt or a paged read silently rewrites later RunEvent replay and every shared +message projection. The cross-backend contract lives on `RunEventStore` in +`runtime/events/store/base.py`; regression coverage is +`tests/test_memory_run_event_snapshot_isolation.py`. + **Event-store mutation fence** (`runtime/events/store/`): every thread mutation — `put`, `put_batch`, `put_if_absent`, `delete_by_thread`, `delete_by_run` — shares one serialization domain: the per-thread `asyncio` lock, plus (on PostgreSQL) the diff --git a/backend/packages/harness/deerflow/runtime/events/store/base.py b/backend/packages/harness/deerflow/runtime/events/store/base.py index 5340433616..3b636eb9bf 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/base.py +++ b/backend/packages/harness/deerflow/runtime/events/store/base.py @@ -92,6 +92,9 @@ class RunEventStore(abc.ABC): may add documented fields such as DbRunEventStore.user_id 6. find_latest_ai_message_run_ids() returns the newest valid AI message event for each requested ID and performs no storage work for empty input + 7. Mutable inputs and returned records are isolated from store-owned state, + including nested content and metadata. Mutating a write input, receipt, + or dict-returning list result cannot change stored records or later reads. """ @abc.abstractmethod @@ -106,14 +109,18 @@ class RunEventStore(abc.ABC): metadata: dict | None = None, created_at: str | None = None, ) -> dict: - """Write an event, auto-assign seq, return the complete record.""" + """Write an event, assign seq, and return a detached record snapshot. + + Mutating input content/metadata or the returned record, including nested + values, must not mutate stored state. + """ @abc.abstractmethod async def put_batch(self, events: list[dict]) -> list[dict]: """Batch-write events. Used by RunJournal flush buffer. - Each dict's keys match put()'s keyword arguments. - Returns complete records with seq assigned. + Each dict's keys match put()'s keyword arguments. Returned records + are detached snapshots, and mutable inputs must not alias stored rows. """ @abc.abstractmethod @@ -135,7 +142,8 @@ class RunEventStore(abc.ABC): The check and write must be serialized with ordinary writers for the thread. Returns ``(record, created)``. This is the durability primitive used by terminal run receipts, whose recovery path may safely retry - after a worker crash. + after a worker crash. Both newly created and existing event receipts + are detached snapshots. Explicit ``user_id`` stamps the owner for contextless worker writes. Stores without user isolation accept it for interface parity. Existing @@ -171,6 +179,8 @@ class RunEventStore(abc.ABC): When both cursors are supplied, return the first ``limit`` records in the exclusive window ``after_seq < seq < before_seq`` (ascending). + Each returned record is a detached snapshot; mutating nested values must + not alter stored state or later reads. ``user_id`` may be passed explicitly by request-independent callers; user-scoped backends must apply it according to their isolation model. @@ -258,7 +268,8 @@ class RunEventStore(abc.ABC): ``metadata["task_id"]``). ``after_seq`` is a forward cursor returning the first ``limit`` records with seq > after_seq, so callers can page through a single subagent task's events without the run-wide ``limit`` truncating - the tail (#3779). ``user_id`` follows the same explicit-caller semantics + the tail (#3779). Returned records are detached snapshots, including nested + content and metadata. ``user_id`` follows the same explicit-caller semantics as :meth:`list_messages`. """ @@ -282,7 +293,8 @@ class RunEventStore(abc.ABC): When both cursors are supplied, return the first ``limit`` records in the exclusive window ``after_seq < seq < before_seq`` (ascending). - ``user_id`` follows the same explicit-caller semantics as + Returned records are detached snapshots, including nested content and + metadata. ``user_id`` follows the same explicit-caller semantics as :meth:`list_messages`. """ diff --git a/backend/packages/harness/deerflow/runtime/events/store/memory.py b/backend/packages/harness/deerflow/runtime/events/store/memory.py index 7e9dc89cf4..4fc8ac4ff9 100644 --- a/backend/packages/harness/deerflow/runtime/events/store/memory.py +++ b/backend/packages/harness/deerflow/runtime/events/store/memory.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio import bisect import weakref +from copy import deepcopy from datetime import UTC, datetime from deerflow.runtime.events.message_identity import message_identity @@ -66,8 +67,8 @@ class MemoryRunEventStore(RunEventStore): "run_id": run_id, "event_type": event_type, "category": category, - "content": content, - "metadata": metadata or {}, + "content": deepcopy(content), + "metadata": deepcopy(metadata) if metadata is not None else {}, "seq": seq, "created_at": created_at or datetime.now(UTC).isoformat(), } @@ -89,21 +90,23 @@ class MemoryRunEventStore(RunEventStore): metadata=None, created_at=None, ): - return self._put_one( - thread_id=thread_id, - run_id=run_id, - event_type=event_type, - category=category, - content=content, - metadata=metadata, - created_at=created_at, + return deepcopy( + self._put_one( + thread_id=thread_id, + run_id=run_id, + event_type=event_type, + category=category, + content=content, + metadata=metadata, + created_at=created_at, + ) ) async def put_batch(self, events): results = [] for ev in events: record = self._put_one(**ev) - results.append(record) + results.append(deepcopy(record)) return results async def put_if_absent( @@ -124,16 +127,18 @@ class MemoryRunEventStore(RunEventStore): raise RunEventLifecycleRejected(f"Run lifecycle no longer admits terminal evidence for {thread_id}/{run_id}") for event in self._events_by_run.get(thread_id, {}).get(run_id, []): if event["event_type"] == event_type: - return event, False + return deepcopy(event), False return ( - self._put_one( - thread_id=thread_id, - run_id=run_id, - event_type=event_type, - category=category, - content=content, - metadata=metadata, - created_at=created_at, + deepcopy( + self._put_one( + thread_id=thread_id, + run_id=run_id, + event_type=event_type, + category=category, + content=content, + metadata=metadata, + created_at=created_at, + ) ), True, ) @@ -148,14 +153,14 @@ class MemoryRunEventStore(RunEventStore): # 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)] + return deepcopy(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] + return deepcopy(messages[max(0, hi - limit) : hi]) else: # Return the latest `limit` records, ascending. - return messages[-limit:] + return deepcopy(messages[-limit:]) async def list_events( self, @@ -177,7 +182,7 @@ class MemoryRunEventStore(RunEventStore): run_events = [e for e in run_events if (e.get("metadata") or {}).get("task_id") == task_id] if after_seq is not None: run_events = [e for e in run_events if e.get("seq", 0) > after_seq] - return run_events[:limit] + return deepcopy(run_events[:limit]) async def list_messages_by_run(self, thread_id, run_id, *, limit=50, before_seq=None, after_seq=None, user_id: str | None | _AutoSentinel = AUTO): # Per-run, messages-only, seq-sorted: the seq window is a contiguous @@ -191,8 +196,8 @@ class MemoryRunEventStore(RunEventStore): # return the last ``limit`` (the latest page, or the page ending just # before ``before_seq``). Matches the prior filter-based semantics. if after_seq is not None: - return window[:limit] - return window[-limit:] + return deepcopy(window[:limit]) + return deepcopy(window[-limit:]) async def get_last_visible_ai_seq_by_run(self, thread_id, run_ids, *, user_id: str | None | _AutoSentinel = AUTO): result: dict[str, int] = {} diff --git a/backend/tests/test_conversation_access_edge_cases.py b/backend/tests/test_conversation_access_edge_cases.py index 00c5cb8f25..e0ad3b87b0 100644 --- a/backend/tests/test_conversation_access_edge_cases.py +++ b/backend/tests/test_conversation_access_edge_cases.py @@ -50,7 +50,7 @@ def test_visible_text_is_parsed_once_per_scan_and_refreshed_on_the_next_read(mon await threads.create("source", user_id="alice") await _put(events, "oldest") await _put(events, "privateolder") - newest = await _put(events, [{"type": "text", "text": "newest"}, {"type": "text", "text": "answer"}]) + await _put(events, [{"type": "text", "text": "newest"}, {"type": "text", "text": "answer"}]) reader, _ = prepare(["source"]) page = json.loads(await reader(thread_id="source", limit=1)) @@ -59,7 +59,8 @@ def test_visible_text_is_parsed_once_per_scan_and_refreshed_on_the_next_read(mon assert calls == {3: 1, 2: 1} calls.clear() - newest["content"]["content"] = "updated answer" + # Model a store-side content refresh; the public receipt is intentionally detached. + events._events["source"][2]["content"]["content"] = "updated answer" refreshed = json.loads(await reader(thread_id="source", limit=1)) assert refreshed["messages"][0]["text"] == "updated answer" assert calls == {3: 1, 2: 1} diff --git a/backend/tests/test_conversation_message_continuation.py b/backend/tests/test_conversation_message_continuation.py index 08dadf0640..72e2ccaf42 100644 --- a/backend/tests/test_conversation_message_continuation.py +++ b/backend/tests/test_conversation_message_continuation.py @@ -199,7 +199,7 @@ def test_offsets_follow_the_current_text_of_a_live_source(): return json.loads(await reader(thread_id="source", message_seq=row["seq"], offset=offset)) at_end, beyond = await read(6000), await read(6001) - row["content"]["content"] = "b" * 5000 # the source is edited between reads + events._events["source"][row["seq"] - 1]["content"]["content"] = "b" * 5000 # model a store-side edit; receipts are detached return at_end, beyond, await read(4000), await read(5500) at_end, beyond, edited, shrunk = asyncio.run(exercise()) diff --git a/backend/tests/test_history_batch_queries.py b/backend/tests/test_history_batch_queries.py index cf104e8bc7..75db6e2958 100644 --- a/backend/tests/test_history_batch_queries.py +++ b/backend/tests/test_history_batch_queries.py @@ -74,7 +74,7 @@ async def test_memory_event_store_defensively_rechecks_message_category(): content={"type": "ai", "content": "visible"}, metadata={"caller": "lead_agent"}, ) - mutated = await store.put( + await store.put( thread_id="t1", run_id="r1", event_type="llm.ai.response", @@ -82,10 +82,9 @@ async def test_memory_event_store_defensively_rechecks_message_category(): content={"type": "ai", "content": "no longer a message"}, metadata={"caller": "lead_agent"}, ) - # Memory projections intentionally share their row dictionaries. Recheck - # category at read time so an accidental mutation cannot violate the same - # contract that the DB and JSONL stores enforce explicitly. - mutated["category"] = "trace" + # Receipts are detached snapshots now. Corrupt the private shared row + # deliberately so this test still exercises the defensive category recheck. + store._messages_by_run["t1"]["r1"][1]["category"] = "trace" assert await store.get_last_visible_ai_seq_by_run("t1", {"r1"}) == {"r1": expected["seq"]} diff --git a/backend/tests/test_memory_run_event_snapshot_isolation.py b/backend/tests/test_memory_run_event_snapshot_isolation.py new file mode 100644 index 0000000000..fdf2e406e5 --- /dev/null +++ b/backend/tests/test_memory_run_event_snapshot_isolation.py @@ -0,0 +1,108 @@ +"""The in-memory event backend must behave like durable snapshot storage.""" + +import pytest + +from deerflow.runtime.events.store.memory import MemoryRunEventStore + + +@pytest.mark.asyncio +@pytest.mark.parametrize("writer", ["put", "put_batch", "put_if_absent"]) +async def test_persisted_events_do_not_alias_writer_inputs_or_receipts(writer): + store = MemoryRunEventStore() + payload = {"parts": [{"text": "original"}]} + metadata = {"audit": {"labels": ["original"]}} + event = { + "thread_id": "thread", + "run_id": "run", + "event_type": "llm.ai.response", + "category": "message", + "content": payload, + "metadata": metadata, + } + + if writer == "put": + receipt = await store.put(**event) + elif writer == "put_batch": + receipt = (await store.put_batch([event]))[0] + else: + receipt, created = await store.put_if_absent(**event) + assert created + + payload["parts"][0]["text"] = "changed by writer" + metadata["audit"]["labels"].append("changed by writer") + receipt["content"]["parts"][0]["text"] = "changed by receipt" + receipt["metadata"]["audit"]["labels"].append("changed by receipt") + + stored = (await store.list_events("thread", "run"))[0] + assert stored["content"] == {"parts": [{"text": "original"}]} + assert stored["metadata"] == {"audit": {"labels": ["original"]}} + + +@pytest.mark.parametrize( + ("reader", "cursor"), + [ + pytest.param("list_events", {}, id="list-events-latest"), + pytest.param("list_events", {"after_seq": 1, "limit": 1}, id="list-events-after-seq"), + pytest.param("list_messages", {}, id="list-messages-latest"), + pytest.param("list_messages", {"before_seq": 3, "limit": 1}, id="list-messages-before-seq"), + pytest.param("list_messages", {"after_seq": 1, "limit": 1}, id="list-messages-after-seq"), + pytest.param("list_messages_by_run", {}, id="list-messages-by-run-latest"), + pytest.param("list_messages_by_run", {"before_seq": 3, "limit": 1}, id="list-messages-by-run-before-seq"), + pytest.param("list_messages_by_run", {"after_seq": 1, "limit": 1}, id="list-messages-by-run-after-seq"), + ], +) +@pytest.mark.asyncio +async def test_reader_results_are_detached_snapshots(reader, cursor): + store = MemoryRunEventStore() + for seq in range(1, 4): + await store.put( + thread_id="thread", + run_id="run", + event_type="llm.ai.response", + category="message", + content={"parts": [f"saved-{seq}"]}, + metadata={"audit": {"tags": [f"saved-{seq}"]}}, + ) + + if reader == "list_events": + results = await store.list_events("thread", "run", **cursor) + elif reader == "list_messages": + results = await store.list_messages("thread", **cursor) + else: + results = await store.list_messages_by_run("thread", "run", **cursor) + + assert results + results[0]["content"]["parts"].append("changed by reader") + results[0]["metadata"]["audit"]["tags"].append("changed by reader") + results[0]["seq"] = -1 + + stored = await store.list_events("thread", "run") + assert [event["seq"] for event in stored] == [1, 2, 3] + for event in stored: + seq = event["seq"] + assert event["content"] == {"parts": [f"saved-{seq}"]} + assert event["metadata"] == {"audit": {"tags": [f"saved-{seq}"]}} + + assert [event["seq"] for event in await store.list_messages("thread")] == [1, 2, 3] + assert [event["seq"] for event in await store.list_messages_by_run("thread", "run")] == [1, 2, 3] + + +@pytest.mark.asyncio +async def test_put_if_absent_existing_receipt_is_not_the_stored_event(): + store = MemoryRunEventStore() + await store.put( + thread_id="thread", + run_id="run", + event_type="run.end", + category="lifecycle", + content={"status": "success"}, + metadata={"audit": {"source": "worker"}}, + ) + existing, created = await store.put_if_absent(thread_id="thread", run_id="run", event_type="run.end", category="lifecycle") + assert not created + existing["metadata"]["audit"]["source"] = "modified" + existing["content"]["status"] = "error" + + saved = (await store.list_events("thread", "run"))[0] + assert saved["content"] == {"status": "success"} + assert saved["metadata"] == {"audit": {"source": "worker"}}