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

@@ -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.

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:

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:**

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"]