fix(agent): return operation input contracts on errors

app/agent/api/arguments.py

@@ -14,6 +14,7 @@ _SCALAR_ADAPTERS = {
     "string": TypeAdapter(str),
 }
 _MISSING = object()
+_CONTRACT_DESCRIPTION_MAX_CHARS = 180
 
 
 def _resolve_schema(schema: dict[str, Any], definitions: dict[str, Any]) -> dict[str, Any]:
@@ -32,6 +33,105 @@ def _resolve_schema(schema: dict[str, Any], definitions: dict[str, Any]) -> dict
     return schema
 
 
+def _contract_description(value: Any) -> str | None:
+    """为错误回执保留有界字段说明，避免把原始请求值带回模型。"""
+    if not isinstance(value, str) or not value.strip():
+        return None
+    text = " ".join(value.split())
+    if len(text) <= _CONTRACT_DESCRIPTION_MAX_CHARS:
+        return text
+    return text[: _CONTRACT_DESCRIPTION_MAX_CHARS - 1].rstrip() + "…"
+
+
+def _contract_node(
+    schema: dict[str, Any],
+    definitions: dict[str, Any],
+    *,
+    depth: int = 0,
+) -> dict[str, Any]:
+    """把单个 JSON Schema 节点压缩成可直接纠错的输入提示。"""
+    try:
+        resolved = _resolve_schema(schema, definitions)
+    except ValueError:
+        return {"type": "unknown"}
+
+    result: dict[str, Any] = {}
+    description = _contract_description(resolved.get("description"))
+    if description:
+        result["description"] = description
+    if "const" in resolved:
+        result["const"] = resolved["const"]
+    if isinstance(resolved.get("enum"), list):
+        result["enum"] = deepcopy(resolved["enum"])
+    if "default" in resolved:
+        result["default"] = deepcopy(resolved["default"])
+    if isinstance(resolved.get("type"), str):
+        result["type"] = resolved["type"]
+    for key in ("minimum", "maximum", "minItems", "maxItems", "minLength", "maxLength"):
+        if key in resolved:
+            result[key] = resolved[key]
+
+    alternatives = resolved.get("oneOf", resolved.get("anyOf"))
+    if isinstance(alternatives, list):
+        result["one_of"] = [
+            _contract_node(item, definitions, depth=depth + 1)
+            for item in alternatives
+            if isinstance(item, dict)
+        ]
+
+    properties = resolved.get("properties")
+    if isinstance(properties, dict):
+        required = resolved.get("required")
+        if isinstance(required, list) and required:
+            result["required"] = [item for item in required if isinstance(item, str)]
+        if depth < 4:
+            result["fields"] = {
+                name: _contract_node(value, definitions, depth=depth + 1)
+                for name, value in properties.items()
+                if isinstance(name, str) and isinstance(value, dict)
+            }
+        else:
+            result["fields"] = sorted(name for name in properties if isinstance(name, str))
+        if resolved.get("additionalProperties") is False:
+            result["additional_properties"] = False
+    elif isinstance(resolved.get("items"), dict) and depth < 4:
+        result["items"] = _contract_node(resolved["items"], definitions, depth=depth + 1)
+    return result
+
+
+def api_input_contract(operation_id: str, schema: dict[str, Any]) -> dict[str, Any]:
+    """返回一个 operation 的有界输入合同，供调用失败时指导模型纠正参数。"""
+    branch = next(
+        (
+            item
+            for item in schema.get("oneOf", [])
+            if isinstance(item, dict)
+            and item.get("properties", {}).get("operation_id", {}).get("const") == operation_id
+        ),
+        None,
+    )
+    if not isinstance(branch, dict):
+        return {"operation_id": operation_id, "available": False}
+    properties = branch.get("properties", {})
+    contract: dict[str, Any] = {
+        "operation_id": operation_id,
+        "allowed_arguments": [
+            name for name in properties if isinstance(name, str) and name != "operation_id"
+        ],
+    }
+    required = branch.get("required")
+    if isinstance(required, list):
+        contract["required_arguments"] = [
+            name for name in required if isinstance(name, str) and name != "operation_id"
+        ]
+    definitions = schema.get("$defs", {})
+    for name in ("path_params", "query", "body"):
+        node = properties.get(name)
+        if isinstance(node, dict):
+            contract[name] = _contract_node(node, definitions)
+    return contract
+
+
 def _value_type(value: Any) -> str:
     """按 JSON 原始类型选择联合分支，布尔值不能误当成整数。"""
     return {

app/agent/middleware/invocation.py

@@ -223,14 +223,18 @@ class InvocationMiddleware(AgentMiddleware):  # type: ignore[misc]
             )
         except InvocationConflictError:
             return self._message(request, "failed", "同一调用 ID 的工具或参数发生变化，未执行。")
