diff --git a/CHANGELOG.md b/CHANGELOG.md index b1851d7..ed089a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。 > 工程口径的完整记录见 `PROGRESS.md` / git log。 +## 0.62.1 — 2026-08-06 + +- 修复外部系统只读 POST 接口因请求体结构信息缺失而反复报参数错误的问题；大查询结果不再被截成半段内容，超出对话容量时会完整暂存并支持按路径、分页和字段继续读取。 + ## 0.62.0 — 2026-08-05 - 外部系统连接不再局限于 Factory MES：管理员现在可以直接配置标准 OpenAPI 系统，并选择用户名密码换取 Token、API Key 或 Bearer Token；用户连接页面会按系统要求动态显示凭据字段。现有 Factory 配置和账号连接无需迁移。 diff --git a/DESIGN.md b/DESIGN.md index 76de1b0..2a38151 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -410,9 +410,11 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB) - Swagger/OpenAPI 是接口契约事实源;Gitea 代码只补业务语义和排障,不覆盖契约。规格/代码内文本一律当不可信数据,不能改写 system/tool 约束。 - Swagger/OpenAPI JSON 不持久化入数据库或文件，连接器按 `definition_id + user_id` 隔离后放在进程内存中缓存 5 分钟；重启自动失效。这样保留实时契约发现，又避免不同身份可见的规格互相污染。 -**工具面**:不把数百个 Swagger operation 全展开为 JSON tool(工具列表膨胀+选择降准),只挂三个 host-side 元工具:`external_system_list`(已连系统 + 管理员查询规划提示),`external_system_search`(按问题搜 operation 摘要 + 置顶管理员推荐入口),`external_system_call`(按 operation_id 调用)。仅当该 user 有 active 连接时注册,密钥不进 sandbox。搜索只展示实际可调用的 GET/HEAD 和已放行 POST；管理员在 definition JSONB 配置 `query_guidance` 与 `recommended_operation_ids`,前者是可信控制面的软路由策略,后者是无需关键词命中的机械发现入口。Factory 默认把 BI dataset list/exec 作为统计聚合入口,日志/明细用于逐条追溯；Swagger 业务文本仍是不可信数据。 +**工具面**:不把数百个 Swagger operation 全展开为 JSON tool(工具列表膨胀+选择降准),只挂四个 host-side 元工具:`external_system_list`(已连系统 + 管理员查询规划提示),`external_system_search`(按问题搜 operation 摘要、解析后的请求 body schema + 置顶管理员推荐入口),`external_system_call`(按 operation_id 调用),`external_system_result_read`(按 `result_ref` + JSON Pointer/分页/字段投影读取大响应)。仅当该 user 有 active 连接时注册,密钥不进 sandbox。搜索只展示实际可调用的 GET/HEAD 和已放行 POST；管理员在 definition JSONB 配置 `query_guidance` 与 `recommended_operation_ids`,前者是可信控制面的软路由策略,后者是无需关键词命中的机械发现入口。Factory 默认把 BI dataset list/exec 作为统计聚合入口,日志/明细用于逐条追溯；Swagger 业务文本仍是不可信数据。 -**明细扫描边界**:单次响应保留字节上限,每次 agent run 另按外部系统累计返回量；Factory connector 将 `page_size` 限在管理员上限,拒绝 `page=0` / `pageoff` 关闭分页。三者防模型通过连续翻日志自行做昂贵聚合,但不改变 Factory 对其他客户端的分页契约。达到边界后工具正向引导回 dataset/聚合接口或缩小查询范围。 +**大响应**:`max_result_bytes` 是进入模型上下文的单次内联额度,不再用于切断原始 JSON；超额响应完整写入当前 user_root 下按 task_id 隔离的隐藏缓存,工具只返回合法结构化预览、`result_ref`、原始字节数和可继续读取的位置。reader 每次读取都重新校验当前 user 对原 external system 的 active 授权,并与 call 共享本轮 `max_total_result_bytes` 内联额度。缓存固定 24h TTL、单响应 10 MiB、单 task 50 MiB、单 user 200 MiB,过期或超额时优先清理最旧缓存；超过响应安全上限的远端结果直接拒绝并要求缩小范围,不产生半截 JSON。这里把“上游响应安全边界”“完整结果保存”“模型上下文额度”拆成三层,既不丢数据,也不靠无限提高上下文额度解决大结果问题。 + +**明细扫描边界**:单次响应保留安全上限与模型内联额度,每次 agent run 另按外部系统累计内联返回量；Factory connector 将 `page_size` 限在管理员上限,拒绝 `page=0` / `pageoff` 关闭分页。三者防模型通过连续翻日志自行做昂贵聚合,但不改变 Factory 对其他客户端的分页契约。达到边界后工具正向引导回 dataset/聚合接口、`result_ref` 分段读取或缩小查询范围。 **状态与 UI（两表）**:`external_system_definitions` 保存管理员维护的可信系统目录、查询规划提示、推荐入口和 `access_mode=all|selected`；这些新增项复用既有 `config` JSONB,无 schema/migration。提示词在 admin 表单里复用通用 dialog 的多行编辑器,不把长文常驻铺在页面。`external_systems` 同时承载指定用户授权和用户密文连接，`pending` 表示已授权但未配置凭据，`active` 才挂工具。管理员撤销指定用户会删除其连接和密文凭据；用户自行断开只清凭据、保留管理员授权。管理后台可新增、编辑、停用目录项，已有用户连接的目录项禁止直接删除。左栏「外部系统」面板只能选择当前用户可见目录、测试连接、替换凭据和断开,不能查看密码。稳定问法沉淀到用户私有 skill 时只写 provider/operation_id/参数规则,永远使用当前提问者的连接执行,共享 skill 不等于共享权限。 diff --git a/PROGRESS.md b/PROGRESS.md index b7dfe7b..99f20d8 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -2,7 +2,7 @@ > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。 -最后更新:2026-08-05(通用 OpenAPI 外部系统与可配置认证,bump 0.62.0) +最后更新:2026-08-06(OpenAPI 请求体契约 + 大响应无损分段读取,bump 0.62.1) --- @@ -21,6 +21,10 @@ ## 已完成关键能力 +### 2026-08-06 + +- **08-06 / 0.62.1 / OpenAPI 请求体契约 + 大响应无损读取**:生产 task `506c6ad0` 暴露 Swagger 2 body `$ref` 未进入搜索结果，模型无法得知 BI dataset exec 需要 `body.query`，并在 400 只剩状态码时连续盲猜；现统一解析 Swagger 2/OpenAPI 3 请求体 schema，返回限长脱敏的上游错误详情。外部响应不再按字节切成半截 JSON，超出模型内联额度时完整写入 user/task 隔离的 24h 缓存并返回 `result_ref`，新增 JSON Pointer、分页和字段投影 reader，单响应/task/user 分别以 10/50/200 MiB 守门；完整 479 项 unittest 全绿(17 skip)，外部系统专项 26 项、Ruff、隔离 mypy、Python 编译与 diff 检查通过。无 schema、migration、依赖或 HTTP API 变化，未写生产 DB。 + ### 2026-08-05 - **08-05 / 0.62.0 / 通用 OpenAPI 外部系统 + 可配置认证**:将 Factory 专用运行态拆为通用 OpenAPI connector、认证 strategy 和 provider preset 注册表；标准 ERP/LIMS 等系统可直接在管理后台选择用户名密码换 Token、API Key 或 Bearer Token，并配置 Token 字段与认证 Header，无需再新增 Python connector。Factory MES 保持内置 preset、既有配置和旧 `{username,password}` HTTP 请求兼容；用户凭据表单按 definition 动态生成，Swagger JSON 不落盘并按 `definition_id+user_id` 隔离缓存 5 分钟。完整 475 项 unittest 全绿(17 skip)，相关 41 项回归、Python/JavaScript 语法及 diff 检查通过；当前环境无可用浏览器实例，真实页面点击留部署后冒烟。无 schema、migration 或依赖变化，未连接生产 DB。 diff --git a/core/__init__.py b/core/__init__.py index ef76fa7..499b09a 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,3 +1,3 @@ # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。 # 改版本只动这一行。 -__version__ = "0.62.0" +__version__ = "0.62.1" diff --git a/core/external_systems/openapi.py b/core/external_systems/openapi.py index c7477a3..73b5176 100644 --- a/core/external_systems/openapi.py +++ b/core/external_systems/openapi.py @@ -17,6 +17,7 @@ from urllib.parse import quote, urljoin, urlparse import httpx from .auth import ExternalAuthError, get_auth_strategy +from .results import MAX_STORED_RESULT_BYTES class OpenApiError(RuntimeError): @@ -26,6 +27,17 @@ class OpenApiError(RuntimeError): _HTTP_METHODS = ("get", "head", "post", "put", "patch", "delete") _SPEC_CACHE: dict[str, tuple[float, dict[str, Any]]] = {} _SPEC_LOCK = Lock() +_SCHEMA_MAX_DEPTH = 5 +_SCHEMA_MAX_PROPERTIES = 50 +_SCHEMA_MAX_ENUM_ITEMS = 30 +_SCHEMA_MAX_NODES = 100 +_SCHEMA_TEXT_MAX_CHARS = 1000 +_ERROR_DETAIL_MAX_CHARS = 2000 +_SENSITIVE_KEY_RE = re.compile( + r"(?:password|passwd|secret|token|api[_-]?key|authorization|cookie|credential)", + re.IGNORECASE, +) + def _bool_value(value: Any, default: bool) -> bool: raw = str(value if value is not None else "").strip().lower() @@ -217,6 +229,189 @@ class OpenApiClient: }) return results + @classmethod + def _compact_schema( + cls, + spec: dict[str, Any], + schema: Any, + *, + depth: int = 0, + seen_refs: frozenset[str] = frozenset(), + budget: Optional[list[int]] = None, + ) -> dict[str, Any]: + """展开本地 schema 引用并限制体量，供模型构造请求而非做完整规范校验。""" + if budget is None: + budget = [_SCHEMA_MAX_NODES] + if budget[0] <= 0: + return {"truncated": True} + budget[0] -= 1 + if not isinstance(schema, dict) or depth > _SCHEMA_MAX_DEPTH: + return {} + ref = schema.get("$ref") + if isinstance(ref, str): + if ref in seen_refs: + return {"$ref": ref, "recursive": True} + prefixes = ( + ("#/definitions/", spec.get("definitions")), + ( + "#/components/schemas/", + (spec.get("components") or {}).get("schemas"), + ), + ) + for prefix, definitions in prefixes: + if ref.startswith(prefix) and isinstance(definitions, dict): + target = definitions.get(ref[len(prefix) :]) + if isinstance(target, dict): + return cls._compact_schema( + spec, + target, + depth=depth + 1, + seen_refs=seen_refs | {ref}, + budget=budget, + ) + return {"$ref": ref} + + compact: dict[str, Any] = {} + for key in ( + "type", + "title", + "description", + "format", + "default", + "nullable", + "x-nullable", + "minimum", + "maximum", + "minLength", + "maxLength", + "pattern", + ): + value = schema.get(key) + if isinstance(value, str): + compact[key] = value[:_SCHEMA_TEXT_MAX_CHARS] + elif ( + isinstance(value, (int, float, bool)) or value is None and key in schema + ): + compact[key] = value + required = schema.get("required") + if isinstance(required, list): + compact["required"] = [ + str(item) for item in required[:_SCHEMA_MAX_PROPERTIES] + ] + enum = schema.get("enum") + if isinstance(enum, list): + compact["enum"] = [ + item[:_SCHEMA_TEXT_MAX_CHARS] if isinstance(item, str) else item + for item in enum[:_SCHEMA_MAX_ENUM_ITEMS] + ] + properties = schema.get("properties") + if isinstance(properties, dict) and depth < _SCHEMA_MAX_DEPTH: + compact["properties"] = { + str(name): cls._compact_schema( + spec, + value, + depth=depth + 1, + seen_refs=seen_refs, + budget=budget, + ) + for name, value in list(properties.items())[:_SCHEMA_MAX_PROPERTIES] + if isinstance(value, dict) + } + items = schema.get("items") + if isinstance(items, dict) and depth < _SCHEMA_MAX_DEPTH: + compact["items"] = cls._compact_schema( + spec, + items, + depth=depth + 1, + seen_refs=seen_refs, + budget=budget, + ) + additional = schema.get("additionalProperties") + if isinstance(additional, bool): + compact["additionalProperties"] = additional + elif isinstance(additional, dict) and depth < _SCHEMA_MAX_DEPTH: + compact["additionalProperties"] = cls._compact_schema( + spec, + additional, + depth=depth + 1, + seen_refs=seen_refs, + budget=budget, + ) + for key in ("allOf", "anyOf", "oneOf"): + variants = schema.get(key) + if isinstance(variants, list) and depth < _SCHEMA_MAX_DEPTH: + compact[key] = [ + cls._compact_schema( + spec, + item, + depth=depth + 1, + seen_refs=seen_refs, + budget=budget, + ) + for item in variants[:10] + if isinstance(item, dict) + ] + return compact + + @classmethod + def _body_contract( + cls, spec: dict[str, Any], operation: dict[str, Any] + ) -> Optional[dict[str, Any]]: + """统一 Swagger 2 body parameter 与 OpenAPI 3 requestBody。""" + for param in operation.get("parameters") or []: + if isinstance(param, dict) and param.get("in") == "body": + return { + "parameter_name": param.get("name") or "body", + "required": bool(param.get("required")), + "content_type": "application/json", + "schema": cls._compact_schema(spec, param.get("schema")), + } + request_body = operation.get("request_body") or operation.get("requestBody") + if not isinstance(request_body, dict): + return None + content = request_body.get("content") + if not isinstance(content, dict) or not content: + return None + content_type = ( + "application/json" if "application/json" in content else next(iter(content)) + ) + media = content.get(content_type) + if not isinstance(media, dict): + return None + return { + "required": bool(request_body.get("required")), + "content_type": content_type, + "schema": cls._compact_schema(spec, media.get("schema")), + } + + @staticmethod + def _error_detail(response: httpx.Response) -> str: + """返回限长、递归脱敏的上游错误详情，帮助模型停止盲猜参数。""" + + def redact(value: Any) -> Any: + if isinstance(value, dict): + return { + str(key): "[REDACTED]" + if _SENSITIVE_KEY_RE.search(str(key)) + else redact(item) + for key, item in value.items() + } + if isinstance(value, list): + return [redact(item) for item in value] + return value + + try: + detail = json.dumps( + redact(response.json()), ensure_ascii=False, default=str + ) + except ValueError: + detail = response.text.strip() + if not detail: + return "" + if len(detail) > _ERROR_DETAIL_MAX_CHARS: + detail = detail[:_ERROR_DETAIL_MAX_CHARS] + "...[truncated]" + return detail + def _spec_base_path(self, spec: dict[str, Any]) -> str: """Return the API path prefix declared by Swagger 2 / OpenAPI 3. @@ -325,6 +520,9 @@ class OpenApiClient: } for p in op["parameters"] if isinstance(p, dict) and "$ref" not in p ] + body_contract = self._body_contract(spec, op) + if body_contract is not None: + compact["body"] = body_contract compact.pop("request_body", None) compact["recommended"] = recommended scored.append(( @@ -411,20 +609,25 @@ class OpenApiClient: except httpx.HTTPError as exc: raise OpenApiError(f"外部系统接口调用失败: {type(exc).__name__}") from exc if response.status_code >= 400: - raise OpenApiError(f"外部系统接口返回 HTTP {response.status_code}") + detail = self._error_detail(response) + suffix = f": {detail}" if detail else "" + raise OpenApiError(f"外部系统接口返回 HTTP {response.status_code}{suffix}") content_type = response.headers.get("content-type", "") try: payload: Any = response.json() if "json" in content_type else response.text except ValueError: payload = response.text encoded = json.dumps(payload, ensure_ascii=False, default=str) - truncated = len(encoded.encode("utf-8")) > self.cfg.max_result_bytes - if truncated: - encoded = encoded.encode("utf-8")[: self.cfg.max_result_bytes].decode("utf-8", "ignore") - payload = encoded + response_bytes = len(encoded.encode("utf-8")) + if response_bytes > MAX_STORED_RESULT_BYTES: + raise OpenApiError( + f"外部系统响应超过安全下载上限({MAX_STORED_RESULT_BYTES} bytes)，" + "请缩小查询范围或使用远端分页/聚合接口" + ) return { "operation_id": operation_id, "status_code": response.status_code, - "truncated": truncated, + "truncated": False, + "response_bytes": response_bytes, "data": payload, } diff --git a/core/external_systems/results.py b/core/external_systems/results.py new file mode 100644 index 0000000..f973a53 --- /dev/null +++ b/core/external_systems/results.py @@ -0,0 +1,258 @@ +"""外部系统大响应的 task 私有缓存与结构化分段读取。""" +from __future__ import annotations + +import json +import re +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from uuid import uuid4 + +from core.file_store import atomic_write_text + +RESULT_CACHE_SUBDIR = ".zcbot_external_results" +RESULT_TTL_SECONDS = 24 * 60 * 60 +MAX_STORED_RESULT_BYTES = 10 * 1024 * 1024 +MAX_TASK_CACHE_BYTES = 50 * 1024 * 1024 +MAX_USER_CACHE_BYTES = 200 * 1024 * 1024 +_REF_RE = re.compile(r"^extres_[0-9a-f]{32}$") +_PREVIEW_MAX_DEPTH = 6 +_PREVIEW_MAX_NODES = 120 +_PREVIEW_MAX_DICT_KEYS = 30 +_PREVIEW_MAX_LIST_ITEMS = 5 +_PREVIEW_MAX_STRING_CHARS = 500 + + +class ExternalResultError(RuntimeError): + pass + + +@dataclass(frozen=True) +class StoredExternalResult: + result_ref: str + original_bytes: int + expires_in_seconds: int + + +def _json_bytes(value: Any) -> bytes: + return json.dumps(value, ensure_ascii=False, default=str).encode("utf-8") + + +def _pointer_escape(value: str) -> str: + return value.replace("~", "~0").replace("/", "~1") + + +def build_result_preview(value: Any) -> tuple[Any, list[dict[str, Any]]]: + """生成合法、定界的 JSON 预览，并列出可继续分段读取的位置。""" + reads: list[dict[str, Any]] = [] + budget = [_PREVIEW_MAX_NODES] + + def visit(item: Any, pointer: str, depth: int) -> Any: + if budget[0] <= 0: + reads.append({"json_pointer": pointer, "reason": "node_budget"}) + return {"$preview_truncated": True} + budget[0] -= 1 + if depth > _PREVIEW_MAX_DEPTH: + reads.append({"json_pointer": pointer, "reason": "depth"}) + return {"$preview_truncated": True} + if isinstance(item, dict): + pairs = list(item.items()) + shown = pairs[:_PREVIEW_MAX_DICT_KEYS] + result: Any = { + str(key): visit( + child, + pointer + "/" + _pointer_escape(str(key)), + depth + 1, + ) + for key, child in shown + } + if len(pairs) > len(shown): + result["$preview_omitted_keys"] = len(pairs) - len(shown) + reads.append({ + "json_pointer": pointer, + "reason": "keys", + "total_keys": len(pairs), + }) + return result + if isinstance(item, list): + shown = item[:_PREVIEW_MAX_LIST_ITEMS] + result = [ + visit(child, pointer + "/" + str(index), depth + 1) + for index, child in enumerate(shown) + ] + if len(item) > len(shown): + reads.append({ + "json_pointer": pointer, + "reason": "items", + "total_items": len(item), + "next_offset": len(shown), + }) + return result + if isinstance(item, str) and len(item) > _PREVIEW_MAX_STRING_CHARS: + reads.append({ + "json_pointer": pointer, + "reason": "string", + "total_chars": len(item), + }) + return item[:_PREVIEW_MAX_STRING_CHARS] + "...[preview]" + return item + + return visit(value, "", 0), reads + + +def resolve_json_pointer(value: Any, pointer: str) -> Any: + """解析 RFC 6901 JSON Pointer；空串表示完整根值。""" + pointer = (pointer or "").strip() + if not pointer: + return value + if not pointer.startswith("/"): + raise ExternalResultError("json_pointer 必须为空或以 / 开头") + current = value + for raw_part in pointer[1:].split("/"): + part = raw_part.replace("~1", "/").replace("~0", "~") + if isinstance(current, dict): + if part not in current: + raise ExternalResultError(f"json_pointer 不存在: {pointer}") + current = current[part] + elif isinstance(current, list): + try: + if not part.isdigit(): + raise ValueError(part) + index = int(part) + current = current[index] + except (ValueError, IndexError) as exc: + raise ExternalResultError(f"json_pointer 不存在: {pointer}") from exc + else: + raise ExternalResultError(f"json_pointer 不存在: {pointer}") + return current + + +class ExternalResultStore: + def __init__(self, user_root: Path, task_id: str): + self.cache_root = Path(user_root) / RESULT_CACHE_SUBDIR + self.root = self.cache_root / str(task_id) + + def _path(self, result_ref: str) -> Path: + if not _REF_RE.fullmatch(result_ref or ""): + raise ExternalResultError("result_ref 无效") + return self.root / f"{result_ref}.json" + + def _sweep(self) -> None: + if not self.cache_root.is_dir(): + return + cutoff = time.time() - RESULT_TTL_SECONDS + for path in self.cache_root.glob("*/extres_*.json"): + try: + if path.stat().st_mtime < cutoff: + path.unlink() + try: + path.parent.rmdir() + except OSError: + pass + except OSError: + continue + + def store(self, system_id: str, result: dict[str, Any]) -> StoredExternalResult: + encoded = _json_bytes(result) + if len(encoded) > MAX_STORED_RESULT_BYTES: + raise ExternalResultError( + f"外部系统响应超过安全保存上限({MAX_STORED_RESULT_BYTES} bytes)" + ) + self.root.mkdir(parents=True, exist_ok=True) + self._sweep() + existing = sorted( + self.root.glob("extres_*.json"), + key=lambda path: path.stat().st_mtime, + ) + total = sum(path.stat().st_size for path in existing) + while existing and total + len(encoded) > MAX_TASK_CACHE_BYTES: + oldest = existing.pop(0) + try: + size = oldest.stat().st_size + oldest.unlink() + total -= size + except OSError: + continue + if total + len(encoded) > MAX_TASK_CACHE_BYTES: + raise ExternalResultError("当前任务的外部结果缓存已达上限") + user_files = sorted( + self.cache_root.glob("*/extres_*.json"), + key=lambda path: path.stat().st_mtime, + ) + user_total = sum(path.stat().st_size for path in user_files) + while user_files and user_total + len(encoded) > MAX_USER_CACHE_BYTES: + oldest = user_files.pop(0) + try: + size = oldest.stat().st_size + oldest.unlink() + user_total -= size + except OSError: + continue + if user_total + len(encoded) > MAX_USER_CACHE_BYTES: + raise ExternalResultError("当前用户的外部结果缓存已达上限") + result_ref = f"extres_{uuid4().hex}" + envelope = { + "version": 1, + "created_at": int(time.time()), + "system_id": str(system_id), + "result": result, + } + atomic_write_text(self._path(result_ref), json.dumps( + envelope, ensure_ascii=False, default=str + )) + return StoredExternalResult( + result_ref=result_ref, + original_bytes=len(encoded), + expires_in_seconds=RESULT_TTL_SECONDS, + ) + + def load(self, result_ref: str) -> dict[str, Any]: + self._sweep() + path = self._path(result_ref) + try: + if time.time() - path.stat().st_mtime > RESULT_TTL_SECONDS: + path.unlink() + raise ExternalResultError("result_ref 已过期") + payload = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise ExternalResultError("result_ref 不存在或已过期") from exc + except (OSError, ValueError) as exc: + raise ExternalResultError("result_ref 无法读取") from exc + if not isinstance(payload, dict) or not isinstance(payload.get("result"), dict): + raise ExternalResultError("result_ref 内容无效") + return payload + + +def project_result( + value: Any, + *, + offset: int = 0, + limit: int = 50, + fields: list[str] | None = None, +) -> dict[str, Any]: + """对缓存节点分页并可选投影字典字段。""" + offset = max(0, int(offset)) + limit = max(1, min(int(limit), 200)) + if fields is not None and not isinstance(fields, list): + raise ExternalResultError("fields 必须是字符串数组") + wanted = [str(field) for field in (fields or [])[:50]] + + def select(item: Any) -> Any: + if not wanted or not isinstance(item, dict): + return item + return {field: item.get(field) for field in wanted if field in item} + + if isinstance(value, list): + page = [select(item) for item in value[offset : offset + limit]] + return { + "type": "array", + "total_items": len(value), + "offset": offset, + "limit": limit, + "has_more": offset + len(page) < len(value), + "data": page, + } + if isinstance(value, dict): + return {"type": "object", "data": select(value)} + return {"type": type(value).__name__, "data": value} diff --git a/core/tool_registry.py b/core/tool_registry.py index d214b55..d931d90 100644 --- a/core/tool_registry.py +++ b/core/tool_registry.py @@ -28,6 +28,7 @@ from tools.documents import DocumentDownloadTool, DocumentListKbTool, DocumentSe from tools.external_systems import ( ExternalSystemCallTool, ExternalSystemListTool, + ExternalSystemResultReadTool, ExternalSystemSearchTool, ) from tools.fs import EditTool, GlobTool, GrepTool, ReadTool, WriteTool @@ -154,10 +155,22 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]: ] def _external_systems() -> list: + result_budget: dict[str, int] = {} return [ ExternalSystemListTool(ctx.uid, **base), ExternalSystemSearchTool(ctx.uid, **base), - ExternalSystemCallTool(ctx.uid, **base), + ExternalSystemCallTool( + ctx.uid, + task_id=ctx.task_id, + result_budget=result_budget, + **wd_base, + ), + ExternalSystemResultReadTool( + ctx.uid, + task_id=ctx.task_id, + result_budget=result_budget, + **wd_base, + ), ] def _load_skill() -> list: diff --git a/tests/test_external_systems.py b/tests/test_external_systems.py index f017758..813106c 100644 --- a/tests/test_external_systems.py +++ b/tests/test_external_systems.py @@ -3,6 +3,7 @@ from __future__ import annotations import json import os import sys +import tempfile import unittest import uuid from copy import deepcopy @@ -59,6 +60,16 @@ def _cfg(*, allowed=frozenset(), recommended=()): _SPEC = { "swagger": "2.0", + "definitions": { + "DataExec": { + "type": "object", + "required": ["query"], + "properties": { + "query": {"title": "查询字典参数", "type": "object"}, + "is_test": {"type": "boolean", "default": False}, + }, + } + }, "paths": { "/api/bi/dataset/": { "get": { @@ -85,7 +96,12 @@ _SPEC = { "summary": "执行只读数据集", "parameters": [ {"name": "code", "in": "path", "required": True, "type": "string"}, - {"name": "payload", "in": "body", "required": True, "schema": {"type": "object"}}, + { + "name": "payload", + "in": "body", + "required": True, + "schema": {"$ref": "#/definitions/DataExec"}, + }, ], } }, @@ -236,6 +252,70 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): ) self.assertTrue(all(item["recommended"] for item in result[:2])) + def test_search_includes_resolved_swagger_body_schema(self): + from core.external_systems.factory import FactoryMesClient + + http = _Http() + client = FactoryMesClient( + "mes-user", "mes-password", _cfg(allowed={"bi_dataset_exec"}) + ) + with patch.object(client, "_client", return_value=http): + result = client.search("执行只读数据集") + operation = next( + item for item in result if item["operation_id"] == "bi_dataset_exec" + ) + self.assertEqual(operation["body"]["parameter_name"], "payload") + self.assertTrue(operation["body"]["required"]) + self.assertEqual( + operation["body"]["schema"]["properties"]["query"]["type"], + "object", + ) + self.assertEqual(operation["body"]["schema"]["required"], ["query"]) + + def test_search_includes_openapi_3_request_body_schema(self): + from core.external_systems.openapi import OpenApiClient + + spec = { + "openapi": "3.0.0", + "components": { + "schemas": { + "ReportQuery": { + "type": "object", + "properties": {"month": {"type": "string"}}, + } + } + }, + "paths": { + "/reports/preview/": { + "post": { + "operationId": "report_preview", + "summary": "预览报表", + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ReportQuery" + } + } + }, + }, + } + } + }, + } + client = OpenApiClient({"token": "private"}, _cfg(allowed={"report_preview"})) + with ( + patch.object(client, "authenticate", return_value={}), + patch.object(client, "_fetch_spec", return_value=spec), + ): + result = client.search("预览报表") + self.assertEqual(result[0]["body"]["content_type"], "application/json") + self.assertEqual( + result[0]["body"]["schema"]["properties"]["month"]["type"], + "string", + ) + def test_get_call_resolves_encoded_path_and_query(self): from core.external_systems.factory import FactoryMesClient @@ -389,12 +469,60 @@ class FactoryOpenApiConnectorTests(unittest.TestCase): result = allowed.call( "bi_dataset_exec", arguments={"code": "quality"}, - body={"batch": "B-1"}, + body={"query": {"batch": "B-1"}}, ) request = [call for call in http.calls if call[0] == "POST" and "/dataset/" in call[1]][0] - self.assertEqual(request[2]["json"], {"batch": "B-1"}) + self.assertEqual(request[2]["json"], {"query": {"batch": "B-1"}}) self.assertFalse(result["truncated"]) + def test_call_preserves_payload_larger_than_inline_limit(self): + from core.external_systems.factory import FactoryMesClient + + class LargeHttp(_Http): + def request(self, method, url, **kwargs): + self.calls.append((method, url, kwargs)) + return _Response(payload={"rows": "x" * 70000}) + + client = FactoryMesClient( + "u", "p", _cfg(allowed={"bi_dataset_exec"}) + ) + with patch.object(client, "_client", return_value=LargeHttp()): + result = client.call( + "bi_dataset_exec", + arguments={"code": "quality"}, + body={"query": {}}, + ) + self.assertEqual(len(result["data"]["rows"]), 70000) + self.assertGreater(result["response_bytes"], client.cfg.max_result_bytes) + self.assertFalse(result["truncated"]) + + def test_call_surfaces_sanitized_upstream_error_detail(self): + from core.external_systems.factory import FactoryMesClient, FactoryMesError + + class ErrorHttp(_Http): + def request(self, method, url, **kwargs): + self.calls.append((method, url, kwargs)) + return _Response( + status_code=400, + payload={ + "query": ["This field is required."], + "access_token": "must-not-leak", + }, + ) + + client = FactoryMesClient("u", "p", _cfg(allowed={"bi_dataset_exec"})) + with patch.object(client, "_client", return_value=ErrorHttp()): + with self.assertRaises(FactoryMesError) as raised: + client.call( + "bi_dataset_exec", + arguments={"code": "yield"}, + body={"wrong": "shape"}, + ) + message = str(raised.exception) + self.assertIn("This field is required.", message) + self.assertIn("[REDACTED]", message) + self.assertNotIn("must-not-leak", message) + def test_rejects_unknown_arguments(self): from core.external_systems.factory import FactoryMesClient, FactoryMesError @@ -446,23 +574,64 @@ class ExternalSystemToolSafetyTests(unittest.TestCase): self.assertEqual(payload["query_guidance"], "统计查询先查看数据集目录") self.assertEqual(payload["recommended_operation_ids"], ["bi_dataset_list"]) - def test_call_discards_result_over_per_run_external_budget(self): - from tools.external_systems import ExternalSystemCallTool + def test_large_call_spills_and_result_reader_pages_without_data_loss(self): + from tools.external_systems import ( + ExternalSystemCallTool, + ExternalSystemResultReadTool, + ) uid = uuid.uuid4() + task_id = uuid.uuid4() + system_id = str(uuid.uuid4()) + budget: dict[str, int] = {} client = SimpleNamespace( - cfg=SimpleNamespace(max_total_result_bytes=32), - call=lambda *args, **kwargs: {"data": "x" * 100}, + cfg=SimpleNamespace(max_result_bytes=512, max_total_result_bytes=4096), + call=lambda *args, **kwargs: { + "operation_id": "detail_list", + "status_code": 200, + "truncated": False, + "data": [{"id": index, "noise": "x" * 100} for index in range(20)], + }, ) - with patch( + with tempfile.TemporaryDirectory() as tmp, patch( "tools.external_systems._row_and_client", return_value=(SimpleNamespace(), client), ): - output = ExternalSystemCallTool(uid).execute( - str(uuid.uuid4()), "detail_list" + call_tool = ExternalSystemCallTool( + uid, + task_id=task_id, + result_budget=budget, + base_dir=Path(tmp), ) - self.assertIn("累计返回量超过上限", output) - self.assertNotIn("x" * 20, output) + spilled = json.loads(call_tool.execute(system_id, "detail_list")) + self.assertFalse(spilled["inline_complete"]) + self.assertFalse(spilled["truncated"]) + self.assertRegex(spilled["result_ref"], r"^extres_[0-9a-f]{32}$") + self.assertNotIn("x" * 100, json.dumps(spilled)) + + read_tool = ExternalSystemResultReadTool( + uid, + task_id=task_id, + result_budget=budget, + base_dir=Path(tmp), + ) + page = json.loads(read_tool.execute( + spilled["result_ref"], + json_pointer="/data", + offset=5, + limit=2, + fields=["id"], + )) + other_task = ExternalSystemResultReadTool( + uid, + task_id=uuid.uuid4(), + base_dir=Path(tmp), + ) + cross_task = other_task.execute(spilled["result_ref"]) + self.assertEqual(page["total_items"], 20) + self.assertEqual(page["data"], [{"id": 5}, {"id": 6}]) + self.assertTrue(page["has_more"]) + self.assertIn("不存在或已过期", cross_task) if __name__ == "__main__": diff --git a/tools/external_systems.py b/tools/external_systems.py index 115c664..82e42bc 100644 --- a/tools/external_systems.py +++ b/tools/external_systems.py @@ -5,6 +5,13 @@ import json from uuid import UUID from core.external_systems.factory import FactoryMesError +from core.external_systems.results import ( + ExternalResultError, + ExternalResultStore, + build_result_preview, + project_result, + resolve_json_pointer, +) from core.external_systems.service import ( ExternalSystemError, client_for_external_system, @@ -86,6 +93,7 @@ class ExternalSystemCallTool(Tool): description = ( "调用已连接外部系统的受控只读 OpenAPI operation。必须使用 search 返回的 operation_id；" "不接受 URL。GET/HEAD 默认可用，POST 仅限管理员声明的只读 operation。" + "大响应会完整保存并返回 result_ref，使用 external_system_result_read 分段读取。" ) parameters = { "type": "object", @@ -99,16 +107,72 @@ class ExternalSystemCallTool(Tool): }, "body": { "type": "object", - "description": "仅对管理员放行的只读 POST 操作提供 JSON body" + "description": ( + "仅对管理员放行的只读 POST 操作提供原始 JSON 请求体；" + "严格遵循 search 返回的 body.schema，不要按 Swagger body 参数名再包一层" + ) }, }, "required": ["system_id", "operation_id"], } - def __init__(self, user_id: UUID, **kwargs): + def __init__( + self, + user_id: UUID, + *, + task_id: UUID | str = "default", + result_budget: dict[str, int] | None = None, + **kwargs, + ): super().__init__(**kwargs) self.user_id = user_id - self._result_bytes: dict[str, int] = {} + self._result_bytes = result_budget if result_budget is not None else {} + self._result_store = ExternalResultStore( + self.user_root or self.base_dir, str(task_id) + ) + + def _bounded_output( + self, + system_id: str, + result: dict, + *, + per_result_limit: int, + total_limit: int, + ) -> str: + rendered = _json(result) + used = self._result_bytes.get(system_id, 0) + remaining = max(0, total_limit - used) + if len(rendered.encode("utf-8")) <= min(per_result_limit, remaining): + self._result_bytes[system_id] = used + len(rendered.encode("utf-8")) + return rendered + + stored = self._result_store.store(system_id, result) + preview, reads = build_result_preview(result) + envelope = { + "operation_id": result.get("operation_id"), + "status_code": result.get("status_code"), + "truncated": False, + "inline_complete": False, + "result_ref": stored.result_ref, + "original_bytes": stored.original_bytes, + "expires_in_seconds": stored.expires_in_seconds, + "preview": preview, + "available_reads": reads, + } + output = _json(envelope) + inline_limit = min(per_result_limit, remaining) + if len(output.encode("utf-8")) > inline_limit: + envelope["preview"] = None + envelope["available_reads"] = reads[:20] + output = _json(envelope) + if len(output.encode("utf-8")) > remaining: + self._result_bytes[system_id] = total_limit + return ( + "[Error] 本轮外部系统内联返回量已达上限。完整结果已缓存为 " + f"{stored.result_ref}，请在下一轮使用 external_system_result_read 读取。" + ) + self._result_bytes[system_id] = used + len(output.encode("utf-8")) + return output def execute( self, @@ -123,19 +187,116 @@ class ExternalSystemCallTool(Tool): used = self._result_bytes.get(system_id, 0) if used >= client.cfg.max_total_result_bytes: return ( - "[Error] 本轮外部系统返回量已达上限。请改用聚合接口或 dataset，" - "不要继续分页拉取日志/明细。" + "[Error] 本轮外部系统内联返回量已达上限。" + "请在下一轮继续查询，或读取之前返回的 result_ref。" ) result = client.call(operation_id, arguments=arguments, body=body) - result_size = len(_json(result).encode("utf-8")) - used += result_size - self._result_bytes[system_id] = used - if used > client.cfg.max_total_result_bytes: - return ( - "[Error] 本轮外部系统累计返回量超过上限，当前结果已丢弃。" - "请改用聚合接口或 dataset，并缩小查询范围。" - ) - return _json(result) - except (ExternalSystemError, FactoryMesError) as exc: + return self._bounded_output( + system_id, + result, + per_result_limit=client.cfg.max_result_bytes, + total_limit=client.cfg.max_total_result_bytes, + ) + except (ExternalSystemError, FactoryMesError, ExternalResultError) as exc: print(f"[WARN] external system call failed: {type(exc).__name__}") return f"[Error] {exc}" + + +class ExternalSystemResultReadTool(Tool): + name = "external_system_result_read" + description = ( + "分段读取 external_system_call 返回的完整大结果。result_ref 仅对当前 task 有效；" + "使用 JSON Pointer 定位数组或对象，并用 offset/limit/fields 控制返回量。" + ) + parameters = { + "type": "object", + "properties": { + "result_ref": {"type": "string"}, + "json_pointer": { + "type": "string", + "default": "", + "description": "RFC 6901 JSON Pointer；空字符串表示根节点", + }, + "offset": {"type": "integer", "minimum": 0, "default": 0}, + "limit": {"type": "integer", "minimum": 1, "maximum": 200, "default": 50}, + "fields": { + "type": "array", + "items": {"type": "string"}, + "maxItems": 50, + "description": "目标是对象或对象数组时，仅返回这些字段", + }, + }, + "required": ["result_ref"], + } + + def __init__( + self, + user_id: UUID, + *, + task_id: UUID | str = "default", + result_budget: dict[str, int] | None = None, + **kwargs, + ): + super().__init__(**kwargs) + self.user_id = user_id + self._result_bytes = result_budget if result_budget is not None else {} + self._result_store = ExternalResultStore( + self.user_root or self.base_dir, str(task_id) + ) + + def execute( + self, + result_ref: str, + json_pointer: str = "", + offset: int = 0, + limit: int = 50, + fields: list[str] | None = None, + **kwargs, + ) -> str: + try: + stored = self._result_store.load(result_ref) + system_id = str(stored.get("system_id") or "") + _, client = _row_and_client(self.user_id, system_id) + used = self._result_bytes.get(system_id, 0) + remaining = client.cfg.max_total_result_bytes - used + if remaining <= 0: + return "[Error] 本轮外部系统内联返回量已达上限，请在下一轮继续读取。" + selected = resolve_json_pointer(stored["result"], json_pointer) + response = { + "result_ref": result_ref, + "json_pointer": json_pointer, + **project_result(selected, offset=offset, limit=limit, fields=fields), + } + output = _json(response) + inline_limit = min(client.cfg.max_result_bytes, remaining) + if len(output.encode("utf-8")) > inline_limit: + preview, reads = build_result_preview(response) + output = _json({ + "result_ref": result_ref, + "json_pointer": json_pointer, + "inline_complete": False, + "preview": preview, + "available_reads": reads, + "hint": "减小 limit、指定更深的 json_pointer 或使用 fields 投影", + }) + if len(output.encode("utf-8")) > inline_limit: + output = _json({ + "result_ref": result_ref, + "json_pointer": json_pointer, + "inline_complete": False, + "hint": "当前分段仍过大；请减小 limit、指定更深的 json_pointer 或使用 fields", + }) + if len(output.encode("utf-8")) > remaining: + self._result_bytes[system_id] = client.cfg.max_total_result_bytes + return "[Error] 本轮外部系统内联返回量已达上限，请在下一轮继续读取。" + self._result_bytes[system_id] = used + len(output.encode("utf-8")) + return output + except ( + ExternalSystemError, + FactoryMesError, + ExternalResultError, + TypeError, + ValueError, + ) as exc: + print(f"[WARN] external system result read failed: {type(exc).__name__}") + return f"[Error] {exc}"