fix: propagate current run failures from wait endpoints (#6217)

* fix: propagate current run failures from wait endpoints

* docs: clarify checkpoint wait errors and SDK behavior
This commit is contained in:
quifox 2026-10-04 07:46:59 +08:00 committed by GitHub
parent e3064f541b
commit 06e9bfd269
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 281 additions and 18 deletions

View File

@ -620,6 +620,8 @@ described above.
Gateway runs automatically enforce native delivery for artifacts created or modified under `/mnt/user-data/outputs`: `present_files` must present at least one output produced by the current run, and the terminal `run.delivery` receipt must be durably recorded. Virtual artifact paths are resolved within the same authenticated user and thread scope that produced the output before the output-directory boundary is validated. Runs that do not produce output artifacts keep ordinary conversational behavior.
Thread-scoped `runs.wait()` calls that finish with `status: error` report the current run error instead of an earlier answer. The asynchronous Python LangGraph SDK raises by default; pass `raise_error=False` to inspect the returned status and error.
DeerFlow's built-in custom events are available through both LangGraph streaming interfaces: native clients can continue subscribing to `stream_mode="custom"`, while callback-based integrations can consume the same payloads as `on_custom_event` records from `astream_events(version="v2")`. The callback event name matches the payload's `type` field.
#### Docker Production Deployment

View File

@ -199,7 +199,7 @@ startup gate rejects process-local memory and JSONL event stores when
- Gateway checkpoint mutations outside run execution must use `services.reserve_checkpoint_write()`, which composes the process-local thread lock with the durable `checkpoint_write` reservation. Manual compaction, `POST /threads/{id}/state`, and both goal mutation routes (`PUT` / `DELETE /threads/{id}/goal`, including creation of a missing goal checkpoint) use this boundary, so an existing run blocks the write and the reservation blocks new reject/interrupt/rollback runs across workers.
- Branch/state-update checkpoints copy only the source checkpoint's persisted `deerflow_agent_name`, never request metadata; missing or malformed bindings stay unbound so compaction fails closed.
- `DELETE /api/threads/{id}` holds a durable `delete` reservation for the whole cleanup and removes thread filesystem data, checkpoints, owner-scoped historical runs, run events, feedback and thread metadata best-effort. Historical-run cleanup removes only `operation_kind="run"` rows, so internal thread-operation rows — including the reservation protecting that request — stay durable until `RunManager.reserve_thread_operation()` exits. This cleans persisted rows; preventing an already-admitted write from re-creating state for a deleted thread is a separate thread-incarnation contract, not part of this cleanup.
- Creator SSE and `/wait` apply `on_disconnect` only to confirmed request disconnect or cancellation; bridge failures propagate without cancelling the run. `/wait` consumes the bridge instead of `record.task`, preventing stale checkpoint responses (#3265).
- Creator SSE and `/wait` apply `on_disconnect` only to confirmed request disconnect or cancellation; bridge failures propagate without cancelling the run. `/wait` consumes bridge END; errors return SDK `__error__`, never stale state.
- 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.

View File

@ -16,8 +16,14 @@ from app.gateway.authz import require_permission
from app.gateway.deps import get_feedback_repo, get_run_event_store, get_run_manager, get_run_store, get_stream_bridge
from app.gateway.pagination import trim_run_message_page
from app.gateway.run_models import RunCreateRequest
from app.gateway.services import abuild_checkpoint_state_accessor, sse_consumer, start_run, wait_for_run_completion
from deerflow.runtime import serialize_channel_values_for_api
from app.gateway.services import (
abuild_checkpoint_state_accessor,
serialize_wait_run_status,
sse_consumer,
start_run,
wait_for_run_completion,
)
from deerflow.runtime import RunStatus, serialize_channel_values_for_api
from deerflow.utils.thread_id import resolve_thread_id
logger = logging.getLogger(__name__)
@ -74,6 +80,9 @@ async def stateless_wait(body: RunCreateRequest, request: Request) -> dict:
if record.task is not None:
completed = await wait_for_run_completion(bridge, record, request, run_mgr)
if completed and record.status == RunStatus.error:
return serialize_wait_run_status(record)
if completed:
try:
accessor, config = await abuild_checkpoint_state_accessor(

View File

@ -51,7 +51,14 @@ from app.gateway.deps import get_current_user, get_feedback_repo, get_run_event_
from app.gateway.internal_auth import INTERNAL_SYSTEM_ROLE, get_trusted_internal_owner_user_id
from app.gateway.pagination import trim_run_message_page
from app.gateway.run_models import RunCreateRequest
from app.gateway.services import abuild_checkpoint_state_accessor, build_thread_checkpoint_state_accessor, sse_consumer, start_run, wait_for_run_completion
from app.gateway.services import (
abuild_checkpoint_state_accessor,
build_thread_checkpoint_state_accessor,
serialize_wait_run_status,
sse_consumer,
start_run,
wait_for_run_completion,
)
from app.gateway.utils import sanitize_log_param
from deerflow.agents.human_input import read_human_input_response
from deerflow.agents.middlewares.dynamic_context_middleware import strip_injected_user_message_id_suffix
@ -1047,13 +1054,20 @@ async def wait_run(
# 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}
return serialize_wait_run_status(record)
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:
record = await _refresh_store_backed_run(run_mgr, record)
if record.status == RunStatus.error:
# A failure before the first checkpoint leaves an earlier answer at
# the thread head. Return this run's error, never that old state.
return serialize_wait_run_status(record)
# 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:
@ -1070,9 +1084,7 @@ async def wait_run(
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}
return serialize_wait_run_status(record)
def _parse_run_page_created_at(value: str) -> str:

View File

@ -2490,6 +2490,14 @@ async def sse_consumer(
await await_drained(run_mgr.cancel(record.run_id))
def serialize_wait_run_status(record: RunRecord) -> dict[str, Any]:
"""Return durable status and the error envelope recognized by the SDK."""
result: dict[str, Any] = {"status": record.status.value, "error": record.error}
if record.status == RunStatus.error:
result["__error__"] = {"error": "RunError", "message": record.error or "Run failed."}
return result
async def wait_for_run_completion(
bridge: StreamBridge,
record: RunRecord,

View File

@ -0,0 +1,122 @@
"""Stateless waits must expose the current run error, not an earlier checkpoint."""
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock
import httpx
import pytest
import pytest_asyncio
from _router_auth_helpers import make_authed_test_app
from langgraph_sdk.client import LangGraphClient
from app.gateway.routers import runs
from deerflow.runtime import DisconnectMode, RunStatus
from deerflow.runtime.runs.manager import RunManager
from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge
@pytest_asyncio.fixture
async def stateless_wait_client(monkeypatch):
bridge = MemoryStreamBridge()
run_manager = RunManager()
record = await run_manager.create(
"thread-1",
"lead_agent",
on_disconnect=DisconnectMode.continue_,
)
snapshot = SimpleNamespace(
config={"configurable": {"checkpoint_id": "old-checkpoint"}},
values={"messages": [{"type": "ai", "content": "Historical answer."}]},
)
accessor = SimpleNamespace(aget=AsyncMock(return_value=snapshot))
state = SimpleNamespace(
terminal_status=RunStatus.success,
terminal_error=None,
)
async def start_run(*args, **kwargs):
async def finish_run():
await run_manager.try_start(record.run_id)
await run_manager.set_status(
record.run_id,
state.terminal_status,
error=state.terminal_error,
persist=False,
)
await bridge.publish_end(record.run_id)
record.task = asyncio.create_task(finish_run())
return record
monkeypatch.setattr(runs, "start_run", start_run)
monkeypatch.setattr(runs, "abuild_checkpoint_state_accessor", AsyncMock(return_value=(accessor, {})))
app = make_authed_test_app()
app.include_router(runs.router)
app.state.stream_bridge = bridge
app.state.run_manager = run_manager
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test/api") as http:
state.client = LangGraphClient(http)
state.record = record
state.run_manager = run_manager
state.snapshot = snapshot
state.accessor = accessor
yield state
if record.task is not None:
await record.task
@pytest.mark.asyncio
@pytest.mark.parametrize("has_checkpoint", [False, True], ids=["no-checkpoint", "old-checkpoint"])
async def test_failed_stateless_wait_raises_through_real_sdk_and_skips_checkpoint(stateless_wait_client, has_checkpoint):
state = stateless_wait_client
state.terminal_status = RunStatus.error
state.terminal_error = "Current model initialization failed."
if not has_checkpoint:
state.snapshot.config = {}
response = await state.client.runs.wait(
None,
"lead_agent",
input={"messages": []},
config={"configurable": {"thread_id": "thread-1"}},
raise_error=False,
)
assert "messages" not in response
assert response["status"] == "error"
assert response["error"] == state.terminal_error
assert response["__error__"]["message"] == state.terminal_error
manager_record = await state.run_manager.get(state.record.run_id)
assert manager_record is state.record
assert manager_record.status == RunStatus.error
assert manager_record.error == state.terminal_error
# The SDK only raises for its __error__ envelope, so check the stock path too.
with pytest.raises(Exception, match="Current model initialization failed"):
await state.client.runs.wait(
None,
"lead_agent",
input={"messages": []},
config={"configurable": {"thread_id": "thread-1"}},
)
state.accessor.aget.assert_not_awaited()
@pytest.mark.asyncio
async def test_successful_stateless_wait_preserves_final_checkpoint_state(stateless_wait_client):
state = stateless_wait_client
state.snapshot.values = {"messages": [{"type": "ai", "content": "Current answer."}]}
response = await state.client.runs.wait(
None,
"lead_agent",
input={"messages": []},
config={"configurable": {"thread_id": "thread-1"}},
)
assert response["messages"][0]["content"] == "Current answer."
assert state.record.status == RunStatus.success
state.accessor.aget.assert_awaited_once()

View File

@ -352,7 +352,11 @@ def test_run_wait_readers_preserve_terminal_error_without_checkpoint() -> None:
thread_result, stateless_result = asyncio.run(_scenario())
expected = {"status": "error", "error": "run failed before checkpoint"}
expected = {
"status": "error",
"error": "run failed before checkpoint",
"__error__": {"error": "RunError", "message": "run failed before checkpoint"},
}
assert thread_result == expected
assert stateless_result == expected
@ -400,7 +404,12 @@ def test_run_wait_readers_preserve_terminal_error_when_accessor_builder_fails(ro
result = asyncio.run(_scenario())
assert result == {"status": "error", "error": "run failed before checkpoint"}
expected = {
"status": "error",
"error": "run failed before checkpoint",
"__error__": {"error": "RunError", "message": "run failed before checkpoint"},
}
assert result == expected
def test_prepare_regenerate_payload_returns_clean_input_and_base_checkpoint():

View File

@ -0,0 +1,89 @@
"""Failed waits must expose the current error instead of historical answers."""
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock
import httpx
import pytest
import pytest_asyncio
from _router_auth_helpers import make_authed_test_app
from langgraph_sdk.client import LangGraphClient
from app.gateway.routers import thread_runs
from deerflow.runtime import DisconnectMode, RunRecord, RunStatus
from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge
@pytest_asyncio.fixture
async def wait_client(monkeypatch):
bridge = MemoryStreamBridge()
record = RunRecord(
run_id="current-run",
thread_id="thread-1",
assistant_id=None,
status=RunStatus.success,
on_disconnect=DisconnectMode.continue_,
)
snapshot = SimpleNamespace(
config={"configurable": {"checkpoint_id": "old-checkpoint"}},
values={"messages": [{"type": "ai", "content": "Historical answer."}]},
)
accessor = SimpleNamespace(aget=AsyncMock(return_value=snapshot))
async def start_run(*args, **kwargs):
# Keep a local task so the real wait helper consumes the bridge's END.
record.task = asyncio.create_task(asyncio.sleep(0))
await bridge.publish_end(record.run_id)
return record
monkeypatch.setattr(thread_runs, "start_run", start_run)
monkeypatch.setattr(thread_runs, "abuild_checkpoint_state_accessor", AsyncMock(return_value=(accessor, {})))
app = make_authed_test_app()
app.include_router(thread_runs.router)
app.state.stream_bridge = bridge
app.state.run_manager = SimpleNamespace(get=AsyncMock(return_value=record))
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test/api") as http:
yield SimpleNamespace(client=LangGraphClient(http), http=http, record=record, snapshot=snapshot, accessor=accessor)
if record.task is not None:
await record.task
@pytest.mark.asyncio
@pytest.mark.parametrize("reused", [False, True], ids=["created", "reused"])
@pytest.mark.parametrize("has_checkpoint", [False, True], ids=["no-checkpoint", "old-checkpoint"])
async def test_failed_wait_raises_through_real_sdk_without_reading_thread_head(wait_client, reused, has_checkpoint):
state = wait_client
state.record.status = RunStatus.error
state.record.error = "Current model initialization failed."
state.record.idempotency_reused = reused
if not has_checkpoint:
state.snapshot.config = {}
response = await state.client.runs.wait("thread-1", "lead_agent", input={"messages": []}, raise_error=False)
assert "messages" not in response
assert response["status"] == "error"
assert response["error"] == state.record.error
assert response["__error__"]["message"] == state.record.error
# The stock SDK raises only for its __error__ envelope, not flat
# status/error responses. Drive it through the actual FastAPI wait route.
with pytest.raises(Exception, match="Current model initialization failed"):
await state.client.runs.wait("thread-1", "lead_agent", input={"messages": []})
state.accessor.aget.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize("reused", [False, True], ids=["created", "reused"])
async def test_successful_wait_preserves_checkpoint_and_reused_status_contract(wait_client, reused):
state = wait_client
state.record.idempotency_reused = reused
state.snapshot.values = {"messages": [{"type": "ai", "content": "Current answer."}]}
response = await state.client.runs.wait("thread-1", "lead_agent", input={"messages": []})
if reused:
assert response == {"status": "success", "error": None}
state.accessor.aget.assert_not_awaited()
else:
assert response["messages"][0]["content"] == "Current answer."
state.accessor.aget.assert_awaited_once()

View File

@ -52,7 +52,9 @@ A mode problem produces different evidence depending on which path hit it.
| Regenerate cannot read the latest checkpoint | log, then HTTP 500 `detail` | `Failed to read latest checkpoint for regenerate thread %s` → `Failed to read latest checkpoint` |
| Regenerate cannot list checkpoint history | log, then HTTP 500 `detail` | `Failed to list checkpoints for regenerate thread %s` → `Failed to inspect checkpoint history` |
The failure payloads differ by route. Run creation does **not** pre-gate on the threads router: `POST /api/threads/{thread_id}/runs` returns a `200` run record that later turns `status: error` with the mismatch text, `POST /runs/stream` returns `200 text/event-stream` and then an `event: error` frame with `{"message": ..., "name": "CheckpointModeMismatchError"}`, and `POST /runs/wait` returns `200` JSON with `{"status": "error", "error": "<mismatch text>"}`. The exception is a thread whose run-event feed is empty but whose checkpoint head exists: `ensure_checkpoint_history_seeded` materializes that head inside `start_run`, where the mode errors are not mapped, so the same mismatch returns a plain `500` before the run is admitted. The stateless `/api/runs/*` routes share that path.
The failure payloads differ by route. Run creation does **not** pre-gate on the threads router: `POST /api/threads/{thread_id}/runs` returns a `200` run record that later turns `status: error` with the mismatch text, `POST /runs/stream` returns `200 text/event-stream` and then an `event: error` frame with `{"message": ..., "name": "CheckpointModeMismatchError"}`, and `POST /runs/wait` returns `200` JSON with `{"status": "error", "error": "<mismatch text>", "__error__": {"error": "RunError", "message": "<mismatch text>"}}`. The exception is a thread whose run-event feed is empty but whose checkpoint head exists: `ensure_checkpoint_history_seeded` materializes that head inside `start_run`, where the mode errors are not mapped, so the same mismatch returns a plain `500` before the run is admitted. The stateless `/api/runs/*` routes share that path.
The wait response remains HTTP `200`. The stock Python SDK defaults to `raise_error=True` and raises an exception with `RunError: <mismatch text>` when it sees `__error__`; pass `raise_error=False` to receive the JSON body and inspect `status`, `error`, and `__error__`.
<Callout type="warning">
The mode gate itself logs nothing. `checkpoint_mode.py` has no logger,

View File

@ -170,8 +170,10 @@ The router states the rationale: a mismatch means the thread's persisted checkpo
| --- | --- | --- | --- |
| `POST /api/threads/{thread_id}/runs` | `200` run record, no pre-gate — but `500` when the thread's run-event feed is empty, so `ensure_checkpoint_history_seeded` materializes the head inside `start_run` first | the run record later has `status: error` and the mismatch text as `error` | `thread_runs:945-960`, `worker:1018-1030,1044-1047`, `worker:1450-1462`, `services:1503-1520,2055` |
| `POST /api/threads/{thread_id}/runs/stream` | `200` `text/event-stream` | an `event: error` frame whose `data` is `{"message": <mismatch text>, "name": "CheckpointModeMismatchError"}` | `worker:1466-1472`, `services:2374-2378` |
| `POST /api/threads/{thread_id}/runs/wait` | `200` JSON, never `409` | `{"status": "error", "error": "Thread requires delta mode; materialize and convert its checkpoints before using full mode."}` | `thread_runs:1018-1076` |
| `POST /api/runs/stream`, `POST /api/runs/wait` (stateless) | same as above | same `{"status", "error"}` body | `runs:59-91` |
| `POST /api/threads/{thread_id}/runs/wait` | `200` JSON, never `409` | `{"status": "error", "error": "Thread requires delta mode; materialize and convert its checkpoints before using full mode.", "__error__": {"error": "RunError", "message": "Thread requires delta mode; materialize and convert its checkpoints before using full mode."}}` | `thread_runs:1018-1076` |
| `POST /api/runs/stream`, `POST /api/runs/wait` (stateless) | same as above | same SSE error frame or wait `{"status", "error", "__error__"}` body | `runs:59-91` |
The wait response remains HTTP `200`. The stock Python SDK defaults to `raise_error=True` and raises an exception with `RunError: <mismatch text>` when it sees `__error__`; pass `raise_error=False` to receive the JSON body and inspect `status`, `error`, and `__error__`.
### Routes with no mode mapping

View File

@ -40,7 +40,9 @@ This chapter is organized by **what you see**. Each entry gives the cause, the f
**Cause**: a process running `full` read a thread whose checkpoints were written in `delta`. The read gate raises `CheckpointModeMismatchError` before materializing, so a full-mode process never silently returns empty state. The routes that return `409` are `GET /api/threads/{thread_id}`, `GET` and `POST /api/threads/{thread_id}/state`, and `POST /api/threads/{thread_id}/history` (all covered by tests); `POST /api/threads/{thread_id}/branches` and `POST .../compact` gate the same way by code path. `compact` also uses `409` for a different reason, with detail `Context compaction is disabled.`
The threads router does not pre-gate run creation, so a mismatch there fails asynchronously: `POST .../runs` returns a `200` run record that later turns `status: error`, `POST .../runs/stream` returns `200 text/event-stream` and then an `event: error` frame, and `POST .../runs/wait` returns `200` JSON with `{"status": "error", "error": "<the mismatch text>"}`. One run path reads a checkpoint before admission, though: when a thread's run-event feed is empty but a checkpoint head exists, `ensure_checkpoint_history_seeded` materializes that head inside `start_run`, where only `ConflictError` and `UnsupportedStrategyError` are mapped, so the mismatch escapes as an unhandled `500` on `POST /api/threads/{thread_id}/runs`, `/runs/stream`, and `/runs/wait`. With the default `run_events.backend: memory` a Gateway restart empties every existing thread's feed, so that branch is reachable after a restart.
The threads router does not pre-gate run creation, so a mismatch there fails asynchronously: `POST .../runs` returns a `200` run record that later turns `status: error`, `POST .../runs/stream` returns `200 text/event-stream` and then an `event: error` frame, and `POST .../runs/wait` returns `200` JSON with `{"status": "error", "error": "<the mismatch text>", "__error__": {"error": "RunError", "message": "<the mismatch text>"}}`. One run path reads a checkpoint before admission, though: when a thread's run-event feed is empty but a checkpoint head exists, `ensure_checkpoint_history_seeded` materializes that head inside `start_run`, where only `ConflictError` and `UnsupportedStrategyError` are mapped, so the mismatch escapes as an unhandled `500` on `POST /api/threads/{thread_id}/runs`, `/runs/stream`, and `/runs/wait`. With the default `run_events.backend: memory` a Gateway restart empties every existing thread's feed, so that branch is reachable after a restart.
The wait response remains HTTP `200`. The stock Python SDK defaults to `raise_error=True` and raises an exception with `RunError: <mismatch text>` when it sees `__error__`; pass `raise_error=False` to receive the JSON body and inspect `status`, `error`, and `__error__`.
In the embedded TUI the client raises in-process before the graph streams: the Textual runner turns it into an `AssistantError` plus `RunEnded` pair, and the headless runners handle it the same way — `--print` prints a concise `Error: ...` line on stderr and exits non-zero, while `--json` appends a terminal `{"type": "error"}` record to the NDJSON stream (diagnostics stay on stderr) and exits non-zero.

View File

@ -52,7 +52,9 @@ import { Callout } from "nextra/components";
| 重新生成无法读取最新检查点 | 日志,随后 HTTP 500 `detail` | `Failed to read latest checkpoint for regenerate thread %s` → `Failed to read latest checkpoint` |
| 重新生成无法列出检查点历史 | 日志,随后 HTTP 500 `detail` | `Failed to list checkpoints for regenerate thread %s` → `Failed to inspect checkpoint history` |
失败载荷随路由不同。创建 run 在 threads 路由上**不会**提前做门禁:`POST /api/threads/{thread_id}/runs` 返回 `200` run 记录,随后变为 `status: error` 并带上不匹配文本;`POST /runs/stream` 返回 `200 text/event-stream`,随后发出 `event: error` 帧,载荷为 `{"message": ..., "name": "CheckpointModeMismatchError"}`;`POST /runs/wait` 返回 `200` JSON,内容为 `{"status": "error", "error": "<不匹配文本>"}`。例外是 run event 流为空但存在检查点头部的线程:`ensure_checkpoint_history_seeded` 会在 `start_run` 内物化该头部,而那里没有映射模式错误,于是同一个不匹配会在 run 被准入之前返回普通的 `500`。无状态的 `/api/runs/*` 路由共享这条路径。
失败载荷随路由不同。创建 run 在 threads 路由上**不会**提前做门禁:`POST /api/threads/{thread_id}/runs` 返回 `200` run 记录,随后变为 `status: error` 并带上不匹配文本;`POST /runs/stream` 返回 `200 text/event-stream`,随后发出 `event: error` 帧,载荷为 `{"message": ..., "name": "CheckpointModeMismatchError"}`;`POST /runs/wait` 返回 `200` JSON,内容为 `{"status": "error", "error": "<不匹配文本>", "__error__": {"error": "RunError", "message": "<不匹配文本>"}}`。例外是 run event 流为空但存在检查点头部的线程:`ensure_checkpoint_history_seeded` 会在 `start_run` 内物化该头部,而那里没有映射模式错误,于是同一个不匹配会在 run 被准入之前返回普通的 `500`。无状态的 `/api/runs/*` 路由共享这条路径。
wait 响应仍为 HTTP `200`。Python SDK 默认使用 `raise_error=True`,识别到 `__error__` 后会抛出异常,消息为 `RunError: <不匹配文本>`;传入 `raise_error=False` 才会返回 JSON 响应体,供调用者读取 `status`、`error` 和 `__error__`。
<Callout type="warning">
模式门自身不记任何日志。`checkpoint_mode.py` 没有 logger,threads

View File

@ -170,8 +170,10 @@ Your config.yaml (version %d) is outdated — the latest version is %d. Run `mak
| --- | --- | --- | --- |
| `POST /api/threads/{thread_id}/runs` | `200` run 记录,无前置门禁 —— 但当线程的 run event 流为空时,`ensure_checkpoint_history_seeded` 会先在 `start_run` 内物化头部,此时返回 `500` | run 记录稍后变为 `status: error`,`error` 为不匹配原文 | `thread_runs:945-960`、`worker:1018-1030,1044-1047`、`worker:1450-1462`、`services:1503-1520,2055` |
| `POST /api/threads/{thread_id}/runs/stream` | `200` `text/event-stream` | 一帧 `event: error`,其 `data` 为 `{"message": <不匹配原文>, "name": "CheckpointModeMismatchError"}` | `worker:1466-1472`、`services:2374-2378` |
| `POST /api/threads/{thread_id}/runs/wait` | `200` JSON,从不返回 `409` | `{"status": "error", "error": "Thread requires delta mode; materialize and convert its checkpoints before using full mode."}` | `thread_runs:1018-1076` |
| `POST /api/runs/stream`、`POST /api/runs/wait`(无状态) | 同上 | 同样的 `{"status", "error"}` 载荷 | `runs:59-91` |
| `POST /api/threads/{thread_id}/runs/wait` | `200` JSON,从不返回 `409` | `{"status": "error", "error": "Thread requires delta mode; materialize and convert its checkpoints before using full mode.", "__error__": {"error": "RunError", "message": "Thread requires delta mode; materialize and convert its checkpoints before using full mode."}}` | `thread_runs:1018-1076` |
| `POST /api/runs/stream`、`POST /api/runs/wait`(无状态) | 同上 | 同样的 SSE 错误帧或 wait `{"status", "error", "__error__"}` 载荷 | `runs:59-91` |
wait 响应仍为 HTTP `200`。Python SDK 默认使用 `raise_error=True`,识别到 `__error__` 后会抛出异常,消息为 `RunError: <不匹配文本>`;传入 `raise_error=False` 才会返回 JSON 响应体,供调用者读取 `status`、`error` 和 `__error__`。
### 未做模式映射的路由

View File

@ -40,7 +40,9 @@ import { Callout } from "nextra/components";
**原因**:一个运行在 `full` 的进程读取了以 `delta` 写入检查点的线程。读门在物化之前抛出 `CheckpointModeMismatchError`,因此 full 模式进程绝不会静默返回空状态。返回 `409` 的路由是 `GET /api/threads/{thread_id}`、`GET` 与 `POST /api/threads/{thread_id}/state`、`POST /api/threads/{thread_id}/history`(均有测试覆盖);`POST /api/threads/{thread_id}/branches` 与 `POST .../compact` 在代码路径上以同样方式门禁。`compact` 也会因另一原因返回 `409`,detail 为 `Context compaction is disabled.`
threads 路由不会对创建 run 提前门禁,因此那里的不匹配改为异步失败:`POST .../runs` 返回 `200` run 记录,随后变为 `status: error`;`POST .../runs/stream` 返回 `200 text/event-stream` 后发出 `event: error` 帧;`POST .../runs/wait` 返回 `200` JSON,内容为 `{"status": "error", "error": "<不匹配文本>"}`。不过有一条 run 路径会在准入前读取检查点:当线程的 run event 流为空但存在检查点头部时,`ensure_checkpoint_history_seeded` 会在 `start_run` 内物化该头部,而那里只映射 `ConflictError` 与 `UnsupportedStrategyError`,因此不匹配会以未处理的 `500` 逃出,作用于 `POST /api/threads/{thread_id}/runs`、`/runs/stream` 与 `/runs/wait`。默认的 `run_events.backend: memory` 会让 Gateway 重启后每个既有线程的流都变空,所以这条分支在重启后可达。
threads 路由不会对创建 run 提前门禁,因此那里的不匹配改为异步失败:`POST .../runs` 返回 `200` run 记录,随后变为 `status: error`;`POST .../runs/stream` 返回 `200 text/event-stream` 后发出 `event: error` 帧;`POST .../runs/wait` 返回 `200` JSON,内容为 `{"status": "error", "error": "<不匹配文本>", "__error__": {"error": "RunError", "message": "<不匹配文本>"}}`。不过有一条 run 路径会在准入前读取检查点:当线程的 run event 流为空但存在检查点头部时,`ensure_checkpoint_history_seeded` 会在 `start_run` 内物化该头部,而那里只映射 `ConflictError` 与 `UnsupportedStrategyError`,因此不匹配会以未处理的 `500` 逃出,作用于 `POST /api/threads/{thread_id}/runs`、`/runs/stream` 与 `/runs/wait`。默认的 `run_events.backend: memory` 会让 Gateway 重启后每个既有线程的流都变空,所以这条分支在重启后可达。
wait 响应仍为 HTTP `200`。Python SDK 默认使用 `raise_error=True`,识别到 `__error__` 后会抛出异常,消息为 `RunError: <不匹配文本>`;传入 `raise_error=False` 才会返回 JSON 响应体,供调用者读取 `status`、`error` 和 `__error__`。
在内嵌 TUI 里,client 在图开始流式输出前就在进程内抛出:Textual runner 会把它变成 `AssistantError` 与 `RunEnded` 的组合;无头入口的处理方式一致——`--print` 在 stderr 上打印一行简洁的 `Error: ...` 并以非零码退出,`--json` 向 NDJSON 流追加一条终止性的 `{"type": "error"}` 记录(诊断文字留在 stderr)并以非零码退出。