From 0f8aa4dfa9cc883970d73dd251a2ea7cbf775f00 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 11 Sep 2026 08:07:54 +0800 Subject: [PATCH] fix(agent): return operation input contracts on errors --- app/agent/api/arguments.py | 100 ++++++++++++++++++++++ app/agent/middleware/invocation.py | 10 ++- app/agent/tools/impl/api.py | 58 +++++++++++-- docs/refactor/agent-codex-parity-plan.md | 2 +- scripts/evaluation/server.py | 83 +++++++++++++++++- skills/moviepilot-api/api/download.md | 4 +- tests/test_agent_api_gateway.py | 36 ++++++-- tests/test_agent_evaluation_server.py | 18 ++++ tests/test_agent_invocation_middleware.py | 4 + tests/test_agent_tool_outcomes.py | 32 ++++++- 10 files changed, 321 insertions(+), 26 deletions(-) diff --git a/app/agent/api/arguments.py b/app/agent/api/arguments.py index a2f397dee..56e78551f 100644 --- a/app/agent/api/arguments.py +++ b/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 { diff --git a/app/agent/middleware/invocation.py b/app/agent/middleware/invocation.py index 625001ef5..e2eb9d55b 100644 --- a/app/agent/middleware/invocation.py +++ b/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__}") diff --git a/app/agent/tools/impl/api.py b/app/agent/tools/impl/api.py index c5e39f740..1e038cfbf 100644 --- a/app/agent/tools/impl/api.py +++ b/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: diff --git a/docs/refactor/agent-codex-parity-plan.md b/docs/refactor/agent-codex-parity-plan.md index e671fa8b2..d64df9d73 100644 --- a/docs/refactor/agent-codex-parity-plan.md +++ b/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/.md` 参数按需读取类别文档；评测目录必须提供同一条链路，不能只返回文件名而不给模型读取能力。operation 输入错误还要返回该 operation 的允许字段、必填字段和类型约束，帮助模型纠正后重试。 - **长任务与新消息**：入站消息在任务运行时仍可接受并进入有界队列；应用到下一模型边界时必须保留 tool-call/tool-result 配对和取消语义，不能只把文本拼到系统提示词。 - **观察证据**：最终答案只能引用宿主记录的工具结果和业务 oracle；“模型说完成”不能替代数据库、下载器、浏览器或进程状态核验。 diff --git a/scripts/evaluation/server.py b/scripts/evaluation/server.py index 2dd26eb2c..664357a9b 100644 --- a/scripts/evaluation/server.py +++ b/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() diff --git a/skills/moviepilot-api/api/download.md b/skills/moviepilot-api/api/download.md index a7e5d80ab..ab834d505 100644 --- a/skills/moviepilot-api/api/download.md +++ b/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`. diff --git a/tests/test_agent_api_gateway.py b/tests/test_agent_api_gateway.py index 588ad819a..332ee4497 100644 --- a/tests/test_agent_api_gateway.py +++ b/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() diff --git a/tests/test_agent_evaluation_server.py b/tests/test_agent_evaluation_server.py index 4aff7826e..5bff0c4c1 100644 --- a/tests/test_agent_evaluation_server.py +++ b/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: """并行连接也共享硬调用上限，超限不会继续读取或改变独立世界。""" diff --git a/tests/test_agent_invocation_middleware.py b/tests/test_agent_invocation_middleware.py index 43f802a8b..52eba8c10 100644 --- a/tests/test_agent_invocation_middleware.py +++ b/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() diff --git a/tests/test_agent_tool_outcomes.py b/tests/test_agent_tool_outcomes.py index 527763d46..b142bc4e8 100644 --- a/tests/test_agent_tool_outcomes.py +++ b/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