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