From a4ede80deb1ea1f73cc2f6e4a465eddab0559bc4 Mon Sep 17 00:00:00 2001 From: ShitK <80999801+ShitK@users.noreply.github.com> Date: Fri, 24 Jul 2026 19:24:24 +0800 Subject: [PATCH] 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 --- README.md | 2 + backend/AGENTS.md | 1 + backend/app/gateway/routers/runs.py | 2 +- backend/app/gateway/routers/thread_runs.py | 24 +- backend/app/gateway/run_models.py | 92 ++++++++ backend/app/gateway/services.py | 31 +-- backend/docs/API.md | 10 +- .../harness/deerflow/runtime/runs/worker.py | 40 +--- .../harness/deerflow/runtime/stream_modes.py | 47 ++++ backend/tests/test_gateway_services.py | 77 +++++++ backend/tests/test_run_request_validation.py | 210 ++++++++++++++++++ frontend/AGENTS.md | 1 + .../src/app/workspace/agents/new/page.tsx | 15 +- frontend/src/core/api/stream-mode.ts | 51 +++-- frontend/src/core/threads/hooks.ts | 41 ++-- .../tests/unit/core/api/stream-mode.test.ts | 71 ++++-- .../unit/core/threads/stream-options.test.ts | 95 ++++++++ .../unit/core/threads/tool-result.test.ts | 60 +++++ 18 files changed, 715 insertions(+), 155 deletions(-) create mode 100644 backend/app/gateway/run_models.py create mode 100644 backend/packages/harness/deerflow/runtime/stream_modes.py create mode 100644 backend/tests/test_run_request_validation.py create mode 100644 frontend/tests/unit/core/threads/stream-options.test.ts create mode 100644 frontend/tests/unit/core/threads/tool-result.test.ts diff --git a/README.md b/README.md index afec2c056..0a0f5d542 100644 --- a/README.md +++ b/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 diff --git a/backend/AGENTS.md b/backend/AGENTS.md index ff2cd14c7..990b2dfc2 100644 --- a/backend/AGENTS.md +++ b/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. diff --git a/backend/app/gateway/routers/runs.py b/backend/app/gateway/routers/runs.py index 23195d301..7ea06ca85 100644 --- a/backend/app/gateway/routers/runs.py +++ b/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 diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index c3acfee8a..b0b0728b1 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/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") diff --git a/backend/app/gateway/run_models.py b/backend/app/gateway/run_models.py new file mode 100644 index 000000000..cfb8d1a7e --- /dev/null +++ b/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 diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 47f0f11a1..44ef5e8f2 100644 --- a/backend/app/gateway/services.py +++ b/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) diff --git a/backend/docs/API.md b/backend/docs/API.md index 7e9677b90..329b32e78 100644 --- a/backend/docs/API.md +++ b/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:** diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index 9712ce8e0..731d68972 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/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 diff --git a/backend/packages/harness/deerflow/runtime/stream_modes.py b/backend/packages/harness/deerflow/runtime/stream_modes.py new file mode 100644 index 000000000..96c68f9d8 --- /dev/null +++ b/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)) diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index 4d89cfebd..e237ce9e0 100644 --- a/backend/tests/test_gateway_services.py +++ b/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 diff --git a/backend/tests/test_run_request_validation.py b/backend/tests/test_run_request_validation.py new file mode 100644 index 000000000..d428e99dc --- /dev/null +++ b/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"] diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 0720af356..7db3826a3 100644 --- a/frontend/AGENTS.md +++ b/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. diff --git a/frontend/src/app/workspace/agents/new/page.tsx b/frontend/src/app/workspace/agents/new/page.tsx index e006cf88a..bbbb0fa34 100644 --- a/frontend/src/app/workspace/agents/new/page.tsx +++ b/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) { diff --git a/frontend/src/core/api/stream-mode.ts b/frontend/src/core/api/stream-mode.ts index 7acae8642..8839176a5 100644 --- a/frontend/src/core/api/stream-mode.ts +++ b/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(); +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(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; } diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index ba2dc7291..efcae9588 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/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(); + 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) { diff --git a/frontend/tests/unit/core/api/stream-mode.test.ts b/frontend/tests/unit/core/api/stream-mode.test.ts index d94980549..bccee1c39 100644 --- a/frontend/tests/unit/core/api/stream-mode.test.ts +++ b/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"], + }); +}); diff --git a/frontend/tests/unit/core/threads/stream-options.test.ts b/frontend/tests/unit/core/threads/stream-options.test.ts new file mode 100644 index 000000000..da9f8adfa --- /dev/null +++ b/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 | undefined; + + rs.resetModules(); + rs.doMock("react", () => ({ + useCallback: unknown>(callback: T) => + callback, + useEffect: () => undefined, + useMemo: (factory: () => T) => factory(), + useRef: (initialValue: T) => ({ current: initialValue }), + useState: (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) => { + 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"); +}); diff --git a/frontend/tests/unit/core/threads/tool-result.test.ts b/frontend/tests/unit/core/threads/tool-result.test.ts new file mode 100644 index 000000000..b4bc855e8 --- /dev/null +++ b/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); +});