-        except ValueError as error:
+        except (TypeError, ValueError) as error:
             detail = summarize_error(error, max_chars=240)
             logger.info(f"工具参数未通过 API 输入合同校验: {detail}")
+            operation_id = str(request.tool_call.get("args", {}).get("operation_id", ""))
+            input_contract = request.tool.get_operation_input_contract(operation_id)
             return self._message(
                 request,
                 "failed",
-                "API 工具参数未通过当前 operation 的输入合同校验。"
-                f"校验信息：{detail}。请重新读取对应 Skill 支持文档，只提交已声明的字段后再调用。",
+                f"{operation_id} 的输入未通过当前 operation 输入合同校验：{detail}。"
+                "请按 input_contract 只提交允许字段并补齐 required 字段后再调用。",
+                operation_id=operation_id,
+                input_contract=input_contract,
             )
         except Exception as error:
             logger.warning(f"工具执行认领失败: {type(error).__name__}")

app/agent/tools/impl/api.py

@@ -8,7 +8,7 @@ from typing import Any, Dict, Optional, Type
 
 from pydantic import BaseModel, Field, PrivateAttr
 
-from app.agent.api.arguments import canonical_api_arguments
+from app.agent.api.arguments import api_input_contract, canonical_api_arguments
 from app.agent.api.executor import ApiExecutionContext, ApiExecutionError, MoviePilotApiExecutor
 from app.agent.policy.api import resolve_api_operation
 from app.agent.policy.contracts import ExecutionOutcome, PrincipalRole
@@ -82,12 +82,9 @@ class MoviePilotApiTool(MoviePilotTool):
         ToolTag.Plugin,
     ]
     description: str = (
-        "Call allowlisted MoviePilot business APIs, including versioned automatic media classification. "
-        "Use the domain Skill to select operation_id, "
-        "parameters, and failure handling. For collection counts, use the smallest documented "
-        "page and read collection.total_count instead of querying the database after item "
-        "truncation. External MCP tools/list exposes one complete oneOf branch per operation. "
-        "Arbitrary URLs, commands, and authentication endpoints are forbidden."
+        "Call allowlisted MoviePilot business APIs through operation-specific input contracts. "
+        "Load the relevant domain Skill before calling and use operation error feedback to correct inputs. "
+        "Arbitrary URLs, commands, authentication endpoints, headers, and tokens are forbidden."
     )
     require_admin: bool = False
     args_schema: Type[BaseModel] = MoviePilotApiInput
@@ -136,6 +133,27 @@ class MoviePilotApiTool(MoviePilotTool):
         validated = MoviePilotApiInput.model_validate(arguments).model_dump(mode="json")
         return canonical_api_arguments(validated, _load_api_mcp_input_schema())
 
+    def get_operation_input_contract(self, operation_id: str) -> dict[str, Any]:
+        """返回单个 operation 的有界参数合同，供失败回执指导模型重试。"""
+        return api_input_contract(operation_id, _load_api_mcp_input_schema())
+
+    def _invalid_input_result(self, operation_id: str, error: Exception) -> str:
+        """把参数合同错误转换为不包含原始请求值的模型纠错回执。"""
+        contract = self.get_operation_input_contract(operation_id)
+        return json.dumps(
+            {
+                "success": False,
+                "error": "invalid_input",
+                "operation_id": operation_id,
+                "message": (
+                    f"{operation_id} 的输入不符合已声明合同（{str(error)}）。"
+                    "请按 input_contract 只提交允许字段并补齐 required 字段后重试。"
+                ),
+                "input_contract": contract,
+            },
+            ensure_ascii=False,
+        )
+
     async def _resolve_superuser_integration_identity(
         self,
     ) -> tuple[str, Optional[str], bool]:
@@ -265,6 +283,17 @@ class MoviePilotApiTool(MoviePilotTool):
                 },
                 ensure_ascii=False,
             )
