fix: allow concurrent and unrestricted Runtime sessions

Local embedded OpenCode serialized otherwise independent conversations, and GoodBuddy could stop long Runtime tasks after fixed tool, activity, model-call, or accumulated display-output limits.

Different conversations can now run concurrently while requests in the same conversation stay ordered. Dynamic MCP tools remain isolated to the current request. OpenCode, Continue, managed ACP, and the Agent model gateway no longer stop native execution at GoodBuddy-defined call counts or prompt-wide output totals; display copies truncate and transport queues apply backpressure instead. Prompt-wide quota fields are removed from the current contracts.

Per-call model settings, provider request timeouts, cancellation, and same-conversation ordering remain unchanged.

Release note: 修复本机 OpenCode 不同会话互相等待的问题,并移除 GoodBuddy 对单次 Runtime 工具、活动和模型调用次数的人为上限;长输出现在只截断展示或施加传输背压,不再取消原生任务。
This commit is contained in:
mesalogo
2026-09-03 17:17:09 +08:00
parent b1e3101857
commit 8b9dbf55df
32 changed files with 687 additions and 549 deletions
+3
View File
@@ -128,6 +128,9 @@ otherwise.
Workspace/Runtime is first used, then reused by other projects in the same Workspace/Runtime is first used, then reused by other projects in the same
process. Managed SSH conversations expose only supported OpenCode choices. process. Managed SSH conversations expose only supported OpenCode choices.
Multiple projects and conversations can run concurrently on one Host; Multiple projects and conversations can run concurrently on one Host;
Agent-owned prompts are not stopped at a fixed model-call count or
prompt-wide output-token total, and Runtime output uses transport
backpressure instead of cancellation when GoodBuddy buffers fill;
recovery separately reports network, Agent, Runtime, committed-event cursor, recovery separately reports network, Agent, Runtime, committed-event cursor,
completion, or failure/retry and blocks only the affected project. A real completion, or failure/retry and blocks only the affected project. A real
Linux x64 Host has covered short and long detach, forced local harness exit, Linux x64 Host has covered short and long detach, forced local harness exit,
+1 -1
View File
@@ -29,7 +29,7 @@
- [x] **直连模型 Runtime**:支持问答、知识总结、受控工具执行和图像生成。 - [x] **直连模型 Runtime**:支持问答、知识总结、受控工具执行和图像生成。
- [x] **直连模型编程 Agent**:本机直连文本模型可在 Execute 模式运行平台 Shell,并可按父请求模式、模型、工作区和能力范围委派一层编程 Subagent;OpenCode、Continue、DeepSeek Harness 和托管 SSH 不重复注入这两个工具。Windows 本机命令与真实模型“修改、测试、修复、复核”闭环已通过,macOS 与 Linux 真机命令仍待对应平台验收。 - [x] **直连模型编程 Agent**:本机直连文本模型可在 Execute 模式运行平台 Shell,并可按父请求模式、模型、工作区和能力范围委派一层编程 Subagent;OpenCode、Continue、DeepSeek Harness 和托管 SSH 不重复注入这两个工具。Windows 本机命令与真实模型“修改、测试、修复、复核”闭环已通过,macOS 与 Linux 真机命令仍待对应平台验收。
- [x] **OpenCode 与 Continue**:使用隔离子进程、环境变量白名单、统一配置、取消、总执行时限、有界流式输出和活动记录;共享进程回收逻辑保留 Windows 完整进程树终止,并对采用独立进程组的 POSIX 子进程执行组回收。聊天状态检查使用一次性探测 Runtime 并在返回后立即回收,不进入按模型与项目隔离的执行 Runtime 缓存;执行 Runtime 仍按项目复用。本机 GoodBuddy 管理的 OpenCode 只对同一会话内的请求保持顺序,不同会话即使属于同一项目也可并行,不同项目同样可并行;请求级动态 MCP 通过默认通配禁用和当前请求精确放行保持隔离。OpenCode 与 Continue 不再按固定工具调用数或活动数中止单次运行,本机展示副本达到容量时只截断后续展示,不取消原生任务。交互提问只由前台对话回答,定时任务、远程通道和委派等后台执行遇到提问时会立即失败并提示改为前台运行,避免无限等待。本机 OpenCode 启动时直接使用已选择的绝对路径,不预先检查文件或运行 `--version`,路径失效时由实际启动返回错误;托管 Linux ARM Host 激活 Runtime 时复用准备阶段已验证的 registry 和 manifest,不重新哈希或检查完整 OpenCode 二进制。 - [x] **OpenCode 与 Continue**:使用隔离子进程、环境变量白名单、统一配置、取消、总执行时限、有界流式输出和活动记录;共享进程回收逻辑保留 Windows 完整进程树终止,并对采用独立进程组的 POSIX 子进程执行组回收。聊天状态检查使用一次性探测 Runtime 并在返回后立即回收,不进入按模型与项目隔离的执行 Runtime 缓存;执行 Runtime 仍按项目复用。本机 GoodBuddy 管理的 OpenCode 只对同一会话内的请求保持顺序,不同会话即使属于同一项目也可并行,不同项目同样可并行;请求级动态 MCP 通过默认通配禁用和当前请求精确放行保持隔离。OpenCode 与 Continue 不再按固定工具调用数或活动数中止单次运行,本机展示副本达到容量时只截断后续展示,不取消原生任务。交互提问只由前台对话回答,定时任务、远程通道和委派等后台执行遇到提问时会立即失败并提示改为前台运行,避免无限等待。本机 OpenCode 启动时直接使用已选择的绝对路径,不预先检查文件或运行 `--version`,路径失效时由实际启动返回错误;托管 Linux ARM Host 激活 Runtime 时复用准备阶段已验证的 registry 和 manifest,不重新哈希或检查完整 OpenCode 二进制。
- [x] **托管 SSH OpenCode 闭环(技术预览)**:该能力由“设置 > 平台功能”中的独立“远程项目(技术预览)”页签控制并默认关闭;关闭时不影响任何本地项目、普通桌面功能或桌面发布。启用后可管理固定 Host Key 的 SSH Host、浏览有界远端目录并创建 Ask/Execute 项目;两种模式都直接启动签名 Runtime,不依赖 bubblewrap。Ask 在 Agent 工具分发边界只允许读取工具,Execute 使用所选 SSH 账号的完整权限。Agent 通过私有 Unix socket 和 ACP v4 持有 accepted Prompt、Provider/工具轮次、Runtime 进程、稳定模型 ledger 与有界语义 transcript;Desktop 退出、网络中断或本机进程结束后 Host 继续执行,重启后只 attach 原 controller/binding/operation 并把 provenance 原子归并回原对话,不重发 Prompt。模型 profile 与 API Key 只在当前 accepted operation 生命周期内进入 Agent 内存,不写入 Renderer、SSH 参数、远端环境或磁盘。远端组件不嵌入桌面包,项目切换只更新本地选择且不会连接或下载;设置页只读取小型签名目录并按 Linux x64/arm64 显示本地版本、在线最新兼容版本和“有更新/已是最新”,用户点击后才下载签名 `.gbagent`,也可导入/导出离线包。每个复合包包含 Agent、固定 Node 与由桌面源码统一维护适配的 OpenCode Runtime;在线来源跟随“关于与更新”的 GitHub/北京 OSS 选择,签名累计目录绑定最低桌面版本、Agent 协议、架构、大小、SHA-256 和固定 URL。公钥 registry 接受等价 JSON 空白和换行,但仍严格校验 schema、Ed25519 key、环境与撤销;目录、包、manifest、payload 签名和流式 SHA-256 不变。缺少某架构包只使该架构托管 SSH 不可用。项目只保存 Host、远端路径、Runtime 选择和模式;首次实际使用 Workspace/Runtime 时读取 Host current identity,同一进程内其他项目复用已确认 identity 和 Agent 连接。托管 SSH 会话只显示支持的 OpenCode 选择。同一 Host 可并行运行多个项目和对话;恢复进度分别显示网络、Agent、Runtime、已提交事件 cursor、完成或失败/重试,只阻塞受影响项目。Linux x64 真实 Host 已覆盖短/长 detach、本机 harness 强制结束、并发、SSH relay 丢失、取消、确定/不确定 Provider 失败、Agent `SIGKILL`/重启和 reopened Desktop SQLite 恢复;成功工具 START/END 各一次,未发现 Prompt、Provider 或工具重放。当前 Agent 源码 lock 为 `0.11.16`,Desktop 发布候选为 `0.12.2`;正式发布状态以 Agent 与 Desktop 的独立发布渠道为准。 - [x] **托管 SSH OpenCode 闭环(技术预览)**:该能力由“设置 > 平台功能”中的独立“远程项目(技术预览)”页签控制并默认关闭;关闭时不影响任何本地项目、普通桌面功能或桌面发布。启用后可管理固定 Host Key 的 SSH Host、浏览有界远端目录并创建 Ask/Execute 项目;两种模式都直接启动签名 Runtime,不依赖 bubblewrap。Ask 在 Agent 工具分发边界只允许读取工具,Execute 使用所选 SSH 账号的完整权限。Agent 通过私有 Unix socket 和 ACP v4 持有 accepted Prompt、Provider/工具轮次、Runtime 进程、稳定模型 ledger 与有界语义 transcript;Desktop 退出、网络中断或本机进程结束后 Host 继续执行,重启后只 attach 原 controller/binding/operation 并把 provenance 原子归并回原对话,不重发 Prompt。模型 profile 与 API Key 只在当前 accepted operation 生命周期内进入 Agent 内存,不写入 Renderer、SSH 参数、远端环境或磁盘。远端组件不嵌入桌面包,项目切换只更新本地选择且不会连接或下载;设置页只读取小型签名目录并按 Linux x64/arm64 显示本地版本、在线最新兼容版本和“有更新/已是最新”,用户点击后才下载签名 `.gbagent`,也可导入/导出离线包。每个复合包包含 Agent、固定 Node 与由桌面源码统一维护适配的 OpenCode Runtime;在线来源跟随“关于与更新”的 GitHub/北京 OSS 选择,签名累计目录绑定最低桌面版本、Agent 协议、架构、大小、SHA-256 和固定 URL。公钥 registry 接受等价 JSON 空白和换行,但仍严格校验 schema、Ed25519 key、环境与撤销;目录、包、manifest、payload 签名和流式 SHA-256 不变。缺少某架构包只使该架构托管 SSH 不可用。项目只保存 Host、远端路径、Runtime 选择和模式;首次实际使用 Workspace/Runtime 时读取 Host current identity,同一进程内其他项目复用已确认 identity 和 Agent 连接。托管 SSH 会话只显示支持的 OpenCode 选择。同一 Host 可并行运行多个项目和对话;Agent-owned Prompt 不因固定模型调用次数或 Prompt 累计输出 Token 数停止,GoodBuddy 输出缓冲区达到容量时使用传输背压而不取消 Runtime;恢复进度分别显示网络、Agent、Runtime、已提交事件 cursor、完成或失败/重试,只阻塞受影响项目。Linux x64 真实 Host 已覆盖短/长 detach、本机 harness 强制结束、并发、SSH relay 丢失、取消、确定/不确定 Provider 失败、Agent `SIGKILL`/重启和 reopened Desktop SQLite 恢复;成功工具 START/END 各一次,未发现 Prompt、Provider 或工具重放。当前 Agent 源码 lock 为 `0.11.16`,Desktop 发布候选为 `0.12.2`;正式发布状态以 Agent 与 Desktop 的独立发布渠道为准。
- [x] **SSH Host 手动环境准备源码链路**:Host Key、认证和系统探针成功后先保存 Host,并只读探测共享 Agent/Runtime;保存 Host、打开项目都不自动安装。Host 卡片只有一个按版本事实显示“安装远程环境”“更新远程环境”或“重新安装”的主按钮,次级 SegmentedControl 选择默认且不持久化的“自动”、Host 下载或 GoodBuddy 传输;“版本匹配”badge 不等同环境健康。自动模式只在 operation/prepare 前探测并择一,显式选择保持有效,任何 prepare、commit 或 adoption 失败都不跨 acquisition 自动 fallback。两种方式把同一签名 compound `.gbagent` 交付到固定 staging 后,共用 control-plane prepare、commit、Agent activate/health、Runtime activate、finalize 与显式 cleanup。GoodBuddy 路径可在同一次操作下载并验证缺失候选、缓存并取得 lease,再有界流式 SFTP 上传一个归档和其中已验证的 bootstrap Node,不把约 294 MiB 整包读入 Main `Buffer`;Host 在解包时完成一次完整 payload 校验。未完成操作只记录暂存 cleanup 所需的 operation ID;下次更新尽力清理旧暂存后重新 prepare,不保存远端 metadata 副本,也不让 cleanup 失败阻塞新更新或回滚健康环境。已有项目在实际使用 Workspace/Runtime 时按需解析 Host current identity 并执行固定 `attach-or-bootstrap`,注册后的 health、capabilities 和 prompt 启动不扫描完整 payload。详见 [设计说明](./docs/features/remote-host/environment-provisioning-technical-design.md)。 - [x] **SSH Host 手动环境准备源码链路**:Host Key、认证和系统探针成功后先保存 Host,并只读探测共享 Agent/Runtime;保存 Host、打开项目都不自动安装。Host 卡片只有一个按版本事实显示“安装远程环境”“更新远程环境”或“重新安装”的主按钮,次级 SegmentedControl 选择默认且不持久化的“自动”、Host 下载或 GoodBuddy 传输;“版本匹配”badge 不等同环境健康。自动模式只在 operation/prepare 前探测并择一,显式选择保持有效,任何 prepare、commit 或 adoption 失败都不跨 acquisition 自动 fallback。两种方式把同一签名 compound `.gbagent` 交付到固定 staging 后,共用 control-plane prepare、commit、Agent activate/health、Runtime activate、finalize 与显式 cleanup。GoodBuddy 路径可在同一次操作下载并验证缺失候选、缓存并取得 lease,再有界流式 SFTP 上传一个归档和其中已验证的 bootstrap Node,不把约 294 MiB 整包读入 Main `Buffer`;Host 在解包时完成一次完整 payload 校验。未完成操作只记录暂存 cleanup 所需的 operation ID;下次更新尽力清理旧暂存后重新 prepare,不保存远端 metadata 副本,也不让 cleanup 失败阻塞新更新或回滚健康环境。已有项目在实际使用 Workspace/Runtime 时按需解析 Host current identity 并执行固定 `attach-or-bootstrap`,注册后的 health、capabilities 和 prompt 启动不扫描完整 payload。详见 [设计说明](./docs/features/remote-host/environment-provisioning-technical-design.md)。
- [ ] **SSH Host 环境准备真实 Host 验收**:源码已可使用当前 format v1 包;仍需完成 GitHub、北京镜像、Linux x64/arm64、取消和离线 GoodBuddy 传输的真实 Host 验收。当前状态不表示已发布或已完成真实 Host 测试。 - [ ] **SSH Host 环境准备真实 Host 验收**:源码已可使用当前 format v1 包;仍需完成 GitHub、北京镜像、Linux x64/arm64、取消和离线 GoodBuddy 传输的真实 Host 验收。当前状态不表示已发布或已完成真实 Host 测试。
- [x] **DeepSeek Harness(预览)**:使用 GoodBuddy 固定 Host 和 OpenAI 兼容模型连接;优先使用管理员提供的连接,否则跟随兼容的默认模型或首个兼容连接,无需单独重复选择。Ask 只允许调用 Host 中真实注册的 `read`、`skill` 以及 Main 管理的 Web Search/Fetch 代理,拒绝插件同名冒充,Execute 放行全部已启用内置及插件工具,并以当前用户权限运行。图像输入跟随所选模型连接的能力声明,文本模型在 Host 或模型调用前拒绝图片,图片模型通过有界内联内容和临时 Attachment Store 接收 JPEG/PNG。 - [x] **DeepSeek Harness(预览)**:使用 GoodBuddy 固定 Host 和 OpenAI 兼容模型连接;优先使用管理员提供的连接,否则跟随兼容的默认模型或首个兼容连接,无需单独重复选择。Ask 只允许调用 Host 中真实注册的 `read`、`skill` 以及 Main 管理的 Web Search/Fetch 代理,拒绝插件同名冒充,Execute 放行全部已启用内置及插件工具,并以当前用户权限运行。图像输入跟随所选模型连接的能力声明,文本模型在 Host 或模型调用前拒绝图片,图片模型通过有界内联内容和临时 Attachment Store 接收 JPEG/PNG。
+16 -9
View File
@@ -202,10 +202,11 @@ Execute 直接启动已签名 Runtime:
- 继承 SSH 账号的正常环境、文件系统、进程和网络能力; - 继承 SSH 账号的正常环境、文件系统、进程和网络能力;
- 不进行 T2/T3、confinement attestation、approval bridge 或逐工具批准。 - 不进行 T2/T3、confinement attestation、approval bridge 或逐工具批准。
两种模式都保留输入/输出字节上限、用户取消和进程组清理。生产 Prompt 不设置固定墙钟 两种模式都保留输入字节上限、用户取消和进程组清理;输出只用有界内存队列与 journal
总时限;只要 Runtime 尚未按自身协议结束,Main 和 Agent 就允许其持续运行。启动、握手、 配合背压,不按 Prompt 累计输出量停止 Runtime。生产 Prompt 不设置固定墙钟总时限;
单个控制 RPC、重连尝试和关闭清理仍使用独立的有界超时,测试可以显式注入短 Prompt 只要 Runtime 尚未按自身协议结束,Main 和 Agent 就允许其持续运行。启动、握手、单个
时限验证取消升级,但该覆盖值不是生产默认。无界 Prompt 继续复用已发布的必填 控制 RPC、重连尝试和关闭清理仍使用独立的有界超时,测试可以显式注入短 Prompt 时限
验证取消升级,但该覆盖值不是生产默认。无界 Prompt 继续复用已发布的必填
`deadlineAt` 字段,以 year-9999 哨兵表达;Main 与 Agent 必须同时理解该哨兵,Agent `deadlineAt` 字段,以 year-9999 哨兵表达;Main 与 Agent 必须同时理解该哨兵,Agent
`0.11.10` 及更早版本仍会按其 manifest 上限夹紧,因此桌面启用这一语义前必须配套更新 `0.11.10` 及更早版本仍会按其 manifest 上限夹紧,因此桌面启用这一语义前必须配套更新
到 Agent `0.11.13` 或更高版本。 到 Agent `0.11.13` 或更高版本。
@@ -234,8 +235,9 @@ Execute 直接启动已签名 Runtime:
- 无法确认外部 Provider 是否已处理的模型调用保持结果未知,避免重复计费或重复副作用。 - 无法确认外部 Provider 是否已处理的模型调用保持结果未知,避免重复计费或重复副作用。
- 只有远端返回 identity 匹配且 `closed: true` 时,Main 才删除持久化 binding;传输失败或未确认 close 会保留 recovery identity。 - 只有远端返回 identity 匹配且 `closed: true` 时,Main 才删除持久化 binding;传输失败或未确认 close 会保留 recovery identity。
- 终态 close 由 Agent 在一个事务中为两个方向记录 sequence high-water tombstone,删除剩余 frame 和 active channel,并归零对应 journal quota。tombstone 用于拒绝迟到的旧 epoch frame,因此当前不会按时间自动裁剪。 - 终态 close 由 Agent 在一个事务中为两个方向记录 sequence high-water tombstone,删除剩余 frame 和 active channel,并归零对应 journal quota。tombstone 用于拒绝迟到的旧 epoch frame,因此当前不会按时间自动裁剪。
- 断线不主动终止正在运行的 Runtime。用户取消、显式关闭、输出边界或 Agent shutdown - 断线不主动终止正在运行的 Runtime。用户取消、显式关闭或 Agent shutdown 才触发停止;
才触发停止;连接恢复只受当前用户取消信号和单次重连控制超时约束,不把 Prompt 已运行 输出队列和 journal 达到容量时暂停读取并等待 ACK 释放容量,不把展示或传输容量转换为
Runtime 取消。连接恢复只受当前用户取消信号和单次重连控制超时约束,不把 Prompt 已运行
时长当作停止条件。 时长当作停止条件。
- Agent 失联期间继续把 Runtime 输出写入本机 journal;重新连接只校验 controller - Agent 失联期间继续把 Runtime 输出写入本机 journal;重新连接只校验 controller
identity、恢复 binding,并从 Main 最后确认的 cursor 后同步缺失数据。已接受的 identity、恢复 binding,并从 Main 最后确认的 cursor 后同步缺失数据。已接受的
@@ -285,9 +287,10 @@ Execute 直接启动已签名 Runtime:
各限制 768 KiB,响应同时核对声明长度与短读。每个 各限制 768 KiB,响应同时核对声明长度与短读。每个
`(binding, operation, roundIndex)` 只 dispatch 一次;已 dispatch 但未证明完整交付的调用 `(binding, operation, roundIndex)` 只 dispatch 一次;已 dispatch 但未证明完整交付的调用
变为 `outcome-unknown`,不会自动重试。 变为 `outcome-unknown`,不会自动重试。
- Prompt 最多 100 个模型调用,并受单次与累计输出 Token 上限保护;这些是资源边界,不是 - Agent 不按固定模型调用次数或 Prompt 累计输出 Token 数停止 Runtime;每次 Provider
Prompt 墙钟总时限。Provider 单次请求仍有独立超时。实际 usage 在语义 transcript 中同步 请求仍遵循所选模型 profile 的单次输出设置和独立请求超时。模型桥协议不包含 Prompt
回 Desktop 的 usage/task 数据。Agent 在所有进程终止路径统一清除 Prompt token 计数。 调用次数或累计 Token 配额字段。实际 usage 在语义 transcript 中同步回 Desktop 的
usage/task 数据。
- helper 可以接收同一 Prompt 内并发到达的模型桥请求;它在单一稳定模型桥上按到达 - helper 可以接收同一 Prompt 内并发到达的模型桥请求;它在单一稳定模型桥上按到达
顺序等待并交付,不返回本地 `bridge-busy`,每个响应只有在 HTTP 完整 flush 后才 顺序等待并交付,不返回本地 `bridge-busy`,每个响应只有在 HTTP 完整 flush 后才
发送 delivery ACK。 发送 delivery ACK。
@@ -400,6 +403,10 @@ model bridge,以及生命周期和恢复逻辑。单元测试、mock、fixture
都未经过外部进程隔离命令。Runtime owner 和隔离目录均已清理,未修改 Host current 都未经过外部进程隔离命令。Runtime owner 和隔离目录均已清理,未修改 Host current
Agent/Runtime。本次只验证能力声明和真实进程启动,没有发送 Prompt,真实文本模型调用 Agent/Runtime。本次只验证能力声明和真实进程启动,没有发送 Prompt,真实文本模型调用
0 次。 0 次。
- 2026-09-03 使用包含本次模型轮次与输出背压修复的当前源码 Agent,在共享 Linux x64
Host 的唯一 `/tmp/goodbuddy-e2e-*` 目录完成 Runtime 激活和 Agent 本机模型 gateway
验证。模型 gateway 通过所选 Anthropic 连接返回预期随机哨兵,HTTP 状态为 200,实际
文本模型调用恰好 1 次;Host 临时目录随后删除,共享 Agent/Runtime 安装和进程未修改。
- 正常 Host 更新路径把 Linux x64 Host 的 Agent 更新为 `0.11.2-e2e.12`,并确认 - 正常 Host 更新路径把 Linux x64 Host 的 Agent 更新为 `0.11.2-e2e.12`,并确认
OpenCode Runtime 已安装版本与所需版本均为 `1.18.9`。 OpenCode Runtime 已安装版本与所需版本均为 `1.18.9`。
- 一条新的 Ask 用户操作只提交一次。OpenCode 先在 build 模型轮次请求一个原生 - 一条新的 Ask 用户操作只提交一次。OpenCode 先在 build 模型轮次请求一个原生
+28 -2
View File
@@ -36,8 +36,6 @@ const profile: AgentPromptModelProfile = {
capabilities: { imageInput: false }, capabilities: { imageInput: false },
limits: { limits: {
maximumOutputTokens: 4_096, maximumOutputTokens: 4_096,
maximumModelCalls: 8,
maximumTotalOutputTokens: 32_768,
requestTimeoutMilliseconds: 5_000 requestTimeoutMilliseconds: 5_000
} }
} }
@@ -124,6 +122,34 @@ describe('AgentModelGateway', () => {
ledger.close() ledger.close()
}) })
it('does not impose prompt-wide model-call or output-token quotas', async () => {
const fetcher = vi.fn<typeof fetch>(async () =>
new Response(
JSON.stringify({
id: 'response',
usage: { output_tokens: 4_096 }
}),
{
status: 200,
headers: { 'content-type': 'application/json' }
}
)
)
const { gateway, ledger } = setup(fetcher)
for (let roundIndex = 0; roundIndex < 101; roundIndex += 1) {
const result = await gateway.dispatch(
{ ...context, roundIndex },
request,
new AbortController().signal
)
await result.acknowledgeDelivery()
}
expect(fetcher).toHaveBeenCalledTimes(101)
ledger.close()
})
it('caps OpenCode chat-completions max_tokens before dispatch', async () => { it('caps OpenCode chat-completions max_tokens before dispatch', async () => {
const fetcher = vi.fn<typeof fetch>(async (_url, init) => { const fetcher = vi.fn<typeof fetch>(async (_url, init) => {
expect(JSON.parse(String(init?.body))).toMatchObject({ expect(JSON.parse(String(init?.body))).toMatchObject({
+1 -50
View File
@@ -313,7 +313,6 @@ export type AgentModelGatewayContext = {
export class AgentModelGateway { export class AgentModelGateway {
readonly #ledger: AgentModelCallLedger readonly #ledger: AgentModelCallLedger
readonly #fetch: typeof fetch readonly #fetch: typeof fetch
readonly #promptOutputTokens = new Map<string, number>()
constructor(options: { constructor(options: {
ledger: AgentModelCallLedger ledger: AgentModelCallLedger
@@ -334,24 +333,10 @@ export class AgentModelGateway {
}> { }> {
signal.throwIfAborted() signal.throwIfAborted()
const request = remoteModelGatewayRequestSchema.parse(requestInput) const request = remoteModelGatewayRequestSchema.parse(requestInput)
const promptKey = `${context.bindingId}\0${context.operationId}`
const usedOutputTokens = this.#promptOutputTokens.get(promptKey) ?? 0
const remainingOutputTokens =
context.profile.limits.maximumTotalOutputTokens - usedOutputTokens
if (remainingOutputTokens < 1) {
throw new AgentModelGatewayError(
'policy',
'Prompt total output token limit is exhausted'
)
}
const reservedOutputTokens = Math.min(
context.profile.limits.maximumOutputTokens,
remainingOutputTokens
)
const prepared = prepareProviderRequest( const prepared = prepareProviderRequest(
context.profile, context.profile,
request, request,
reservedOutputTokens context.profile.limits.maximumOutputTokens
) )
const requestDigest = sha256(prepared.canonicalRequest) const requestDigest = sha256(prepared.canonicalRequest)
const callId = createAgentModelCallId(context) const callId = createAgentModelCallId(context)
@@ -444,14 +429,6 @@ export class AgentModelGateway {
headers, headers,
bodyBase64: Buffer.from(bytes).toString('base64') bodyBase64: Buffer.from(bytes).toString('base64')
}) })
this.#promptOutputTokens.set(
promptKey,
usedOutputTokens +
Math.min(
reservedOutputTokens,
responseOutputTokens(bytes) ?? reservedOutputTokens
)
)
this.#ledger.complete(callId) this.#ledger.complete(callId)
let finalized = false let finalized = false
return { return {
@@ -474,9 +451,6 @@ export class AgentModelGateway {
} }
} }
finalizePrompt(bindingId: string, operationId: string): void {
this.#promptOutputTokens.delete(`${bindingId}\0${operationId}`)
}
} }
export function createAgentModelCallId( export function createAgentModelCallId(
@@ -598,29 +572,6 @@ function prepareProviderRequest(
} }
} }
function responseOutputTokens(body: Uint8Array): number | undefined {
let value: unknown
try {
value = JSON.parse(new TextDecoder().decode(body))
} catch {
return undefined
}
if (value === null || typeof value !== 'object' || Array.isArray(value)) {
return undefined
}
const usage = (value as { usage?: unknown }).usage
if (usage === null || typeof usage !== 'object' || Array.isArray(usage)) {
return undefined
}
const record = usage as Record<string, unknown>
const tokens = record.output_tokens ?? record.completion_tokens
return typeof tokens === 'number' &&
Number.isSafeInteger(tokens) &&
tokens >= 0
? tokens
: undefined
}
function providerUrl(profile: AgentPromptModelProfile): URL { function providerUrl(profile: AgentPromptModelProfile): URL {
const url = new URL(profile.baseUrl) const url = new URL(profile.baseUrl)
const path = url.pathname.replace(/\/+$/u, '') const path = url.pathname.replace(/\/+$/u, '')
@@ -254,6 +254,56 @@ describe('direct Linux stdio Runtime ownership', () => {
} }
}) })
it('uses stdout backpressure instead of stopping when output queues fill', async () => {
const registry = createRegistry()
const child = fakeChild()
const launched = identity()
const sendSignal = vi.fn<typeof process.kill>(() => true)
let releaseFirst!: () => void
const firstConsumed = new Promise<void>((resolve) => {
releaseFirst = resolve
})
const owner = await launchDirectLinuxStdioProcessOwner({
manifest: manifest(),
profile: profile(),
identity: { launchId: 'launch-output', processId: 'process-output' },
installationId: 'installation-1',
registry,
deadlineAt: UNBOUNDED_REMOTE_PROMPT_DEADLINE,
maximumInputBytes: 1024,
maximumOutputQueueChunks: 1,
platform: 'linux',
spawn: () => {
queueMicrotask(() => child.emit('spawn'))
return child
},
randomOwnerToken: () => 'a'.repeat(32),
readProcessIdentity: async () => launched,
sendSignal
})
const pause = vi.spyOn(child.stdout, 'pause')
const resume = vi.spyOn(child.stdout, 'resume')
const received: string[] = []
owner.subscribeOutput(async ({ data }) => {
received.push(Buffer.from(data).toString())
if (received.length === 1) {
await firstConsumed
}
})
child.stdout.write('one')
await vi.waitFor(() => expect(pause).toHaveBeenCalled())
child.stdout.write('two')
releaseFirst()
await vi.waitFor(() => expect(received).toEqual(['one', 'two']))
expect(resume).toHaveBeenCalled()
expect(sendSignal).not.toHaveBeenCalled()
child.exitCode = 0
child.emit('close', 0, null)
registry.close()
})
it('startup reconciles only owners from the current installation', async () => { it('startup reconciles only owners from the current installation', async () => {
const registry = createRegistry() const registry = createRegistry()
const reservation = { const reservation = {
@@ -170,13 +170,13 @@ export class DirectLinuxStdioProcessOwner {
#maximumPendingStdinBytes: number #maximumPendingStdinBytes: number
#inputBytes = 0 #inputBytes = 0
#pendingInputBytes = 0 #pendingInputBytes = 0
#outputBytes = 0
#queuedOutputBytes = 0 #queuedOutputBytes = 0
#writeTail: Promise<void> = Promise.resolve() #writeTail: Promise<void> = Promise.resolve()
#deadlineTimer?: NodeJS.Timeout #deadlineTimer?: NodeJS.Timeout
#promptGeneration = 1 #promptGeneration = 1
#promptActive = true #promptActive = true
#draining = false #draining = false
#outputPaused = false
#closed = false #closed = false
#stdinClosed = false #stdinClosed = false
#stopPromise?: Promise<void> #stopPromise?: Promise<void>
@@ -280,7 +280,6 @@ export class DirectLinuxStdioProcessOwner {
this.#maximumPendingStdinBytes = this.#maximumInputBytes this.#maximumPendingStdinBytes = this.#maximumInputBytes
this.#inputBytes = 0 this.#inputBytes = 0
this.#pendingInputBytes = 0 this.#pendingInputBytes = 0
this.#outputBytes = 0
this.#promptActive = true this.#promptActive = true
this.#promptGeneration += 1 this.#promptGeneration += 1
this.#scheduleDeadline( this.#scheduleDeadline(
@@ -449,24 +448,17 @@ export class DirectLinuxStdioProcessOwner {
#acceptOutput(chunk: Buffer | Uint8Array): void { #acceptOutput(chunk: Buffer | Uint8Array): void {
if (this.#closed || chunk.byteLength === 0) return if (this.#closed || chunk.byteLength === 0) return
if (
this.#outputBytes + chunk.byteLength > this.#maximumOutputBytes ||
this.#outputQueue.length >= this.#maximumOutputQueueChunks ||
this.#queuedOutputBytes + chunk.byteLength > this.#maximumOutputBytes
) {
this.#child.stdout.pause()
this.#outputQueue.splice(0)
this.#queuedOutputBytes = 0
void this.stop({
reason: 'output-quota',
deadlineAt: new Date(this.#now() + DEFAULT_STOP_TIMEOUT_MS).toISOString()
}).catch(() => undefined)
return
}
const copy = Uint8Array.from(chunk) const copy = Uint8Array.from(chunk)
this.#outputBytes += copy.byteLength
this.#queuedOutputBytes += copy.byteLength this.#queuedOutputBytes += copy.byteLength
this.#outputQueue.push(copy) this.#outputQueue.push(copy)
if (
this.#outputQueue.length >= this.#maximumOutputQueueChunks ||
(this.#maximumOutputBytes > 0 &&
this.#queuedOutputBytes >= this.#maximumOutputBytes)
) {
this.#child.stdout.pause()
this.#outputPaused = true
}
this.#scheduleDrain() this.#scheduleDrain()
} }
@@ -484,6 +476,15 @@ export class DirectLinuxStdioProcessOwner {
for (const listener of [...this.#outputListeners]) { for (const listener of [...this.#outputListeners]) {
await listener({ stream: 'stdout', data: data.slice() }) await listener({ stream: 'stdout', data: data.slice() })
} }
if (
this.#outputPaused &&
this.#outputQueue.length < this.#maximumOutputQueueChunks &&
(this.#maximumOutputBytes === 0 ||
this.#queuedOutputBytes < this.#maximumOutputBytes)
) {
this.#outputPaused = false
this.#child.stdout.resume()
}
} }
} catch { } catch {
void this.stop({ void this.stop({
+79 -5
View File
@@ -7,6 +7,7 @@ import {
import type { RemoteRuntimeBundleManifest } from '../shared/remote-runtime-launch-contracts' import type { RemoteRuntimeBundleManifest } from '../shared/remote-runtime-launch-contracts'
import type { ControllerLease } from './controller-registry' import type { ControllerLease } from './controller-registry'
import { AgentModelGatewayError } from './agent-model-gateway' import { AgentModelGatewayError } from './agent-model-gateway'
import { EventJournalCapacityError } from './event-journal'
import type { ProtocolMethodContext } from './protocol-server' import type { ProtocolMethodContext } from './protocol-server'
import type { import type {
ModelBridgeBrokerDispatch, ModelBridgeBrokerDispatch,
@@ -43,8 +44,16 @@ class MemoryAcpJournal implements RuntimeAcpJournal {
}> = [] }> = []
mainAckSequence = '0' mainAckSequence = '0'
failAppend = false failAppend = false
capacityBlocked = false
appendAcpFrame(input: (typeof this.frames)[number]): { created: boolean } { appendAcpFrame(input: (typeof this.frames)[number]): { created: boolean } {
if (this.capacityBlocked) {
throw new EventJournalCapacityError(
'journal capacity',
'stream',
false
)
}
if (this.failAppend) { if (this.failAppend) {
throw new Error('journal capacity') throw new Error('journal capacity')
} }
@@ -415,8 +424,6 @@ function harness(input: {
capabilities: { imageInput: false }, capabilities: { imageInput: false },
limits: { limits: {
maximumOutputTokens: 4_096, maximumOutputTokens: 4_096,
maximumModelCalls: 100,
maximumTotalOutputTokens: 409_600,
requestTimeoutMilliseconds: 60_000 requestTimeoutMilliseconds: 60_000
} }
}, },
@@ -541,6 +548,52 @@ describe('RuntimeAcpBackend', () => {
expect(fixture.blobSink).not.toHaveBeenCalled() expect(fixture.blobSink).not.toHaveBeenCalled()
}) })
it('allows more than 100 Agent-local model calls in one prompt', async () => {
const dispatch = vi.fn(async () => ({
response: {
status: 200,
headers: {},
bodyBase64: Buffer.from('{}').toString('base64')
},
acknowledgeDelivery: vi.fn(async () => undefined),
failDelivery: vi.fn()
}))
const fixture = harness({
agentOwned: true,
modelGateway: {
dispatch,
finalizePrompt: vi.fn()
}
})
await open(fixture)
await invoke(
fixture,
'runtime/preparePrompt',
fixture.preparation()
)
for (let index = 0; index < 101; index += 1) {
const exchange = await fixture.bridgeDispatch!(
{
method: 'POST',
path: '/v1/responses',
headers: { 'content-type': 'application/json' },
bodyBase64: Buffer.from('{}').toString('base64')
},
{
requestId: `provider-request-${index}`,
signal: new AbortController().signal
}
)
if (!('acknowledgeDelivery' in exchange)) {
throw new Error('Expected an acknowledged model exchange')
}
await exchange.acknowledgeDelivery()
}
expect(dispatch).toHaveBeenCalledTimes(101)
})
it('runs open, prepare, ACP delivery, durable output, cursors, and close', async () => { it('runs open, prepare, ACP delivery, durable output, cursors, and close', async () => {
const fixture = harness() const fixture = harness()
const opened = await invoke( const opened = await invoke(
@@ -1162,7 +1215,7 @@ describe('RuntimeAcpBackend', () => {
} }
}) })
it('stops the process when its output quota is exceeded', async () => { it('does not stop the process when cumulative output exceeds the display budget', async () => {
const fixture = harness({ maximumOutputBytes: 3 }) const fixture = harness({ maximumOutputBytes: 3 })
await open(fixture) await open(fixture)
await invoke( await invoke(
@@ -1172,9 +1225,30 @@ describe('RuntimeAcpBackend', () => {
) )
await expect( await expect(
fixture.process.emit('stdout', 'four') fixture.process.emit('stdout', 'four')
).rejects.toMatchObject({ code: 'output-quota' }) ).resolves.toBeUndefined()
expect(fixture.process.stops).toEqual(['output-quota']) expect(fixture.process.stops).toEqual([])
expect(fixture.outputFrames).toEqual(['four'])
})
it('waits for durable journal capacity instead of stopping Runtime output', async () => {
const fixture = harness()
await open(fixture)
await invoke(
fixture,
'runtime/preparePrompt',
fixture.preparation()
)
fixture.journal.capacityBlocked = true
const output = fixture.process.emit('stdout', 'later')
await new Promise<void>((resolve) => setTimeout(resolve, 10))
expect(fixture.process.stops).toEqual([])
expect(fixture.outputFrames).toEqual([]) expect(fixture.outputFrames).toEqual([])
fixture.journal.capacityBlocked = false
await output
expect(fixture.process.stops).toEqual([])
expect(fixture.outputFrames).toEqual(['later'])
}) })
it('stops with output quota when the detached durable journal is exhausted', async () => { it('stops with output quota when the detached durable journal is exhausted', async () => {
+34 -57
View File
@@ -64,6 +64,7 @@ import {
type AgentModelGateway type AgentModelGateway
} from './agent-model-gateway' } from './agent-model-gateway'
import { openCodeModelBridgeModelId } from './model-bridge-helper' import { openCodeModelBridgeModelId } from './model-bridge-helper'
import { EventJournalCapacityError } from './event-journal'
const ZERO_CURSOR = '0' const ZERO_CURSOR = '0'
const DEFAULT_MAXIMUM_BINDINGS = 128 const DEFAULT_MAXIMUM_BINDINGS = 128
@@ -82,6 +83,7 @@ const DEFAULT_MAXIMUM_OPERATIONS_PER_BINDING = 1_000
const DEFAULT_DISPOSE_TIMEOUT_MS = 10_000 const DEFAULT_DISPOSE_TIMEOUT_MS = 10_000
const OWNED_PROMPT_CANCEL_GRACE_MS = 1_000 const OWNED_PROMPT_CANCEL_GRACE_MS = 1_000
const OWNED_PROMPT_START_TIMEOUT_MS = 2 * 60_000 const OWNED_PROMPT_START_TIMEOUT_MS = 2 * 60_000
const ACP_JOURNAL_RETRY_MILLISECONDS = 25
export type RuntimeAcpBackendErrorCode = export type RuntimeAcpBackendErrorCode =
| 'capacity' | 'capacity'
@@ -317,8 +319,6 @@ type BindingState = {
activeOperationId?: string activeOperationId?: string
terminalState?: 'completed' | 'failed' | 'cancelled' | 'interrupted' terminalState?: 'completed' | 'failed' | 'cancelled' | 'interrupted'
inputBytes: number inputBytes: number
outputBytes: number
maximumOutputBytes?: number
nextOutputSequence: bigint nextOutputSequence: bigint
pendingInput: number pendingInput: number
pendingOutput: number pendingOutput: number
@@ -721,8 +721,6 @@ export class RuntimeAcpBackend {
existing.stopRequested = false existing.stopRequested = false
existing.terminalState = undefined existing.terminalState = undefined
existing.inputBytes = 0 existing.inputBytes = 0
existing.outputBytes = 0
existing.maximumOutputBytes = undefined
existing.nextOutputSequence = 1n existing.nextOutputSequence = 1n
existing.pendingInput = 0 existing.pendingInput = 0
existing.pendingOutput = 0 existing.pendingOutput = 0
@@ -774,7 +772,6 @@ export class RuntimeAcpBackend {
stopRequested: false, stopRequested: false,
operations: new Map(), operations: new Map(),
inputBytes: 0, inputBytes: 0,
outputBytes: 0,
nextOutputSequence: 1n, nextOutputSequence: 1n,
pendingInput: 0, pendingInput: 0,
pendingOutput: 0, pendingOutput: 0,
@@ -1234,11 +1231,6 @@ export class RuntimeAcpBackend {
} }
} }
binding.inputBytes = 0 binding.inputBytes = 0
binding.outputBytes = 0
binding.maximumOutputBytes = Math.min(
preparation.budget.maximumOutputBytes,
verified.manifest.limits.maximumPromptOutputBytes
)
if (!binding.operations.has(preparation.operationId)) { if (!binding.operations.has(preparation.operationId)) {
binding.operations.set(preparation.operationId, { binding.operations.set(preparation.operationId, {
preparationDigest, preparationDigest,
@@ -1308,10 +1300,6 @@ export class RuntimeAcpBackend {
) )
} }
await this.#closeModelBridge(binding, true) await this.#closeModelBridge(binding, true)
this.#options.modelGateway?.finalizePrompt(
binding.request.bindingId,
operation.operationId
)
if (binding.poisoned) { if (binding.poisoned) {
throw new RuntimeAcpBackendError( throw new RuntimeAcpBackendError(
'Model response delivery was not proven before prompt completion', 'Model response delivery was not proven before prompt completion',
@@ -1336,8 +1324,6 @@ export class RuntimeAcpBackend {
prepared.completion = completion prepared.completion = completion
binding.activeOperationId = undefined binding.activeOperationId = undefined
binding.inputBytes = 0 binding.inputBytes = 0
binding.outputBytes = 0
binding.maximumOutputBytes = undefined
if (binding.deadlineTimer !== undefined) { if (binding.deadlineTimer !== undefined) {
clearTimeout(binding.deadlineTimer) clearTimeout(binding.deadlineTimer)
binding.deadlineTimer = undefined binding.deadlineTimer = undefined
@@ -1527,10 +1513,6 @@ export class RuntimeAcpBackend {
return return
} }
await this.#closeModelBridge(binding, true) await this.#closeModelBridge(binding, true)
this.#options.modelGateway?.finalizePrompt(
binding.request.bindingId,
operationId
)
try { try {
await binding.process.completePrompt() await binding.process.completePrompt()
} catch { } catch {
@@ -1568,8 +1550,6 @@ export class RuntimeAcpBackend {
prepared.modelProfile = undefined prepared.modelProfile = undefined
binding.activeOperationId = undefined binding.activeOperationId = undefined
binding.inputBytes = 0 binding.inputBytes = 0
binding.outputBytes = 0
binding.maximumOutputBytes = undefined
if (binding.deadlineTimer !== undefined) { if (binding.deadlineTimer !== undefined) {
clearTimeout(binding.deadlineTimer) clearTimeout(binding.deadlineTimer)
binding.deadlineTimer = undefined binding.deadlineTimer = undefined
@@ -1673,12 +1653,6 @@ export class RuntimeAcpBackend {
'closed' 'closed'
) )
} }
if (binding.nextModelRound >= profile.limits.maximumModelCalls) {
throw new RuntimeAcpBackendError(
'Prompt model call limit reached',
'capacity'
)
}
const roundIndex = binding.nextModelRound const roundIndex = binding.nextModelRound
let exchanged let exchanged
try { try {
@@ -1990,21 +1964,6 @@ export class RuntimeAcpBackend {
) )
} }
this.#assertBeforeDeadline(binding) this.#assertBeforeDeadline(binding)
const budget = this.#activeBudget(binding)
const maximumOutputBytes = Math.min(
budget.maximumOutputBytes,
binding.maximumOutputBytes ?? budget.maximumOutputBytes
)
if (binding.outputBytes + output.data.byteLength > maximumOutputBytes) {
await this.#stopAndReconcile(binding, 'output-quota').catch(
() => undefined
)
throw new RuntimeAcpBackendError(
'Runtime output quota reached',
'output-quota'
)
}
binding.outputBytes += output.data.byteLength
for ( for (
let offset = 0; let offset = 0;
offset < output.data.byteLength; offset < output.data.byteLength;
@@ -2017,14 +1976,31 @@ export class RuntimeAcpBackend {
try { try {
await this.#enqueueOutput(binding, payload, async () => { await this.#enqueueOutput(binding, payload, async () => {
const sequence = binding.nextOutputSequence.toString() const sequence = binding.nextOutputSequence.toString()
this.#options.journal.appendAcpFrame({ while (true) {
controllerId: binding.controllerId, try {
bindingId: binding.request.bindingId, this.#options.journal.appendAcpFrame({
channelEpoch: binding.openResult.channelEpoch, controllerId: binding.controllerId,
direction: 'runtime-to-main', bindingId: binding.request.bindingId,
sequence, channelEpoch: binding.openResult.channelEpoch,
payload direction: 'runtime-to-main',
}) sequence,
payload
})
break
} catch (error) {
if (!(error instanceof EventJournalCapacityError)) {
throw error
}
if (binding.state !== 'running') {
throw new RuntimeAcpBackendError(
'Runtime output arrived after the binding stopped',
'closed'
)
}
this.#assertBeforeDeadline(binding)
await delay(ACP_JOURNAL_RETRY_MILLISECONDS)
}
}
binding.nextOutputSequence += 1n binding.nextOutputSequence += 1n
if (binding.transportState === 'live') { if (binding.transportState === 'live') {
try { try {
@@ -2287,12 +2263,6 @@ export class RuntimeAcpBackend {
} }
#finishProcess(binding: BindingState): void { #finishProcess(binding: BindingState): void {
if (binding.activeOperationId !== undefined) {
this.#options.modelGateway?.finalizePrompt(
binding.request.bindingId,
binding.activeOperationId
)
}
if (binding.process !== undefined) { if (binding.process !== undefined) {
this.#processIdentityBindings.delete( this.#processIdentityBindings.delete(
processIdentityKey(binding.process.identity) processIdentityKey(binding.process.identity)
@@ -2750,6 +2720,13 @@ function processIdentityKey(identity: RuntimeAcpProcessIdentity): string {
].join('\0') ].join('\0')
} }
async function delay(milliseconds: number): Promise<void> {
await new Promise<void>((resolve) => {
const timer = setTimeout(resolve, milliseconds)
timer.unref?.()
})
}
function digestSecretBearingPreparation( function digestSecretBearingPreparation(
preparation: RemotePromptOperationPreparation preparation: RemotePromptOperationPreparation
): string { ): string {
+33 -21
View File
@@ -1544,7 +1544,7 @@ describe('AcpRemoteRuntime', () => {
}) })
}) })
it('bounds native subagent output per event and per request', async () => { it('bounds native subagent output per event without a request-wide stop', async () => {
const client: { value?: AgentSideConnection } = {} const client: { value?: AgentSideConnection } = {}
const server = fakeServer({ const server = fakeServer({
prompt: async ({ sessionId }) => { prompt: async ({ sessionId }) => {
@@ -1620,10 +1620,12 @@ describe('AcpRemoteRuntime', () => {
async () => 'once' async () => 'once'
) )
) )
).rejects.toThrow('远端 Runtime 请求输出超过安全上限') ).resolves.toContainEqual(
expect.objectContaining({ type: 'done' })
)
}) })
it('rejects more than 100 distinct remote tool calls', async () => { it('accepts more than 100 distinct remote tool calls', async () => {
const client: { value?: AgentSideConnection } = {} const client: { value?: AgentSideConnection } = {}
const server = fakeServer({ const server = fakeServer({
prompt: async ({ sessionId }) => { prompt: async ({ sessionId }) => {
@@ -1634,7 +1636,7 @@ describe('AcpRemoteRuntime', () => {
sessionUpdate: 'tool_call', sessionUpdate: 'tool_call',
toolCallId: `call-${index}`, toolCallId: `call-${index}`,
title: 'tool', title: 'tool',
status: 'in_progress' status: 'completed'
} }
}) })
} }
@@ -1643,17 +1645,18 @@ describe('AcpRemoteRuntime', () => {
}) })
client.value = server.client client.value = server.client
await expect( const events = await collect(
collect( runtime(server).run(
runtime(server).run( request,
request, new AbortController().signal,
new AbortController().signal, async () => 'once'
async () => 'once'
)
) )
).rejects.toThrow(
'远端 Runtime 单次运行的工具调用超过 100 个'
) )
expect(
events.filter((event) => event.type === 'tool')
).toHaveLength(101)
expect(events.at(-1)).toMatchObject({ type: 'done' })
}) })
it('does not impose a default wall-clock deadline on remote prompts', async () => { it('does not impose a default wall-clock deadline on remote prompts', async () => {
@@ -2235,7 +2238,7 @@ describe('AcpRemoteRuntime', () => {
]) ])
}) })
it('cancels and truncates a flood when the consumer is slow', async () => { it('backpressures a flood without cancelling the Runtime', async () => {
const store = new MemoryRuntimeSessionBindingStore() const store = new MemoryRuntimeSessionBindingStore()
const client: { value?: AgentSideConnection } = {} const client: { value?: AgentSideConnection } = {}
const cancel = vi.fn(async () => {}) const cancel = vi.fn(async () => {})
@@ -2276,15 +2279,26 @@ describe('AcpRemoteRuntime', () => {
done: false done: false
}) })
await new Promise((resolve) => setTimeout(resolve, 10)) await new Promise((resolve) => setTimeout(resolve, 10))
await expect(stream.next()).rejects.toThrow('输出已截断') const events = []
while (true) {
const next = await stream.next()
if (next.done) {
break
}
events.push(next.value)
}
expect(
events.filter((event) => event.type === 'text')
).toHaveLength(20)
expect(events.at(-1)).toMatchObject({ type: 'done' })
expect(server.setInboundPaused).toHaveBeenCalledWith(true) expect(server.setInboundPaused).toHaveBeenCalledWith(true)
expect(server.setInboundPaused).not.toHaveBeenCalledWith(false) expect(server.setInboundPaused).toHaveBeenCalledWith(false)
expect(cancel).toHaveBeenCalledOnce() expect(cancel).not.toHaveBeenCalled()
await expect( await expect(
store.getByConversation(request.conversationId) store.getByConversation(request.conversationId)
).resolves.toMatchObject({ ).resolves.toMatchObject({
state: 'outcome-unknown', state: 'ready',
activePromptOperationId: request.requestId activePromptOperationId: undefined
}) })
}) })
@@ -2504,8 +2518,6 @@ describe('AcpRemoteRuntime Agent-owned prompts', () => {
capabilities: { imageInput: false }, capabilities: { imageInput: false },
limits: { limits: {
maximumOutputTokens: 4_096, maximumOutputTokens: 4_096,
maximumModelCalls: 100,
maximumTotalOutputTokens: 409_600,
requestTimeoutMilliseconds: 60_000 requestTimeoutMilliseconds: 60_000
} }
} }
+4 -55
View File
@@ -15,9 +15,6 @@ import {
import type { import type {
AgentRuntimeStatus AgentRuntimeStatus
} from '../../shared/contracts' } from '../../shared/contracts'
import {
maximumConversationToolActivities
} from '../../shared/assistant-contracts'
import { import {
assertRemotePromptAcceptanceMatchesPreparation, assertRemotePromptAcceptanceMatchesPreparation,
REMOTE_SEMANTIC_TRANSCRIPT_LIMITS, REMOTE_SEMANTIC_TRANSCRIPT_LIMITS,
@@ -148,7 +145,6 @@ type ActivePrompt = {
updates: QueuedSessionUpdate[] updates: QueuedSessionUpdate[]
pendingUpdateBytes: number pendingUpdateBytes: number
inboundPaused: boolean inboundPaused: boolean
outputCharacters: number
toolCalls: Map< toolCalls: Map<
string, string,
{ {
@@ -1846,19 +1842,6 @@ export class AcpRemoteRuntime implements AgentRuntime {
this.options.maxPendingUpdateBytes ?? this.options.maxPendingUpdateBytes ??
DEFAULT_MAX_PENDING_UPDATE_BYTES DEFAULT_MAX_PENDING_UPDATE_BYTES
) )
if (
prompt.updates.length >= maximumItems ||
prompt.pendingUpdateBytes + bytes > maximumBytes
) {
const overflow = new Error(
'远端 Runtime 待消费事件超过安全上限,输出已截断且请求状态未知'
)
prompt.interruption = overflow
prompt.interrupt?.(overflow)
prompt.wake?.()
prompt.wake = undefined
throw overflow
}
let consume: (() => void) | undefined let consume: (() => void) | undefined
const consumed = new Promise<void>((resolve) => { const consumed = new Promise<void>((resolve) => {
consume = resolve consume = resolve
@@ -2030,19 +2013,13 @@ export class AcpRemoteRuntime implements AgentRuntime {
}) })
} }
private limitEvent( private limitEvent(event: RuntimePublicEvent): RuntimePublicEvent {
prompt: ActivePrompt,
event: RuntimePublicEvent
): RuntimePublicEvent {
const maximum = const maximum =
this.options.maxEventCharacters ?? this.options.maxEventCharacters ??
DEFAULT_MAX_EVENT_CHARACTERS DEFAULT_MAX_EVENT_CHARACTERS
let characters = 0
if (event.type === 'text' || event.type === 'reasoning') { if (event.type === 'text' || event.type === 'reasoning') {
characters = event.delta.length if (event.delta.length > maximum) {
if (characters > maximum) {
event = { ...event, delta: `${event.delta.slice(0, maximum)}…` } event = { ...event, delta: `${event.delta.slice(0, maximum)}…` }
characters = maximum + 1
} }
} else if (event.type === 'tool') { } else if (event.type === 'tool') {
const fieldMaximum = Math.max(1, Math.floor(maximum / 3)) const fieldMaximum = Math.max(1, Math.floor(maximum / 3))
@@ -2056,10 +2033,6 @@ export class AcpRemoteRuntime implements AgentRuntime {
output: bounded(event.output), output: bounded(event.output),
error: bounded(event.error) error: bounded(event.error)
} }
characters =
(event.input?.length ?? 0) +
(event.output?.length ?? 0) +
(event.error?.length ?? 0)
} else if (event.type === 'subagent') { } else if (event.type === 'subagent') {
const fieldMaximum = Math.max(1, Math.floor(maximum / 2)) const fieldMaximum = Math.max(1, Math.floor(maximum / 2))
const bounded = (value: string | undefined): string | undefined => const bounded = (value: string | undefined): string | undefined =>
@@ -2071,10 +2044,6 @@ export class AcpRemoteRuntime implements AgentRuntime {
output: bounded(event.output), output: bounded(event.output),
error: bounded(event.error) error: bounded(event.error)
} }
characters =
(event.reason?.length ?? 0) +
(event.output?.length ?? 0) +
(event.error?.length ?? 0)
} else if (event.type === 'status') { } else if (event.type === 'status') {
if (event.message.length > maximum) { if (event.message.length > maximum) {
event = { event = {
@@ -2082,15 +2051,6 @@ export class AcpRemoteRuntime implements AgentRuntime {
message: `${event.message.slice(0, maximum)}…` message: `${event.message.slice(0, maximum)}…`
} }
} }
characters = event.message.length
}
prompt.outputCharacters += characters
if (
prompt.outputCharacters >
(this.options.maxRequestOutputBytes ??
DEFAULT_MAX_REQUEST_OUTPUT_BYTES)
) {
throw new Error('远端 Runtime 请求输出超过安全上限')
} }
return event return event
} }
@@ -2120,15 +2080,6 @@ export class AcpRemoteRuntime implements AgentRuntime {
) { ) {
const callId = update.toolCallId.slice(0, 256) const callId = update.toolCallId.slice(0, 256)
const previous = prompt.toolCalls.get(callId) const previous = prompt.toolCalls.get(callId)
if (
!previous &&
prompt.toolCalls.size >=
maximumConversationToolActivities
) {
throw new Error(
'远端 Runtime 单次运行的工具调用超过 100 个'
)
}
const retainedMaximum = const retainedMaximum =
this.options.maxEventCharacters ?? this.options.maxEventCharacters ??
DEFAULT_MAX_EVENT_CHARACTERS DEFAULT_MAX_EVENT_CHARACTERS
@@ -2359,7 +2310,6 @@ export class AcpRemoteRuntime implements AgentRuntime {
updates: [], updates: [],
pendingUpdateBytes: 0, pendingUpdateBytes: 0,
inboundPaused: false, inboundPaused: false,
outputCharacters: 0,
toolCalls: new Map(), toolCalls: new Map(),
open: true, open: true,
completed: false, completed: false,
@@ -2559,7 +2509,7 @@ export class AcpRemoteRuntime implements AgentRuntime {
) )
const mapped = this.mapUpdate(prompt, notification.update) const mapped = this.mapUpdate(prompt, notification.update)
if (mapped !== undefined) { if (mapped !== undefined) {
publicEvents.push(this.limitEvent(prompt, mapped)) publicEvents.push(this.limitEvent(mapped))
} }
} else if (transcriptEvent.kind === 'prompt-terminal') { } else if (transcriptEvent.kind === 'prompt-terminal') {
const terminal = transcriptTerminal( const terminal = transcriptTerminal(
@@ -2764,7 +2714,6 @@ export class AcpRemoteRuntime implements AgentRuntime {
updates: [], updates: [],
pendingUpdateBytes: 0, pendingUpdateBytes: 0,
inboundPaused: false, inboundPaused: false,
outputCharacters: 0,
toolCalls: new Map(), toolCalls: new Map(),
open: true, open: true,
completed: false, completed: false,
@@ -2867,7 +2816,7 @@ export class AcpRemoteRuntime implements AgentRuntime {
} }
const event = this.mapUpdate(prompt, update) const event = this.mapUpdate(prompt, update)
if (event) { if (event) {
yield this.limitEvent(prompt, event) yield this.limitEvent(event)
} }
} }
await this.awaitOperation( await this.awaitOperation(
+70 -49
View File
@@ -1176,7 +1176,7 @@ describe('ContinueHostAdapter', () => {
expect(killed).toBe(true) expect(killed).toBe(true)
}) })
it('fails when the patched host reports dropped stream events', async () => { it('completes from final history when the patched stream is truncated', async () => {
const distribution = await createDistribution() const distribution = await createDistribution()
let stateRequests = 0 let stateRequests = 0
vi.stubGlobal( vi.stubGlobal(
@@ -1185,11 +1185,23 @@ describe('ContinueHostAdapter', () => {
if (String(input).endsWith('/state')) { if (String(input).endsWith('/state')) {
stateRequests += 1 stateRequests += 1
return Response.json({ return Response.json({
session: { history: [] }, session: {
isProcessing: stateRequests > 1, history:
stateRequests === 1
? []
: [
{
message: {
role: 'assistant',
content: 'CONTINUE_OK'
}
}
]
},
isProcessing: false,
messageQueueLength: 0, messageQueueLength: 0,
pendingPermission: null, pendingPermission: null,
goodbuddyEventsOverflow: stateRequests > 1 goodbuddyEventsOverflow: stateRequests === 2
}) })
} }
return Response.json({}) return Response.json({})
@@ -1224,27 +1236,15 @@ describe('ContinueHostAdapter', () => {
new AbortController().signal, new AbortController().signal,
async () => 'deny' async () => 'deny'
) )
).rejects.toThrow('流式事件超过安全限制') ).resolves.toMatchObject({
text: 'CONTINUE_OK',
streamTruncated: true
})
}) })
it.each([ it(
{ 'truncates cumulative streamed bytes without stopping the run',
label: 'event count', async () => {
limits: {
maximumStreamEvents: 1,
maximumStreamEventBytes: 10_000
}
},
{
label: 'event bytes',
limits: {
maximumStreamEvents: 10,
maximumStreamEventBytes: 60
}
}
])(
'enforces cumulative streamed $label across state polls',
async ({ limits }) => {
const distribution = await createDistribution() const distribution = await createDistribution()
let stateRequests = 0 let stateRequests = 0
vi.stubGlobal( vi.stubGlobal(
@@ -1253,8 +1253,20 @@ describe('ContinueHostAdapter', () => {
if (String(input).endsWith('/state')) { if (String(input).endsWith('/state')) {
stateRequests += 1 stateRequests += 1
return Response.json({ return Response.json({
session: { history: [] }, session: {
isProcessing: stateRequests > 1, history:
stateRequests < 3
? []
: [
{
message: {
role: 'assistant',
content: '1234567890CONTINUE_OK'
}
}
]
},
isProcessing: stateRequests === 2,
messageQueueLength: 0, messageQueueLength: 0,
pendingPermission: null, pendingPermission: null,
goodbuddyEvents: goodbuddyEvents:
@@ -1291,8 +1303,7 @@ describe('ContinueHostAdapter', () => {
} }
}, },
{ {
...limits, maximumStreamEventBytes: 60,
maximumToolCalls: 100,
terminateProcessTree: vi.fn().mockResolvedValue(undefined) terminateProcessTree: vi.fn().mockResolvedValue(undefined)
} }
) )
@@ -1308,14 +1319,18 @@ describe('ContinueHostAdapter', () => {
} }
} }
) )
).rejects.toThrow('流式事件超过安全限制') ).resolves.toMatchObject({
text: '1234567890CONTINUE_OK',
streamedText: true,
streamTruncated: true
})
expect(forwarded).toEqual([ expect(forwarded).toEqual([
{ type: 'text', delta: '1234567890' } { type: 'text', delta: '1234567890' }
]) ])
} }
) )
it('enforces cumulative unique tool calls before forwarding a later batch', async () => { it('forwards more than 100 unique tool calls', async () => {
const distribution = await createDistribution() const distribution = await createDistribution()
let stateRequests = 0 let stateRequests = 0
vi.stubGlobal( vi.stubGlobal(
@@ -1324,20 +1339,30 @@ describe('ContinueHostAdapter', () => {
if (String(input).endsWith('/state')) { if (String(input).endsWith('/state')) {
stateRequests += 1 stateRequests += 1
return Response.json({ return Response.json({
session: { history: [] }, session: {
isProcessing: stateRequests > 1, history:
stateRequests === 1
? []
: [
{
message: {
role: 'assistant',
content: 'TOOLS_OK'
}
}
]
},
isProcessing: false,
messageQueueLength: 0, messageQueueLength: 0,
pendingPermission: null, pendingPermission: null,
goodbuddyEvents: goodbuddyEvents:
stateRequests > 1 stateRequests === 2
? [ ? Array.from({ length: 101 }, (_, index) => ({
{ type: 'tool',
type: 'tool', callId: `call-${index + 1}`,
callId: `call-${stateRequests}`, name: 'Bash',
name: 'Bash', state: 'completed'
state: 'running' }))
}
]
: [] : []
}) })
} }
@@ -1369,9 +1394,7 @@ describe('ContinueHostAdapter', () => {
} }
}, },
{ {
maximumStreamEvents: 10, maximumStreamEventBytes: 1_000_000,
maximumStreamEventBytes: 10_000,
maximumToolCalls: 1,
terminateProcessTree: vi.fn().mockResolvedValue(undefined) terminateProcessTree: vi.fn().mockResolvedValue(undefined)
} }
) )
@@ -1387,11 +1410,11 @@ describe('ContinueHostAdapter', () => {
} }
} }
) )
).rejects.toThrow('工具调用超过 100 个') ).resolves.toMatchObject({ text: 'TOOLS_OK' })
expect(forwarded).toHaveLength(1) expect(forwarded).toHaveLength(101)
expect(forwarded[0]).toMatchObject({ expect(forwarded.at(-1)).toMatchObject({
type: 'tool', type: 'tool',
tool: { callId: 'call-2' } tool: { callId: 'call-101' }
}) })
}) })
@@ -1461,9 +1484,7 @@ describe('ContinueHostAdapter', () => {
} }
}, },
{ {
maximumStreamEvents: 10,
maximumStreamEventBytes: 10_000, maximumStreamEventBytes: 10_000,
maximumToolCalls: 10,
terminateProcessTree terminateProcessTree
} }
) )
+19 -56
View File
@@ -26,9 +26,6 @@ import type {
RuntimeNativeSnapshot, RuntimeNativeSnapshot,
RuntimeSettings RuntimeSettings
} from '../../shared/contracts' } from '../../shared/contracts'
import {
maximumConversationToolActivities
} from '../../shared/assistant-contracts'
import { import {
continueConfigurationPresetSchema, continueConfigurationPresetSchema,
runtimeNativeInventoryLimits, runtimeNativeInventoryLimits,
@@ -69,9 +66,7 @@ const maximumConfiguredMcpServers =
runtimeNativeInventoryLimits.mcpServers runtimeNativeInventoryLimits.mcpServers
const maximumConfiguredRules = runtimeNativeInventoryLimits.rules const maximumConfiguredRules = runtimeNativeInventoryLimits.rules
const maximumConfiguredPrompts = runtimeNativeInventoryLimits.prompts const maximumConfiguredPrompts = runtimeNativeInventoryLimits.prompts
const maximumStreamEvents = 5_000
const maximumStreamEventBytes = 2 * 1024 * 1024 const maximumStreamEventBytes = 2 * 1024 * 1024
const maximumToolCalls = maximumConversationToolActivities
const defaultControlRequestTimeoutMs = 30_000 const defaultControlRequestTimeoutMs = 30_000
const knowledgeMcpName = 'goodbuddy-knowledge' const knowledgeMcpName = 'goodbuddy-knowledge'
const customMcpName = 'goodbuddy-custom-mcp' const customMcpName = 'goodbuddy-custom-mcp'
@@ -142,7 +137,7 @@ const continueHostQuestionSchema = z
const stateSchema = z.object({ const stateSchema = z.object({
session: z.object({ session: z.object({
history: z.array(z.unknown()).max(5_000), history: z.array(z.unknown()),
usage: sessionUsageSchema.optional() usage: sessionUsageSchema.optional()
}), }),
isProcessing: z.boolean(), isProcessing: z.boolean(),
@@ -152,12 +147,11 @@ const stateSchema = z.object({
toolName: z.string().min(1).max(128), toolName: z.string().min(1).max(128),
toolArgs: z.record(z.string(), z.unknown()), toolArgs: z.record(z.string(), z.unknown()),
requestId: z.string().min(1).max(256), requestId: z.string().min(1).max(256),
toolCallPreview: z.array(z.unknown()).max(100).optional() toolCallPreview: z.array(z.unknown()).optional()
}) })
.nullable(), .nullable(),
goodbuddyEvents: z goodbuddyEvents: z
.array(continueHostStreamEventSchema) .array(continueHostStreamEventSchema)
.max(maximumStreamEvents)
.optional(), .optional(),
goodbuddyEventsOverflow: z.boolean().optional(), goodbuddyEventsOverflow: z.boolean().optional(),
goodbuddyQuestion: continueHostQuestionSchema.nullable().optional() goodbuddyQuestion: continueHostQuestionSchema.nullable().optional()
@@ -199,6 +193,7 @@ export type ContinueHostTool = {
export type ContinueHostRunResult = { export type ContinueHostRunResult = {
text: string text: string
streamedText?: true streamedText?: true
streamTruncated?: true
usage?: ContinueHostUsage usage?: ContinueHostUsage
tools?: ContinueHostTool[] tools?: ContinueHostTool[]
} }
@@ -249,9 +244,7 @@ export type ContinueHostAdapterOptions = {
export type ContinueHostAdapterDependencies = { export type ContinueHostAdapterDependencies = {
terminateProcessTree: typeof terminateProcessTreeAndWait terminateProcessTree: typeof terminateProcessTreeAndWait
maximumStreamEvents: number
maximumStreamEventBytes: number maximumStreamEventBytes: number
maximumToolCalls: number
controlRequestTimeoutMs: number controlRequestTimeoutMs: number
} }
@@ -713,9 +706,6 @@ function extractContinueTools(
if (!callId || typeof name !== 'string' || !name.trim()) { if (!callId || typeof name !== 'string' || !name.trim()) {
continue continue
} }
if (!tools.has(callId) && tools.size >= 100) {
throw new Error('Continue 单次运行的工具调用超过 100 个')
}
const status = state.status const status = state.status
const normalizedState = const normalizedState =
status === 'done' || status === 'completed' status === 'done' || status === 'completed'
@@ -834,9 +824,7 @@ export class ContinueHostAdapter {
) { ) {
this.dependencies = { this.dependencies = {
terminateProcessTree: terminateProcessTreeAndWait, terminateProcessTree: terminateProcessTreeAndWait,
maximumStreamEvents,
maximumStreamEventBytes, maximumStreamEventBytes,
maximumToolCalls,
controlRequestTimeoutMs: defaultControlRequestTimeoutMs, controlRequestTimeoutMs: defaultControlRequestTimeoutMs,
...dependencies ...dependencies
} }
@@ -985,7 +973,7 @@ export class ContinueHostAdapter {
patched = replaceExactly( patched = replaceExactly(
patched, patched,
streamCallbacksMarker, streamCallbacksMarker,
'a={onContent:u=>{if(!u)return;let l=String(u);e.goodbuddyEventsBytes+=Buffer.byteLength(l);l.length<=1e5&&e.goodbuddyEvents.length<5e3&&e.goodbuddyEventsBytes<=2097152?e.goodbuddyEvents.push({type:"text",delta:l}):e.goodbuddyEventsOverflow=!0},onContentComplete:u=>{},onToolStart:(u,l,c)=>{if(!c)return;let d={type:"tool",callId:c,name:u,state:"running",input:(()=>{try{return JSON.stringify(l).slice(0,4e3)}catch{return"[无法序列化]"}})()};e.goodbuddyEventsBytes+=Buffer.byteLength(JSON.stringify(d));e.goodbuddyEvents.length<5e3&&e.goodbuddyEventsBytes<=2097152?e.goodbuddyEvents.push(d):e.goodbuddyEventsOverflow=!0},onToolResult:(u,l,c,d)=>{if(!d)return;let p={type:"tool",callId:d,name:l,state:c==="done"?"completed":"failed",output:String(u).slice(0,16e3)};e.goodbuddyEventsBytes+=Buffer.byteLength(JSON.stringify(p));e.goodbuddyEvents.length<5e3&&e.goodbuddyEventsBytes<=2097152?e.goodbuddyEvents.push(p):e.goodbuddyEventsOverflow=!0},onToolError:(u,l,c)=>{if(!c)return;let d={type:"tool",callId:c,name:l??"unknown",state:"failed",error:String(u).slice(0,1e3)};e.goodbuddyEventsBytes+=Buffer.byteLength(JSON.stringify(d));e.goodbuddyEvents.length<5e3&&e.goodbuddyEventsBytes<=2097152?e.goodbuddyEvents.push(d):e.goodbuddyEventsOverflow=!0},onToolPermissionRequest:' 'a={onContent:u=>{if(!u)return;let l=String(u);e.goodbuddyEventsBytes+=Buffer.byteLength(l);l.length<=1e5&&e.goodbuddyEventsBytes<=2097152?e.goodbuddyEvents.push({type:"text",delta:l}):e.goodbuddyEventsOverflow=!0},onContentComplete:u=>{},onToolStart:(u,l,c)=>{if(!c)return;let d={type:"tool",callId:c,name:u,state:"running",input:(()=>{try{return JSON.stringify(l).slice(0,4e3)}catch{return"[无法序列化]"}})()};e.goodbuddyEventsBytes+=Buffer.byteLength(JSON.stringify(d));e.goodbuddyEventsBytes<=2097152?e.goodbuddyEvents.push(d):e.goodbuddyEventsOverflow=!0},onToolResult:(u,l,c,d)=>{if(!d)return;let p={type:"tool",callId:d,name:l,state:c==="done"?"completed":"failed",output:String(u).slice(0,16e3)};e.goodbuddyEventsBytes+=Buffer.byteLength(JSON.stringify(p));e.goodbuddyEventsBytes<=2097152?e.goodbuddyEvents.push(p):e.goodbuddyEventsOverflow=!0},onToolError:(u,l,c)=>{if(!c)return;let d={type:"tool",callId:c,name:l??"unknown",state:"failed",error:String(u).slice(0,1e3)};e.goodbuddyEventsBytes+=Buffer.byteLength(JSON.stringify(d));e.goodbuddyEventsBytes<=2097152?e.goodbuddyEvents.push(d):e.goodbuddyEventsOverflow=!0},onToolPermissionRequest:'
) )
patched = replaceExactly( patched = replaceExactly(
patched, patched,
@@ -1584,7 +1572,7 @@ export class ContinueHostAdapter {
let observedTools: ContinueHostTool[] = [] let observedTools: ContinueHostTool[] = []
const reportedQuestionIds = new Set<string>() const reportedQuestionIds = new Set<string>()
let streamedText = false let streamedText = false
let streamEventCount = 0 let streamTruncated = false
let streamEventBytes = 0 let streamEventBytes = 0
const observedToolCallIds = new Set<string>() const observedToolCallIds = new Set<string>()
try { try {
@@ -1635,38 +1623,28 @@ export class ContinueHostAdapter {
signal: executionSignal signal: executionSignal
}) })
) )
if (state.goodbuddyEventsOverflow) { streamTruncated ||= state.goodbuddyEventsOverflow === true
throw new Error('Continue 宿主流式事件超过安全限制') let streamEvents = state.goodbuddyEvents ?? []
}
const streamEvents = state.goodbuddyEvents ?? []
const batchStreamEventBytes = Buffer.byteLength( const batchStreamEventBytes = Buffer.byteLength(
JSON.stringify(streamEvents) JSON.stringify(streamEvents)
) )
const nextStreamEventCount =
streamEventCount + streamEvents.length
const nextStreamEventBytes = const nextStreamEventBytes =
streamEventBytes + batchStreamEventBytes streamEventBytes + batchStreamEventBytes
if (
nextStreamEventBytes >
this.dependencies.maximumStreamEventBytes
) {
streamTruncated = true
streamEvents = []
} else {
streamEventBytes = nextStreamEventBytes
}
const nextToolCallIds = new Set(observedToolCallIds) const nextToolCallIds = new Set(observedToolCallIds)
for (const event of streamEvents) { for (const event of streamEvents) {
if (event.type === 'tool') { if (event.type === 'tool') {
nextToolCallIds.add(event.callId) nextToolCallIds.add(event.callId)
} }
} }
if (
nextStreamEventCount >
this.dependencies.maximumStreamEvents ||
nextStreamEventBytes >
this.dependencies.maximumStreamEventBytes
) {
throw new Error('Continue 宿主流式事件超过安全限制')
}
if (
nextToolCallIds.size > this.dependencies.maximumToolCalls
) {
throw new Error('Continue 单次运行的工具调用超过 100 个')
}
streamEventCount = nextStreamEventCount
streamEventBytes = nextStreamEventBytes
for (const callId of nextToolCallIds) { for (const callId of nextToolCallIds) {
observedToolCallIds.add(callId) observedToolCallIds.add(callId)
} }
@@ -1677,11 +1655,6 @@ export class ContinueHostAdapter {
for (const tool of historyTools) { for (const tool of historyTools) {
observedToolCallIds.add(tool.callId) observedToolCallIds.add(tool.callId)
} }
if (
observedToolCallIds.size > this.dependencies.maximumToolCalls
) {
throw new Error('Continue 单次运行的工具调用超过 100 个')
}
observedTools = mergeContinueTools( observedTools = mergeContinueTools(
observedTools, observedTools,
historyTools historyTools
@@ -1744,12 +1717,6 @@ export class ContinueHostAdapter {
} }
const pending = state.pendingPermission const pending = state.pendingPermission
if (pending && !handledPermissionIds.has(pending.requestId)) { if (pending && !handledPermissionIds.has(pending.requestId)) {
if (
handledPermissionIds.size >=
this.dependencies.maximumToolCalls
) {
throw new Error('Continue 单次运行的工具调用超过 100 个')
}
handledPermissionIds.add(pending.requestId) handledPermissionIds.add(pending.requestId)
const pendingCallId = const pendingCallId =
observedTools.find( observedTools.find(
@@ -1761,13 +1728,6 @@ export class ContinueHostAdapter {
if ( if (
!observedTools.some((tool) => tool.callId === pendingCallId) !observedTools.some((tool) => tool.callId === pendingCallId)
) { ) {
if (
!observedToolCallIds.has(pendingCallId) &&
observedToolCallIds.size >=
this.dependencies.maximumToolCalls
) {
throw new Error('Continue 单次运行的工具调用超过 100 个')
}
observedToolCallIds.add(pendingCallId) observedToolCallIds.add(pendingCallId)
observedTools = [ observedTools = [
...observedTools, ...observedTools,
@@ -1829,6 +1789,9 @@ export class ContinueHostAdapter {
return { return {
text, text,
...(streamedText ? { streamedText: true as const } : {}), ...(streamedText ? { streamedText: true as const } : {}),
...(streamTruncated
? { streamTruncated: true as const }
: {}),
...(usage ? { usage } : {}), ...(usage ? { usage } : {}),
...(observedTools.length > 0 ...(observedTools.length > 0
? { tools: observedTools } ? { tools: observedTools }
+10 -6
View File
@@ -1134,7 +1134,7 @@ describe('ContinueAgentRuntime', () => {
).rejects.toThrow('已失效或不存在') ).rejects.toThrow('已失效或不存在')
}) })
it('fails instead of silently dropping an overflowing stream queue', async () => { it('drains more than 1,000 queued stream events without stopping the run', async () => {
mocks.runHost.mockImplementation( mocks.runHost.mockImplementation(
async ( async (
_prompt, _prompt,
@@ -1160,11 +1160,15 @@ describe('ContinueAgentRuntime', () => {
new AbortController().signal new AbortController().signal
) )
await expect(async () => { const events: RuntimeEvent[] = []
for await (const _event of stream) { for await (const event of stream) {
void _event events.push(event)
} }
}).rejects.toThrow('流式事件积压超过安全限制')
expect(
events.filter((event) => event.type === 'text')
).toHaveLength(1_001)
expect(events.at(-1)).toMatchObject({ type: 'done' })
}) })
it('aborts the host run when stream consumption ends early', async () => { it('aborts the host run when stream consumption ends early', async () => {
+19 -7
View File
@@ -64,7 +64,6 @@ export type ContinueRuntimeOptions = {
// The prompt reaches the Continue host through a local HTTP POST body, so no // The prompt reaches the Continue host through a local HTTP POST body, so no
// platform command-line limit applies to it. // platform command-line limit applies to it.
const MAX_CONTINUE_PROMPT_CHARACTERS = 128_000 const MAX_CONTINUE_PROMPT_CHARACTERS = 128_000
const MAX_QUEUED_STREAM_EVENTS = 1_000
const scopedReadToolNameSet = new Set<string>(scopedReadToolNames) const scopedReadToolNameSet = new Set<string>(scopedReadToolNames)
function continueToolFailureMessage(tool: ContinueHostTool): string { function continueToolFailureMessage(tool: ContinueHostTool): string {
@@ -548,6 +547,7 @@ export class ContinueAgentRuntime implements AgentRuntime {
} }
let result: ContinueHostRunResult let result: ContinueHostRunResult
const emittedTools = new Map<string, ContinueHostTool>() const emittedTools = new Map<string, ContinueHostTool>()
let emittedText = ''
const requestQuestionIds = new Set<string>() const requestQuestionIds = new Set<string>()
try { try {
const host = this.getHostAdapter( const host = this.getHostAdapter(
@@ -574,9 +574,6 @@ export class ContinueAgentRuntime implements AgentRuntime {
const hostController = new AbortController() const hostController = new AbortController()
const hostSignal = AbortSignal.any([signal, hostController.signal]) const hostSignal = AbortSignal.any([signal, hostController.signal])
const onEvent = (event: ContinueHostStreamEvent): void => { const onEvent = (event: ContinueHostStreamEvent): void => {
if (queuedEvents.length >= MAX_QUEUED_STREAM_EVENTS) {
throw new Error('Continue 流式事件积压超过安全限制')
}
queuedEvents.push(event) queuedEvents.push(event)
wakeStream?.() wakeStream?.()
wakeStream = undefined wakeStream = undefined
@@ -623,6 +620,7 @@ export class ContinueAgentRuntime implements AgentRuntime {
emittedTools.set(event.tool.callId, event.tool) emittedTools.set(event.tool.callId, event.tool)
} }
if (event.type === 'text') { if (event.type === 'text') {
emittedText += event.delta
yield { yield {
requestId: request.requestId, requestId: request.requestId,
type: 'text', type: 'text',
@@ -742,11 +740,25 @@ export class ContinueAgentRuntime implements AgentRuntime {
yield finalEvent yield finalEvent
} }
} }
if (!result.streamedText) { if (result.streamTruncated) {
yield { yield {
requestId: request.requestId, requestId: request.requestId,
type: 'text', type: 'status',
delta: result.text message:
'Continue 输出较长,已截断部分流式展示;任务已继续完成'
}
}
if (!result.streamedText || result.streamTruncated) {
const remainingText =
result.streamTruncated && result.text.startsWith(emittedText)
? result.text.slice(emittedText.length)
: result.text
if (remainingText) {
yield {
requestId: request.requestId,
type: 'text',
delta: remainingText
}
} }
} }
if (result.usage) { if (result.usage) {
+81 -29
View File
@@ -2035,7 +2035,7 @@ describe('OpenCodeRuntime embedded permission mediation', () => {
await runtime.dispose() await runtime.dispose()
}) })
it('fails before emitting text that exceeds the aggregate output budget', async () => { it('truncates oversized text display without stopping the run', async () => {
const setup = runClient([ const setup = runClient([
{ {
id: 'first-text', id: 'first-text',
@@ -2056,29 +2056,50 @@ describe('OpenCodeRuntime embedded permission mediation', () => {
field: 'text', field: 'text',
delta: 'b'.repeat(600_000) delta: 'b'.repeat(600_000)
} }
},
{
id: 'idle',
type: 'session.idle',
properties: { sessionID: 'session-1' }
} }
]) ])
const runtime = embeddedRuntime(setup.client) const runtime = embeddedRuntime(setup.client)
const stream = runtime.run( const events = await collectRun(runtime, 'execute')
{ const text = events
requestId: '3f496642-f47d-4e0a-8944-a32c77b0d6ef', .flatMap((event) => event.type === 'text' ? [event.delta] : [])
conversationId: 'conversation-1', .join('')
prompt: 'test',
workMode: 'execute'
},
new AbortController().signal
)
await stream.next() expect(Buffer.byteLength(text)).toBe(1024 * 1024)
const firstText = await stream.next() expect(events).toContainEqual({
expect(firstText.value).toMatchObject({ requestId: '3f496642-f47d-4e0a-8944-a32c77b0d6ef',
type: 'text', type: 'status',
delta: expect.stringMatching(/^a+$/u) message:
'OpenCode 输出较长,已停止继续展示文本;任务仍在继续'
}) })
await expect(stream.next()).rejects.toThrow( expect(events.at(-1)).toMatchObject({ type: 'done' })
'文本与推理输出超过 1 MB 安全限制' expect(setup.session.abort).not.toHaveBeenCalled()
) await runtime.dispose()
expect(setup.session.abort).toHaveBeenCalled() })
it('completes more than 100 distinct tool calls in one run', async () => {
const setup = runClient([
...Array.from({ length: 101 }, (_, index) =>
completedToolEvent(`call-${index + 1}`)
),
{
id: 'idle',
type: 'session.idle',
properties: { sessionID: 'session-1' }
}
])
const runtime = embeddedRuntime(setup.client)
const events = await collectRun(runtime, 'execute')
expect(
events.filter((event) => event.type === 'tool')
).toHaveLength(101)
expect(events.at(-1)).toMatchObject({ type: 'done' })
await runtime.dispose() await runtime.dispose()
}) })
@@ -2184,6 +2205,8 @@ describe('OpenCodeRuntime embedded permission mediation', () => {
read: false, read: false,
write: false, write: false,
bash: false, bash: false,
'goodbuddy-data-*': false,
'goodbuddy-custom-*': false,
[knowledgeToolId]: true [knowledgeToolId]: true
} }
}), }),
@@ -2264,7 +2287,7 @@ describe('OpenCodeRuntime embedded permission mediation', () => {
await runtime.dispose() await runtime.dispose()
}) })
it('serializes overlapping embedded MCP registration and discovery', async () => { it('runs embedded conversations concurrently with request-scoped MCP registrations', async () => {
const setup = runClient([ const setup = runClient([
{ {
id: 'idle', id: 'idle',
@@ -2357,13 +2380,19 @@ describe('OpenCodeRuntime embedded permission mediation', () => {
'conversation-two', 'conversation-two',
'second-token' 'second-token'
) )
await new Promise((resolve) => setTimeout(resolve, 10)) try {
expect(mcpAdd).toHaveBeenCalledTimes(1) await vi.waitFor(
() => expect(mcpAdd).toHaveBeenCalledTimes(2),
resolveFirstAdd() { timeout: 500 }
await first )
await vi.waitFor(() => expect(mcpAdd).toHaveBeenCalledTimes(2)) await vi.waitFor(
await second () => expect(setup.session.promptAsync).toHaveBeenCalledOnce(),
{ timeout: 500 }
)
} finally {
resolveFirstAdd()
}
await Promise.all([first, second])
expect( expect(
mcpAdd.mock.calls.map( mcpAdd.mock.calls.map(
([input]) => ([input]) =>
@@ -2483,6 +2512,8 @@ describe('OpenCodeRuntime embedded permission mediation', () => {
write: false, write: false,
bash: false, bash: false,
task: false, task: false,
'goodbuddy-data-*': false,
'goodbuddy-custom-*': false,
skill: true skill: true
} }
}), }),
@@ -2557,11 +2588,30 @@ describe('OpenCodeRuntime embedded permission mediation', () => {
title: 'GoodBuddy 对话', title: 'GoodBuddy 对话',
directory: process.cwd(), directory: process.cwd(),
permission: [ permission: [
{ permission: '*', pattern: '*', action: 'allow' } { permission: '*', pattern: '*', action: 'allow' },
{
permission: 'goodbuddy-data-*',
pattern: '*',
action: 'deny'
},
{
permission: 'goodbuddy-custom-*',
pattern: '*',
action: 'deny'
}
] ]
}, },
{ signal: expect.any(AbortSignal) } { signal: expect.any(AbortSignal) }
) )
expect(session.promptAsync).toHaveBeenCalledWith(
expect.objectContaining({
tools: {
'goodbuddy-data-*': false,
'goodbuddy-custom-*': false
}
}),
expect.anything()
)
expect(permissionReply).toHaveBeenCalledOnce() expect(permissionReply).toHaveBeenCalledOnce()
expect(permissionReply).toHaveBeenCalledWith( expect(permissionReply).toHaveBeenCalledWith(
{ {
@@ -2926,7 +2976,9 @@ describe('OpenCodeRuntime embedded permission mediation', () => {
read: false, read: false,
write: false, write: false,
bash: false, bash: false,
task: false task: false,
'goodbuddy-data-*': false,
'goodbuddy-custom-*': false
} }
}), }),
expect.anything() expect.anything()
+76 -86
View File
@@ -22,9 +22,6 @@ import type {
AgentRuntimeStatus, AgentRuntimeStatus,
SubagentEvent SubagentEvent
} from '../../shared/contracts' } from '../../shared/contracts'
import {
maximumConversationToolActivities
} from '../../shared/assistant-contracts'
import { import {
boundedRuntimeIdentifierSchema, boundedRuntimeIdentifierSchema,
runtimeNativeInventoryLimits, runtimeNativeInventoryLimits,
@@ -78,7 +75,6 @@ const MAX_PERMISSION_PATTERNS = 32
const MAX_PERMISSION_PATTERN_LENGTH = 1_024 const MAX_PERMISSION_PATTERN_LENGTH = 1_024
const MAX_PERMISSION_PATTERNS_BYTES = 8 * 1_024 const MAX_PERMISSION_PATTERNS_BYTES = 8 * 1_024
const MAX_PERMISSION_METADATA_BYTES = 8 * 1_024 const MAX_PERMISSION_METADATA_BYTES = 8 * 1_024
const MAX_TOOL_CALLS_PER_RUN = maximumConversationToolActivities
const MAX_EXECUTION_OUTPUT_BYTES = 1024 * 1024 const MAX_EXECUTION_OUTPUT_BYTES = 1024 * 1024
const MAX_QUESTION_REQUEST_BYTES = 32 * 1_024 const MAX_QUESTION_REQUEST_BYTES = 32 * 1_024
const MAX_QUESTIONS_PER_REQUEST = 4 const MAX_QUESTIONS_PER_REQUEST = 4
@@ -99,6 +95,15 @@ const TEMPORARY_MCP_PREFIXES = [
'goodbuddy-data-', 'goodbuddy-data-',
'goodbuddy-custom-' 'goodbuddy-custom-'
] as const ] as const
const temporaryMcpToolOverrides = Object.fromEntries(
TEMPORARY_MCP_PREFIXES.map((prefix) => [`${prefix}*`, false])
)
const temporaryMcpDenyRules: PermissionRuleset =
TEMPORARY_MCP_PREFIXES.map((prefix) => ({
permission: `${prefix}*`,
pattern: '*',
action: 'deny'
}))
const OPENCODE_INTERNAL_TOOL_IDS = new Set(['invalid']) const OPENCODE_INTERNAL_TOOL_IDS = new Set(['invalid'])
const OPENCODE_BUILTIN_TOOL_KINDS: Readonly< const OPENCODE_BUILTIN_TOOL_KINDS: Readonly<
Partial<Record<string, RuntimeNativeTool['kind']>> Partial<Record<string, RuntimeNativeTool['kind']>>
@@ -719,7 +724,6 @@ export class OpenCodeRuntime implements AgentRuntime {
upstreamQuestionId: string upstreamQuestionId: string
} }
>() >()
private embeddedRunTail: Promise<void> = Promise.resolve()
private readonly conversationRunTails = new Map<string, Promise<void>>() private readonly conversationRunTails = new Map<string, Promise<void>>()
private readonly dependencies: OpenCodeRuntimeDependencies private readonly dependencies: OpenCodeRuntimeDependencies
@@ -751,36 +755,6 @@ export class OpenCodeRuntime implements AgentRuntime {
return this.usesEmbeddedPermissionMediation() return this.usesEmbeddedPermissionMediation()
} }
private async acquireEmbeddedRun(
signal: AbortSignal
): Promise<() => void> {
signal.throwIfAborted()
const previous = this.embeddedRunTail
let release!: () => void
const current = new Promise<void>((resolve) => {
release = resolve
})
this.embeddedRunTail = previous.then(
() => current,
() => current
)
let abort!: () => void
const aborted = new Promise<never>((_resolve, reject) => {
abort = () => reject(signal.reason)
})
signal.addEventListener('abort', abort, { once: true })
try {
await Promise.race([previous, aborted])
signal.throwIfAborted()
return release
} catch (error) {
release()
throw error
} finally {
signal.removeEventListener('abort', abort)
}
}
private async acquireConversationRun( private async acquireConversationRun(
conversationId: string, conversationId: string,
signal: AbortSignal signal: AbortSignal
@@ -1835,12 +1809,8 @@ export class OpenCodeRuntime implements AgentRuntime {
const executionSignal = deadline const executionSignal = deadline
? AbortSignal.any([signal, deadline.signal]) ? AbortSignal.any([signal, deadline.signal])
: signal : signal
let releaseEmbedded: (() => void) | undefined
let releaseConversation: (() => void) | undefined let releaseConversation: (() => void) | undefined
try { try {
releaseEmbedded = this.usesEmbeddedPermissionMediation()
? await this.acquireEmbeddedRun(executionSignal)
: undefined
releaseConversation = await this.acquireConversationRun( releaseConversation = await this.acquireConversationRun(
request.conversationId, request.conversationId,
executionSignal executionSignal
@@ -1856,7 +1826,6 @@ export class OpenCodeRuntime implements AgentRuntime {
clearTimeout(deadlineTimer) clearTimeout(deadlineTimer)
} }
releaseConversation?.() releaseConversation?.()
releaseEmbedded?.()
} }
} }
@@ -2054,6 +2023,9 @@ export class OpenCodeRuntime implements AgentRuntime {
request.workMode === 'execute' request.workMode === 'execute'
? [ ? [
...executePermissionRules, ...executePermissionRules,
...(this.usesEmbeddedPermissionMediation()
? temporaryMcpDenyRules
: []),
...nativeSkillPermissionRules, ...nativeSkillPermissionRules,
...knowledgeToolIds.map((toolId) => ({ ...knowledgeToolIds.map((toolId) => ({
permission: toolId, permission: toolId,
@@ -2075,7 +2047,7 @@ export class OpenCodeRuntime implements AgentRuntime {
...readOnlyPermissionRules, ...readOnlyPermissionRules,
...nativeSkillPermissionRules ...nativeSkillPermissionRules
] ]
let disabledTools: Record<string, boolean> | undefined let toolOverrides: Record<string, boolean> | undefined
if (request.workMode !== 'execute') { if (request.workMode !== 'execute') {
const tools = await this.controlRequest( const tools = await this.controlRequest(
'读取工具清单', '读取工具清单',
@@ -2089,15 +2061,23 @@ export class OpenCodeRuntime implements AgentRuntime {
if (tools.error || !tools.data) { if (tools.error || !tools.data) {
throw new Error('OpenCode 无法确认工具已禁用,已阻止只读请求') throw new Error('OpenCode 无法确认工具已禁用,已阻止只读请求')
} }
disabledTools = { toolOverrides = {
...Object.fromEntries( ...Object.fromEntries(
tools.data.map((toolId) => [toolId, false]) tools.data.map((toolId) => [toolId, false])
), ),
...temporaryMcpToolOverrides,
...Object.fromEntries( ...Object.fromEntries(
knowledgeToolIds.map((toolId) => [toolId, true]) knowledgeToolIds.map((toolId) => [toolId, true])
), ),
...(nativeSkillIds.length > 0 ? { skill: true } : {}) ...(nativeSkillIds.length > 0 ? { skill: true } : {})
} }
} else if (this.usesEmbeddedPermissionMediation()) {
toolOverrides = {
...temporaryMcpToolOverrides,
...Object.fromEntries(
knowledgeToolIds.map((toolId) => [toolId, true])
)
}
} }
const session = await this.getSessionId( const session = await this.getSessionId(
client, client,
@@ -2165,12 +2145,28 @@ export class OpenCodeRuntime implements AgentRuntime {
const reasoningPartIds = new Set<string>() const reasoningPartIds = new Set<string>()
const reportedQuestionIds = new Map<string, string>() const reportedQuestionIds = new Map<string, string>()
let aggregateOutputBytes = 0 let aggregateOutputBytes = 0
const consumeOutputBudget = (delta: string): void => { let outputTruncated = false
aggregateOutputBytes += Buffer.byteLength(delta) const retainOutputDelta = (
if (aggregateOutputBytes > MAX_EXECUTION_OUTPUT_BYTES) { delta: string
throw new Error( ): { delta: string; truncated: boolean } => {
'OpenCode 单次运行的文本与推理输出超过 1 MB 安全限制' if (outputTruncated) {
) return { delta: '', truncated: false }
}
const bytes = Buffer.from(delta)
const remaining =
MAX_EXECUTION_OUTPUT_BYTES - aggregateOutputBytes
if (bytes.byteLength <= remaining) {
aggregateOutputBytes += bytes.byteLength
return { delta, truncated: false }
}
outputTruncated = true
aggregateOutputBytes = MAX_EXECUTION_OUTPUT_BYTES
return {
delta: bytes
.subarray(0, Math.max(0, remaining))
.toString('utf8')
.replace(/\uFFFD+$/u, ''),
truncated: true
} }
} }
let hasResponseTextAfterFailure = false let hasResponseTextAfterFailure = false
@@ -2228,8 +2224,8 @@ export class OpenCodeRuntime implements AgentRuntime {
nativeSkillIds.length > 0 nativeSkillIds.length > 0
? undefined ? undefined
: this.options.skillInstructions || undefined, : this.options.skillInstructions || undefined,
...(disabledTools ...(toolOverrides
? { tools: disabledTools } ? { tools: toolOverrides }
: {}), : {}),
parts: [ parts: [
{ type: 'text' as const, text: promptText }, { type: 'text' as const, text: promptText },
@@ -2276,7 +2272,6 @@ export class OpenCodeRuntime implements AgentRuntime {
'thinking' 'thinking'
].includes(event.properties.field) ].includes(event.properties.field)
if (reasoning || event.properties.field === 'text') { if (reasoning || event.properties.field === 'text') {
consumeOutputBudget(event.properties.delta)
if ( if (
!reasoning && !reasoning &&
/\S/u.test(event.properties.delta) && /\S/u.test(event.properties.delta) &&
@@ -2286,10 +2281,21 @@ export class OpenCodeRuntime implements AgentRuntime {
) { ) {
hasResponseTextAfterFailure = true hasResponseTextAfterFailure = true
} }
yield { const retained = retainOutputDelta(event.properties.delta)
requestId: request.requestId, if (retained.delta) {
type: reasoning ? 'reasoning' : 'text', yield {
delta: event.properties.delta requestId: request.requestId,
type: reasoning ? 'reasoning' : 'text',
delta: retained.delta
}
}
if (retained.truncated) {
yield {
requestId: request.requestId,
type: 'status',
message:
'OpenCode 输出较长,已停止继续展示文本;任务仍在继续'
}
} }
} }
} }
@@ -2307,12 +2313,6 @@ export class OpenCodeRuntime implements AgentRuntime {
throw new Error('OpenCode 工具调用 ID 格式无效') throw new Error('OpenCode 工具调用 ID 格式无效')
} }
const toolName = part.tool.slice(0, 200) const toolName = part.tool.slice(0, 200)
if (
!toolStates.has(callId) &&
toolStates.size >= MAX_TOOL_CALLS_PER_RUN
) {
throw new Error('OpenCode 单次运行的工具调用超过 100 个')
}
const state = const state =
part.state.status === 'error' ? 'failed' : part.state.status part.state.status === 'error' ? 'failed' : part.state.status
if (state === 'failed') { if (state === 'failed') {
@@ -2372,11 +2372,21 @@ export class OpenCodeRuntime implements AgentRuntime {
event.properties.sessionID === sessionId && event.properties.sessionID === sessionId &&
event.properties.delta event.properties.delta
) { ) {
consumeOutputBudget(event.properties.delta) const retained = retainOutputDelta(event.properties.delta)
yield { if (retained.delta) {
requestId: request.requestId, yield {
type: 'reasoning', requestId: request.requestId,
delta: event.properties.delta type: 'reasoning',
delta: retained.delta
}
}
if (retained.truncated) {
yield {
requestId: request.requestId,
type: 'status',
message:
'OpenCode 输出较长,已停止继续展示文本;任务仍在继续'
}
} }
} }
@@ -2469,14 +2479,6 @@ export class OpenCodeRuntime implements AgentRuntime {
properties.id.length <= MAX_PERMISSION_NAME_LENGTH && properties.id.length <= MAX_PERMISSION_NAME_LENGTH &&
!repliedPermissionIds.has(properties.id) !repliedPermissionIds.has(properties.id)
) { ) {
if (
repliedPermissionIds.size >= MAX_TOOL_CALLS_PER_RUN
) {
throw new Error(
'OpenCode 单次运行的权限请求超过 100 个',
{ cause: error }
)
}
const permissionId = properties.id const permissionId = properties.id
repliedPermissionIds.add(properties.id) repliedPermissionIds.add(properties.id)
const rejection = await this.controlRequest( const rejection = await this.controlRequest(
@@ -2505,21 +2507,12 @@ export class OpenCodeRuntime implements AgentRuntime {
if (repliedPermissionIds.has(permissionRequest.id)) { if (repliedPermissionIds.has(permissionRequest.id)) {
continue continue
} }
if (repliedPermissionIds.size >= MAX_TOOL_CALLS_PER_RUN) {
throw new Error('OpenCode 单次运行的权限请求超过 100 个')
}
repliedPermissionIds.add(permissionRequest.id) repliedPermissionIds.add(permissionRequest.id)
const callId = ( const callId = (
permissionRequest.tool?.callID ?? permissionRequest.id permissionRequest.tool?.callID ?? permissionRequest.id
) )
const toolName = permissionRequest.permission.slice(0, 200) const toolName = permissionRequest.permission.slice(0, 200)
if (
!toolStates.has(callId) &&
toolStates.size >= MAX_TOOL_CALLS_PER_RUN
) {
throw new Error('OpenCode 单次运行的工具调用超过 100 个')
}
toolStates.set(callId, { name: toolName, state: 'pending' }) toolStates.set(callId, { name: toolName, state: 'pending' })
yield { yield {
requestId: request.requestId, requestId: request.requestId,
@@ -2772,10 +2765,8 @@ export class OpenCodeRuntime implements AgentRuntime {
const executionSignal = deadline const executionSignal = deadline
? AbortSignal.any([signal, deadline.signal]) ? AbortSignal.any([signal, deadline.signal])
: signal : signal
let releaseEmbedded: (() => void) | undefined
let releaseConversation: (() => void) | undefined let releaseConversation: (() => void) | undefined
try { try {
releaseEmbedded = await this.acquireEmbeddedRun(executionSignal)
releaseConversation = await this.acquireConversationRun( releaseConversation = await this.acquireConversationRun(
request.conversationId, request.conversationId,
executionSignal executionSignal
@@ -2932,7 +2923,6 @@ export class OpenCodeRuntime implements AgentRuntime {
clearTimeout(deadlineTimer) clearTimeout(deadlineTimer)
} }
releaseConversation?.() releaseConversation?.()
releaseEmbedded?.()
} }
} }
@@ -1256,6 +1256,11 @@ describe.runIf(enabled)('runtime end-to-end', () => {
) )
.join('') .join('')
).toContain('Prism Relay') ).toContain('Prism Relay')
const modelCalls = events.filter(
(event) => event.type === 'model-usage'
).length
expect(modelCalls).toBe(2)
console.info(`REAL_OPENCODE_MCP_MODEL_CALLS=${modelCalls}`)
} finally { } finally {
await runtime.dispose() await runtime.dispose()
await gateway.dispose() await gateway.dispose()
@@ -159,6 +159,43 @@ describe('AssistantDatabase', () => {
unchanged.close() unchanged.close()
}) })
it('persists more than 100 tool activities in one message', async () => {
const database = await createDatabase()
const conversationId = '00000000-0000-4000-8000-000000000901'
const messageId = '00000000-0000-4000-8000-000000000902'
const tools = Array.from({ length: 101 }, (_, index) => ({
callId: `call-${index + 1}`,
name: 'read',
state: 'completed' as const,
summary: `Tool ${index + 1}`
}))
database.saveLocalConversations([
{
header: {
id: conversationId,
title: '长工具链',
updatedAt: 1
},
messages: [
{
id: messageId,
role: 'assistant',
content: '',
createdAt: 1,
state: 'complete',
tools
}
]
}
])
expect(
database.getConversation(conversationId)?.messages[0]?.tools
).toHaveLength(101)
database.close()
})
it('lists projects by creation time with newer projects last', async () => { it('lists projects by creation time with newer projects last', async () => {
vi.useFakeTimers() vi.useFakeTimers()
vi.setSystemTime(new Date('2026-08-07T00:00:00.000Z')) vi.setSystemTime(new Date('2026-08-07T00:00:00.000Z'))
+11 -20
View File
@@ -7,8 +7,6 @@ import {
conversationMessageSchema, conversationMessageSchema,
conversationSnapshotSchema, conversationSnapshotSchema,
expertCreateSchema, expertCreateSchema,
maximumConversationSubagentActivities,
maximumConversationToolActivities,
normalizeInteractiveWorkMode, normalizeInteractiveWorkMode,
persistedProjectExecutionSpaceSchema, persistedProjectExecutionSpaceSchema,
projectCreateSchema, projectCreateSchema,
@@ -1170,7 +1168,6 @@ function serializeConversationMessageMetadata(
} }
const maximumRecoveredMessageLength = 1_000_000 const maximumRecoveredMessageLength = 1_000_000
const maximumRecoveredMessageBlocks = 500
function appendRecoveredMessageBlock( function appendRecoveredMessageBlock(
blocks: ConversationMessageBlock[] | undefined, blocks: ConversationMessageBlock[] | undefined,
@@ -1190,7 +1187,7 @@ function appendRecoveredMessageBlock(
maximumRecoveredMessageLength maximumRecoveredMessageLength
) )
} }
} else if (current.length < maximumRecoveredMessageBlocks) { } else {
current.push({ current.push({
id: randomUUID(), id: randomUUID(),
type, type,
@@ -1221,12 +1218,10 @@ function upsertRecoveredToolBlock(
: block : block
) )
} }
return blocks.length >= maximumRecoveredMessageBlocks return [
? blocks ...blocks,
: [ { id: randomUUID(), type: 'tool' as const, tool }
...blocks, ]
{ id: randomUUID(), type: 'tool' as const, tool }
]
} }
function upsertRecoveredSubagentBlock( function upsertRecoveredSubagentBlock(
@@ -1262,12 +1257,10 @@ function upsertRecoveredSubagentBlock(
: block : block
) )
} }
return blocks.length >= maximumRecoveredMessageBlocks return [
? blocks ...blocks,
: [ { id: randomUUID(), type: 'subagent' as const, childTaskId }
...blocks, ]
{ id: randomUUID(), type: 'subagent' as const, childTaskId }
]
} }
function terminalizeRecoveredToolBlocks( function terminalizeRecoveredToolBlocks(
@@ -1377,7 +1370,7 @@ function reduceRecoveredAgentEvent(
) )
if (index >= 0) { if (index >= 0) {
tools[index] = tool tools[index] = tool
} else if (tools.length < maximumConversationToolActivities) { } else {
tools.push(tool) tools.push(tool)
} }
next = { next = {
@@ -1410,9 +1403,7 @@ function reduceRecoveredAgentEvent(
) )
if (index >= 0) { if (index >= 0) {
subagents[index] = subagent subagents[index] = subagent
} else if ( } else {
subagents.length < maximumConversationSubagentActivities
) {
subagents.push(subagent) subagents.push(subagent)
} }
next = { next = {
@@ -15,7 +15,7 @@ const profile: ResolvedModelProfile = {
} }
describe('managed Agent prompt model setup', () => { describe('managed Agent prompt model setup', () => {
it('validates and clones a bounded Agent prompt profile', () => { it('validates and clones an Agent prompt profile without prompt-wide quotas', () => {
const setup = createManagedModelBridge({ profile }) const setup = createManagedModelBridge({ profile })
expect(setup.profile).toEqual({ expect(setup.profile).toEqual({
@@ -32,8 +32,6 @@ describe('managed Agent prompt model setup', () => {
capabilities: { imageInput: true }, capabilities: { imageInput: true },
limits: { limits: {
maximumOutputTokens: 8_192, maximumOutputTokens: 8_192,
maximumModelCalls: 100,
maximumTotalOutputTokens: 819_200,
requestTimeoutMilliseconds: 60_000 requestTimeoutMilliseconds: 60_000
} }
}) })
@@ -8,7 +8,6 @@ import { createResolvedModelProfileDigest } from '../agent/remote-model-gateway'
import type { ResolvedModelProfile } from '../runtime-settings-store' import type { ResolvedModelProfile } from '../runtime-settings-store'
const DEFAULT_MAXIMUM_OUTPUT_TOKENS = 32_000 const DEFAULT_MAXIMUM_OUTPUT_TOKENS = 32_000
const MAXIMUM_MODEL_CALLS_PER_PROMPT = 100
const MODEL_REQUEST_TIMEOUT_MILLISECONDS = 60_000 const MODEL_REQUEST_TIMEOUT_MILLISECONDS = 60_000
/** /**
@@ -61,11 +60,6 @@ export function createManagedModelBridge(options: {
}, },
limits: { limits: {
maximumOutputTokens, maximumOutputTokens,
maximumModelCalls: MAXIMUM_MODEL_CALLS_PER_PROMPT,
maximumTotalOutputTokens: Math.min(
10_000_000,
maximumOutputTokens * MAXIMUM_MODEL_CALLS_PER_PROMPT
),
requestTimeoutMilliseconds: requestTimeoutMilliseconds:
MODEL_REQUEST_TIMEOUT_MILLISECONDS MODEL_REQUEST_TIMEOUT_MILLISECONDS
} }
@@ -226,8 +226,6 @@ function harness(
capabilities: { imageInput: false }, capabilities: { imageInput: false },
limits: { limits: {
maximumOutputTokens: 4_096, maximumOutputTokens: 4_096,
maximumModelCalls: 100,
maximumTotalOutputTokens: 409_600,
requestTimeoutMilliseconds: 60_000 requestTimeoutMilliseconds: 60_000
} }
} }
@@ -729,8 +729,6 @@ describe('ProtocolRemoteRuntimeChannel', () => {
capabilities: { imageInput: false }, capabilities: { imageInput: false },
limits: { limits: {
maximumOutputTokens: 4_096, maximumOutputTokens: 4_096,
maximumModelCalls: 100,
maximumTotalOutputTokens: 409_600,
requestTimeoutMilliseconds: 60_000 requestTimeoutMilliseconds: 60_000
} }
}, },
+31
View File
@@ -3694,6 +3694,37 @@ describe('App', () => {
expect(screen.getByText('项目:默认项目')).toHaveClass('scope-badge') expect(screen.getByText('项目:默认项目')).toHaveClass('scope-badge')
}) })
it('keeps more than 100 tool activities in one response', async () => {
render(<App />)
fireEvent.change(screen.getByLabelText('向 GoodBuddy 提问'), {
target: { value: '执行长工具链' }
})
fireEvent.click(await screen.findByLabelText('发送'))
await waitFor(() => expect(run).toHaveBeenCalledOnce())
const request = run.mock.calls[0]?.[0]
if (!request) {
throw new Error('Missing request')
}
act(() => {
for (let index = 0; index < 101; index += 1) {
agentListener?.({
requestId: request.requestId,
type: 'tool',
callId: `call-${index + 1}`,
name: 'read',
state: 'completed',
summary: `OpenCode 工具:read ${index + 1}`
})
}
})
expect(
screen.getByRole('region', { name: '工具执行,共 101 项' })
).toBeInTheDocument()
})
it('keeps adjacent streamed text blocks exact in StrictMode', async () => { it('keeps adjacent streamed text blocks exact in StrictMode', async () => {
render( render(
<StrictMode> <StrictMode>
+2 -21
View File
@@ -128,7 +128,6 @@ import {
conversationMessageBlocksSchema, conversationMessageBlocksSchema,
conversationSubagentActivitySchema, conversationSubagentActivitySchema,
interactiveWorkModes, interactiveWorkModes,
maximumConversationSubagentActivities,
normalizeInteractiveWorkMode, normalizeInteractiveWorkMode,
projectChannelLabels projectChannelLabels
} from '../../shared/assistant-contracts' } from '../../shared/assistant-contracts'
@@ -568,7 +567,6 @@ const minimumPrimaryWorkspaceWidth = 480
const primarySidebarKeyboardResizeStep = 16 const primarySidebarKeyboardResizeStep = 16
const maxMessageContentLength = 1_000_000 const maxMessageContentLength = 1_000_000
const maxMessageBlocks = 500
function getPrimarySidebarWidthLimits(viewportWidth: number): { function getPrimarySidebarWidthLimits(viewportWidth: number): {
minimum: number minimum: number
@@ -638,9 +636,6 @@ function appendMessageContentBlock(
} }
return current return current
} }
if (current.length >= maxMessageBlocks) {
return current
}
current.push({ current.push({
id: crypto.randomUUID(), id: crypto.randomUUID(),
type, type,
@@ -670,9 +665,6 @@ function upsertMessageToolBlock(
: block : block
) )
} }
if (blocks.length >= maxMessageBlocks) {
return blocks
}
return [ return [
...blocks, ...blocks,
{ {
@@ -718,9 +710,6 @@ function upsertMessageSubagentBlock(
: block : block
) )
} }
if (blocks.length >= maxMessageBlocks) {
return blocks
}
return [ return [
...blocks, ...blocks,
{ {
@@ -1327,8 +1316,6 @@ function isConversation(value: unknown): value is Conversation {
))) && ))) &&
(entry.subagents === undefined || (entry.subagents === undefined ||
(Array.isArray(entry.subagents) && (Array.isArray(entry.subagents) &&
entry.subagents.length <=
maximumConversationSubagentActivities &&
entry.subagents.every( entry.subagents.every(
(subagent) => (subagent) =>
conversationSubagentActivitySchema.safeParse(subagent) conversationSubagentActivitySchema.safeParse(subagent)
@@ -4564,10 +4551,7 @@ function App(): React.JSX.Element {
? current.map((task) => ? current.map((task) =>
task.id === event.childTaskId ? childTask : task task.id === event.childTaskId ? childTask : task
) )
: [...current, childTask].slice( : [...current, childTask]
0,
maximumConversationSubagentActivities
)
}) })
} }
if (event.runtimeCallId) { if (event.runtimeCallId) {
@@ -4638,10 +4622,7 @@ function App(): React.JSX.Element {
} }
if (index >= 0) { if (index >= 0) {
subagents[index] = subagent subagents[index] = subagent
} else if ( } else {
subagents.length <
maximumConversationSubagentActivities
) {
subagents.push(subagent) subagents.push(subagent)
} }
const runtimeCallId = event.runtimeCallId const runtimeCallId = event.runtimeCallId
+47
View File
@@ -4,6 +4,7 @@ import {
builtInDefaultProjectSeedDescription, builtInDefaultProjectSeedDescription,
builtInDefaultProjectSeedName, builtInDefaultProjectSeedName,
conversationBranchInputSchema, conversationBranchInputSchema,
conversationMessageSchema,
conversationSubagentActivitySchema, conversationSubagentActivitySchema,
conversationSnapshotSchema, conversationSnapshotSchema,
isUntouchedBuiltInDefaultProject, isUntouchedBuiltInDefaultProject,
@@ -172,6 +173,52 @@ describe('project execution space contracts', () => {
}) })
}) })
describe('conversation activity contracts', () => {
it('accepts more than 100 tool and subagent activities', () => {
const tools = Array.from({ length: 101 }, (_, index) => ({
callId: `call-${index + 1}`,
name: 'read',
state: 'completed' as const,
summary: `Tool ${index + 1}`
}))
const subagents = Array.from({ length: 101 }, (_, index) => ({
childTaskId: `00000000-0000-4000-8000-${String(index + 1).padStart(12, '0')}`,
routingMode: 'native' as const,
actor: {
kind: 'direct-model' as const,
label: '编程 Subagent' as const
},
state: 'completed' as const
}))
expect(
conversationMessageSchema.parse({
id: '00000000-0000-4000-8000-000000000999',
role: 'assistant',
content: '',
blocks: tools.map((tool, index) => ({
id: `10000000-0000-4000-8000-${String(index + 1).padStart(12, '0')}`,
type: 'tool',
tool
})),
createdAt: 1,
state: 'complete',
tools,
subagents
})
).toMatchObject({
tools: expect.arrayContaining([
expect.objectContaining({ callId: 'call-101' })
]),
subagents: expect.arrayContaining([
expect.objectContaining({
childTaskId: '00000000-0000-4000-8000-000000000101'
})
])
})
})
})
describe('conversation branch contracts', () => { describe('conversation branch contracts', () => {
it('accepts bounded branch provenance and rejects renderer extras', () => { it('accepts bounded branch provenance and rejects renderer extras', () => {
const sourceConversationId = const sourceConversationId =
+5 -14
View File
@@ -182,9 +182,9 @@ export const conversationMessageBlockSchema = z.discriminatedUnion('type', [
.strict() .strict()
]) ])
export const conversationMessageBlocksSchema = z export const conversationMessageBlocksSchema = z.array(
.array(conversationMessageBlockSchema) conversationMessageBlockSchema
.max(500) )
export type ConversationToolActivity = z.infer< export type ConversationToolActivity = z.infer<
typeof conversationToolActivitySchema typeof conversationToolActivitySchema
@@ -193,9 +193,6 @@ export type ConversationMessageBlock = z.infer<
typeof conversationMessageBlockSchema typeof conversationMessageBlockSchema
> >
export const maximumConversationToolActivities = 100
export const maximumConversationSubagentActivities = 100
export const subagentActorSchema = z export const subagentActorSchema = z
.object({ .object({
kind: z.literal('direct-model'), kind: z.literal('direct-model'),
@@ -277,14 +274,8 @@ export const conversationMessageSchema = z
.array(conversationContextCompressionMarkerSchema) .array(conversationContextCompressionMarkerSchema)
.max(2) .max(2)
.optional(), .optional(),
tools: z tools: z.array(conversationToolActivitySchema).optional(),
.array(conversationToolActivitySchema) subagents: z.array(conversationSubagentActivitySchema).optional(),
.max(maximumConversationToolActivities)
.optional(),
subagents: z
.array(conversationSubagentActivitySchema)
.max(maximumConversationSubagentActivities)
.optional(),
sources: z.array(z.string().max(8_192)).max(100).optional(), sources: z.array(z.string().max(8_192)).max(100).optional(),
sourceReferences: z sourceReferences: z
.array( .array(
+2 -2
View File
@@ -191,14 +191,14 @@ describe('model bridge contracts and codec', () => {
apiKey: 'must-not-cross' apiKey: 'must-not-cross'
}) })
).toThrow() ).toThrow()
expect( expect(() =>
modelBridgePolicySchema.parse({ modelBridgePolicySchema.parse({
...policy, ...policy,
maximumOutputTokens: 65_537, maximumOutputTokens: 65_537,
maximumModelCalls: 8, maximumModelCalls: 8,
maximumTotalOutputTokens: 32_768 maximumTotalOutputTokens: 32_768
}) })
).toMatchObject(policy) ).toThrow()
await expect( await expect(
createModelBridgeRequestMessage({ createModelBridgeRequestMessage({
identity, identity,
-17
View File
@@ -126,8 +126,6 @@ export const agentPromptModelProfileSchema = z
limits: z limits: z
.object({ .object({
maximumOutputTokens: z.number().int().min(1).max(1_000_000), maximumOutputTokens: z.number().int().min(1).max(1_000_000),
maximumModelCalls: z.number().int().min(1).max(10_000),
maximumTotalOutputTokens: z.number().int().min(1).max(10_000_000),
requestTimeoutMilliseconds: z requestTimeoutMilliseconds: z
.number() .number()
.int() .int()
@@ -158,16 +156,6 @@ export const agentPromptModelProfileSchema = z
message: 'Model provider does not match its wire protocol' message: 'Model provider does not match its wire protocol'
}) })
} }
if (
profile.limits.maximumOutputTokens >
profile.limits.maximumTotalOutputTokens
) {
context.addIssue({
code: 'custom',
path: ['limits', 'maximumOutputTokens'],
message: 'Per-call output limit exceeds the prompt total'
})
}
}) })
export type AgentPromptModelProfile = z.infer< export type AgentPromptModelProfile = z.infer<
typeof agentPromptModelProfileSchema typeof agentPromptModelProfileSchema
@@ -178,11 +166,6 @@ export const modelBridgePolicySchema = z
protocol: modelBridgeModelProtocolSchema, protocol: modelBridgeModelProtocolSchema,
model: modelNameSchema, model: modelNameSchema,
modelProfileDigest: sha256DigestSchema, modelProfileDigest: sha256DigestSchema,
// Accepted only so persisted pre-removal bindings can be read and
// retired normally. Current code neither emits nor enforces these fields.
maximumOutputTokens: z.number().int().min(1).optional(),
maximumModelCalls: z.number().int().min(1).optional(),
maximumTotalOutputTokens: z.number().int().min(1).optional(),
supportsImageInput: z.boolean() supportsImageInput: z.boolean()
}) })
.strict() .strict()
@@ -116,20 +116,20 @@ describe('model call operation contracts', () => {
).toThrow() ).toThrow()
}) })
it('accepts legacy quota fields without requiring them', () => { it('rejects prompt-wide quota fields', () => {
const request = { const request = {
identity, identity,
requestDigest: `sha256:${'a'.repeat(64)}`, requestDigest: `sha256:${'a'.repeat(64)}`,
...policy ...policy
} }
expect(prepareModelCallSchema.parse(request)).toEqual(request) expect(prepareModelCallSchema.parse(request)).toEqual(request)
expect( expect(() =>
prepareModelCallSchema.parse({ prepareModelCallSchema.parse({
...request, ...request,
maximumOutputTokens: 65_537, maximumOutputTokens: 65_537,
maximumModelCalls: 100_000, maximumModelCalls: 100_000,
maximumTotalOutputTokens: 1_000_000_000 maximumTotalOutputTokens: 1_000_000_000
}) })
).toMatchObject(request) ).toThrow()
}) })
}) })
+1 -9
View File
@@ -60,12 +60,7 @@ export const prepareModelCallSchema = z
.object({ .object({
identity: modelCallIdentitySchema, identity: modelCallIdentitySchema,
requestDigest: modelCallDigestSchema, requestDigest: modelCallDigestSchema,
modelProfileDigest: modelCallDigestSchema, modelProfileDigest: modelCallDigestSchema
// Legacy ledger fields are accepted during migration but no longer
// constrain dispatch.
maximumOutputTokens: positiveCountSchema.optional(),
maximumModelCalls: positiveCountSchema.optional(),
maximumTotalOutputTokens: positiveCountSchema.optional()
}) })
.strict() .strict()
export type PrepareModelCall = z.infer<typeof prepareModelCallSchema> export type PrepareModelCall = z.infer<typeof prepareModelCallSchema>
@@ -164,9 +159,6 @@ export const modelCallRecordSchema = z
identity: modelCallIdentitySchema, identity: modelCallIdentitySchema,
requestDigest: modelCallDigestSchema, requestDigest: modelCallDigestSchema,
modelProfileDigest: modelCallDigestSchema, modelProfileDigest: modelCallDigestSchema,
maximumOutputTokens: positiveCountSchema.optional(),
maximumModelCalls: positiveCountSchema.optional(),
maximumTotalOutputTokens: positiveCountSchema.optional(),
status: modelCallStatusSchema, status: modelCallStatusSchema,
dispatchMetadata: modelCallDispatchMetadataSchema.optional(), dispatchMetadata: modelCallDispatchMetadataSchema.optional(),
terminalEvidence: modelCallTerminalEvidenceSchema.optional(), terminalEvidence: modelCallTerminalEvidenceSchema.optional(),