From 80848837b7d059610dc24055684f9753ae452192 Mon Sep 17 00:00:00 2001
From: Vanzeren <53075619+Vanzeren@users.noreply.github.com>
Date: Fri, 31 Jul 2026 17:25:07 +0800
Subject: [PATCH] feat(gateway): seed checkpoint history (#4590)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
* feat(runtime): seed empty run-event feed from checkpoint history
Threads created before the journaled run-event model hold their history
only in the LangGraph checkpoint. Before the first journaled run, backfill
an empty run-event message feed from the existing checkpoint head so
legacy history receives earlier thread-global seq numbers and remains
visible in the unified feed. Threads with no checkpoint or an already
populated feed skip the path.
The seed guard resolves the user explicitly instead of relying on the
store's AUTO default, which raises without a user contextvar (scheduler
launch path on the DB event store).
* docs: document checkpoint history seeding in thread runs
Before the first journaled run, an empty run-event message feed is
seeded from an existing checkpoint head so legacy checkpoint-only
history stays visible with earlier thread-global sequence numbers.
* fix(gateway): make checkpoint-history seed guard thread-scoped
The emptiness guard filtered by the current user whenever one was in
context, answering "does this user have any messages?" rather than
"has this thread's feed ever been journaled?". Seed rows stamped with
a different principal (NULL for ownerless seeds, or another user on a
shared NULL-owner thread) were invisible to the guard, so each new
principal re-seeded a duplicate history. Pass user_id=None
unconditionally; None also opts out of AUTO resolution, so the
ownerless scheduler path still cannot raise.
Adds a DbRunEventStore-backed regression test (the MemoryRunEventStore
tests cannot catch this — the memory store ignores user_id) proving the
ownerless-seed -> authenticated-run sequence seeds exactly once.
---
backend/AGENTS.md | 2 +-
backend/app/gateway/services.py | 63 ++++
.../harness/deerflow/runtime/journal.py | 62 +++-
backend/tests/test_gateway_services.py | 287 +++++++++++++++++-
.../test_stateless_runs_owner_isolation.py | 5 +
5 files changed, 395 insertions(+), 24 deletions(-)
diff --git a/backend/AGENTS.md b/backend/AGENTS.md
index 2ce7a1e60..1f303d537 100644
--- a/backend/AGENTS.md
+++ b/backend/AGENTS.md
@@ -495,7 +495,7 @@ Localhost persistence deliberately reads the direct request `Host` and ignores `
| **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - serve artifacts; active content types (`text/html`, `application/xhtml+xml`, `image/svg+xml`) are always forced as download attachments to reduce XSS risk; `?download=true` still forces download for other file types |
| **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`...`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing |
| **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) |
-| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block; `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens |
+| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, an empty run-event message feed is seeded from an existing checkpoint head so legacy checkpoint-only history receives earlier thread-global sequence numbers and remains visible after the new run; a thread with no checkpoint or an already-populated feed skips this compatibility path. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens |
| **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific |
| **Runs** (`/api/runs`) | `POST /stream` - stateless run + SSE; `POST /wait` - stateless run + block; `GET /{rid}/messages` - paginated messages by run_id `{data, has_more}` (cursor: `after_seq`/`before_seq`); `GET /{rid}/feedback` - list feedback by run_id |
| **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503, keeping the delivery recorded as failed for manual/API/scripted redelivery (GitHub does not automatically retry any failed delivery, 5xx included); permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. |
diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py
index 44c39f2a2..6d30f494e 100644
--- a/backend/app/gateway/services.py
+++ b/backend/app/gateway/services.py
@@ -61,6 +61,7 @@ from deerflow.runtime.checkpoint_mode import (
)
from deerflow.runtime.checkpoint_state import graph_state_schema
from deerflow.runtime.goal import goal_thread_lock
+from deerflow.runtime.journal import build_checkpoint_history_seed_events
from deerflow.runtime.runs.naming import resolve_root_run_name
from deerflow.runtime.secret_context import (
LegacyRunMetadataSecretError,
@@ -983,6 +984,63 @@ async def apply_checkpoint_to_run_config(
configurable["checkpoint_map"] = checkpoint_map
+async def ensure_checkpoint_history_seeded(
+ request: Request,
+ *,
+ thread_id: str,
+ assistant_id: str | None,
+) -> None:
+ """Backfill an empty run-event feed from an existing checkpoint head.
+
+ No-op unless the feed is empty AND a checkpoint head with messages
+ exists — i.e. a legacy checkpoint-only thread facing its first journaled
+ run. This is a migration shim: remove it once pre-journal threads are no
+ longer a supported upgrade source. The info log on a successful seed is
+ the observability hook for that decision — when it stops appearing, the
+ shim is dead.
+ """
+ event_store = request.app.state.run_event_store
+ # The emptiness check is deliberately thread-scoped, never user-scoped:
+ # seed rows may be stamped with a different principal (NULL for ownerless
+ # seeds, or another user on a shared NULL-owner thread), so a user-scoped
+ # query would miss them and re-seed a duplicate history per principal.
+ # Passing user_id=None also opts out of AUTO resolution explicitly, which
+ # would raise when no user contextvar is set (e.g. the scheduler launch
+ # path for ownerless internal tasks).
+ if await event_store.list_messages(thread_id, limit=1, user_id=None):
+ return
+
+ checkpoint_config = {
+ "configurable": {
+ "thread_id": thread_id,
+ "checkpoint_ns": "",
+ }
+ }
+ if await get_checkpointer(request).aget_tuple(checkpoint_config) is None:
+ return
+
+ accessor, config = build_checkpoint_state_accessor(
+ request,
+ thread_id=thread_id,
+ assistant_id=assistant_id,
+ )
+ snapshot = await accessor.aget(config)
+ values = getattr(snapshot, "values", None)
+ messages = values.get("messages") if isinstance(values, dict) else None
+ if not isinstance(messages, list) or not messages:
+ return
+
+ events = build_checkpoint_history_seed_events(
+ messages,
+ thread_id=thread_id,
+ run_id_prefix=f"checkpoint-seed-{thread_id}",
+ )
+ if not events:
+ return
+ await event_store.put_batch(events)
+ logger.info("Seeded %d checkpoint-history events for thread %s", len(events), thread_id)
+
+
# ---------------------------------------------------------------------------
# Run lifecycle
# ---------------------------------------------------------------------------
@@ -1156,6 +1214,11 @@ async def start_run(
try:
async with goal_thread_lock(thread_id):
+ await ensure_checkpoint_history_seeded(
+ request,
+ thread_id=thread_id,
+ assistant_id=body.assistant_id,
+ )
record = await run_mgr.create_or_reject(
thread_id,
body.assistant_id,
diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py
index 7583d07d0..6f3007fa1 100644
--- a/backend/packages/harness/deerflow/runtime/journal.py
+++ b/backend/packages/harness/deerflow/runtime/journal.py
@@ -87,24 +87,16 @@ def _coerce_seed_message(message: Any) -> Any:
return message
-def build_branch_history_seed_events(
+def _build_history_seed_events(
messages: Sequence[Any],
*,
thread_id: str,
run_id_prefix: str,
- parent_thread_id: str,
+ seed_metadata: Mapping[str, Any],
) -> list[dict]:
- """Serialize a branch checkpoint's messages into run-event message rows.
+ """Serialize checkpoint messages into run-event rows.
- Thread branching copies checkpoint state, but the thread feed
- (``list_messages`` / ``GET /threads/{id}/messages/page``) reads the
- run-event store — which a fresh branch has no rows in, so the inherited
- history vanishes from the UI as soon as the branch's first run refreshes
- the feed (#4380). Seeding the branch's run_events from the same
- checkpoint snapshot the branch was created from keeps the feed
- consistent with what the branch actually contains.
-
- Rows are grouped into one synthetic run per inherited turn
+ Rows are grouped into one synthetic run per checkpoint turn
(``{run_id_prefix}-{n}``), a new turn starting at every persisted human
message — the same boundary a real run has, since a run begins with a
human input (including the allowlisted hidden ``ask_clarification``
@@ -117,9 +109,9 @@ def build_branch_history_seed_events(
id per turn confines the drop to the turn actually regenerated.
Mirrors RunJournal's message-event contract so seeded rows are
- indistinguishable from journaled ones except by the ``branch_seed``
- marker: same event types, ``category="message"``, ``content=
- message.model_dump()``, the human-input persistence rule
+ indistinguishable from journaled ones except by the supplied seed metadata:
+ same event types, ``category="message"``, ``content=message.model_dump()``,
+ the human-input persistence rule
(``_should_persist_human_input_message``), the original-user-text
restoration, and the same treatment of ``hide_from_ui`` AI/tool rows —
RunJournal persists them (``on_llm_end`` / ``_persist_tool_result_message``
@@ -136,7 +128,6 @@ def build_branch_history_seed_events(
"""
events: list[dict] = []
created_at = datetime.now(UTC).isoformat()
- seed_metadata = {"branch_seed": True, "branch_parent_thread_id": parent_thread_id}
# Messages ahead of the first human turn (none in practice) stay in turn 0.
turn_index = 0
for raw_message in messages:
@@ -175,6 +166,45 @@ def build_branch_history_seed_events(
return events
+def build_branch_history_seed_events(
+ messages: Sequence[Any],
+ *,
+ thread_id: str,
+ run_id_prefix: str,
+ parent_thread_id: str,
+) -> list[dict]:
+ """Serialize inherited branch history into the branch's empty event feed."""
+ return _build_history_seed_events(
+ messages,
+ thread_id=thread_id,
+ run_id_prefix=run_id_prefix,
+ seed_metadata={
+ "branch_seed": True,
+ "branch_parent_thread_id": parent_thread_id,
+ },
+ )
+
+
+def build_checkpoint_history_seed_events(
+ messages: Sequence[Any],
+ *,
+ thread_id: str,
+ run_id_prefix: str,
+) -> list[dict]:
+ """Serialize legacy checkpoint history for a thread's empty event feed.
+
+ Reuse the branch seed's message normalization and per-turn synthetic run
+ grouping, but stamp migration-specific metadata so these rows are not
+ misidentified as history inherited from another thread.
+ """
+ return _build_history_seed_events(
+ messages,
+ thread_id=thread_id,
+ run_id_prefix=run_id_prefix,
+ seed_metadata={"checkpoint_history_seed": True},
+ )
+
+
class RunJournal(BaseCallbackHandler):
"""LangChain callback handler that captures events to RunEventStore."""
diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py
index 842546991..9d283a3b6 100644
--- a/backend/tests/test_gateway_services.py
+++ b/backend/tests/test_gateway_services.py
@@ -12,6 +12,7 @@ import pytest
from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL
from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
+from deerflow.runtime.events.store.memory import MemoryRunEventStore
@pytest.fixture
@@ -38,7 +39,7 @@ def _make_start_run_request(run_manager, *, thread_store=None, auth_source=None)
run_manager=run_manager,
checkpointer=InMemorySaver(),
store=store,
- run_event_store=SimpleNamespace(),
+ run_event_store=MemoryRunEventStore(),
run_events_config=None,
thread_store=thread_store or MemoryThreadMetaStore(store),
)
@@ -918,7 +919,6 @@ def test_apply_checkpoint_to_run_config_writes_checkpoint_fields():
config = {"configurable": {"thread_id": "thread-1"}}
asyncio.run(apply_checkpoint_to_run_config(config, body=body, thread_id="thread-1", request=request))
-
assert checkpointer.seen_config == {
"configurable": {
"thread_id": "thread-1",
@@ -932,6 +932,279 @@ def test_apply_checkpoint_to_run_config_writes_checkpoint_fields():
assert config["configurable"]["checkpoint_map"] == {"": "ckpt-1"}
+@pytest.mark.anyio
+async def test_seeded_checkpoint_messages_precede_the_first_new_run_messages():
+ from unittest.mock import AsyncMock, patch
+
+ from langchain_core.messages import AIMessage, HumanMessage
+
+ from app.gateway.services import ensure_checkpoint_history_seeded
+
+ event_store = MemoryRunEventStore()
+ checkpointer = SimpleNamespace(
+ aget_tuple=AsyncMock(return_value=SimpleNamespace(checkpoint={})),
+ )
+ snapshot = SimpleNamespace(
+ values={
+ "messages": [
+ HumanMessage(id="legacy-human", content="old question"),
+ AIMessage(id="legacy-ai", content="old answer"),
+ ]
+ }
+ )
+ accessor = SimpleNamespace(aget=AsyncMock(return_value=snapshot))
+ request = SimpleNamespace(
+ app=SimpleNamespace(
+ state=SimpleNamespace(
+ checkpointer=checkpointer,
+ run_event_store=event_store,
+ )
+ )
+ )
+
+ with patch(
+ "app.gateway.services.build_checkpoint_state_accessor",
+ return_value=(accessor, {"configurable": {"thread_id": "thread-1"}}),
+ ):
+ await ensure_checkpoint_history_seeded(
+ request,
+ thread_id="thread-1",
+ assistant_id="lead_agent",
+ )
+
+ for message_type, message_id in (
+ ("human", "new-human"),
+ ("ai", "new-ai"),
+ ):
+ await event_store.put(
+ thread_id="thread-1",
+ run_id="new-run",
+ event_type=("llm.human.input" if message_type == "human" else "llm.ai.response"),
+ category="message",
+ content={
+ "type": message_type,
+ "id": message_id,
+ "content": message_id,
+ "additional_kwargs": {},
+ },
+ metadata={"caller": "lead_agent"},
+ )
+
+ rows = await event_store.list_messages("thread-1", limit=10)
+ assert [row["content"]["id"] for row in rows] == [
+ "legacy-human",
+ "legacy-ai",
+ "new-human",
+ "new-ai",
+ ]
+ assert [row["seq"] for row in rows] == [1, 2, 3, 4]
+ assert {row["run_id"] for row in rows[:2]} == {"checkpoint-seed-thread-1-1"}
+ assert all(row["metadata"].get("checkpoint_history_seed") is True for row in rows[:2])
+
+
+@pytest.mark.anyio
+async def test_checkpoint_history_seed_skips_new_thread_without_checkpoint():
+ from unittest.mock import AsyncMock, patch
+
+ from app.gateway.services import ensure_checkpoint_history_seeded
+
+ event_store = SimpleNamespace(
+ list_messages=AsyncMock(return_value=[]),
+ put_batch=AsyncMock(),
+ )
+ checkpointer = SimpleNamespace(aget_tuple=AsyncMock(return_value=None))
+ request = SimpleNamespace(
+ app=SimpleNamespace(
+ state=SimpleNamespace(
+ checkpointer=checkpointer,
+ run_event_store=event_store,
+ )
+ )
+ )
+
+ with patch(
+ "app.gateway.services.build_checkpoint_state_accessor",
+ side_effect=AssertionError("new threads should not build an accessor"),
+ ):
+ await ensure_checkpoint_history_seeded(
+ request,
+ thread_id="thread-1",
+ assistant_id="lead_agent",
+ )
+
+ event_store.put_batch.assert_not_awaited()
+
+
+@pytest.mark.anyio
+async def test_checkpoint_history_seed_is_skipped_when_journal_already_has_messages():
+ from unittest.mock import AsyncMock, patch
+
+ from app.gateway.services import ensure_checkpoint_history_seeded
+
+ event_store = SimpleNamespace(
+ list_messages=AsyncMock(return_value=[{"seq": 1}]),
+ put_batch=AsyncMock(),
+ )
+ request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(run_event_store=event_store)))
+
+ with patch(
+ "app.gateway.services.build_checkpoint_state_accessor",
+ side_effect=AssertionError("checkpoint state should not be loaded"),
+ ):
+ await ensure_checkpoint_history_seeded(
+ request,
+ thread_id="thread-1",
+ assistant_id="lead_agent",
+ )
+
+ event_store.put_batch.assert_not_awaited()
+
+
+@pytest.mark.anyio
+@pytest.mark.no_auto_user
+async def test_checkpoint_history_seed_guard_tolerates_missing_user_context():
+ """Scheduler/internal launch paths can run without a user contextvar.
+ DbRunEventStore resolves user_id=AUTO strictly and raises in that case;
+ the seed guard must pass user_id=None explicitly instead of aborting the
+ run."""
+ from unittest.mock import AsyncMock
+
+ from app.gateway.services import ensure_checkpoint_history_seeded
+ from deerflow.runtime.user_context import AUTO, _AutoSentinel
+
+ captured: dict[str, object] = {}
+
+ async def list_messages(thread_id, *, limit=50, before_seq=None, after_seq=None, user_id=AUTO):
+ # Mirror DbRunEventStore: AUTO with no user context raises.
+ if isinstance(user_id, _AutoSentinel):
+ raise RuntimeError("list_messages called with user_id=AUTO but no user context is set")
+ captured["user_id"] = user_id
+ return []
+
+ event_store = SimpleNamespace(list_messages=list_messages, put_batch=AsyncMock())
+ checkpointer = SimpleNamespace(aget_tuple=AsyncMock(return_value=None))
+ request = SimpleNamespace(
+ app=SimpleNamespace(
+ state=SimpleNamespace(
+ checkpointer=checkpointer,
+ run_event_store=event_store,
+ )
+ )
+ )
+
+ await ensure_checkpoint_history_seeded(
+ request,
+ thread_id="thread-1",
+ assistant_id="lead_agent",
+ )
+
+ assert captured["user_id"] is None
+ event_store.put_batch.assert_not_awaited()
+
+
+@pytest.mark.anyio
+async def test_checkpoint_history_seed_guard_is_thread_scoped_under_user_context():
+ """Regression: the emptiness guard must stay thread-scoped (user_id=None)
+ even when a user is authenticated. Seed rows stamped by another principal
+ (or NULL) are invisible to a user-scoped query, which would re-seed a
+ duplicate history per principal."""
+ from unittest.mock import AsyncMock, patch
+
+ from app.gateway.services import ensure_checkpoint_history_seeded
+ from deerflow.runtime.user_context import AUTO
+
+ captured: dict[str, object] = {}
+
+ async def list_messages(thread_id, *, limit=50, before_seq=None, after_seq=None, user_id=AUTO):
+ captured["user_id"] = user_id
+ return [{"seq": 1}]
+
+ event_store = SimpleNamespace(list_messages=list_messages, put_batch=AsyncMock())
+ request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(run_event_store=event_store)))
+
+ with patch(
+ "app.gateway.services.build_checkpoint_state_accessor",
+ side_effect=AssertionError("checkpoint state should not be loaded"),
+ ):
+ await ensure_checkpoint_history_seeded(
+ request,
+ thread_id="thread-1",
+ assistant_id="lead_agent",
+ )
+
+ assert captured["user_id"] is None
+ event_store.put_batch.assert_not_awaited()
+
+
+@pytest.mark.anyio
+@pytest.mark.no_auto_user
+async def test_checkpoint_history_seed_runs_exactly_once_across_principals(tmp_path):
+ """DbRunEventStore regression: an ownerless seed stamps rows with
+ user_id=NULL; a later authenticated run on the same thread must still
+ see them and skip re-seeding (the MemoryRunEventStore-based tests above
+ cannot catch this because the memory store ignores user_id)."""
+ from unittest.mock import AsyncMock, patch
+
+ from langchain_core.messages import AIMessage, HumanMessage
+
+ from app.gateway.services import ensure_checkpoint_history_seeded
+ from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+ from deerflow.runtime.events.store.db import DbRunEventStore
+ from deerflow.runtime.user_context import reset_current_user, set_current_user
+
+ url = f"sqlite+aiosqlite:///{tmp_path / 'events.db'}"
+ await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
+ try:
+ event_store = DbRunEventStore(get_session_factory())
+ checkpointer = SimpleNamespace(
+ aget_tuple=AsyncMock(return_value=SimpleNamespace(checkpoint={})),
+ )
+ snapshot = SimpleNamespace(
+ values={
+ "messages": [
+ HumanMessage(id="legacy-human", content="old question"),
+ AIMessage(id="legacy-ai", content="old answer"),
+ ]
+ }
+ )
+ accessor = SimpleNamespace(aget=AsyncMock(return_value=snapshot))
+ request = SimpleNamespace(
+ app=SimpleNamespace(
+ state=SimpleNamespace(
+ checkpointer=checkpointer,
+ run_event_store=event_store,
+ )
+ )
+ )
+
+ with patch(
+ "app.gateway.services.build_checkpoint_state_accessor",
+ return_value=(accessor, {"configurable": {"thread_id": "thread-1"}}),
+ ):
+ # First seed: ownerless (no user contextvar) — rows stamped NULL.
+ await ensure_checkpoint_history_seeded(
+ request,
+ thread_id="thread-1",
+ assistant_id="lead_agent",
+ )
+ # Second attempt: authenticated user on the same thread — must
+ # see the NULL-stamped rows and skip.
+ token = set_current_user(SimpleNamespace(id="user-a"))
+ try:
+ await ensure_checkpoint_history_seeded(
+ request,
+ thread_id="thread-1",
+ assistant_id="lead_agent",
+ )
+ finally:
+ reset_current_user(token)
+
+ rows = await event_store.list_messages("thread-1", limit=100, user_id=None)
+ assert [row["content"]["id"] for row in rows] == ["legacy-human", "legacy-ai"]
+ finally:
+ await close_engine()
+
+
def test_apply_checkpoint_to_run_config_rejects_missing_checkpoint():
import asyncio
from types import SimpleNamespace
@@ -1362,7 +1635,7 @@ async def _capture_start_run_graph_input(body, *, auth_source=None):
run_manager=run_manager,
checkpointer=InMemorySaver(),
store=InMemoryStore(),
- run_event_store=SimpleNamespace(),
+ run_event_store=MemoryRunEventStore(),
run_events_config=None,
thread_store=MemoryThreadMetaStore(InMemoryStore()),
)
@@ -1403,7 +1676,7 @@ def _make_start_run_persistence_context():
run_manager=RunManager(store=run_store),
checkpointer=InMemorySaver(),
store=InMemoryStore(),
- run_event_store=SimpleNamespace(),
+ run_event_store=MemoryRunEventStore(),
run_events_config=None,
thread_store=thread_store,
checkpoint_channel_mode="full",
@@ -1656,7 +1929,7 @@ def test_start_run_uses_internal_owner_header_for_persistence(_stub_app_config):
run_manager=run_manager,
checkpointer=InMemorySaver(),
store=InMemoryStore(),
- run_event_store=SimpleNamespace(),
+ run_event_store=MemoryRunEventStore(),
run_events_config=None,
thread_store=thread_store,
)
@@ -1745,7 +2018,7 @@ def test_start_run_stamps_internal_owner_guardrail_attribution(_stub_app_config)
run_manager=run_manager,
checkpointer=InMemorySaver(),
store=InMemoryStore(),
- run_event_store=SimpleNamespace(),
+ run_event_store=MemoryRunEventStore(),
run_events_config=None,
thread_store=thread_store,
)
@@ -1827,7 +2100,7 @@ def test_start_run_session_caller_anti_forgery(_stub_app_config):
run_manager=run_manager,
checkpointer=InMemorySaver(),
store=InMemoryStore(),
- run_event_store=SimpleNamespace(),
+ run_event_store=MemoryRunEventStore(),
run_events_config=None,
thread_store=thread_store,
)
diff --git a/backend/tests/test_stateless_runs_owner_isolation.py b/backend/tests/test_stateless_runs_owner_isolation.py
index 6d6521238..b5ab980cf 100644
--- a/backend/tests/test_stateless_runs_owner_isolation.py
+++ b/backend/tests/test_stateless_runs_owner_isolation.py
@@ -81,9 +81,14 @@ def _client(user):
app.state.thread_store = _make_thread_store()
app.state.stream_bridge = MagicMock()
app.state.checkpointer = MagicMock()
+ # start_run's checkpoint-history seeding runs before admission: give the
+ # store/checkpointer async stubs so the seed path sees an empty feed and
+ # no checkpoint head, then skips.
+ app.state.checkpointer.aget_tuple = AsyncMock(return_value=None)
app.state.store = MagicMock()
app.state.run_events_config = None
app.state.run_event_store = MagicMock()
+ app.state.run_event_store.list_messages = AsyncMock(return_value=[])
run_manager = MagicMock()
run_manager.create_or_reject = AsyncMock(side_effect=ConflictError("sentinel: owner check passed"))
app.state.run_manager = run_manager