Compare commits
2 Commits
80fbc7ab4a
...
e81005435e
| Author | SHA1 | Date |
|---|---|---|
|
|
e81005435e | |
|
|
bce896a21f |
|
|
@ -5,6 +5,10 @@
|
||||||
> 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。
|
> 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。
|
||||||
> 工程口径的完整记录见 `PROGRESS.md` / git log。
|
> 工程口径的完整记录见 `PROGRESS.md` / git log。
|
||||||
|
|
||||||
|
## 0.62.1 — 2026-08-06
|
||||||
|
|
||||||
|
- 修复外部系统只读 POST 接口因请求体结构信息缺失而反复报参数错误的问题;大查询结果不再被截成半段内容,超出对话容量时会完整暂存并支持按路径、分页和字段继续读取。
|
||||||
|
|
||||||
## 0.62.0 — 2026-08-05
|
## 0.62.0 — 2026-08-05
|
||||||
|
|
||||||
- 外部系统连接不再局限于 Factory MES:管理员现在可以直接配置标准 OpenAPI 系统,并选择用户名密码换取 Token、API Key 或 Bearer Token;用户连接页面会按系统要求动态显示凭据字段。现有 Factory 配置和账号连接无需迁移。
|
- 外部系统连接不再局限于 Factory MES:管理员现在可以直接配置标准 OpenAPI 系统,并选择用户名密码换取 Token、API Key 或 Bearer Token;用户连接页面会按系统要求动态显示凭据字段。现有 Factory 配置和账号连接无需迁移。
|
||||||
|
|
|
||||||
26
DESIGN.md
26
DESIGN.md
|
|
@ -410,14 +410,36 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB)
|
||||||
- Swagger/OpenAPI 是接口契约事实源;Gitea 代码只补业务语义和排障,不覆盖契约。规格/代码内文本一律当不可信数据,不能改写 system/tool 约束。
|
- Swagger/OpenAPI 是接口契约事实源;Gitea 代码只补业务语义和排障,不覆盖契约。规格/代码内文本一律当不可信数据,不能改写 system/tool 约束。
|
||||||
- Swagger/OpenAPI JSON 不持久化入数据库或文件,连接器按 `definition_id + user_id` 隔离后放在进程内存中缓存 5 分钟;重启自动失效。这样保留实时契约发现,又避免不同身份可见的规格互相污染。
|
- 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 不等于共享权限。
|
**状态与 UI(两表)**:`external_system_definitions` 保存管理员维护的可信系统目录、查询规划提示、推荐入口和 `access_mode=all|selected`;这些新增项复用既有 `config` JSONB,无 schema/migration。提示词在 admin 表单里复用通用 dialog 的多行编辑器,不把长文常驻铺在页面。`external_systems` 同时承载指定用户授权和用户密文连接,`pending` 表示已授权但未配置凭据,`active` 才挂工具。管理员撤销指定用户会删除其连接和密文凭据;用户自行断开只清凭据、保留管理员授权。管理后台可新增、编辑、停用目录项,已有用户连接的目录项禁止直接删除。左栏「外部系统」面板只能选择当前用户可见目录、测试连接、替换凭据和断开,不能查看密码。稳定问法沉淀到用户私有 skill 时只写 provider/operation_id/参数规则,永远使用当前提问者的连接执行,共享 skill 不等于共享权限。
|
||||||
|
|
||||||
**不选**:①zcbot 直连 Factory DB(绕过现有 RBAC/审计,只读仍可越权/拖垮主库);②固定几个查询模板(把 agent 降成菜单,无法利用 Factory 已有广泛 API);③直接复用 Factory `ichat` 自由 SQL 原型(字符串安全判断不构成边界,且使用默认 DB 凭据);④自动把相似问题生成并上线新代码工具(候选配方可自动生成,可执行能力仍需工具门控/人审)。
|
**不选**:①zcbot 直连 Factory DB(绕过现有 RBAC/审计,只读仍可越权/拖垮主库);②固定几个查询模板(把 agent 降成菜单,无法利用 Factory 已有广泛 API);③直接复用 Factory `ichat` 自由 SQL 原型(字符串安全判断不构成边界,且使用默认 DB 凭据);④自动把相似问题生成并上线新代码工具(候选配方可自动生成,可执行能力仍需工具门控/人审)。
|
||||||
|
|
||||||
|
### 8.15 OpenWorker 对照:动作治理与可恢复的人类介入(design,2026-08-06)
|
||||||
|
|
||||||
|
**调研结论**:OpenWorker 与 zcbot 同样采用单 agent loop、模型自由、文件产物、skill 渐进披露、持久会话、定时任务和外部连接器;这些不是 zcbot 的新增方向。真正值得借鉴的是它把「agent 能做什么」与「人此刻是否在线」拆成两条正交轴:统一权限引擎先按动作风险决定 allow/deny/ask,审批、问题、通知再进入同一个可持久化 attention inbox;人在 Web 或消息渠道任一端回应后,原调用幂等恢复。对 zcbot 的增量因此不是再造 agent/工作流框架,而是在现有 loop、scheduler、channel、external systems 之上补一层**动作治理控制面**。参考实现以 `andrewyng/openworker` 2026-08-06 的 `permissions.py`、`risk.py`、`inbox.py`、`audit.py`、`selfwake.py`、`automation/models.py`、`mcp/tools.py` 为样本;项目仍在 beta,只借心智与不变量,不复制其本地单机 JSON/SQLite 实现。
|
||||||
|
|
||||||
|
**P0 决策候选——统一 `ActionPolicy`**:所有有副作用的工具在执行前经过一个机械决策点,风险至少分四类:`read`(读文件/知识库/外部查询,默认允许)、`workspace_mutation`(用户工作目录内写产物,默认允许,继续由沙箱/路径边界/产物机检兜底)、`external_effect`(发邮件/微信、MES/LIMS 写入、创建远端对象,默认 ask)、`destructive_or_privileged`(删除远端对象、扩大授权、修改持久许可,始终 ask)。**不照搬 OpenWorker 的“shell/本地写默认逐次问”**:zcbot 的 shell 在 per-user sandbox 内且文档生产会高频写工作区,逐次审批只会制造确认疲劳;本层按 blast radius 而非工具名字分类。风险元数据由平台可信注册表声明,不能信 skill、OpenAPI/MCP 描述或模型自报;未知外部工具 fail closed。可再提供 read-only discuss/plan 模式,但它必须由同一策略层机械拒绝副作用,不能只靠 prompt。
|
||||||
|
|
||||||
|
**P0 决策候选——统一 Attention Inbox**:`ask_user` 继续服务「2-4 个方向选择、结束本轮等下一条用户消息」的轻交互;新增 attention item 服务「某个在途动作暂停后从原 tool_call 恢复」。PG 是唯一事实源,最小状态机 `pending -> resolved|expired|cancelled`,以 `(task_id, tool_call_id)` 唯一保证重连/重启不重复提问,resolve 用条件更新实现 first-responder-wins。item 至少记录 kind(approval/question/notification)、脱敏请求摘要、resolution、来源渠道和时间;Web SSE、企业微信/个人微信只是同一 item 的展示/响应 transport,不各存一份状态。删除 task、取消 run、授权被管理员撤回时确定性关闭关联 pending item;恢复前重跑权限判断,防等待期间策略或用户权限已经变化。**边界**:不能把待审批工具参数作为普通 user 文本让模型重新解释,批准的是被冻结且可校验的具体动作;凭据和完整敏感正文不入 item。
|
||||||
|
|
||||||
|
**P0 决策候选——外部动作审计**:新增独立 `action_audit_events`(不挤进回答费用口径的 `usage_events`,也不拿 toolfail 代替),记录 user/task/run、provider/tool/operation、风险级别、决策依据、approval item/rule、执行状态、目标资源标识、脱敏 args/result preview 与时间。审计回答「谁在什么任务中、凭哪条授权、对哪个对象做了什么、结果如何」;token/password/secret、邮件/消息正文、浏览器输入、外部响应全文机械脱敏或不落库。只读查询可按采样/高价值 operation 记,外部写与拒绝/审批必须全记。该项是 §8.14 从查询扩到写操作前的 hard prerequisite。
|
||||||
|
|
||||||
|
**P1——定时任务的精确目标长期授权**:无人值守任务不能靠“整个工具永远允许”。借鉴 OpenWorker 的 task-scoped standing rule,授权归具体 scheduled job,形态为 `provider + operation_id/tool + normalized_target`;删除/停用 job 或管理员撤权即失效。`normalized_target` 的组成字段由可信 provider definition 声明(如 recipient/channel_id/plant_id/dataset_id),模型不能自行挑字段,不支持通配符;创建任务时 consent card 同时展示将读取的数据与将写入的精确目标。shell、任意文件删除及无法提取稳定目标的动作不授长期许可,每次仍 ask。现有定时查询与确定性 notify 不受影响;只有未来开放外部写时才启用该契约。
|
||||||
|
|
||||||
|
**P1——显式 self-wake,修订 §8.13 的绝对边界但不改默认**:保留「后台进程完成后只通知人、不自动续跑」为默认;仅当用户明确要求连续科研闭环,或 agent 显式调用类似 `wake_on_process(proc_id)` 时,允许进程终态触发一次新 run,注入机械完成事件、exit code 和输出路径后继续分析。设每任务自动恢复次数、token/费用预算、截止时间与取消开关;日志仍由 agent 按需读,不把全文注入上下文;后续外部副作用照常进 Attention Inbox。目标场景是「启动模拟/拟合 -> 等完成 -> 检查收敛 -> 出图和结论」,不是通用 workflow/job graph。没有真实中间计算需求信号前不实施。
|
||||||
|
|
||||||
|
**P2——MCP 只作受控连接协议补充**:保留 §8.14 OpenAPI 元工具为院内 MES/ERP/LIMS 主入口;MCP 用于确有需求的标准 SaaS/第三方服务。服务器地址和 OAuth/密钥由管理员 definition 管,普通用户不能填任意 URL;启动/刷新时发现 schema 后仍过 pinned allowlist、风险元数据、响应体积与审计边界。**不照搬“一 MCP tool 一 JSON tool 全展开”**:继续复用 `external_system_search/call` 的延迟发现心智,把 MCP tool 映射为稳定 `provider/tool_id`,避免几十上百个 schema 常驻上下文和供应商新增能力后自动越权。协议内容、tool description 和返回值均是不可信数据。
|
||||||
|
|
||||||
|
**明确不借**:①Tauri 桌面壳、本地 secret store 和 JSON/SQLite 状态——OpenWorker 是个人单机,zcbot 是多用户 Web + PG + 蓝绿;②为展示广度铺 25+ 通用 SaaS connector——优先院内 MES/LIMS/设备/知识与企微的真实需求;③多 persona/多 agent 编排——职责隔离继续由 skill 承担,§6/§8.11 的证据门槛不变;④逐次审批沙箱 shell/工作区写入——确认疲劳且不增加外部 blast-radius 安全;⑤直接复制 beta 项目代码——并发、身份、持久化和恢复不变量不同。
|
||||||
|
|
||||||
|
**落地顺序/触发**:先 `ActionPolicy -> Attention Inbox(Web) -> 渠道响应 -> action audit`,四者作为一个完整外部写安全闭环;再按真实无人值守写需求增加 exact-target standing rule,按真实长计算中间步骤增加 self-wake;有明确第三方 MCP 接入对象后再做 adapter。任何阶段都不得先开放外部写、再用 prompt 要求模型“记得询问”补安全边界。实现时需回写 §3.1 loop、§7.5 sandbox/tool registry、§8.5 scheduler、§8.7 channel、§8.13 bg proc 与 §8.14 external systems 的最终契约,并以 migration 保持现有 API/数据兼容。
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## 附录:DeepSeek V4 关键事实(2026-04-24)
|
## 附录:DeepSeek V4 关键事实(2026-04-24)
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,7 @@
|
||||||
|
|
||||||
> 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。
|
> 配合 `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
|
### 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。
|
- **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。
|
||||||
|
|
|
||||||
|
|
@ -1,3 +1,3 @@
|
||||||
# zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。
|
# zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。
|
||||||
# 改版本只动这一行。
|
# 改版本只动这一行。
|
||||||
__version__ = "0.62.0"
|
__version__ = "0.62.1"
|
||||||
|
|
|
||||||
|
|
@ -17,6 +17,7 @@ from urllib.parse import quote, urljoin, urlparse
|
||||||
import httpx
|
import httpx
|
||||||
|
|
||||||
from .auth import ExternalAuthError, get_auth_strategy
|
from .auth import ExternalAuthError, get_auth_strategy
|
||||||
|
from .results import MAX_STORED_RESULT_BYTES
|
||||||
|
|
||||||
|
|
||||||
class OpenApiError(RuntimeError):
|
class OpenApiError(RuntimeError):
|
||||||
|
|
@ -26,6 +27,17 @@ class OpenApiError(RuntimeError):
|
||||||
_HTTP_METHODS = ("get", "head", "post", "put", "patch", "delete")
|
_HTTP_METHODS = ("get", "head", "post", "put", "patch", "delete")
|
||||||
_SPEC_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}
|
_SPEC_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}
|
||||||
_SPEC_LOCK = Lock()
|
_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:
|
def _bool_value(value: Any, default: bool) -> bool:
|
||||||
raw = str(value if value is not None else "").strip().lower()
|
raw = str(value if value is not None else "").strip().lower()
|
||||||
|
|
@ -217,6 +229,189 @@ class OpenApiClient:
|
||||||
})
|
})
|
||||||
return results
|
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:
|
def _spec_base_path(self, spec: dict[str, Any]) -> str:
|
||||||
"""Return the API path prefix declared by Swagger 2 / OpenAPI 3.
|
"""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
|
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.pop("request_body", None)
|
||||||
compact["recommended"] = recommended
|
compact["recommended"] = recommended
|
||||||
scored.append((
|
scored.append((
|
||||||
|
|
@ -411,20 +609,25 @@ class OpenApiClient:
|
||||||
except httpx.HTTPError as exc:
|
except httpx.HTTPError as exc:
|
||||||
raise OpenApiError(f"外部系统接口调用失败: {type(exc).__name__}") from exc
|
raise OpenApiError(f"外部系统接口调用失败: {type(exc).__name__}") from exc
|
||||||
if response.status_code >= 400:
|
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", "")
|
content_type = response.headers.get("content-type", "")
|
||||||
try:
|
try:
|
||||||
payload: Any = response.json() if "json" in content_type else response.text
|
payload: Any = response.json() if "json" in content_type else response.text
|
||||||
except ValueError:
|
except ValueError:
|
||||||
payload = response.text
|
payload = response.text
|
||||||
encoded = json.dumps(payload, ensure_ascii=False, default=str)
|
encoded = json.dumps(payload, ensure_ascii=False, default=str)
|
||||||
truncated = len(encoded.encode("utf-8")) > self.cfg.max_result_bytes
|
response_bytes = len(encoded.encode("utf-8"))
|
||||||
if truncated:
|
if response_bytes > MAX_STORED_RESULT_BYTES:
|
||||||
encoded = encoded.encode("utf-8")[: self.cfg.max_result_bytes].decode("utf-8", "ignore")
|
raise OpenApiError(
|
||||||
payload = encoded
|
f"外部系统响应超过安全下载上限({MAX_STORED_RESULT_BYTES} bytes),"
|
||||||
|
"请缩小查询范围或使用远端分页/聚合接口"
|
||||||
|
)
|
||||||
return {
|
return {
|
||||||
"operation_id": operation_id,
|
"operation_id": operation_id,
|
||||||
"status_code": response.status_code,
|
"status_code": response.status_code,
|
||||||
"truncated": truncated,
|
"truncated": False,
|
||||||
|
"response_bytes": response_bytes,
|
||||||
"data": payload,
|
"data": payload,
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -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}
|
||||||
|
|
@ -28,6 +28,7 @@ from tools.documents import DocumentDownloadTool, DocumentListKbTool, DocumentSe
|
||||||
from tools.external_systems import (
|
from tools.external_systems import (
|
||||||
ExternalSystemCallTool,
|
ExternalSystemCallTool,
|
||||||
ExternalSystemListTool,
|
ExternalSystemListTool,
|
||||||
|
ExternalSystemResultReadTool,
|
||||||
ExternalSystemSearchTool,
|
ExternalSystemSearchTool,
|
||||||
)
|
)
|
||||||
from tools.fs import EditTool, GlobTool, GrepTool, ReadTool, WriteTool
|
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:
|
def _external_systems() -> list:
|
||||||
|
result_budget: dict[str, int] = {}
|
||||||
return [
|
return [
|
||||||
ExternalSystemListTool(ctx.uid, **base),
|
ExternalSystemListTool(ctx.uid, **base),
|
||||||
ExternalSystemSearchTool(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:
|
def _load_skill() -> list:
|
||||||
|
|
|
||||||
|
|
@ -3,6 +3,7 @@ from __future__ import annotations
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
|
import tempfile
|
||||||
import unittest
|
import unittest
|
||||||
import uuid
|
import uuid
|
||||||
from copy import deepcopy
|
from copy import deepcopy
|
||||||
|
|
@ -59,6 +60,16 @@ def _cfg(*, allowed=frozenset(), recommended=()):
|
||||||
|
|
||||||
_SPEC = {
|
_SPEC = {
|
||||||
"swagger": "2.0",
|
"swagger": "2.0",
|
||||||
|
"definitions": {
|
||||||
|
"DataExec": {
|
||||||
|
"type": "object",
|
||||||
|
"required": ["query"],
|
||||||
|
"properties": {
|
||||||
|
"query": {"title": "查询字典参数", "type": "object"},
|
||||||
|
"is_test": {"type": "boolean", "default": False},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
},
|
||||||
"paths": {
|
"paths": {
|
||||||
"/api/bi/dataset/": {
|
"/api/bi/dataset/": {
|
||||||
"get": {
|
"get": {
|
||||||
|
|
@ -85,7 +96,12 @@ _SPEC = {
|
||||||
"summary": "执行只读数据集",
|
"summary": "执行只读数据集",
|
||||||
"parameters": [
|
"parameters": [
|
||||||
{"name": "code", "in": "path", "required": True, "type": "string"},
|
{"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]))
|
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):
|
def test_get_call_resolves_encoded_path_and_query(self):
|
||||||
from core.external_systems.factory import FactoryMesClient
|
from core.external_systems.factory import FactoryMesClient
|
||||||
|
|
||||||
|
|
@ -389,12 +469,60 @@ class FactoryOpenApiConnectorTests(unittest.TestCase):
|
||||||
result = allowed.call(
|
result = allowed.call(
|
||||||
"bi_dataset_exec",
|
"bi_dataset_exec",
|
||||||
arguments={"code": "quality"},
|
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]
|
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"])
|
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):
|
def test_rejects_unknown_arguments(self):
|
||||||
from core.external_systems.factory import FactoryMesClient, FactoryMesError
|
from core.external_systems.factory import FactoryMesClient, FactoryMesError
|
||||||
|
|
||||||
|
|
@ -446,23 +574,64 @@ class ExternalSystemToolSafetyTests(unittest.TestCase):
|
||||||
self.assertEqual(payload["query_guidance"], "统计查询先查看数据集目录")
|
self.assertEqual(payload["query_guidance"], "统计查询先查看数据集目录")
|
||||||
self.assertEqual(payload["recommended_operation_ids"], ["bi_dataset_list"])
|
self.assertEqual(payload["recommended_operation_ids"], ["bi_dataset_list"])
|
||||||
|
|
||||||
def test_call_discards_result_over_per_run_external_budget(self):
|
def test_large_call_spills_and_result_reader_pages_without_data_loss(self):
|
||||||
from tools.external_systems import ExternalSystemCallTool
|
from tools.external_systems import (
|
||||||
|
ExternalSystemCallTool,
|
||||||
|
ExternalSystemResultReadTool,
|
||||||
|
)
|
||||||
|
|
||||||
uid = uuid.uuid4()
|
uid = uuid.uuid4()
|
||||||
|
task_id = uuid.uuid4()
|
||||||
|
system_id = str(uuid.uuid4())
|
||||||
|
budget: dict[str, int] = {}
|
||||||
client = SimpleNamespace(
|
client = SimpleNamespace(
|
||||||
cfg=SimpleNamespace(max_total_result_bytes=32),
|
cfg=SimpleNamespace(max_result_bytes=512, max_total_result_bytes=4096),
|
||||||
call=lambda *args, **kwargs: {"data": "x" * 100},
|
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",
|
"tools.external_systems._row_and_client",
|
||||||
return_value=(SimpleNamespace(), client),
|
return_value=(SimpleNamespace(), client),
|
||||||
):
|
):
|
||||||
output = ExternalSystemCallTool(uid).execute(
|
call_tool = ExternalSystemCallTool(
|
||||||
str(uuid.uuid4()), "detail_list"
|
uid,
|
||||||
|
task_id=task_id,
|
||||||
|
result_budget=budget,
|
||||||
|
base_dir=Path(tmp),
|
||||||
)
|
)
|
||||||
self.assertIn("累计返回量超过上限", output)
|
spilled = json.loads(call_tool.execute(system_id, "detail_list"))
|
||||||
self.assertNotIn("x" * 20, output)
|
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__":
|
if __name__ == "__main__":
|
||||||
|
|
|
||||||
|
|
@ -5,6 +5,13 @@ import json
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
|
|
||||||
from core.external_systems.factory import FactoryMesError
|
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 (
|
from core.external_systems.service import (
|
||||||
ExternalSystemError,
|
ExternalSystemError,
|
||||||
client_for_external_system,
|
client_for_external_system,
|
||||||
|
|
@ -86,6 +93,7 @@ class ExternalSystemCallTool(Tool):
|
||||||
description = (
|
description = (
|
||||||
"调用已连接外部系统的受控只读 OpenAPI operation。必须使用 search 返回的 operation_id;"
|
"调用已连接外部系统的受控只读 OpenAPI operation。必须使用 search 返回的 operation_id;"
|
||||||
"不接受 URL。GET/HEAD 默认可用,POST 仅限管理员声明的只读 operation。"
|
"不接受 URL。GET/HEAD 默认可用,POST 仅限管理员声明的只读 operation。"
|
||||||
|
"大响应会完整保存并返回 result_ref,使用 external_system_result_read 分段读取。"
|
||||||
)
|
)
|
||||||
parameters = {
|
parameters = {
|
||||||
"type": "object",
|
"type": "object",
|
||||||
|
|
@ -99,16 +107,72 @@ class ExternalSystemCallTool(Tool):
|
||||||
},
|
},
|
||||||
"body": {
|
"body": {
|
||||||
"type": "object",
|
"type": "object",
|
||||||
"description": "仅对管理员放行的只读 POST 操作提供 JSON body"
|
"description": (
|
||||||
|
"仅对管理员放行的只读 POST 操作提供原始 JSON 请求体;"
|
||||||
|
"严格遵循 search 返回的 body.schema,不要按 Swagger body 参数名再包一层"
|
||||||
|
)
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
"required": ["system_id", "operation_id"],
|
"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)
|
super().__init__(**kwargs)
|
||||||
self.user_id = user_id
|
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(
|
def execute(
|
||||||
self,
|
self,
|
||||||
|
|
@ -123,19 +187,116 @@ class ExternalSystemCallTool(Tool):
|
||||||
used = self._result_bytes.get(system_id, 0)
|
used = self._result_bytes.get(system_id, 0)
|
||||||
if used >= client.cfg.max_total_result_bytes:
|
if used >= client.cfg.max_total_result_bytes:
|
||||||
return (
|
return (
|
||||||
"[Error] 本轮外部系统返回量已达上限。请改用聚合接口或 dataset,"
|
"[Error] 本轮外部系统内联返回量已达上限。"
|
||||||
"不要继续分页拉取日志/明细。"
|
"请在下一轮继续查询,或读取之前返回的 result_ref。"
|
||||||
)
|
)
|
||||||
result = client.call(operation_id, arguments=arguments, body=body)
|
result = client.call(operation_id, arguments=arguments, body=body)
|
||||||
result_size = len(_json(result).encode("utf-8"))
|
return self._bounded_output(
|
||||||
used += result_size
|
system_id,
|
||||||
self._result_bytes[system_id] = used
|
result,
|
||||||
if used > client.cfg.max_total_result_bytes:
|
per_result_limit=client.cfg.max_result_bytes,
|
||||||
return (
|
total_limit=client.cfg.max_total_result_bytes,
|
||||||
"[Error] 本轮外部系统累计返回量超过上限,当前结果已丢弃。"
|
|
||||||
"请改用聚合接口或 dataset,并缩小查询范围。"
|
|
||||||
)
|
)
|
||||||
return _json(result)
|
except (ExternalSystemError, FactoryMesError, ExternalResultError) as exc:
|
||||||
except (ExternalSystemError, FactoryMesError) as exc:
|
|
||||||
print(f"[WARN] external system call failed: {type(exc).__name__}")
|
print(f"[WARN] external system call failed: {type(exc).__name__}")
|
||||||
return f"[Error] {exc}"
|
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}"
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue