fix(runtime): reject unsupported run options and stream modes (#4430)

* fix(runtime): reject unsupported run options

* fix(runtime): align SDK run compatibility

* fix(frontend): avoid unsupported events stream mode

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -876,6 +876,8 @@ client.get_goal("thread-1")       # {"goal": {...}} or {"goal": None}
 client.clear_goal("thread-1")
 ```
 
+The HTTP Gateway accepts `values`, `messages-tuple`, `updates`, `debug`, `tasks`, `checkpoints`, and `custom` stream modes. Unsupported modes such as `messages` and `events`, unsupported non-default run options such as webhooks, delayed execution, or `multitask_strategy="enqueue"`, and undeclared SDK options such as checkpoint durability overrides return `422` before execution instead of being silently ignored or downgraded.
+
 All dict-returning methods are validated against Gateway Pydantic response models in CI (`TestGatewayConformance`), ensuring the embedded client stays in sync with the HTTP API schemas. See `backend/packages/harness/deerflow/client.py` for full API documentation.
 
 ## Scheduled Tasks

backend/AGENTS.md

@@ -436,6 +436,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.
 - `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/routers/runs.py

@@ -16,7 +16,7 @@ from fastapi.responses import StreamingResponse
 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.routers.thread_runs import RunCreateRequest
+from app.gateway.run_models import RunCreateRequest
 from app.gateway.services import build_checkpoint_state_accessor, sse_consumer, start_run, wait_for_run_completion
 from deerflow.runtime import serialize_channel_values_for_api
 

backend/app/gateway/routers/thread_runs.py

@@ -34,6 +34,7 @@ from app.gateway.checkpoint_lineage import (
 )
 from app.gateway.deps import get_current_user, 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 build_checkpoint_state_accessor, build_thread_checkpoint_state_accessor, sse_consumer, start_run, wait_for_run_completion
 from app.gateway.utils import sanitize_log_param
 from deerflow.runtime import CancelOutcome, RunRecord, RunStatus, serialize_channel_values_for_api
@@ -78,29 +79,6 @@ def compute_run_durations(runs) -> dict[str, int]:
 # ---------------------------------------------------------------------------
 
 
-class RunCreateRequest(BaseModel):
-    assistant_id: str | None = Field(default=None, description="Agent / assistant to use")
-    input: dict[str, Any] | None = Field(default=None, description="Graph input (e.g. {messages: [...]})")
-    command: dict[str, Any] | None = Field(default=None, description="LangGraph Command")
-    metadata: dict[str, Any] | None = Field(default=None, description="Run metadata")
-    config: dict[str, Any] | None = Field(default=None, description="RunnableConfig overrides")
-    context: dict[str, Any] | None = Field(default=None, description="DeerFlow context overrides (model_name, thinking_enabled, etc.)")
-    webhook: str | None = Field(default=None, description="Completion callback URL")
-    checkpoint_id: str | None = Field(default=None, description="Resume from checkpoint")
-    checkpoint: dict[str, Any] | None = Field(default=None, description="Full checkpoint object")
-    interrupt_before: list[str] | Literal["*"] | None = Field(default=None, description="Nodes to interrupt before")
-    interrupt_after: list[str] | Literal["*"] | None = Field(default=None, description="Nodes to interrupt after")
-    stream_mode: list[str] | str | None = Field(default=None, description="Stream mode(s)")
-    stream_subgraphs: bool = Field(default=False, description="Include subgraph events")
-    stream_resumable: bool | None = Field(default=None, description="SSE resumable mode")
-    on_disconnect: Literal["cancel", "continue"] = Field(default="cancel", description="Behaviour on SSE disconnect")
-    on_completion: Literal["delete", "keep"] = Field(default="keep", description="Delete temp thread on completion")
-    multitask_strategy: Literal["reject", "rollback", "interrupt", "enqueue"] = Field(default="reject", description="Concurrency strategy")
-    after_seconds: float | None = Field(default=None, description="Delayed execution")
-    if_not_exists: Literal["reject", "create"] = Field(default="create", description="Thread creation policy")
-    feedback_keys: list[str] | None = Field(default=None, description="LangSmith feedback keys")
-
-
 class RegeneratePrepareRequest(BaseModel):
     message_id: str = Field(..., min_length=1, description="Assistant message id to regenerate")
 

backend/app/gateway/run_models.py

@@ -0,0 +1,92 @@
+"""Shared request models for the LangGraph-compatible run boundary."""
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator
+from pydantic_core import PydanticCustomError
+
+from deerflow.runtime.stream_modes import RunStreamMode, UnsupportedStreamModeError, normalize_stream_modes
+
+
+class RunCreateRequest(BaseModel):
+    """Validated run request used by both HTTP and internal launch paths."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    assistant_id: str | None = Field(default=None, description="Agent / assistant to use")
+    input: dict[str, Any] | None = Field(default=None, description="Graph input (e.g. {messages: [...]})")
+    command: dict[str, Any] | None = Field(default=None, description="LangGraph Command")
+    metadata: dict[str, Any] | None = Field(default=None, description="Run metadata")
+    config: dict[str, Any] | None = Field(default=None, description="RunnableConfig overrides")
+    context: dict[str, Any] | None = Field(default=None, description="DeerFlow context overrides (model_name, thinking_enabled, etc.)")
+    webhook: None = Field(default=None, description="Compatibility placeholder; completion callbacks are not supported")
+    checkpoint_id: str | None = Field(default=None, description="Resume from checkpoint")
+    checkpoint: dict[str, Any] | None = Field(default=None, description="Full checkpoint object")
+    interrupt_before: list[str] | Literal["*"] | None = Field(default=None, description="Nodes to interrupt before")
+    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")
+    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")
+    after_seconds: None = Field(default=None, description="Compatibility placeholder; delayed execution is not supported")
+    if_not_exists: Literal["create"] = Field(default="create", description="Compatibility default; missing threads are created")
+    feedback_keys: None = Field(default=None, description="Compatibility placeholder; feedback key collection is not supported")
+
+    @field_validator(
+        "webhook",
+        "stream_resumable",
+        "on_completion",
+        "multitask_strategy",
+        "after_seconds",
+        "if_not_exists",
+        "feedback_keys",
+        mode="before",
+    )
+    @classmethod
+    def reject_unsupported_run_options(cls, value: Any, info: ValidationInfo) -> Any:
+        if info.field_name in {"multitask_strategy", "if_not_exists"} and not isinstance(value, str):
+            return value
+
+        supported_defaults = {
+            "webhook": None,
+            "stream_resumable": None,
+            "on_completion": None,
+            "multitask_strategy": {"reject", "rollback", "interrupt"},
+            "after_seconds": None,
+            "if_not_exists": "create",
+            "feedback_keys": None,
+        }
+        supported = supported_defaults[info.field_name]
+        if isinstance(supported, set):
+            is_supported = isinstance(value, str) and value in supported
+        else:
+            is_supported = value == supported
+        if not is_supported:
+            raise PydanticCustomError(
+                "unsupported_run_option",
+                "Run option '{option}' is not supported by DeerFlow",
+                {"option": info.field_name},
+            )
+        return value
+
+    @field_validator("stream_mode", mode="before")
+    @classmethod
+    def reject_unsupported_stream_modes(cls, value: Any) -> Any:
+        if value is None:
+            return value
+        if not isinstance(value, str) and (not isinstance(value, list) or not all(isinstance(mode, str) for mode in value)):
+            return value
+        try:
+            normalize_stream_modes(value)
+        except UnsupportedStreamModeError as exc:
+            modes = ", ".join(exc.modes)
+            raise PydanticCustomError(
+                "unsupported_stream_mode",
+                "Unsupported stream mode(s): {modes}",
+                {"modes": modes},
+            ) from exc
+        return value

backend/app/gateway/services.py

@@ -28,6 +28,7 @@ from app.gateway.internal_auth import (
     get_internal_user,
     get_trusted_internal_owner_user_id,
 )
+from app.gateway.run_models import RunCreateRequest
 from app.gateway.utils import sanitize_log_param
 from deerflow.agents.middlewares.dynamic_context_middleware import _DYNAMIC_CONTEXT_REMINDER_KEY, _REMINDER_DATE_KEY
 from deerflow.agents.middlewares.view_image_middleware import _IMAGE_CONTEXT_MESSAGE_MARKER_KEY
@@ -56,6 +57,7 @@ from deerflow.runtime.checkpoint_state import graph_state_schema
 from deerflow.runtime.goal import goal_thread_lock
 from deerflow.runtime.runs.naming import resolve_root_run_name
 from deerflow.runtime.secret_context import redact_config_secrets
+from deerflow.runtime.stream_modes import normalize_stream_modes
 from deerflow.runtime.user_context import reset_current_user, set_current_user
 from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
 
@@ -125,18 +127,6 @@ async def _terminal_record_stream_missing(bridge: StreamBridge, record: RunRecor
 # ---------------------------------------------------------------------------
 
 
-def normalize_stream_modes(raw: list[str] | str | None) -> list[str]:
-    """Normalize the stream_mode parameter to a list.
-
-    Default matches what ``useStream`` expects: values + messages-tuple.
-    """
-    if raw is None:
-        return ["values"]
-    if isinstance(raw, str):
-        return [raw]
-    return raw if raw else ["values"]
-
-
 def _strip_external_message_metadata(message: Any) -> Any:
     """Remove server-owned metadata from an untrusted input message."""
     if not isinstance(message, BaseMessage):
@@ -884,7 +874,7 @@ async def apply_checkpoint_to_run_config(
 
 
 async def start_run(
-    body: Any,
+    body: RunCreateRequest,
     thread_id: str,
     request: Request,
 ) -> RunRecord:
@@ -893,13 +883,13 @@ async def start_run(
     Parameters
     ----------
     body : RunCreateRequest
-        The validated request body (typed as Any to avoid circular import
-        with the router module that defines the Pydantic model).
+        The validated request body shared by HTTP and internal launch paths.
     thread_id : str
         Target thread.
     request : Request
         FastAPI request — used to retrieve singletons from ``app.state``.
     """
+    stream_modes = normalize_stream_modes(body.stream_mode)
     bridge = get_stream_bridge(request)
     run_mgr = get_run_manager(request)
     run_ctx = get_run_context(request)
@@ -1019,8 +1009,6 @@ async def start_run(
             request_context=getattr(body, "context", None),
         )
 
-        stream_modes = normalize_stream_modes(body.stream_mode)
-
         task = asyncio.create_task(
             run_agent(
                 bridge,
@@ -1070,10 +1058,7 @@ async def launch_scheduled_thread_run(
             ),
             cookies={},
         )
-    # SimpleNamespace stands in for the Pydantic run-request body that the
-    # HTTP path parses. If start_run gains a new body.* attribute that it reads
-    # directly, add the matching field here so the scheduler path stays in sync.
-    body = SimpleNamespace(
+    body = RunCreateRequest(
         assistant_id=assistant_id,
         input={"messages": [{"role": "user", "content": prompt}]},
         command=None,
@@ -1093,10 +1078,10 @@ async def launch_scheduled_thread_run(
         stream_subgraphs=False,
         stream_resumable=None,
         on_disconnect="continue",
-        on_completion="keep",
+        on_completion=None,
         multitask_strategy="reject",
         after_seconds=None,
-        if_not_exists="reject",
+        if_not_exists="create",
         feedback_keys=None,
     )
     record = await start_run(body, thread_id, request)

backend/docs/API.md

@@ -103,8 +103,14 @@ Content-Type: application/json
 ```
 
 **Stream Mode Compatibility:**
-- Use: `values`, `messages-tuple`, `custom`, `updates`, `events`, `debug`, `tasks`, `checkpoints`
-- Do not use: `tools` (deprecated/invalid in current `langgraph-api` and will trigger schema validation errors)
+- Use: `values`, `messages-tuple`, `custom`, `updates`, `debug`, `tasks`, `checkpoints`
+- Unsupported modes, including `messages`, `events`, and `tools`, return `422` before a run is created. DeerFlow never substitutes `values` for an unsupported mode.
+
+**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"`
+- Undeclared SDK options, including `checkpoint_during` and `durability`, also return `422` instead of being silently discarded
 
 **Recursion Limit:**
 

backend/packages/harness/deerflow/runtime/runs/worker.py

@@ -7,7 +7,7 @@ Uses ``graph.astream(stream_mode=[...])`` which gives correct full-state
 snapshots for ``values`` mode, proper ``{node: writes}`` for ``updates``,
 and ``(chunk, metadata)`` tuples for ``messages`` mode.
 
-Note: ``events`` mode is not supported through the gateway — it requires
+Note: ``events`` mode is rejected by the gateway — it requires
 ``graph.astream_events()`` which cannot simultaneously produce ``values``
 snapshots.  The JS open-source LangGraph API server works around this via
 internal checkpoint callbacks that are not exposed in the Python public API.
@@ -63,6 +63,7 @@ from deerflow.runtime.goal import (
 )
 from deerflow.runtime.serialization import serialize
 from deerflow.runtime.stream_bridge import StreamBridge
+from deerflow.runtime.stream_modes import normalize_stream_modes, to_langgraph_stream_modes
 from deerflow.runtime.user_context import get_effective_user_id, resolve_runtime_user_id
 from deerflow.trace_context import (
     DEERFLOW_TRACE_METADATA_KEY,
@@ -102,8 +103,6 @@ async def _checkpoint_thread_lock(thread_id: str) -> AsyncIterator[None]:
         yield
 
 
-# Valid stream_mode values for LangGraph's graph.astream()
-_VALID_LG_MODES = {"values", "updates", "checkpoints", "tasks", "debug", "messages", "custom"}
 # Keep this streaming policy separate from middleware write-authorization sets.
 _LARGE_FILE_TOOL_NAMES = frozenset({"str_replace", "write_file"})
 _LARGE_FILE_TOOL_BATCH_SIZE = 32
@@ -397,7 +396,6 @@ async def run_agent(
 
     run_id = record.run_id
     thread_id = record.thread_id
-    requested_modes: set[str] = set(stream_modes or ["values"])
     pre_run_checkpoint_id: str | None = None
     pre_run_workspace_snapshot: WorkspaceSnapshot | None = None
     workspace_changes_user_id: str | None = None
@@ -420,14 +418,10 @@ async def run_agent(
     # finally is safe even if an exception fires before streaming begins.
     subagent_events: _SubagentEventBuffer | None = None
 
-    # Track whether "events" was requested but skipped
-    if "events" in requested_modes:
-        logger.info(
-            "Run %s: 'events' stream_mode not supported in gateway (requires astream_events + checkpoint callbacks). Skipping.",
-            run_id,
-        )
-
     try:
+        normalized_stream_modes = normalize_stream_modes(stream_modes)
+        requested_modes: set[str] = set(normalized_stream_modes)
+        lg_modes = to_langgraph_stream_modes(normalized_stream_modes)
         await run_manager.wait_for_prior_finalizing(thread_id, run_id)
         mode = ctx.checkpoint_channel_mode
         inject_checkpoint_mode(config, mode)
@@ -616,30 +610,6 @@ async def run_agent(
         if interrupt_after:
             agent.interrupt_after_nodes = interrupt_after
 
-        # 6. Build LangGraph stream_mode list
-        #    "events" is NOT a valid astream mode — skip it
-        #    "messages-tuple" maps to LangGraph's "messages" mode
-        lg_modes: list[str] = []
-        for m in requested_modes:
-            if m == "messages-tuple":
-                lg_modes.append("messages")
-            elif m == "events":
-                # Skipped — see log above
-                continue
-            elif m in _VALID_LG_MODES:
-                lg_modes.append(m)
-        if not lg_modes:
-            lg_modes = ["values"]
-
-        # Deduplicate while preserving order
-        seen: set[str] = set()
-        deduped: list[str] = []
-        for m in lg_modes:
-            if m not in seen:
-                seen.add(m)
-                deduped.append(m)
-        lg_modes = deduped
-
         logger.info("Run %s: streaming with modes %s (requested: %s)", run_id, lg_modes, requested_modes)
 
         # Buffer subagent step events and persist them in batches (#3779) instead

backend/packages/harness/deerflow/runtime/stream_modes.py

@@ -0,0 +1,47 @@
+"""Supported stream modes for the LangGraph-compatible runtime boundary."""
+
+from __future__ import annotations
+
+from typing import Literal, get_args
+
+type RunStreamMode = Literal[
+    "values",
+    "messages-tuple",
+    "updates",
+    "debug",
+    "tasks",
+    "checkpoints",
+    "custom",
+]
+
+SUPPORTED_RUN_STREAM_MODES: frozenset[str] = frozenset(get_args(RunStreamMode.__value__))
+
+
+class UnsupportedStreamModeError(ValueError):
+    """Raised when a caller requests a stream mode DeerFlow cannot honor."""
+
+    def __init__(self, modes: list[str]) -> None:
+        self.modes = tuple(dict.fromkeys(modes))
+        super().__init__(f"Unsupported stream mode(s): {', '.join(self.modes)}")
+
+
+def normalize_stream_modes(raw: list[str] | str | None) -> list[str]:
+    """Normalize and validate public run stream modes."""
+    if raw is None:
+        modes = ["values"]
+    elif isinstance(raw, str):
+        modes = [raw]
+    else:
+        modes = raw or ["values"]
+
+    unsupported = [mode if isinstance(mode, str) else type(mode).__name__ for mode in modes if not isinstance(mode, str) or mode not in SUPPORTED_RUN_STREAM_MODES]
+    if unsupported:
+        raise UnsupportedStreamModeError(unsupported)
+    return modes
+
+
+def to_langgraph_stream_modes(raw: list[str] | str | None) -> list[str]:
+    """Map public run modes to ``graph.astream`` modes without silent fallback."""
+    modes = normalize_stream_modes(raw)
+    mapped = ["messages" if mode == "messages-tuple" else mode for mode in modes]
+    return list(dict.fromkeys(mapped))

backend/tests/test_gateway_services.py

@@ -79,6 +79,28 @@ def test_normalize_stream_modes_empty_list():
     assert normalize_stream_modes([]) == ["values"]
 
 
+@pytest.mark.parametrize("raw", ["messages", "events", "tools", ["values", "events"]])
+def test_normalize_stream_modes_rejects_unsupported_modes(raw):
+    from app.gateway.services import normalize_stream_modes
+
+    with pytest.raises(ValueError, match="Unsupported stream mode"):
+        normalize_stream_modes(raw)
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        ("messages-tuple", ["messages"]),
+        (["values", "messages-tuple", "messages-tuple", "values"], ["values", "messages"]),
+        (["updates", "custom"], ["updates", "custom"]),
+    ],
+)
+def test_to_langgraph_stream_modes_maps_alias_and_deduplicates(raw, expected):
+    from deerflow.runtime.stream_modes import to_langgraph_stream_modes
+
+    assert to_langgraph_stream_modes(raw) == expected
+
+
 def test_normalize_input_none():
     from app.gateway.services import normalize_input
 
@@ -1438,15 +1460,19 @@ def test_launch_scheduled_thread_run_marks_context_non_interactive(_stub_app_con
     from types import SimpleNamespace
     from unittest.mock import patch
 
+    from app.gateway.routers.thread_runs import RunCreateRequest
     from app.gateway.services import launch_scheduled_thread_run
 
     async def _scenario():
         captured: dict[str, object] = {}
 
         async def fake_start_run(body, thread_id, request):
+            captured["body"] = body
             captured["thread_id"] = thread_id
             captured["context"] = body.context
             captured["metadata"] = body.metadata
+            captured["if_not_exists"] = body.if_not_exists
+            captured["on_completion"] = body.on_completion
             return SimpleNamespace(run_id="run-1", thread_id=thread_id)
 
         with patch("app.gateway.services.start_run", side_effect=fake_start_run):
@@ -1463,8 +1489,11 @@ def test_launch_scheduled_thread_run_marks_context_non_interactive(_stub_app_con
     captured, result = asyncio.run(_scenario())
 
     assert captured["thread_id"] == "thread-scheduled"
+    assert isinstance(captured["body"], RunCreateRequest)
     assert captured["context"] == {"non_interactive": True, "user_id": "user-1"}
     assert captured["metadata"] == {"scheduled_task_id": "task-1"}
+    assert captured["if_not_exists"] == "create"
+    assert captured["on_completion"] is None
     assert result == {"run_id": "run-1", "thread_id": "thread-scheduled"}
 
 
@@ -1757,6 +1786,54 @@ class TestInjectAuthenticatedUserContextAuthz:
             inject_authenticated_user_context(config, _make_request_with_auth_source("session"))
 
 
+@pytest.mark.asyncio
+async def test_run_agent_invalid_stream_mode_finalizes_run_before_graph_invocation():
+    import asyncio
+    from types import SimpleNamespace
+    from unittest.mock import AsyncMock, MagicMock
+
+    from deerflow.runtime.runs.manager import RunManager
+    from deerflow.runtime.runs.schemas import RunStatus
+    from deerflow.runtime.runs.worker import RunContext, run_agent
+
+    run_manager = RunManager()
+    record = await run_manager.create("thread-invalid-stream-mode")
+    bridge = SimpleNamespace(
+        publish=AsyncMock(),
+        publish_end=AsyncMock(),
+        cleanup=AsyncMock(),
+    )
+    agent_factory = MagicMock()
+
+    await run_agent(
+        bridge,
+        run_manager,
+        record,
+        ctx=RunContext(checkpointer=None),
+        agent_factory=agent_factory,
+        graph_input={"messages": []},
+        config={"configurable": {"thread_id": record.thread_id}},
+        stream_modes=["events"],
+    )
+    await asyncio.sleep(0)
+
+    assert record.status == RunStatus.error
+    assert record.error == "Unsupported stream mode(s): events"
+    agent_factory.assert_not_called()
+    bridge.publish.assert_awaited_once_with(
+        record.run_id,
+        "error",
+        {
+            "message": "Unsupported stream mode(s): events",
+            "name": "UnsupportedStreamModeError",
+        },
+    )
+    bridge.publish_end.assert_awaited_once_with(record.run_id)
+    bridge.cleanup.assert_awaited_once_with(record.run_id, delay=60)
+    replacement = await run_manager.create_or_reject(record.thread_id)
+    assert replacement.run_id != record.run_id
+
+
 @pytest.mark.asyncio
 async def test_run_agent_full_mode_rejects_delta_before_graph_invocation():
     import asyncio

backend/tests/test_run_request_validation.py

@@ -0,0 +1,210 @@
+"""Contract tests for the LangGraph-compatible run request boundary."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.gateway.routers import runs
+
+SUPPORTED_STREAM_MODES = {
+    "values",
+    "messages-tuple",
+    "updates",
+    "debug",
+    "tasks",
+    "checkpoints",
+    "custom",
+}
+
+
+@pytest.fixture
+def client() -> TestClient:
+    app = FastAPI()
+    app.include_router(runs.router)
+    return TestClient(app, raise_server_exceptions=False)
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("webhook", "https://example.com/callback"),
+        ("stream_resumable", False),
+        ("on_completion", "complete"),
+        ("on_completion", "continue"),
+        ("on_completion", "keep"),
+        ("on_completion", "delete"),
+        ("after_seconds", 1.5),
+        ("if_not_exists", "reject"),
+        ("feedback_keys", []),
+        ("multitask_strategy", "enqueue"),
+    ],
+)
+def test_run_request_rejects_each_unsupported_option_with_exact_422(
+    client: TestClient,
+    field: str,
+    value: Any,
+) -> None:
+    response = client.post("/api/runs/stream", json={field: value})
+
+    assert response.status_code == 422
+    assert response.json() == {
+        "detail": [
+            {
+                "type": "unsupported_run_option",
+                "loc": ["body", field],
+                "msg": f"Run option '{field}' is not supported by DeerFlow",
+                "input": value,
+                "ctx": {"option": field},
+            }
+        ]
+    }
+
+
+@pytest.mark.parametrize(
+    "stream_mode",
+    [
+        "messages",
+        "events",
+        "tools",
+        ["values", "events"],
+        ["events", "tools"],
+    ],
+)
+def test_run_request_rejects_unsupported_stream_modes_with_exact_422(
+    client: TestClient,
+    stream_mode: str | list[str],
+) -> None:
+    response = client.post("/api/runs/stream", json={"stream_mode": stream_mode})
+    requested = [stream_mode] if isinstance(stream_mode, str) else stream_mode
+    unsupported = list(dict.fromkeys(mode for mode in requested if mode not in SUPPORTED_STREAM_MODES))
+    mode_list = ", ".join(unsupported)
+
+    assert response.status_code == 422
+    assert response.json() == {
+        "detail": [
+            {
+                "type": "unsupported_stream_mode",
+                "loc": ["body", "stream_mode"],
+                "msg": f"Unsupported stream mode(s): {mode_list}",
+                "input": stream_mode,
+                "ctx": {"modes": mode_list},
+            }
+        ]
+    }
+
+
+def test_run_request_keeps_supported_modes_and_compatibility_defaults() -> None:
+    from app.gateway.routers.thread_runs import RunCreateRequest
+
+    body = RunCreateRequest(
+        stream_mode=list(SUPPORTED_STREAM_MODES),
+        webhook=None,
+        stream_resumable=None,
+        on_completion=None,
+        after_seconds=None,
+        if_not_exists="create",
+        feedback_keys=None,
+    )
+
+    assert set(body.stream_mode or []) == SUPPORTED_STREAM_MODES
+    assert body.on_completion is None
+    assert body.if_not_exists == "create"
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {"multitask_strategy": []},
+        {"stream_mode": [{}]},
+    ],
+)
+def test_malformed_option_types_remain_validation_errors(client: TestClient, payload: dict[str, Any]) -> None:
+    response = client.post("/api/runs/stream", json=payload)
+
+    assert response.status_code == 422
+
+
+@pytest.mark.parametrize(
+    "field",
+    [
+        "multitask_strategy",
+        "if_not_exists",
+    ],
+)
+def test_string_run_options_keep_native_type_errors(client: TestClient, field: str) -> None:
+    response = client.post("/api/runs/stream", json={field: []})
+
+    assert response.status_code == 422
+    assert response.json()["detail"][0]["type"] == "literal_error"
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("checkpoint_during", False),
+        ("durability", "sync"),
+    ],
+)
+def test_run_request_rejects_unknown_sdk_options_with_exact_422(
+    client: TestClient,
+    field: str,
+    value: Any,
+) -> None:
+    response = client.post("/api/runs/stream", json={field: value})
+
+    assert response.status_code == 422
+    assert response.json() == {
+        "detail": [
+            {
+                "type": "extra_forbidden",
+                "loc": ["body", field],
+                "msg": "Extra inputs are not permitted",
+                "input": value,
+            }
+        ]
+    }
+
+
+def test_openapi_stream_mode_enum_matches_runtime_support() -> None:
+    app = FastAPI()
+    app.include_router(runs.router)
+    openapi = app.openapi()
+    schemas = openapi["components"]["schemas"]
+    stream_mode_schema = schemas["RunCreateRequest"]["properties"]["stream_mode"]
+
+    enums: list[str] = []
+
+    def collect_enums(value: Any) -> None:
+        if isinstance(value, dict):
+            enum = value.get("enum")
+            if isinstance(enum, list):
+                enums.extend(item for item in enum if isinstance(item, str))
+            for child in value.values():
+                collect_enums(child)
+        elif isinstance(value, list):
+            for child in value:
+                collect_enums(child)
+
+    collect_enums(stream_mode_schema)
+    collect_enums(schemas["RunStreamMode"])
+
+    assert set(enums) == SUPPORTED_STREAM_MODES
+    assert "events" not in enums
+
+
+def test_openapi_run_option_schema_exposes_only_supported_values() -> None:
+    app = FastAPI()
+    app.include_router(runs.router)
+    schema = app.openapi()["components"]["schemas"]["RunCreateRequest"]
+    properties = schema["properties"]
+
+    assert schema["additionalProperties"] is False
+    for field in ("webhook", "stream_resumable", "after_seconds", "feedback_keys"):
+        assert properties[field]["type"] == "null"
+    assert properties["on_completion"]["type"] == "null"
+    assert properties["if_not_exists"]["const"] == "create"
+    assert properties["multitask_strategy"]["enum"] == ["reject", "rollback", "interrupt"]

frontend/AGENTS.md

@@ -89,6 +89,7 @@ Tool-calling AI messages can contain user-visible text as well as `tool_calls`.
 - **Thread hooks** (`useThreadStream`, `useSubmitThread`, `useThreads`) are the primary API interface
 - **Thread routes** — construct Web UI chat paths through `core/threads/utils.ts::pathOfThread()`, which percent-encodes both custom agent names and thread IDs before inserting them into route segments
 - **LangGraph client** is a singleton obtained via `getAPIClient()` in `core/api/`
+- **Run stream options** are sanitized by `core/api/stream-mode.ts`: the Gateway-supported set is `values`, `messages-tuple`, `updates`, `debug`, `tasks`, `checkpoints`, and `custom`; any request containing an unsupported mode throws before HTTP instead of being partially forwarded or silently defaulting to `values`. `streamResumable` is retained by thread hooks only for SDK-side reconnect bookkeeping but stripped before the HTTP request because the Gateway does not implement resumable SSE. Keep this boundary aligned with the backend request schema; `messages` and `events` are not supported and must not be forwarded.
 - **Streaming Markdown rendering** is owned by `core/streamdown`: Streamdown's `animated` / `isAnimating` API handles incremental word animation, while the shared `streamdownRenderingPlugins` config registers the named code-highlighting and Mermaid plugins required by Streamdown 2.5. Keep wrappers and derived configs wired to that shared object; do not reintroduce a rehype plugin that wraps every word, because reparsing a growing block remounts old words and replays their animation.
 - **Environment validation** uses `@t3-oss/env-nextjs` with Zod schemas (`src/env.js`). Skip with `SKIP_ENV_VALIDATION=1`
 - **Subtask step history and runtime metadata** (`core/tasks/`) — the subtask card shows a subagent's full step timeline (#3779): its assistant reasoning turns interleaved with the tools it ran. `Subtask.steps[]` is accumulated live from `task_running` events (appended via `mergeSteps`, not overwritten) and backfilled on expand for historical runs by `fetchSubtaskSteps`, which pages the events endpoint scoped to one task (GET `/runs/{runId}/events?event_types=subagent.step&task_id=…&after_seq=…`) until a short page, so the run-wide limit can't truncate the timeline. `task_started` carries the effective `model_name`; `task_running` carries a cumulative usage snapshot after each completed LLM call. `core/tasks/lifecycle.ts` normalizes these additive events, and `computeNextSubtask` keeps the largest cumulative total so replayed or late SSE frames cannot double-count or roll the folded card backward. Terminal ToolMessage metadata (`subagent_model_name` / `subagent_token_usage`) restores the same values from normal history after reload; no per-card event fetch is needed. `core/tasks/steps.ts` is the pure step model: `messageToStep` (live), `eventsToSteps` (reload), `mergeSteps` (dedup by `message_index`), and `stepsForDisplay` (what the card renders — keeps tool steps + AI steps with text, drops the trailing final-answer AI step when completed since it's shown as `result`). `core/tasks/context.tsx`'s `useUpdateSubtask` applies updates against a `tasksRef` mirroring the latest state (not a closure snapshot), so a late-resolving `fetchSubtaskSteps` backfill merges into current state instead of clobbering SSE steps or sibling subtasks that arrived meanwhile. The owning `run_id` is carried onto history content messages in `buildVisibleHistoryMessages` so the card can resolve the events endpoint.

frontend/src/app/workspace/agents/new/page.tsx

@@ -46,7 +46,7 @@ import {
 } from "@/core/messages/human-input";
 import { isHiddenFromUIMessage } from "@/core/messages/utils";
 import { safeLocalStorage } from "@/core/settings/local";
-import { useThreadStream } from "@/core/threads/hooks";
+import { hasToolResult, useThreadStream } from "@/core/threads/hooks";
 import { uuid } from "@/core/utils/uuid";
 import { isIMEComposing } from "@/lib/ime";
 import { cn } from "@/lib/utils";
@@ -100,13 +100,14 @@ export default function NewAgentPage() {
       mode: "flash",
       is_bootstrap: true,
     },
-    onFinish() {
-      if (!agent && setupAgentStatus === "requested") {
-        setSetupAgentStatus("idle");
+    onFinish(state) {
+      if (agent || setupAgentStatus !== "requested") {
+        return;
+      }
+      if (!agentName || !hasToolResult(state.messages, "setup_agent")) {
+        setSetupAgentStatus("idle");
+        return;
       }
-    },
-    onToolEnd({ name }) {
-      if (name !== "setup_agent" || !agentName) return;
       setSetupAgentStatus("completed");
       void getAgentWithRetry(agentName).then((fetched) => {
         if (fetched) {

frontend/src/core/api/stream-mode.ts

@@ -1,9 +1,7 @@
 const SUPPORTED_RUN_STREAM_MODES = new Set([
   "values",
-  "messages",
   "messages-tuple",
   "updates",
-  "events",
   "debug",
   "tasks",
   "checkpoints",
@@ -11,6 +9,7 @@ const SUPPORTED_RUN_STREAM_MODES = new Set([
 ] as const);
 
 const warnedUnsupportedStreamModes = new Set<string>();
+let warnedUnsupportedStreamResumable = false;
 
 export function warnUnsupportedStreamModes(
   modes: string[],
@@ -29,40 +28,48 @@ export function warnUnsupportedStreamModes(
   }
 
   warn(
-    `[deer-flow] Dropped unsupported LangGraph stream mode(s): ${unseenModes.join(", ")}`,
+    `[deer-flow] Rejected unsupported LangGraph stream mode(s): ${unseenModes.join(", ")}`,
   );
 }
 
 export function sanitizeRunStreamOptions<T>(options: T): T {
-  if (
-    typeof options !== "object" ||
-    options === null ||
-    !("streamMode" in options)
-  ) {
+  if (typeof options !== "object" || options === null) {
     return options;
   }
 
+  let sanitizedOptions: T = options;
+  if ("streamResumable" in options) {
+    const withoutStreamResumable = { ...options };
+    delete withoutStreamResumable.streamResumable;
+    sanitizedOptions = withoutStreamResumable as T;
+
+    if (!warnedUnsupportedStreamResumable) {
+      warnedUnsupportedStreamResumable = true;
+      console.warn(
+        "[deer-flow] Dropped unsupported LangGraph run option: streamResumable",
+      );
+    }
+  }
+
+  if (!("streamMode" in options)) {
+    return sanitizedOptions;
+  }
+
   const streamMode = options.streamMode;
   if (streamMode == null) {
-    return options;
+    return sanitizedOptions;
   }
 
   const requestedModes = Array.isArray(streamMode) ? streamMode : [streamMode];
-  const sanitizedModes = requestedModes.filter((mode) =>
-    SUPPORTED_RUN_STREAM_MODES.has(mode),
-  );
-
-  if (sanitizedModes.length === requestedModes.length) {
-    return options;
-  }
-
   const droppedModes = requestedModes.filter(
     (mode) => !SUPPORTED_RUN_STREAM_MODES.has(mode),
   );
-  warnUnsupportedStreamModes(droppedModes);
+  if (droppedModes.length > 0) {
+    warnUnsupportedStreamModes(droppedModes);
+    throw new Error(
+      `[deer-flow] Unsupported LangGraph stream mode(s): ${droppedModes.join(", ")}`,
+    );
+  }
 
-  return {
-    ...options,
-    streamMode: Array.isArray(streamMode) ? sanitizedModes : sanitizedModes[0],
-  };
+  return sanitizedOptions;
 }

frontend/src/core/threads/hooks.ts

@@ -44,11 +44,6 @@ import type {
   ThreadTokenUsageResponse,
 } from "./types";
 
-export type ToolEndEvent = {
-  name: string;
-  data: unknown;
-};
-
 export type ThreadStreamOptions = {
   threadId?: string | null | undefined;
   displayThreadId?: string | null | undefined;
@@ -57,7 +52,6 @@ export type ThreadStreamOptions = {
   onSend?: (threadId: string) => void;
   onStart?: (threadId: string, runId: string) => void;
   onFinish?: (state: AgentThreadState) => void;
-  onToolEnd?: (event: ToolEndEvent) => void;
 };
 
 type SendMessageOptions = {
@@ -96,6 +90,27 @@ type RegeneratePrepareResponse = {
   target_run_id: string;
 };
 
+export function hasToolResult(messages: Message[], toolName: string): boolean {
+  const matchingToolCallIds = new Set<string>();
+  for (const message of messages) {
+    if (message.type !== "ai") {
+      continue;
+    }
+    for (const toolCall of message.tool_calls ?? []) {
+      if (toolCall.name === toolName && toolCall.id) {
+        matchingToolCallIds.add(toolCall.id);
+      }
+    }
+  }
+
+  return messages.some(
+    (message) =>
+      message.type === "tool" &&
+      (message.name === toolName ||
+        matchingToolCallIds.has(message.tool_call_id)),
+  );
+}
+
 export function buildThreadSubmitMessages({
   text,
   additionalKwargs,
@@ -959,7 +974,6 @@ export function useThreadStream({
   onSend,
   onStart,
   onFinish,
-  onToolEnd,
 }: ThreadStreamOptions) {
   const { t } = useI18n();
   const currentViewThreadId = displayThreadId ?? threadId ?? null;
@@ -990,7 +1004,6 @@ export function useThreadStream({
     onSend,
     onStart,
     onFinish,
-    onToolEnd,
   });
 
   const {
@@ -1005,8 +1018,8 @@ export function useThreadStream({
 
   // Keep listeners ref updated with latest callbacks
   useEffect(() => {
-    listeners.current = { onSend, onStart, onFinish, onToolEnd };
-  }, [onSend, onStart, onFinish, onToolEnd]);
+    listeners.current = { onSend, onStart, onFinish };
+  }, [onSend, onStart, onFinish]);
 
   useEffect(() => {
     const normalizedThreadId = threadId ?? null;
@@ -1097,14 +1110,6 @@ export function useThreadStream({
           .catch(() => ({}));
       }
     },
-    onLangChainEvent(event) {
-      if (event.event === "on_tool_end") {
-        listeners.current.onToolEnd?.({
-          name: event.name,
-          data: event.data,
-        });
-      }
-    },
     onUpdateEvent(data) {
       const _messages = getSummarizationMiddlewareMessages(data);
       if (_messages && _messages.length >= 2) {

frontend/tests/unit/core/api/stream-mode.test.ts

@@ -2,33 +2,34 @@ import { expect, test } from "@rstest/core";
 
 import { sanitizeRunStreamOptions } from "@/core/api/stream-mode";
 
-test("drops unsupported stream modes from array payloads", () => {
-  const sanitized = sanitizeRunStreamOptions({
-    streamMode: [
-      "values",
-      "messages-tuple",
-      "custom",
-      "updates",
-      "events",
-      "tools",
-    ],
-  });
-
-  expect(sanitized.streamMode).toEqual([
-    "values",
-    "messages-tuple",
-    "custom",
-    "updates",
-    "events",
-  ]);
+test("rejects mixed supported and unsupported stream modes", () => {
+  expect(() =>
+    sanitizeRunStreamOptions({
+      streamMode: ["values", "events", "tools"],
+    }),
+  ).toThrow("Unsupported LangGraph stream mode(s): events, tools");
 });
 
-test("drops unsupported stream modes from scalar payloads", () => {
-  const sanitized = sanitizeRunStreamOptions({
-    streamMode: "tools",
-  });
+test("rejects payloads when every requested stream mode is unsupported", () => {
+  expect(() =>
+    sanitizeRunStreamOptions({
+      streamMode: ["events", "tools"],
+    }),
+  ).toThrow("Unsupported LangGraph stream mode(s): events, tools");
 
-  expect(sanitized.streamMode).toBeUndefined();
+  expect(() =>
+    sanitizeRunStreamOptions({
+      streamMode: "tools",
+    }),
+  ).toThrow("Unsupported LangGraph stream mode(s): tools");
+});
+
+test("rejects messages because the Gateway only supports messages-tuple framing", () => {
+  expect(() =>
+    sanitizeRunStreamOptions({
+      streamMode: "messages",
+    }),
+  ).toThrow("Unsupported LangGraph stream mode(s): messages");
 });
 
 test("keeps payloads without streamMode untouched", () => {
@@ -38,3 +39,25 @@ test("keeps payloads without streamMode untouched", () => {
 
   expect(sanitizeRunStreamOptions(options)).toBe(options);
 });
+
+test("strips streamResumable before sending run options to the API", () => {
+  const sanitized = sanitizeRunStreamOptions({
+    streamResumable: true,
+    streamSubgraphs: true,
+  });
+
+  expect(sanitized).toEqual({
+    streamSubgraphs: true,
+  });
+});
+
+test("sanitizes streamResumable while preserving valid stream modes", () => {
+  const sanitized = sanitizeRunStreamOptions({
+    streamResumable: true,
+    streamMode: ["values", "custom"],
+  });
+
+  expect(sanitized).toEqual({
+    streamMode: ["values", "custom"],
+  });
+});

frontend/tests/unit/core/threads/stream-options.test.ts

@@ -0,0 +1,95 @@
+import { afterEach, expect, test, rs } from "@rstest/core";
+
+async function captureThreadStreamOptions() {
+  let capturedOptions: Record<string, unknown> | undefined;
+
+  rs.resetModules();
+  rs.doMock("react", () => ({
+    useCallback: <T extends (...args: never[]) => unknown>(callback: T) =>
+      callback,
+    useEffect: () => undefined,
+    useMemo: <T>(factory: () => T) => factory(),
+    useRef: <T>(initialValue: T) => ({ current: initialValue }),
+    useState: <T>(initialValue: T | (() => T)) => [
+      typeof initialValue === "function"
+        ? (initialValue as () => T)()
+        : initialValue,
+      rs.fn(),
+    ],
+  }));
+  rs.doMock("@tanstack/react-query", () => ({
+    useInfiniteQuery: () => ({
+      data: { pages: [] },
+      error: null,
+      fetchNextPage: rs.fn(),
+      hasNextPage: false,
+      isFetchingNextPage: false,
+      isLoading: false,
+    }),
+    useMutation: rs.fn(),
+    useQuery: rs.fn(),
+    useQueryClient: () => ({
+      invalidateQueries: rs.fn(),
+      setQueriesData: rs.fn(),
+    }),
+  }));
+  rs.doMock("@langchain/langgraph-sdk/react", () => ({
+    useStream: (options: Record<string, unknown>) => {
+      capturedOptions = options;
+      return {
+        isLoading: false,
+        messages: [],
+        stop: rs.fn(),
+        submit: rs.fn(),
+        values: { title: "", messages: [] },
+      };
+    },
+  }));
+  rs.doMock("@/core/api", () => ({
+    getAPIClient: () => ({}),
+  }));
+  rs.doMock("@/core/i18n/hooks", () => ({
+    useI18n: () => ({
+      t: {
+        pages: { newChat: "New chat" },
+        uploads: { uploadingFiles: "Uploading files" },
+      },
+    }),
+  }));
+  rs.doMock("@/core/tasks/context", () => ({
+    useUpdateSubtask: () => rs.fn(),
+  }));
+
+  const { useThreadStream } = await import("@/core/threads/hooks");
+  function ThreadStreamCapture() {
+    useThreadStream({
+      context: {
+        mode: "flash",
+      },
+      isMock: true,
+    } as never);
+    return null;
+  }
+  ThreadStreamCapture();
+
+  return capturedOptions;
+}
+
+afterEach(() => {
+  rs.doUnmock("react");
+  rs.doUnmock("@tanstack/react-query");
+  rs.doUnmock("@langchain/langgraph-sdk/react");
+  rs.doUnmock("@/core/api");
+  rs.doUnmock("@/core/i18n/hooks");
+  rs.doUnmock("@/core/tasks/context");
+  rs.resetModules();
+});
+
+test("does not subscribe to unsupported LangGraph events mode", async () => {
+  const options = await captureThreadStreamOptions();
+
+  expect(options).toBeDefined();
+  expect(options).not.toHaveProperty("onLangChainEvent");
+  expect(options).toHaveProperty("onUpdateEvent");
+  expect(options).toHaveProperty("onCustomEvent");
+});

frontend/tests/unit/core/threads/tool-result.test.ts

@@ -0,0 +1,60 @@
+import type { Message } from "@langchain/langgraph-sdk";
+import { expect, test } from "@rstest/core";
+
+import { hasToolResult } from "@/core/threads/hooks";
+
+test("recognizes a completed tool from its ToolMessage", () => {
+  const messages = [
+    {
+      type: "ai",
+      content: "",
+      tool_calls: [{ id: "call-1", name: "setup_agent", args: {} }],
+    },
+    {
+      type: "tool",
+      content: "Agent saved",
+      name: "setup_agent",
+      tool_call_id: "call-1",
+    },
+  ] as Message[];
+
+  expect(hasToolResult(messages, "setup_agent")).toBe(true);
+});
+
+test("does not treat a pending call or another tool result as completed", () => {
+  const pending = [
+    {
+      type: "ai",
+      content: "",
+      tool_calls: [{ id: "call-1", name: "setup_agent", args: {} }],
+    },
+  ] as Message[];
+  const otherTool = [
+    {
+      type: "tool",
+      content: "Done",
+      name: "web_search",
+      tool_call_id: "call-2",
+    },
+  ] as Message[];
+
+  expect(hasToolResult(pending, "setup_agent")).toBe(false);
+  expect(hasToolResult(otherTool, "setup_agent")).toBe(false);
+});
+
+test("matches a ToolMessage without a name through its tool call id", () => {
+  const messages = [
+    {
+      type: "ai",
+      content: "",
+      tool_calls: [{ id: "call-1", name: "setup_agent", args: {} }],
+    },
+    {
+      type: "tool",
+      content: "Agent saved",
+      tool_call_id: "call-1",
+    },
+  ] as Message[];
+
+  expect(hasToolResult(messages, "setup_agent")).toBe(true);
+});