+        try:
+            arguments: dict[str, Any] = {"operation_id": operation_id}
+            if path_params is not None:
+                arguments["path_params"] = path_params
+            if query is not None:
+                arguments["query"] = query
+            if body is not None:
+                arguments["body"] = body
+            self.canonical_arguments(arguments)
+        except (TypeError, ValueError) as error:
+            return self._invalid_input_result(operation_id, error)
         try:
             requires_system_admin = spec.required_role is PrincipalRole.SYSTEM_ADMIN
             executor, is_admin = await self._get_executor(
@@ -285,6 +314,21 @@ class MoviePilotApiTool(MoviePilotTool):
                 query=query,
                 body=body,
             )
+            try:
+                payload = json.loads(result)
+            except (TypeError, ValueError):
+                payload = None
+            if (
+                isinstance(payload, dict)
+                and payload.get("error") == "api_error"
+                and payload.get("status_code") in {400, 422}
+            ):
+                payload["operation_id"] = operation_id
+                payload["message"] = (
+                    f"{operation_id} 被 API 拒绝，请根据 input_contract 修正输入后重试。"
+                )
+                payload["input_contract"] = self.get_operation_input_contract(operation_id)
+                result = json.dumps(payload, ensure_ascii=False)
             if operation_id == "scheduler.run" and inspect_tool_result(result) is ExecutionOutcome.SUCCEEDED:
                 payload = json.loads(result)
                 if isinstance(payload, dict) and payload.get("success") is True:

docs/refactor/agent-codex-parity-plan.md

@@ -48,7 +48,7 @@ uv run --locked --no-sync python -m scripts.evaluation \
 - **命令行**：普通一次性命令与后台终端分别支持 pipe/PTY、共享 cwd/shell/login/环境和 UTF-8；输入写入、空写入、EOF、分页、短写、interrupt、kill、超时、取消和进程组收尾都有结构化终态。
 - **浏览器**：工具目录必须明确导航、读取、交互、等待和截图的动作与权限；浏览器会话、页面状态和失败重试由宿主持有，不能由模型字符串冒领。
 - **子代理**：主 Agent 自动选择通用子代理；专用画像只有在 held-out 任务上证明提高成功率、减少调用或降低副作用风险时才保留。子代理不能自行发送消息、执行高影响写操作或继承兄弟任务句柄。
-- **API Skill 与合同读取**：`skills/moviepilot-api/SKILL.md` 只负责路由索引和工作流；`api/*.md` 保留按类别拆分的完整合同，`api/models.md` 保留共享模型。真实 Agent 先用 `read_skill` 得到 supporting-file 清单，再用受权限根约束的 `read_file` 按需读取类别文档；评测目录必须提供同一条链路，不能只返回文件名而不给模型读取能力。
+- **API Skill 与合同读取**：`skills/moviepilot-api/SKILL.md` 只负责路由索引和工作流；`api/*.md` 各自包含完整操作合同及该类别需要的 Body Models，不再依赖单独的 `api/models.md`。真实 Agent 先用 `read_skill` 得到 supporting-file 清单，再用同一个工具的 `file=api/<category>.md` 参数按需读取类别文档；评测目录必须提供同一条链路，不能只返回文件名而不给模型读取能力。operation 输入错误还要返回该 operation 的允许字段、必填字段和类型约束，帮助模型纠正后重试。
 - **长任务与新消息**：入站消息在任务运行时仍可接受并进入有界队列；应用到下一模型边界时必须保留 tool-call/tool-result 配对和取消语义，不能只把文本拼到系统提示词。
 - **观察证据**：最终答案只能引用宿主记录的工具结果和业务 oracle；“模型说完成”不能替代数据库、下载器、浏览器或进程状态核验。
 

scripts/evaluation/server.py

@@ -70,6 +70,56 @@ class _Session:
     initialized: bool = False
 
 
