From 06e9bfd269044578603c95e9a5d9fc74c51a6b6f Mon Sep 17 00:00:00 2001 From: quifox <289420841+quifox@users.noreply.github.com> Date: Sun, 4 Oct 2026 07:46:59 +0800 Subject: [PATCH] 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 --- README.md | 2 + backend/app/gateway/AGENTS.md | 2 +- backend/app/gateway/routers/runs.py | 13 +- backend/app/gateway/routers/thread_runs.py | 22 +++- backend/app/gateway/services.py | 8 ++ .../tests/test_stateless_run_wait_errors.py | 122 ++++++++++++++++++ .../tests/test_thread_regenerate_prepare.py | 13 +- backend/tests/test_thread_run_wait_errors.py | 89 +++++++++++++ .../en/harness/checkpoints/observability.mdx | 4 +- .../en/harness/checkpoints/reference.mdx | 6 +- .../harness/checkpoints/troubleshooting.mdx | 4 +- .../zh/harness/checkpoints/observability.mdx | 4 +- .../zh/harness/checkpoints/reference.mdx | 6 +- .../harness/checkpoints/troubleshooting.mdx | 4 +- 14 files changed, 281 insertions(+), 18 deletions(-) create mode 100644 backend/tests/test_stateless_run_wait_errors.py create mode 100644 backend/tests/test_thread_run_wait_errors.py diff --git a/README.md b/README.md index 39a065473d..2a46a1197d 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 351b7820a0..b6d535e5a2 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -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. diff --git a/backend/app/gateway/routers/runs.py b/backend/app/gateway/routers/runs.py index c16aa66a40..dbf25d769a 100644 --- a/backend/app/gateway/routers/runs.py +++ b/backend/app/gateway/routers/runs.py @@ -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( diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index d66a2da464..6a49c69bf5 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -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: diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 73d0cdf7bd..c84e827ca7 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -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, diff --git a/backend/tests/test_stateless_run_wait_errors.py b/backend/tests/test_stateless_run_wait_errors.py new file mode 100644 index 0000000000..eb42f9fac9 --- /dev/null +++ b/backend/tests/test_stateless_run_wait_errors.py @@ -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() diff --git a/backend/tests/test_thread_regenerate_prepare.py b/backend/tests/test_thread_regenerate_prepare.py index 19ad45471b..94e66b31c3 100644 --- a/backend/tests/test_thread_regenerate_prepare.py +++ b/backend/tests/test_thread_regenerate_prepare.py @@ -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(): diff --git a/backend/tests/test_thread_run_wait_errors.py b/backend/tests/test_thread_run_wait_errors.py new file mode 100644 index 0000000000..ecc314fe4a --- /dev/null +++ b/backend/tests/test_thread_run_wait_errors.py @@ -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() diff --git a/frontend/src/content/en/harness/checkpoints/observability.mdx b/frontend/src/content/en/harness/checkpoints/observability.mdx index f0f61246a6..8b9631b959 100644 --- a/frontend/src/content/en/harness/checkpoints/observability.mdx +++ b/frontend/src/content/en/harness/checkpoints/observability.mdx @@ -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": ""}`. 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": "", "__error__": {"error": "RunError", "message": ""}}`. 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: ` when it sees `__error__`; pass `raise_error=False` to receive the JSON body and inspect `status`, `error`, and `__error__`. The mode gate itself logs nothing. `checkpoint_mode.py` has no logger, diff --git a/frontend/src/content/en/harness/checkpoints/reference.mdx b/frontend/src/content/en/harness/checkpoints/reference.mdx index 8fa502f2d9..db1629b0db 100644 --- a/frontend/src/content/en/harness/checkpoints/reference.mdx +++ b/frontend/src/content/en/harness/checkpoints/reference.mdx @@ -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": , "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: ` when it sees `__error__`; pass `raise_error=False` to receive the JSON body and inspect `status`, `error`, and `__error__`. ### Routes with no mode mapping diff --git a/frontend/src/content/en/harness/checkpoints/troubleshooting.mdx b/frontend/src/content/en/harness/checkpoints/troubleshooting.mdx index b9f886fcca..148b3cd11c 100644 --- a/frontend/src/content/en/harness/checkpoints/troubleshooting.mdx +++ b/frontend/src/content/en/harness/checkpoints/troubleshooting.mdx @@ -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": ""}`. 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": "", "__error__": {"error": "RunError", "message": ""}}`. 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: ` 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. diff --git a/frontend/src/content/zh/harness/checkpoints/observability.mdx b/frontend/src/content/zh/harness/checkpoints/observability.mdx index 7920290546..33269c17fa 100644 --- a/frontend/src/content/zh/harness/checkpoints/observability.mdx +++ b/frontend/src/content/zh/harness/checkpoints/observability.mdx @@ -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__`。 模式门自身不记任何日志。`checkpoint_mode.py` 没有 logger,threads diff --git a/frontend/src/content/zh/harness/checkpoints/reference.mdx b/frontend/src/content/zh/harness/checkpoints/reference.mdx index b7721ec775..0980ae505f 100644 --- a/frontend/src/content/zh/harness/checkpoints/reference.mdx +++ b/frontend/src/content/zh/harness/checkpoints/reference.mdx @@ -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__`。 ### 未做模式映射的路由 diff --git a/frontend/src/content/zh/harness/checkpoints/troubleshooting.mdx b/frontend/src/content/zh/harness/checkpoints/troubleshooting.mdx index 413dd0e91c..8b1ee915b2 100644 --- a/frontend/src/content/zh/harness/checkpoints/troubleshooting.mdx +++ b/frontend/src/content/zh/harness/checkpoints/troubleshooting.mdx @@ -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)并以非零码退出。