From e646188ab468103c801af5a769fd94211d273820 Mon Sep 17 00:00:00 2001 From: Aari Date: Sun, 26 Jul 2026 15:00:42 +0800 Subject: [PATCH] fix(runtime): accept the SDK's default stream_resumable=false (#4468) RunCreateRequest declared stream_resumable as None-only, but langgraph_sdk defaults it to False (not None) and its payload filter only drops None, so every SDK request carried "stream_resumable": false and was rejected with 422. False means "no resumable stream", which is what DeerFlow already serves. This broke every IM channel run (runs.stream and runs.create both send the field; only runs.wait omits it) and any external langgraph_sdk client. The web frontend was unaffected because its compatibility wrapper does not send it. An explicit true still returns 422. The new regression test drives a real SDK client to capture its default payload and posts that to the HTTP boundary, so a future non-None SDK default cannot regress this silently. Fixes #4466 --- backend/AGENTS.md | 2 +- backend/app/gateway/run_models.py | 18 ++++++-- backend/docs/API.md | 3 +- backend/tests/test_run_request_validation.py | 47 +++++++++++++++++++- 4 files changed, 63 insertions(+), 7 deletions(-) diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 35241fcb3..59226f6b9 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -452,7 +452,7 @@ are size-limited; binary, large, and sensitive-looking paths are persisted as metadata only. **RunManager / RunStore contract**: -- LangGraph-compatible run requests validate their supported subset before creating a run. `runtime/stream_modes.py` is the shared backend contract for public stream modes and the worker's `graph.astream` mapping; the public `messages-tuple` mode maps to LangGraph's internal `messages` mode, while public `messages`, `events`, and other unsupported modes are rejected instead of being dropped or replaced with `values`. `app/gateway/run_models.py::RunCreateRequest` is shared by HTTP and internal scheduled launch paths, retains only truthful compatibility defaults for unimplemented options (`if_not_exists="create"` plus `None` placeholders), returns 422 for unsupported values including `on_completion="complete"`, `on_completion="continue"`, and `multitask_strategy="enqueue"`, and forbids undeclared SDK options so fields such as `checkpoint_during` and `durability` cannot be silently discarded. +- LangGraph-compatible run requests validate their supported subset before creating a run. `runtime/stream_modes.py` is the shared backend contract for public stream modes and the worker's `graph.astream` mapping; the public `messages-tuple` mode maps to LangGraph's internal `messages` mode, while public `messages`, `events`, and other unsupported modes are rejected instead of being dropped or replaced with `values`. `app/gateway/run_models.py::RunCreateRequest` is shared by HTTP and internal scheduled launch paths, retains only truthful compatibility defaults for unimplemented options (`if_not_exists="create"` plus `None` placeholders), returns 422 for unsupported values including `on_completion="complete"`, `on_completion="continue"`, and `multitask_strategy="enqueue"`, and forbids undeclared SDK options so fields such as `checkpoint_during` and `durability` cannot be silently discarded. A placeholder must still accept the stock SDK's own default: `langgraph_sdk` drops only `None` from its run payload, so `stream_resumable=False` reaches every request and means "non-resumable", which is what DeerFlow serves — rejecting it 422'd every IM channel run (#4466). `tests/test_run_request_validation.py::test_gateway_accepts_langgraph_sdk_default_payload` pins the real SDK payload against this boundary; channel tests mock the SDK client and cannot catch this class of drift. - `RunManager.get()` is async; direct callers must `await` it. - The history batch helpers `list_successful_regenerate_sources()` and `get_many_by_thread()` default to `user_id=AUTO`: they resolve the request user and fail closed when no user context exists. Migration/admin callers that intentionally need an unscoped read must pass `user_id=None` explicitly. - When a persistent `RunStore` is configured, `get()` and `list_by_thread()` hydrate historical runs from the store. In-memory records win for the same `run_id` so task, abort, and stream-control state stays attached to active local runs. diff --git a/backend/app/gateway/run_models.py b/backend/app/gateway/run_models.py index cfb8d1a7e..08e648501 100644 --- a/backend/app/gateway/run_models.py +++ b/backend/app/gateway/run_models.py @@ -28,7 +28,7 @@ class RunCreateRequest(BaseModel): interrupt_after: list[str] | Literal["*"] | None = Field(default=None, description="Nodes to interrupt after") stream_mode: list[RunStreamMode] | RunStreamMode | None = Field(default=None, description="Supported stream mode(s)") stream_subgraphs: bool = Field(default=False, description="Include subgraph events") - stream_resumable: None = Field(default=None, description="Compatibility placeholder; resumable SSE is not supported") + stream_resumable: Literal[False] | None = Field(default=None, description="Compatibility placeholder; only the SDK's non-resumable default (null/false) is accepted") on_disconnect: Literal["cancel", "continue"] = Field(default="cancel", description="Behaviour on SSE disconnect") on_completion: None = Field(default=None, description="Compatibility placeholder; completion behavior is not supported") multitask_strategy: Literal["reject", "rollback", "interrupt"] = Field(default="reject", description="Concurrency strategy") @@ -38,7 +38,6 @@ class RunCreateRequest(BaseModel): @field_validator( "webhook", - "stream_resumable", "on_completion", "multitask_strategy", "after_seconds", @@ -53,7 +52,6 @@ class RunCreateRequest(BaseModel): supported_defaults = { "webhook": None, - "stream_resumable": None, "on_completion": None, "multitask_strategy": {"reject", "rollback", "interrupt"}, "after_seconds": None, @@ -73,6 +71,20 @@ class RunCreateRequest(BaseModel): ) return value + @field_validator("stream_resumable", mode="before") + @classmethod + def reject_resumable_streams(cls, value: Any) -> Any: + # LangGraph SDK clients always send this field (its default is ``False``, which the + # payload's ``None`` filter keeps). ``False`` asks for the non-resumable stream + # DeerFlow already serves, so only an explicit ``True`` requests the unsupported feature. + if value is None or value is False: + return value + raise PydanticCustomError( + "unsupported_run_option", + "Run option '{option}' is not supported by DeerFlow", + {"option": "stream_resumable"}, + ) + @field_validator("stream_mode", mode="before") @classmethod def reject_unsupported_stream_modes(cls, value: Any) -> Any: diff --git a/backend/docs/API.md b/backend/docs/API.md index 329b32e78..b23ddeb9d 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -109,7 +109,8 @@ Content-Type: application/json **Run Option Compatibility:** - Supported concurrency strategies: `reject`, `rollback`, and `interrupt` - Compatibility default: `if_not_exists="create"`; this matches DeerFlow's current behavior -- Unsupported options return `422`: `webhook`, `stream_resumable`, `after_seconds`, `feedback_keys`, any non-null `on_completion` value (including the SDK values `"complete"` and `"continue"`), `if_not_exists="reject"`, and `multitask_strategy="enqueue"` +- Unsupported options return `422`: `webhook`, `stream_resumable=true`, `after_seconds`, `feedback_keys`, any non-null `on_completion` value (including the SDK values `"complete"` and `"continue"`), `if_not_exists="reject"`, and `multitask_strategy="enqueue"` +- `stream_resumable=false` is accepted: it is the LangGraph SDK's default and requests the non-resumable stream DeerFlow already serves - Undeclared SDK options, including `checkpoint_during` and `durability`, also return `422` instead of being silently discarded **Recursion Limit:** diff --git a/backend/tests/test_run_request_validation.py b/backend/tests/test_run_request_validation.py index d428e99dc..1e8516e7a 100644 --- a/backend/tests/test_run_request_validation.py +++ b/backend/tests/test_run_request_validation.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio from typing import Any import pytest @@ -32,7 +33,7 @@ def client() -> TestClient: ("field", "value"), [ ("webhook", "https://example.com/callback"), - ("stream_resumable", False), + ("stream_resumable", True), ("on_completion", "complete"), ("on_completion", "continue"), ("on_completion", "keep"), @@ -115,6 +116,47 @@ def test_run_request_keeps_supported_modes_and_compatibility_defaults() -> None: assert body.if_not_exists == "create" +def _sdk_default_payload(method: str) -> dict[str, Any]: + """Capture the body a stock ``langgraph_sdk`` client sends for a default run.""" + from langgraph_sdk.client import get_client + + client = get_client(url="http://gateway.invalid") + captured: dict[str, Any] = {} + + def capture(*_args: Any, json: dict[str, Any] | None = None, **_kwargs: Any) -> None: + captured.update(json or {}) + + async def capture_post(*args: Any, **kwargs: Any) -> dict[str, Any]: + capture(*args, **kwargs) + return {} + + if method == "stream": + client.runs.http.stream = capture # type: ignore[method-assign] + # ``stream()`` is a sync factory: the payload is built and handed to the + # transport before the returned async iterator is consumed. + client.runs.stream("thread-id", "deerflow", input={"messages": []}) + else: + client.runs.http.post = capture_post # type: ignore[method-assign] + asyncio.run(client.runs.create("thread-id", "deerflow", input={"messages": []})) + return captured + + +@pytest.mark.parametrize("method", ["stream", "create"]) +def test_gateway_accepts_langgraph_sdk_default_payload(client: TestClient, method: str) -> None: + """The SDK's own defaults must never be rejected as unsupported options (#4466). + + ``stream_resumable`` defaults to ``False`` rather than ``None``, so the SDK's + drop-``None`` payload filter always forwards it; every IM channel run goes + through this path. + """ + payload = _sdk_default_payload(method) + + assert payload["stream_resumable"] is False, "SDK contract changed; update this test" + response = client.post("/api/runs/stream", json=payload) + + assert response.status_code != 422, response.text + + @pytest.mark.parametrize( "payload", [ @@ -203,8 +245,9 @@ def test_openapi_run_option_schema_exposes_only_supported_values() -> None: properties = schema["properties"] assert schema["additionalProperties"] is False - for field in ("webhook", "stream_resumable", "after_seconds", "feedback_keys"): + for field in ("webhook", "after_seconds", "feedback_keys"): assert properties[field]["type"] == "null" + assert properties["stream_resumable"]["anyOf"] == [{"const": False, "type": "boolean"}, {"type": "null"}] assert properties["on_completion"]["type"] == "null" assert properties["if_not_exists"]["const"] == "create" assert properties["multitask_strategy"]["enum"] == ["reject", "rollback", "interrupt"]