+def _operation_input_contract(schema: dict[str, Any], operation_id: str) -> dict[str, Any]:
+    """压缩一个 operation 的参数字段，保持评测服务不依赖 MoviePilot 运行时。"""
+    branch = next(
+        (
+            item for item in schema.get("oneOf", [])
+            if isinstance(item, dict)
+            and item.get("properties", {}).get("operation_id", {}).get("const") == operation_id
+        ),
+        None,
+    )
+    if not isinstance(branch, dict):
+        return {"operation_id": operation_id, "available": False}
+    properties = branch.get("properties", {})
+    contract: dict[str, Any] = {
+        "operation_id": operation_id,
+        "allowed_arguments": [name for name in properties if name != "operation_id"],
+        "required_arguments": [
+            name for name in branch.get("required", []) if name != "operation_id"
+        ],
+    }
+    for name in ("path_params", "query", "body"):
+        node = properties.get(name)
+        if not isinstance(node, dict):
+            continue
+        shape: dict[str, Any] = {}
+        if isinstance(node.get("type"), str):
+            shape["type"] = node["type"]
+        if isinstance(node.get("required"), list) and node["required"]:
+            shape["required"] = list(node["required"])
+        fields = node.get("properties")
+        if isinstance(fields, dict):
+            shape["fields"] = {}
+            for field, declaration in fields.items():
+                if not isinstance(declaration, dict):
+                    continue
+                field_shape: dict[str, Any] = {}
+                for key in ("type", "enum", "const", "default", "$ref"):
+                    if key in declaration:
+                        field_shape[key] = deepcopy(declaration[key])
+                alternatives = declaration.get("anyOf", declaration.get("oneOf"))
+                if isinstance(alternatives, list):
+                    field_shape["one_of"] = [
+                        {key: deepcopy(option[key]) for key in ("type", "enum", "const") if key in option}
+                        for option in alternatives if isinstance(option, dict)
+                    ]
+                shape["fields"][field] = field_shape
+        contract[name] = shape
+    return contract
+
+
 class _LocalServer(uvicorn.Server):  # type: ignore[misc]  # follow_imports=skip 不展开外部 Server 基类。
     """让控制器拥有服务启停，嵌入运行时不替换进程信号处理器。"""
 
@@ -113,6 +163,7 @@ class EvaluationMcpServer:
         self._results: OrderedDict[str, _StoredResult] = OrderedDict()
         self._skill_root = Path(root_dir) if root_dir is not None else Path(__file__).resolve().parents[2]
         self._skill = self._load_skill(self._skill_root)
+        self._api_contract_schema = self._load_api_contract(self._skill_root)
 
     @property
     def endpoint(self) -> str:
