From 9ad79baf97e940c1c737383873f2bf1200d0211a Mon Sep 17 00:00:00 2001 From: wutongyuonce <147830929+wutongyuonce@users.noreply.github.com> Date: Tue, 8 Sep 2026 14:40:05 +0800 Subject: [PATCH] feat(gateway): support idempotent thread runs (#5258) * feat(gateway): support idempotent thread runs Accept Idempotency-Key on thread-scoped create, stream, and wait endpoints, scoped by owner and thread before durable admission.\n\nRefs #5257. * fix(gateway): handle idempotent run reuse on wait and stream Reused store-only records have no local task. /wait now waits on the bridge when it can observe the stream, and otherwise returns durable status instead of a stale checkpoint. A reused terminal stream that has been evicted emits gap/reload_durable_state. Replay is bound to the original input and assistant_id. * fix(gateway): 409 reused in-flight streams on this worker A store-only running record on a process-local bridge has no owner stream. POST /runs/stream used to subscribe anyway, which created an empty log and waited forever. Match join: 409 unless the run is already terminal, so missing-stream retries can still emit gap. * fix(gateway): keep observer joins off the idempotent stream-gap path sse_consumer keyed missing-stream gap on the sticky idempotency_reused flag, so a later join of a terminal run inherited it. Gate that branch on apply_on_disconnect, which already separates creating streams from joins. Document the retry outcomes clients have to handle. * fix(gateway): gate missing-stream gap on creating retry Reuse apply_on_disconnect to pick gap vs end changed sse_consumer default path, so a missing stream started returning gap for default callers and for out-of-scope POST /api/runs/stream. Keep that branch behind emit_gap_on_missing_stream and pass it only from thread-scoped /runs/stream on this request reuse. * fix(gateway): keep wait reuse off later checkpoints Direct handler calls were crashing because FastAPI Header() leaked in as the Python default. Bind Idempotency-Key with Annotated so the default is None, and ignore non-str keys. A reused completed /wait was still reading the latest thread checkpoint. After a later run on the same thread that is the later run's result. Return durable status instead of claiming the head as this run's output. * fix(gateway): snapshot wait reuse and refresh store status idempotency_reused lives on the shared cached record. Capture it before awaiting completion so an overlapping retry cannot suppress the original creating /wait checkpoint. A store-only peer record still holds admission-time status after the owner publishes END. Refresh durable status/error before returning them. --- backend/app/gateway/AGENTS.md | 1 + backend/app/gateway/routers/thread_runs.py | 149 +++- backend/app/gateway/services.py | 29 + backend/docs/API.md | 31 + backend/tests/test_sse_observer_disconnect.py | 7 +- backend/tests/test_thread_run_idempotency.py | 792 ++++++++++++++++++ 6 files changed, 993 insertions(+), 16 deletions(-) create mode 100644 backend/tests/test_thread_run_idempotency.py diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index f1223fccc..21e1d4151 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -128,6 +128,7 @@ startup gate rejects process-local memory and JSONL event stores when - `POST /wait` (both thread-scoped and `/api/runs/wait`) drains the stream bridge via `wait_for_run_completion()` instead of bare `await record.task`, so it honours the run's `on_disconnect` setting and cancels the background run on real client disconnect rather than returning a stale checkpoint (issue #3265). - Memory and Redis `StreamBridge` implementations retain only `stream_bridge.queue_maxsize` data events. A syntactically valid `Last-Event-ID` older than the retained watermark, or a live subscriber that falls behind it, yields `StreamGap` before any partial replay. `sse_consumer` maps that control item to an id-less SSE `gap` payload (`stream_replay_gap`) and intentionally leaves the run active; internal `/wait` consumers resume from its latest retained ID because they only need terminal completion. Redis checks bounds plus the non-blocking read in one transaction, using blocking `XREAD` only as a wake-up before repeating the atomic snapshot. For a no-cursor subscriber that established a wait on an empty stream, the first wake response remains provisional until that next snapshot verifies its tail is still retained; this closes the pre-first-delivery trimming window without changing malformed-cursor live tailing. The correctness tradeoff is one three-command snapshot pipeline per poll plus the blocking wake round trip while idle. Malformed cursor behavior remains backend-specific. Memory treats a syntactically numeric cursor below its watermark conservatively as a gap even when the evicted timestamp can no longer be verified; unknown ids at or above the watermark retain the legacy replay-from-earliest policy. - Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup and lease-driven periodic orphan recovery share one Gateway stream-terminalization path: after `RunManager` durably marks a run `error` with `stop_reason=orphan_recovered`, Gateway publishes `END_SENTINEL` and schedules stream cleanup. The periodic store scan, per-row status writes, and Gateway callback run as one supervised single-flight task, so a slow pass is skipped at the next interval instead of piling up or pausing the sole lease-renewal loop. Store retries have bounded attempts/backoff; an individual operation still relies on the database driver/pool timeout. `RunManager.shutdown()` gives active user runs priority within its shared deadline, then drains or cancels orphan recovery. Gateway tracks delayed recovered-stream cleanups and converts unfinished delays to immediate deletes before closing the bridge; the Redis TTL remains the outage safety net. Only startup recovery, before the runtime yields to requests, projects the latest affected thread to `error`; periodic recovery deliberately avoids that non-atomic projection because `ThreadMetaStore` has no `latest_run_id` conditional-update contract. Store-only SSE and `/wait` consumers wait for the bridge's real END marker after an ordinary durable terminal status, because status persistence can precede tail events. The explicit `orphan_recovered` signal is the only heartbeat fallback: its publisher is known to be gone, so it supplies the liveness boundary if END publication fails or the retained key expires. Malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Keep cross-component recovery orchestration in Gateway through the generic `RunManager.on_orphans_recovered` callback; do not introduce a harness-to-app dependency. Callback failure warnings include every recovered `run_id` so operators can identify rows whose Gateway-side terminalization needs inspection. +- Thread-scoped run creation accepts an optional `Idempotency-Key` header on create, stream, and wait. Gateway hashes the caller key with the authenticated owner and `thread_id` before passing it to `RunManager`, whose persistence index is process-wide; never pass an unscoped external key to that index. The same scoped key is shared across `/runs`, `/runs/stream`, and `/runs/wait`; a reused admission whose stored `input` or `assistant_id` differs from the retry returns 409. `/wait` must not treat `task is None` as completion: `store_only` records without a cross-process bridge return durable `status`/`error` instead of serializing the current checkpoint; otherwise wait on the bridge. An idempotent reuse must not serialize the latest thread checkpoint as this run's result — a later run on the same thread may have advanced the head — so reused `/wait` returns durable `status`/`error`. Capture that reuse decision before awaiting completion; `idempotency_reused` is sticky on the shared cached record and an overlapping retry must not suppress the original creating request's checkpoint. After observing completion, refresh store-backed `status`/`error` before returning them — a hydrated peer record still holds admission-time fields. A creating-endpoint retry of a terminal record whose stream is gone emits SSE `gap`/`stream_replay_gap` with `recovery: reload_durable_state` rather than a bare `end`; observer joins of that same record still emit `end`. That gap is opt-in via `sse_consumer(..., emit_gap_on_missing_stream=True)` from thread-scoped `/runs/stream` on this request's reuse — do not key it off `apply_on_disconnect` or the sticky `idempotency_reused` flag. Default `sse_consumer` callers, including stateless `/api/runs/stream`, still emit `end`. A reused still-running `store_only` record on a process-local bridge returns 409 from `/stream` with no `Retry-After`, matching `join`. Missing headers preserve ordinary non-idempotent admission. Stateless `/api/runs/*` stays outside this contract because a request without an explicit thread creates a fresh temporary thread before admission. - Thread-scoped run creation accepts `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching. In `delta` checkpoint mode the worker rewrites that fork into a linear head write before the graph starts (see "A delta-mode run cannot fork" under Checkpoint Channel Modes), because delta state for a fork replays the abandoned sibling's writes. - Thread-scoped Gateway runs evaluate an active `ThreadState.goal` after the visible turn completes. `runtime/goal.py` asks a non-thinking evaluator model to judge only visible conversation evidence and return a typed blocker; the evaluator model is created once per run and reused across hidden continuation checks. The evaluator runs after the graph root's tracing scope has already closed, so `create_goal_evaluator_model`/`evaluate_goal_completion` attach their own model-level tracing callbacks (`attach_tracing=True`) and inject Langfuse trace metadata (`thread_id`/`user_id`/`deerflow_trace_id`) directly onto the `ainvoke` call — the same standalone-caller pattern as `oneshot_llm.run_oneshot_llm` and `MemoryUpdater` (see Tracing System below). Satisfied goals are cleared; every non-satisfied evaluation — continuable or stand-down — is persisted with `last_evaluation` (the blocker, reason, and evidence summary; outcomes that stop the loop additionally record a `stand_down_reason` for observability), but only `goal_not_met_yet` evaluations are streamed as hidden `HumanMessage` continuations, and only when a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the thread did not change during evaluation, and the no-progress breaker has not fired. The continuation cap is 8 — a hard maximum in the `0`–`8` range; callers requesting more are clamped (`set_goal`/TUI) or rejected with 422 (`PUT /goal`). The no-progress breaker keys on the latest visible assistant evidence (not the evaluator's free-text reason, which an LLM rewords every turn), so two consecutive continuations that add no new visible assistant output stop the loop after 2 attempts. Model-response cleanup helpers such as think-block stripping and code-fence stripping live in `deerflow.utils.llm_text` so `runtime/goal.py` and Gateway suggestion parsing share the same JSON-prep behavior. - Run event stream changes must keep producer code, `deerflow/constants.py`, `runtime/events/catalog.py`, `contracts/run_event_stream_contract.json`, `backend/docs/RUN_EVENT_STREAM.md`, and `tests/test_run_event_stream_contract.py` in sync. The dependency-free constants module owns the persisted envelope limits (`event_type` 32 characters, `category` 16) and cross-layer workspace event identity; the catalog owns validated runtime definitions and categories. Dynamic middleware tags are limited to 21 characters after the `middleware:` prefix. The JSON contract owns payload schemas, backend-specific storage semantics, legacy aliases, and compatibility rules; conformance tests require both views and all producer groups to agree. `run.end.content` remains opaque and may retain nested Python values in memory while JSONL/database stores stringify non-JSON nested values, so consumers must not assume backend-identical nested output representations. diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index a274bc5ed..bda521fa7 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -12,14 +12,15 @@ works without modification. from __future__ import annotations import asyncio +import hashlib import logging import re import uuid from copy import deepcopy from datetime import UTC, datetime -from typing import Any, Literal +from typing import Annotated, Any, Literal -from fastapi import APIRouter, Depends, HTTPException, Query, Request +from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request from fastapi.responses import Response, StreamingResponse from langchain_core.messages import BaseMessage from pydantic import BaseModel, Field @@ -66,6 +67,60 @@ _UNSAFE_REGENERATE_LINEAGE_DETAIL = "Could not safely resolve the checkpoint bef THREAD_MESSAGE_LEGACY_SCAN_BATCH = 201 +IdempotencyKeyHeader = Annotated[ + str | None, + Header( + alias="Idempotency-Key", + max_length=255, + description="Retry key for idempotent run admission within this thread", + ), +] + + +def _scope_http_run_idempotency_key(request: Request, thread_id: str, key: str | None) -> str | None: + """Namespace a caller key for the process-wide run idempotency index.""" + if not isinstance(key, str): + return None + key = key.strip() + if not key: + raise HTTPException(status_code=422, detail="Idempotency-Key must not be blank") + owner_id = get_trusted_internal_owner_user_id(request) + if owner_id is None: + user = getattr(request.state, "user", None) + user_id = getattr(user, "id", None) + owner_id = str(user_id) if user_id is not None else get_effective_user_id() + digest = hashlib.sha256(f"{owner_id}\0{thread_id}\0{key}".encode()).hexdigest() + return f"http-run:{digest}" + + +async def _refresh_store_backed_run(run_mgr: Any, record: Any) -> Any: + """Overlay durable status/error onto a hydrated store-only record.""" + if not getattr(record, "store_only", False): + return record + store = getattr(run_mgr, "_store", None) + get = getattr(store, "get", None) + if get is None: + return record + try: + row = get(record.run_id) + if hasattr(row, "__await__"): + row = await row + except Exception: + logger.exception("Failed to refresh store-backed run %s", getattr(record, "run_id", None)) + return record + if not isinstance(row, dict): + return record + raw_status = row.get("status") + if raw_status: + try: + record.status = RunStatus(raw_status) + except ValueError: + pass + if "error" in row: + record.error = row.get("error") + return record + + def _is_duration_only_checkpoint(checkpoint_tuple: Any) -> bool: return is_duration_only_checkpoint(checkpoint_tuple) @@ -861,15 +916,30 @@ async def prepare_edit_regenerate_run( @router.post("/{thread_id}/runs", response_model=RunResponse) @require_permission("runs", "create", owner_check=True, require_existing=True) -async def create_run(thread_id: ThreadId, body: RunCreateRequest, request: Request) -> RunResponse: +async def create_run( + thread_id: ThreadId, + body: RunCreateRequest, + request: Request, + idempotency_key: IdempotencyKeyHeader = None, +) -> RunResponse: """Create a background run (returns immediately).""" - record = await start_run(body, thread_id, request) + record = await start_run( + body, + thread_id, + request, + idempotency_key=_scope_http_run_idempotency_key(request, thread_id, idempotency_key), + ) return _record_to_response(record) @router.post("/{thread_id}/runs/stream") @require_permission("runs", "create", owner_check=True, require_existing=True) -async def stream_run(thread_id: ThreadId, body: RunCreateRequest, request: Request) -> StreamingResponse: +async def stream_run( + thread_id: ThreadId, + body: RunCreateRequest, + request: Request, + idempotency_key: IdempotencyKeyHeader = None, +) -> StreamingResponse: """Create a run and stream events via SSE. The response includes a ``Content-Location`` header with the run's @@ -878,10 +948,32 @@ async def stream_run(thread_id: ThreadId, body: RunCreateRequest, request: Reque """ bridge = get_stream_bridge(request) run_mgr = get_run_manager(request) - record = await start_run(body, thread_id, request) + record = await start_run( + body, + thread_id, + request, + idempotency_key=_scope_http_run_idempotency_key(request, thread_id, idempotency_key), + ) + + # Same shape join already rejects: a reused store-only handle on a + # process-local bridge has no owner stream. Subscribing would create an + # empty log and wait forever. Terminal reuse still goes through + # sse_consumer with emit_gap_on_missing_stream so a missing stream emits + # gap rather than a bare end. First-time creates keep the default `end`. + if record.store_only and not bridge.supports_cross_process and record.status in (RunStatus.pending, RunStatus.running): + raise HTTPException( + status_code=409, + detail=f"Run {record.run_id} is not active on this worker and cannot be streamed", + ) return StreamingResponse( - sse_consumer(bridge, record, request, run_mgr), + sse_consumer( + bridge, + record, + request, + run_mgr, + emit_gap_on_missing_stream=record.idempotency_reused, + ), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", @@ -897,17 +989,46 @@ async def stream_run(thread_id: ThreadId, body: RunCreateRequest, request: Reque @router.post("/{thread_id}/runs/wait", response_model=dict) @require_permission("runs", "create", owner_check=True, require_existing=True) -async def wait_run(thread_id: ThreadId, body: RunCreateRequest, request: Request) -> dict: - """Create a run and block until it completes, returning the final state.""" +async def wait_run( + thread_id: ThreadId, + body: RunCreateRequest, + request: Request, + idempotency_key: IdempotencyKeyHeader = None, +) -> dict: + """Create a run and block until it completes, returning the final state. + + A reused in-flight run that this worker cannot observe returns the durable + status without blocking. A reused completed run also returns durable + status: the latest thread checkpoint may belong to a later run. + """ bridge = get_stream_bridge(request) run_mgr = get_run_manager(request) - record = await start_run(body, thread_id, request) + record = await start_run( + body, + thread_id, + request, + idempotency_key=_scope_http_run_idempotency_key(request, thread_id, idempotency_key), + ) + # Capture before waiting: create_or_reject mutates the shared cached + # record's idempotency_reused flag, so an overlapping retry must not + # change this request's checkpoint-vs-status decision. + reused = bool(getattr(record, "idempotency_reused", False)) - completed = True - if record.task is not None: + # Reused/hydrated records have no local task. Wait on the bridge when this + # worker can observe it; otherwise return durable status rather than + # serializing whatever checkpoint happens to exist. + if getattr(record, "store_only", False) and not getattr(bridge, "supports_cross_process", False): + record = await _refresh_store_backed_run(run_mgr, record) + return {"status": record.status.value, "error": record.error} + + if record.task is not None or getattr(record, "store_only", False): completed = await wait_for_run_completion(bridge, record, request, run_mgr) + else: + completed = True - if completed: + # Idempotent reuse is not bound to a run-specific checkpoint id. The latest + # thread head may be a later run, so do not claim it as this run's result. + if completed and not reused: try: accessor, config = build_checkpoint_state_accessor( request, @@ -921,6 +1042,8 @@ async def wait_run(thread_id: ThreadId, body: RunCreateRequest, request: Request except Exception: logger.exception("Failed to fetch final state for run %s", record.run_id) + if completed: + record = await _refresh_store_backed_run(run_mgr, record) return {"status": record.status.value, "error": record.error} diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index e6102c43c..dbcbddfdd 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -1479,6 +1479,12 @@ async def start_run( ) if record.idempotency_reused: + stored = record.kwargs or {} + if stored.get("input") != body.input or record.assistant_id != body.assistant_id: + raise HTTPException( + status_code=409, + detail="Idempotency-Key already used with a different request", + ) return record worker = run_after_metadata(record) @@ -1673,6 +1679,7 @@ async def sse_consumer( run_mgr: RunManager, *, apply_on_disconnect: bool = True, + emit_gap_on_missing_stream: bool = False, ): """Async generator that yields SSE frames from the bridge. @@ -1687,9 +1694,31 @@ async def sse_consumer( connection, and a read-only observer closing a join must not cancel the run (a runs:read-only credential would otherwise cancel without runs:cancel just by disconnecting). + + ``emit_gap_on_missing_stream`` is a separate creating-retry signal, default + ``False``. ``create_or_reject`` sets ``record.idempotency_reused`` on the + shared cached record and never clears it, so this function must not read + that flag. Thread-scoped ``/runs/stream`` passes True only for this + request's reuse; default callers (joins, stateless ``/api/runs/stream``, + tests) keep ``end`` when a terminal record's stream is gone. """ last_event_id = request.headers.get("Last-Event-ID") if await _terminal_record_stream_missing(bridge, record): + if emit_gap_on_missing_stream: + # Creating-endpoint retry: a bare `end` looks like the run + # produced nothing. Point the client at durable state instead. + yield format_sse( + "gap", + { + "code": "stream_replay_gap", + "run_id": record.run_id, + "requested_event_id": last_event_id, + "earliest_available_event_id": None, + "latest_available_event_id": None, + "recovery": "reload_durable_state", + }, + ) + return yield format_sse("end", None) return diff --git a/backend/docs/API.md b/backend/docs/API.md index f7a90e621..3284e9fe6 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -189,8 +189,38 @@ Execute the agent with input. ```http POST /api/langgraph/threads/{thread_id}/runs Content-Type: application/json +Idempotency-Key: # optional ``` +The thread-scoped create, stream, and wait endpoints accept an optional +`Idempotency-Key` header. Retrying with the same authenticated user, `thread_id`, +and key reuses the existing run instead of executing the input again. The key is +shared across `/runs`, `/runs/stream`, and `/runs/wait` for a given user and +thread, so the same key string cannot back two different calls even across those +endpoints. Reuse is bound to the original `input` and `assistant_id`; a retry +that changes either returns 409. Generate a new key for every intentional user +action; reuse a key only when retrying that same action after an uncertain HTTP +result. Keys may be at most 255 characters. Stateless `/api/langgraph/runs/*` +endpoints do not support this header because requests without an explicit thread +create a new temporary conversation. + +Retrying a still-running run that this worker cannot stream returns 409 from +`/runs/stream` (`Run ... is not active on this worker and cannot be streamed`) +with no `Retry-After`. The same shape on `/runs/wait` returns 200 +`{"status": "", "error": ...}` without blocking for a final +state. Retrying a finished run through `/runs/wait` also returns that durable +status payload rather than the latest thread checkpoint: a later run on the +same thread may have advanced the head, and `/wait` does not claim that head +as this run's result. That status is the durable row after completion, not +the hydrated record from admission time. The original creating `/wait` still +returns this run's checkpoint even if a retry overlaps while it is waiting. Retrying a finished run whose SSE log is gone emits a `gap` frame +(`stream_replay_gap`, `recovery: reload_durable_state`) on the creating +`/runs/stream` endpoint and closes without an `end` frame; reload durable +thread/run state instead of treating the stream as empty. Observer joins of +that same run still end with `end`. Stateless `/api/langgraph/runs/stream` +does not accept this header and keeps the existing missing-stream close of +`end`; the `gap` signal is only on a thread-scoped creating retry. + **Request Body:** ```json { @@ -290,6 +320,7 @@ Stream responses in real-time. ```http POST /api/langgraph/threads/{thread_id}/runs/stream Content-Type: application/json +Idempotency-Key: # optional ``` Same request body as Create Run. Returns SSE stream. diff --git a/backend/tests/test_sse_observer_disconnect.py b/backend/tests/test_sse_observer_disconnect.py index 99e90db97..dd4e8694e 100644 --- a/backend/tests/test_sse_observer_disconnect.py +++ b/backend/tests/test_sse_observer_disconnect.py @@ -106,9 +106,10 @@ def test_join_routes_wire_sse_consumer_as_observers(): thread_runs_source = inspect.getsource(thread_runs) # join_run + the shared existing-run stream implementation assert thread_runs_source.count("sse_consumer(bridge, record, request, run_mgr, apply_on_disconnect=False)") == 2 - # stream_run — the creator's create-and-stream endpoint - assert thread_runs_source.count("sse_consumer(bridge, record, request, run_mgr),") == 1 + # stream_run — creating retry opts into missing-stream gap; first create does not + assert "emit_gap_on_missing_stream=record.idempotency_reused" in thread_runs_source runs_source = inspect.getsource(runs_router) - # stateless create-and-stream — also a creator stream + # stateless create-and-stream — creator on_disconnect policy, not the retry gap assert "sse_consumer(bridge, record, request, run_mgr, apply_on_disconnect=False)" not in runs_source + assert "emit_gap_on_missing_stream" not in runs_source diff --git a/backend/tests/test_thread_run_idempotency.py b/backend/tests/test_thread_run_idempotency.py new file mode 100644 index 000000000..cd4538100 --- /dev/null +++ b/backend/tests/test_thread_run_idempotency.py @@ -0,0 +1,792 @@ +"""HTTP contract tests for idempotent thread-run creation (issue #5257).""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch +from uuid import uuid4 + +import pytest +from _router_auth_helpers import call_unwrapped, make_authed_test_app +from fastapi import HTTPException +from fastapi.testclient import TestClient + +from app.gateway.auth.models import User +from app.gateway.routers import thread_runs +from app.gateway.run_models import RunCreateRequest +from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config +from deerflow.runtime import DisconnectMode, RunManager, RunRecord, RunStatus +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.runs.store.memory import MemoryRunStore + + +def _user(email: str) -> User: + return User(email=email, password_hash="x", system_role="user", id=uuid4()) + + +def _run(run_id: str, thread_id: str) -> RunRecord: + return RunRecord( + run_id=run_id, + thread_id=thread_id, + assistant_id=None, + status=RunStatus.success, + on_disconnect=DisconnectMode.continue_, + error=run_id, + ) + + +def _make_client(monkeypatch, user: User, admissions: dict[str, RunRecord]) -> TestClient: + async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False): + del body, request, require_existing_thread + if idempotency_key is not None and idempotency_key in admissions: + record = admissions[idempotency_key] + record.idempotency_reused = True + return record + record = _run(f"run-{len(admissions) + 1}", thread_id) + admissions[idempotency_key or f"unkeyed-{record.run_id}"] = record + return record + + monkeypatch.setattr(thread_runs, "start_run", fake_start_run) + app = make_authed_test_app(user_factory=lambda: user) + app.include_router(thread_runs.router) + app.state.stream_bridge = MagicMock(stream_exists=AsyncMock(return_value=False)) + app.state.run_manager = MagicMock() + return TestClient(app) + + +def test_same_idempotency_key_reuses_thread_run(monkeypatch): + admissions: dict[str, RunRecord] = {} + client = _make_client(monkeypatch, _user("alice@example.com"), admissions) + url = "/api/threads/thread-1/runs" + headers = {"Idempotency-Key": "send-message-1"} + + first = client.post(url, json={"input": {"messages": []}}, headers=headers) + retry = client.post(url, json={"input": {"messages": []}}, headers=headers) + + assert first.status_code == 200, first.text + assert retry.status_code == 200, retry.text + assert retry.json()["run_id"] == first.json()["run_id"] + + +def test_same_idempotency_key_reuses_stream_run(monkeypatch): + admissions: dict[str, RunRecord] = {} + client = _make_client(monkeypatch, _user("alice@example.com"), admissions) + url = "/api/threads/thread-1/runs/stream" + headers = {"Idempotency-Key": "send-message-1"} + + first = client.post(url, json={"input": {"messages": []}}, headers=headers) + retry = client.post(url, json={"input": {"messages": []}}, headers=headers) + + assert first.status_code == 200, first.text + assert retry.status_code == 200, retry.text + assert retry.headers["Content-Location"] == first.headers["Content-Location"] + assert "event: end" in first.text + assert "event: gap" not in first.text + assert "event: gap" in retry.text + assert "stream_replay_gap" in retry.text + assert "reload_durable_state" in retry.text + assert "event: end" not in retry.text + + +def test_same_idempotency_key_reuses_wait_run(monkeypatch): + admissions: dict[str, RunRecord] = {} + client = _make_client(monkeypatch, _user("alice@example.com"), admissions) + url = "/api/threads/thread-1/runs/wait" + headers = {"Idempotency-Key": "send-message-1"} + + first = client.post(url, json={"input": {"messages": []}}, headers=headers) + retry = client.post(url, json={"input": {"messages": []}}, headers=headers) + + assert first.status_code == 200, first.text + assert retry.status_code == 200, retry.text + assert retry.json()["error"] == first.json()["error"] + + +def test_idempotency_key_is_scoped_to_thread(monkeypatch): + admissions: dict[str, RunRecord] = {} + client = _make_client(monkeypatch, _user("alice@example.com"), admissions) + headers = {"Idempotency-Key": "send-message-1"} + + first = client.post("/api/threads/thread-1/runs", json={}, headers=headers) + second = client.post("/api/threads/thread-2/runs", json={}, headers=headers) + + assert first.status_code == 200, first.text + assert second.status_code == 200, second.text + assert second.json()["run_id"] != first.json()["run_id"] + + +def test_idempotency_key_is_scoped_to_authenticated_user(monkeypatch): + admissions: dict[str, RunRecord] = {} + alice = _make_client(monkeypatch, _user("alice@example.com"), admissions) + bob = _make_client(monkeypatch, _user("bob@example.com"), admissions) + url = "/api/threads/thread-1/runs" + headers = {"Idempotency-Key": "send-message-1"} + + first = alice.post(url, json={}, headers=headers) + second = bob.post(url, json={}, headers=headers) + + assert first.status_code == 200, first.text + assert second.status_code == 200, second.text + assert second.json()["run_id"] != first.json()["run_id"] + + +def test_missing_idempotency_key_keeps_creating_runs(monkeypatch): + admissions: dict[str, RunRecord] = {} + client = _make_client(monkeypatch, _user("alice@example.com"), admissions) + url = "/api/threads/thread-1/runs" + + first = client.post(url, json={}) + second = client.post(url, json={}) + + assert first.status_code == 200, first.text + assert second.status_code == 200, second.text + assert second.json()["run_id"] != first.json()["run_id"] + + +def test_different_idempotency_keys_create_different_runs(monkeypatch): + admissions: dict[str, RunRecord] = {} + client = _make_client(monkeypatch, _user("alice@example.com"), admissions) + url = "/api/threads/thread-1/runs" + + first = client.post(url, json={}, headers={"Idempotency-Key": "send-message-1"}) + second = client.post(url, json={}, headers={"Idempotency-Key": "send-message-2"}) + + assert first.status_code == 200, first.text + assert second.status_code == 200, second.text + assert second.json()["run_id"] != first.json()["run_id"] + + +def test_blank_idempotency_key_is_rejected(monkeypatch): + client = _make_client(monkeypatch, _user("alice@example.com"), {}) + + response = client.post( + "/api/threads/thread-1/runs", + json={}, + headers={"Idempotency-Key": " "}, + ) + + assert response.status_code == 422 + + +def test_oversized_idempotency_key_is_rejected(monkeypatch): + client = _make_client(monkeypatch, _user("alice@example.com"), {}) + + response = client.post( + "/api/threads/thread-1/runs", + json={}, + headers={"Idempotency-Key": "x" * 256}, + ) + + assert response.status_code == 422 + + +class _LocalBridge: + supports_cross_process = False + + async def stream_exists(self, run_id): + del run_id + return False + + +class _StaleSnapshot: + config = {"configurable": {"checkpoint_id": "cp-previous"}} + values = {"messages": [{"type": "ai", "content": "PREVIOUS_TURN"}]} + + +def test_wait_reused_store_only_run_does_not_return_stale_checkpoint(monkeypatch): + """A reused running record has no local task; /wait must not serialize the current checkpoint.""" + + async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False): + del body, request, idempotency_key, require_existing_thread + return RunRecord( + run_id="run-live", + thread_id=thread_id, + assistant_id=None, + status=RunStatus.running, + on_disconnect=DisconnectMode.continue_, + store_only=True, + idempotency_reused=True, + ) + + async def fake_aget(config): + del config + return _StaleSnapshot() + + monkeypatch.setattr(thread_runs, "start_run", fake_start_run) + monkeypatch.setattr( + thread_runs, + "build_checkpoint_state_accessor", + lambda *args, **kwargs: (SimpleNamespace(aget=fake_aget), {}), + ) + monkeypatch.setattr(thread_runs, "serialize_channel_values_for_api", lambda values: values) + + app = make_authed_test_app(user_factory=lambda: _user("alice@example.com")) + app.include_router(thread_runs.router) + app.state.stream_bridge = _LocalBridge() + app.state.run_manager = MagicMock() + + with TestClient(app) as client: + response = client.post( + "/api/threads/thread-1/runs/wait", + json={"input": {"messages": []}}, + headers={"Idempotency-Key": "send-message-1"}, + ) + + assert response.status_code == 200, response.text + assert response.json() == {"status": "running", "error": None} + assert "PREVIOUS_TURN" not in response.text + + +def test_wait_reused_completed_run_does_not_return_later_checkpoint(monkeypatch): + """A locally cached completed reuse must not serialize a later thread head.""" + + async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False): + del body, request, idempotency_key, require_existing_thread + return RunRecord( + run_id="run-a", + thread_id=thread_id, + assistant_id=None, + status=RunStatus.success, + on_disconnect=DisconnectMode.continue_, + store_only=False, + idempotency_reused=True, + ) + + async def fake_aget(config): + del config + return SimpleNamespace( + config={"configurable": {"checkpoint_id": "cp-later"}}, + values={"messages": [{"type": "ai", "content": "LATER_RUN_RESULT"}]}, + ) + + monkeypatch.setattr(thread_runs, "start_run", fake_start_run) + monkeypatch.setattr( + thread_runs, + "build_checkpoint_state_accessor", + lambda *args, **kwargs: (SimpleNamespace(aget=fake_aget), {}), + ) + monkeypatch.setattr(thread_runs, "serialize_channel_values_for_api", lambda values: values) + + app = make_authed_test_app(user_factory=lambda: _user("alice@example.com")) + app.include_router(thread_runs.router) + app.state.stream_bridge = _LocalBridge() + app.state.run_manager = MagicMock() + + with TestClient(app) as client: + response = client.post( + "/api/threads/thread-1/runs/wait", + json={"input": {"messages": []}}, + headers={"Idempotency-Key": "send-message-1"}, + ) + + assert response.status_code == 200, response.text + assert response.json() == {"status": "success", "error": None} + assert "LATER_RUN_RESULT" not in response.text + + +@pytest.mark.anyio +async def test_wait_original_request_keeps_checkpoint_when_retry_overlaps(): + """An overlapping retry must not suppress the original creating /wait result.""" + from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge + + bridge = MemoryStreamBridge() + record = RunRecord( + run_id="run-a", + thread_id="thread-1", + assistant_id=None, + status=RunStatus.running, + on_disconnect=DisconnectMode.continue_, + store_only=False, + idempotency_reused=False, + ) + record.task = asyncio.create_task(asyncio.Event().wait()) + snapshot = SimpleNamespace( + config={"configurable": {"checkpoint_id": "cp-a"}}, + values={"messages": [{"type": "ai", "content": "FIRST_RUN_RESULT"}]}, + ) + request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False)) + + async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False): + del body, thread_id, request, idempotency_key, require_existing_thread + return record + + async def fake_aget(config): + del config + return snapshot + + with ( + patch.object(thread_runs, "start_run", fake_start_run), + patch.object(thread_runs, "get_stream_bridge", return_value=bridge), + patch.object(thread_runs, "get_run_manager", return_value=MagicMock()), + patch.object( + thread_runs, + "build_checkpoint_state_accessor", + lambda *args, **kwargs: (SimpleNamespace(aget=fake_aget), {}), + ), + patch.object(thread_runs, "serialize_channel_values_for_api", lambda values: values), + ): + wait_task = asyncio.create_task( + call_unwrapped( + thread_runs.wait_run, + "thread-1", + RunCreateRequest(input={"messages": []}), + request, + ) + ) + await asyncio.sleep(0.05) + record.idempotency_reused = True + record.status = RunStatus.success + await bridge.publish_end(record.run_id) + result = await asyncio.wait_for(wait_task, timeout=2) + + record.task.cancel() + with pytest.raises(asyncio.CancelledError): + await record.task + + assert result["messages"][0]["content"] == "FIRST_RUN_RESULT" + + +@pytest.mark.anyio +async def test_wait_peer_refreshes_status_after_owner_completes(): + """A cross-worker reuse must not keep admission-time running after END.""" + from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge + + store = MemoryRunStore() + owner = RunManager(store=store, worker_id="worker-a") + peer = RunManager(store=store, worker_id="worker-b") + bridge = MemoryStreamBridge() + bridge.supports_cross_process = True + input_payload = {"messages": [{"role": "user", "content": "hello"}]} + first = await owner.create_or_reject( + "thread-1", + user_id=None, + idempotency_key="http-run:same", + kwargs={"input": input_payload, "config": None}, + ) + await owner.set_status(first.run_id, RunStatus.running) + reused = await peer.create_or_reject( + "thread-1", + user_id=None, + idempotency_key="http-run:same", + kwargs={"input": input_payload, "config": None}, + ) + assert reused.run_id == first.run_id + assert reused.store_only is True + assert reused.idempotency_reused is True + assert reused.status == RunStatus.running + request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False)) + + async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False): + del body, thread_id, request, idempotency_key, require_existing_thread + return reused + + with ( + patch.object(thread_runs, "start_run", fake_start_run), + patch.object(thread_runs, "get_stream_bridge", return_value=bridge), + patch.object(thread_runs, "get_run_manager", return_value=peer), + patch.object( + thread_runs, + "build_checkpoint_state_accessor", + side_effect=AssertionError("reused wait must not read latest checkpoint"), + ), + ): + wait_task = asyncio.create_task( + call_unwrapped( + thread_runs.wait_run, + "thread-1", + RunCreateRequest(input=input_payload), + request, + ) + ) + await asyncio.sleep(0.05) + await owner.set_status(first.run_id, RunStatus.success) + await bridge.publish_end(first.run_id) + result = await asyncio.wait_for(wait_task, timeout=2) + + assert result == {"status": "success", "error": None} + assert reused.status == RunStatus.success + + +def test_scope_http_run_idempotency_key_ignores_header_default(): + """Direct handler calls pass FastAPI's Header() object, not None.""" + from fastapi.params import Header as HeaderParam + + request = SimpleNamespace(state=SimpleNamespace(user=None)) + assert thread_runs._scope_http_run_idempotency_key(request, "thread-1", HeaderParam(default=None)) is None + assert thread_runs._scope_http_run_idempotency_key(request, "thread-1", None) is None + + +def test_stream_reused_store_only_running_run_returns_409(monkeypatch): + """A reused running record on a process-local bridge must not hang on an empty stream.""" + + async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False): + del body, request, idempotency_key, require_existing_thread + return RunRecord( + run_id="run-live", + thread_id=thread_id, + assistant_id=None, + status=RunStatus.running, + on_disconnect=DisconnectMode.continue_, + store_only=True, + idempotency_reused=True, + ) + + monkeypatch.setattr(thread_runs, "start_run", fake_start_run) + + app = make_authed_test_app(user_factory=lambda: _user("alice@example.com")) + app.include_router(thread_runs.router) + app.state.stream_bridge = _LocalBridge() + app.state.run_manager = MagicMock() + + with TestClient(app) as client: + response = client.post( + "/api/threads/thread-1/runs/stream", + json={"input": {"messages": []}}, + headers={"Idempotency-Key": "send-message-1"}, + ) + + assert response.status_code == 409, response.text + assert "not active on this worker" in response.json()["detail"] + + +@pytest.mark.anyio +async def test_sse_consumer_reused_terminal_missing_stream_yields_gap(): + from app.gateway.services import sse_consumer + + record = RunRecord( + run_id="run-done", + thread_id="thread-1", + assistant_id=None, + status=RunStatus.success, + on_disconnect=DisconnectMode.continue_, + store_only=True, + idempotency_reused=True, + ) + request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False)) + + frames = [ + frame + async for frame in sse_consumer( + _LocalBridge(), + record, + request, + MagicMock(), + emit_gap_on_missing_stream=True, + ) + ] + + assert len(frames) == 1 + assert frames[0].startswith("event: gap\n") + assert "stream_replay_gap" in frames[0] + assert "reload_durable_state" in frames[0] + assert "event: end" not in frames[0] + + +@pytest.mark.anyio +async def test_sse_consumer_observer_join_keeps_end_after_sticky_reuse_flag(): + """Observer joins must not inherit create_or_reject's sticky reuse flag.""" + from app.gateway.services import sse_consumer + + record = RunRecord( + run_id="run-done", + thread_id="thread-1", + assistant_id=None, + status=RunStatus.success, + on_disconnect=DisconnectMode.continue_, + idempotency_reused=True, + ) + request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False)) + + frames = [frame async for frame in sse_consumer(_LocalBridge(), record, request, MagicMock(), apply_on_disconnect=False)] + + assert len(frames) == 1 + assert frames[0].startswith("event: end\n") + assert "event: gap" not in frames[0] + + +@pytest.mark.anyio +async def test_sse_consumer_default_path_keeps_end_after_sticky_reuse_flag(): + """Default sse_consumer, including stateless /api/runs/stream, must not emit gap + just because create_or_reject left idempotency_reused set, or because + apply_on_disconnect still defaults to True. + """ + from app.gateway.services import sse_consumer + + record = RunRecord( + run_id="run-done", + thread_id="thread-1", + assistant_id=None, + status=RunStatus.success, + on_disconnect=DisconnectMode.continue_, + store_only=True, + idempotency_reused=True, + ) + request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False)) + + frames = [frame async for frame in sse_consumer(_LocalBridge(), record, request, MagicMock())] + + assert len(frames) == 1 + assert frames[0].startswith("event: end\n") + assert "event: gap" not in frames[0] + + +@pytest.mark.anyio +async def test_sse_consumer_missing_stream_gap_requires_explicit_flag(): + """apply_on_disconnect must not select gap vs end by itself.""" + from app.gateway.services import sse_consumer + + record = RunRecord( + run_id="run-done", + thread_id="thread-1", + assistant_id=None, + status=RunStatus.success, + on_disconnect=DisconnectMode.continue_, + store_only=True, + ) + request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False)) + + default_frames = [frame async for frame in sse_consumer(_LocalBridge(), record, request, MagicMock(), apply_on_disconnect=True)] + gap_frames = [ + frame + async for frame in sse_consumer( + _LocalBridge(), + record, + request, + MagicMock(), + apply_on_disconnect=False, + emit_gap_on_missing_stream=True, + ) + ] + + assert default_frames[0].startswith("event: end\n") + assert gap_frames[0].startswith("event: gap\n") + + +@pytest.mark.anyio +async def test_observer_join_stays_end_after_real_manager_reuse(): + """Join of a terminal missing stream stays `end` after a later key reuse. + + ``create_or_reject`` sets ``idempotency_reused`` on the cached record that + ``RunManager.get()`` returns. Observer joins read that same object; the + missing-stream branch must still follow ``emit_gap_on_missing_stream``, + not the sticky flag or ``apply_on_disconnect``. + """ + from app.gateway.services import sse_consumer + + store = MemoryRunStore() + manager = RunManager(store=store, worker_id="worker-a") + first = await manager.create_or_reject( + "thread-1", + user_id=None, + idempotency_key="http-run:same", + ) + await manager.set_status(first.run_id, RunStatus.success) + request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False)) + + async def _frames(*, apply_on_disconnect: bool = True, emit_gap_on_missing_stream: bool = False): + record = await manager.get(first.run_id) + assert record is not None + return [ + frame + async for frame in sse_consumer( + _LocalBridge(), + record, + request, + manager, + apply_on_disconnect=apply_on_disconnect, + emit_gap_on_missing_stream=emit_gap_on_missing_stream, + ) + ] + + before = await _frames(apply_on_disconnect=False) + assert before[0].startswith("event: end\n") + + reused = await manager.create_or_reject( + "thread-1", + user_id=None, + idempotency_key="http-run:same", + ) + assert reused.run_id == first.run_id + assert reused.idempotency_reused is True + + after = await _frames(apply_on_disconnect=False) + assert after[0].startswith("event: end\n") + assert "event: gap" not in after[0] + + after_default = await _frames() + assert after_default[0].startswith("event: end\n") + assert "event: gap" not in after_default[0] + + creating = await _frames(emit_gap_on_missing_stream=True) + assert creating[0].startswith("event: gap\n") + assert "event: end" not in creating[0] + + +def _make_start_run_request(run_manager): + from langgraph.checkpoint.memory import InMemorySaver + from langgraph.store.memory import InMemoryStore + + from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore + + store = InMemoryStore() + return SimpleNamespace( + headers={}, + state=SimpleNamespace(auth_source=None, user=None), + app=SimpleNamespace( + state=SimpleNamespace( + stream_bridge=SimpleNamespace(), + run_manager=run_manager, + checkpointer=InMemorySaver(), + store=store, + run_event_store=MemoryRunEventStore(), + run_events_config=None, + thread_store=MemoryThreadMetaStore(store), + ) + ), + ) + + +@pytest.fixture +def _stub_app_config(): + set_app_config(AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}})) + yield + reset_app_config() + + +@pytest.mark.anyio +async def test_start_run_reuses_store_backed_running_row_without_attaching_worker(_stub_app_config): + from app.gateway.services import start_run + + input_payload = {"messages": [{"role": "user", "content": "hello"}]} + store = MemoryRunStore() + owner = RunManager(store=store, worker_id="worker-a") + peer = RunManager(store=store, worker_id="worker-b") + first = await owner.create_or_reject( + "thread-1", + user_id=None, + idempotency_key="http-run:same", + kwargs={"input": input_payload, "config": None}, + ) + + attached = False + + async def fake_run_agent(*args, **kwargs): + del args, kwargs + nonlocal attached + attached = True + + with ( + patch("app.gateway.services.resolve_agent_factory", return_value=object()), + patch("app.gateway.services.run_agent", side_effect=fake_run_agent), + ): + record = await start_run( + RunCreateRequest(input=input_payload), + "thread-1", + _make_start_run_request(peer), + idempotency_key="http-run:same", + ) + + assert record.run_id == first.run_id + assert record.idempotency_reused is True + assert record.store_only is True + assert record.task is None + assert attached is False + + +@pytest.mark.anyio +async def test_start_run_rejects_reused_key_with_different_input(_stub_app_config): + from app.gateway.services import start_run + + store = MemoryRunStore() + owner = RunManager(store=store, worker_id="worker-a") + peer = RunManager(store=store, worker_id="worker-b") + await owner.create_or_reject( + "thread-1", + user_id=None, + idempotency_key="http-run:same", + kwargs={"input": {"messages": [{"role": "user", "content": "summarize"}]}, "config": None}, + ) + + with ( + patch("app.gateway.services.resolve_agent_factory", return_value=object()), + patch("app.gateway.services.run_agent", side_effect=AssertionError("worker must not attach")), + pytest.raises(HTTPException) as excinfo, + ): + await start_run( + RunCreateRequest(input={"messages": [{"role": "user", "content": "translate"}]}), + "thread-1", + _make_start_run_request(peer), + idempotency_key="http-run:same", + ) + + assert excinfo.value.status_code == 409 + assert "different request" in str(excinfo.value.detail) + + +@pytest.mark.anyio +async def test_wait_retry_after_later_run_does_not_return_later_checkpoint(monkeypatch): + """Complete two runs, then retry the first key: /wait must not return run B.""" + first_input = {"messages": [{"role": "user", "content": "one"}]} + later_input = {"messages": [{"role": "user", "content": "two"}]} + store = MemoryRunStore() + manager = RunManager(store=store, worker_id="worker-a") + first = await manager.create_or_reject( + "thread-1", + user_id=None, + idempotency_key="http-run:first", + kwargs={"input": first_input, "config": None}, + ) + await manager.set_status(first.run_id, RunStatus.success) + later = await manager.create_or_reject( + "thread-1", + user_id=None, + idempotency_key="http-run:later", + kwargs={"input": later_input, "config": None}, + ) + await manager.set_status(later.run_id, RunStatus.success) + reused = await manager.create_or_reject( + "thread-1", + user_id=None, + idempotency_key="http-run:first", + kwargs={"input": first_input, "config": None}, + ) + assert reused.run_id == first.run_id + assert reused.idempotency_reused is True + assert reused.store_only is False + assert reused.status == RunStatus.success + + async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False): + del body, thread_id, request, idempotency_key, require_existing_thread + return reused + + async def fake_aget(config): + del config + return SimpleNamespace( + config={"configurable": {"checkpoint_id": "cp-later"}}, + values={"messages": [{"type": "ai", "content": "LATER_RUN_RESULT"}]}, + ) + + monkeypatch.setattr(thread_runs, "start_run", fake_start_run) + monkeypatch.setattr( + thread_runs, + "build_checkpoint_state_accessor", + lambda *args, **kwargs: (SimpleNamespace(aget=fake_aget), {}), + ) + monkeypatch.setattr(thread_runs, "serialize_channel_values_for_api", lambda values: values) + + app = make_authed_test_app(user_factory=lambda: _user("alice@example.com")) + app.include_router(thread_runs.router) + app.state.stream_bridge = _LocalBridge() + app.state.run_manager = manager + + with TestClient(app) as client: + response = client.post( + "/api/threads/thread-1/runs/wait", + json={"input": first_input}, + headers={"Idempotency-Key": "first"}, + ) + + assert response.status_code == 200, response.text + assert response.json() == {"status": "success", "error": None} + assert "LATER_RUN_RESULT" not in response.text