@@ -325,9 +376,9 @@ class EvaluationMcpServer:
         """只公开与 MoviePilot 任务输入等价的工具合同，不注入场景答案。"""
         return [
             {"name": "moviepilot_api", "description": (
-                "Call allowlisted MoviePilot business APIs. Use the domain Skill to select operation_id, parameters, and failure handling. "
-                "For collection counts, use the smallest documented page and read collection.total_count instead of querying the database after item truncation. "
-                "Arbitrary URLs, commands, and authentication endpoints are forbidden."
+                "Call allowlisted MoviePilot business APIs through operation-specific input contracts. "
+                "Load the relevant domain Skill before calling and use operation error feedback to correct inputs. "
+                "Arbitrary URLs, commands, authentication endpoints, headers, and tokens are forbidden."
             ), "inputSchema": deepcopy(API_INPUT_SCHEMA)},
             {"name": "read_skill", "description": "Read the named MoviePilot Skill or one of its listed supporting documents. Available skill: moviepilot-api.", "inputSchema": {
                 "type": "object", "required": ["name"], "additionalProperties": False, "properties": {
@@ -364,6 +415,7 @@ class EvaluationMcpServer:
                 else:
                     self._world_calls += 1
                     result = self._world.execute(**arguments)
+                result = self._attach_input_contract(result, arguments)
             elif name == "read_skill":
                 result = self._read_skill(arguments)
             elif name == "read_tool_result":
@@ -397,6 +449,31 @@ class EvaluationMcpServer:
         ), "truncated": truncated,
             "truncation_message": "SKILL.md exceeds 512 KiB; content contains only the first 512 KiB." if truncated else None}
 
+    @staticmethod
+    def _load_api_contract(root: Path) -> dict[str, Any]:
+        """读取与生产网关相同的 operation schema，供评测错误回执提供纠正提示。"""
+        path = root / "app/agent/policy/resources/api_mcp_schema.json"
+        payload = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(payload, dict) or not isinstance(payload.get("oneOf"), list):
+            raise ValueError("moviepilot_api operation schema 无效")
+        return payload
+
+    def _attach_input_contract(self, result: Any, arguments: dict[str, Any]) -> Any:
+        """为 operation 失败回执附加允许字段和必填字段，不回显错误请求值。"""
+        if not isinstance(result, dict) or result.get("success") is not False:
+            return result
+        operation_id = arguments.get("operation_id")
+        if type(operation_id) is not str:
+            return result
+        payload = dict(result)
+        payload["operation_id"] = operation_id
+        payload["message"] = (
+            f"{operation_id} 调用失败，请按 input_contract 只提交允许字段并补齐 required 字段后重试。"
+            f" 原因：{str(result.get('message', ''))[:240]}"
+        )
+        payload["input_contract"] = _operation_input_contract(self._api_contract_schema, operation_id)
+        return payload
+
     def _prune_results(self) -> None:
         """匹配 MoviePilot 的 15 分钟、8 项、4MiB 会话归档约束。"""
         now = time.monotonic()

skills/moviepilot-api/api/download.md

@@ -23,8 +23,8 @@ For `download.add`, `torrent_in` is an object whose minimum usable fields are:
 `enclosure` is the direct string magnet or torrent URL; do not wrap it in another
 `url` object. Keep `media_source` and `media_id` as sibling fields beside
 `torrent_in`. Do not use a top-level `url`, `media_info`, or nested `media_info`
-object. Load `api/models.md` when the full optional `TorrentInfo` fields are
-needed.
+object. The complete optional `TorrentInfo` fields are included in this
+category document under **Body Models** when they are needed.
 
 ### `download.artist_collection`
 `POST /api/v1/download/artist-collection`; policy effect: `external_side_effect`.

tests/test_agent_api_gateway.py

@@ -420,18 +420,16 @@ def test_gateway_forwards_structured_arguments_to_api_executor() -> None:
     result = asyncio.run(
         gateway.run(
             operation_id="media.search",
-            path_params={"media_type": "movie"},
-            query={"page": 1},
-            body={"title": "示例"},
+            query={"page": 1, "title": "示例"},
         )
     )
 
     assert json.loads(result)["success"] is True
     executor.execute.assert_awaited_once_with(
         "media.search",
-        path_params={"media_type": "movie"},
-        query={"page": 1},
-        body={"title": "示例"},
+        path_params=None,
+        query={"page": 1, "title": "示例"},
+        body=None,
     )
 
 
@@ -445,6 +443,32 @@ def test_gateway_rejects_unknown_operation() -> None:
     assert "unknown_operation" in result
 
 
+def test_gateway_returns_operation_contract_when_input_is_invalid() -> None:
+    """单个 operation 的参数错误必须返回可直接纠正调用的精确合同。"""
+    executor = AsyncMock()
+    gateway = MoviePilotApiTool(
+        session_id="session",
+        user_id="1",
+        executor=executor,
+    )
+
+    result = asyncio.run(
+        gateway.run(
+            operation_id="site.list",
+            query={"status": "enabled"},
+        )
+    )
+    payload = json.loads(result)
+
+    assert payload["error"] == "invalid_input"
+    assert payload["operation_id"] == "site.list"
+    assert payload["input_contract"]["allowed_arguments"] == ["query"]
+    status = payload["input_contract"]["query"]["fields"]["status"]
+    assert status["enum"] == ["active", "inactive", "all"]
+    assert "enabled" not in json.dumps(payload, ensure_ascii=False)
+    executor.execute.assert_not_awaited()
+
+
 def test_gateway_rejects_admin_operation_for_non_admin_before_http() -> None:
     """管理员 operation 必须在网关层拒绝普通用户，不能只依赖最终端点。"""
     executor = AsyncMock()

tests/test_agent_evaluation_server.py

@@ -152,6 +152,24 @@ async def test_skill_supporting_document_uses_read_skill_without_file_tool() ->
             assert invalid["error"] == "skill_file_not_found"
 
 
+@pytest.mark.asyncio
+async def test_invalid_operation_input_returns_operation_contract() -> None:
+    """仿真工具报错时应把当前 operation 的正确输入合同交给模型。"""
+    async with EvaluationMcpServer(EvaluationWorld("dedup_existing")) as server:
+        async with _client(server) as client:
+            result, payload = await _tool(
+                client,
+                "moviepilot_api",
+                {"operation_id": "site.list", "query": {"status": "enabled"}},
+            )
+            assert result["isError"] is True
+            assert payload["operation_id"] == "site.list"
+            assert payload["input_contract"]["query"]["fields"]["status"]["enum"] == [
+                "active", "inactive", "all",
+            ]
+            assert "enabled" not in json.dumps(payload, ensure_ascii=False)
+
+
 @pytest.mark.asyncio
 async def test_world_call_limit_is_global_across_sessions_and_parallel_requests() -> None:
     """并行连接也共享硬调用上限，超限不会继续读取或改变独立世界。"""

tests/test_agent_invocation_middleware.py

@@ -424,6 +424,10 @@ async def test_api_forbidden_query_keys_do_not_execute_or_claim(invocation_runti
     result = await _invoke(graph)
     assert _tool_messages(result)[0].status == "error"
     assert "输入合同校验" in _tool_messages(result)[0].content
+    payload = json.loads(_tool_messages(result)[0].content)
+    assert payload["operation_id"] == "plugin.install"
+    assert "path_params" in payload["input_contract"]["allowed_arguments"]
+    assert "query" in payload["input_contract"]["allowed_arguments"]
     assert records() == []
     run.assert_not_awaited()
 

tests/test_agent_tool_outcomes.py

@@ -180,6 +180,19 @@ def _api_tool(request: AsyncMock) -> MoviePilotApiTool:
     return tool
 
 
+def _valid_api_arguments(operation: str) -> dict:
+    """为传输层测试提供每个 operation 的最小合法输入。"""
+    return {
+        "download.add": {"body": {"torrent_in": {
+            "title": "Test Movie", "enclosure": "https://example.invalid/test.torrent",
+        }}},
+        "subscription.update": {"body": {"id": 1}},
+        "system.restart": {},
+        "subscription.list": {},
+        "storage.list": {"body": {"path": "/"}},
+    }[operation]
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(("operation", "expected"), [
     ("download.add", ExecutionOutcome.UNKNOWN),
@@ -192,7 +205,10 @@ async def test_api_transport_failure_preserves_unknown_mutations(monkeypatch, op
     """API 传输异常对写操作保留未知结果，包含副作用 GET 及只读 POST 的策略区分。"""
     monkeypatch.setattr("app.agent.api.executor.create_access_token", lambda **_kwargs: "test-token")
     request = AsyncMock(side_effect=httpx2.ReadError("private-api-response-body"))
-    result = await _api_tool(request).run(operation_id=operation)
+    result = await _api_tool(request).run(
+        operation_id=operation,
+        **_valid_api_arguments(operation),
+    )
     assert inspect_tool_result(result) is expected
     assert "private-api-response-body" not in result
     request.assert_awaited_once()
@@ -208,7 +224,10 @@ async def test_api_preflight_validation_and_http_rejection_are_definite_failures
     missing_path = await tool.run(operation_id="subscription.delete")
     assert inspect_tool_result(missing_path) is ExecutionOutcome.FAILED
     request.assert_not_awaited()
-    rejected = await tool.run(operation_id="download.add")
+    rejected = await tool.run(
+        operation_id="download.add",
+        **_valid_api_arguments("download.add"),
+    )
     assert inspect_tool_result(rejected) is ExecutionOutcome.FAILED
     assert json.loads(rejected)["status_code"] == 403
     response.aclose.assert_awaited_once()
@@ -225,7 +244,10 @@ async def test_api_unreadable_response_retains_observed_http_status(monkeypatch,
         raise ValueError("private-response-body")
 
     response = SimpleNamespace(status_code=status_code, headers={}, json=unreadable, aclose=AsyncMock())
-    result = await _api_tool(AsyncMock(return_value=response)).run(operation_id="download.add")
+    result = await _api_tool(AsyncMock(return_value=response)).run(
+        operation_id="download.add",
+        **_valid_api_arguments("download.add"),
+    )
     assert inspect_tool_result(result) is expected
     assert "private-response-body" not in result
     response.aclose.assert_awaited_once()
@@ -283,7 +305,9 @@ async def test_api_submission_annotation_preserves_failure_and_synchronous_workf
     payload = {"success": success, "message": "original", "data": None}
     response = SimpleNamespace(status_code=200, headers={}, json=lambda: payload, aclose=AsyncMock())
     result = await _api_tool(AsyncMock(return_value=response)).run(
-        operation_id=operation, path_params={"workflow_id": 1} if operation == "workflow.run" else {},
+        operation_id=operation,
+        path_params={"workflow_id": 1} if operation == "workflow.run" else {},
+        query={"jobid": "test-job"} if operation == "scheduler.run" else None,
     )
     assert inspect_tool_result(result) is expected
     assert json.loads(result) == payload