diff --git a/.env.example b/.env.example index 6e8911fa0..fc1149085 100644 --- a/.env.example +++ b/.env.example @@ -50,6 +50,16 @@ CODEX_PORT=8080 # Codex 上游传输：默认 http；可设 ws 强制使用 WebSocket；auto 在 HTTP 入站下仍走 http # CODEX_UPSTREAM_TRANSPORT=http +# Codex Multi-Agent v2 第三方子 Agent 明文兼容（实验性，默认关闭）。 +# 仅对官方 Codex 客户端、原生 Codex 账号、且请求含 collaboration 下的 +# spawn_agent / send_message / followup_task 时生效：移除 message 的 +# encrypted 标记并把命名空间临时改名为 collaboration-optimize，响应再还原。 +# 开启后无法区分子 Agent 提供方，匹配的 v2 协作消息都会走明文；上游可能按 +# 非标准 schema 拒绝或触发风控，请先在隔离实例/账号池灰度。不解密历史密文； +# 冲突、无效 JSON 或工具数量超限时保持原生路径。 +# 详见 docs/CODEX_MULTI_AGENT_V2_COMPAT.md。 +# CODEX_MULTI_AGENT_V2_COMPAT=false + # 旧版 WebSocket 开关（兼容保留；推荐改用 CODEX_UPSTREAM_TRANSPORT=ws） # USE_WEBSOCKET=true diff --git a/README.md b/README.md index 76867d52f..4859ca7bf 100644 --- a/README.md +++ b/README.md @@ -206,6 +206,7 @@ Antigravity accounts are managed as a dedicated Google channel with browser/impo | [Usage Guide](docs/USAGE.md) | Client setup, SDK examples, media workflows, and troubleshooting | `docs/USAGE.md` | | [API Documentation](docs/API.md) | API endpoints, request and response examples, error codes | `docs/API.md` | | [Antigravity Integration](docs/ANTIGRAVITY.md) | Google OAuth and experimental API Key channel, models, risks, and protocol status | `docs/ANTIGRAVITY.md` | +| [Codex Multi-Agent v2 Compatibility](docs/CODEX_MULTI_AGENT_V2_COMPAT.md) | Opt-in plaintext compatibility for third-party subagents, risks, rollout, rollback | `docs/CODEX_MULTI_AGENT_V2_COMPAT.md` | | [Deployment Guide](docs/DEPLOYMENT.md) | Deployment modes, upgrade guide, backup and restore | `docs/DEPLOYMENT.md` | | [Configuration Guide](docs/CONFIGURATION.md) | Environment variables, system settings, configuration priority | `docs/CONFIGURATION.md` | | [Architecture](docs/ARCHITECTURE.md) | System architecture, scheduling algorithm, storage design | `docs/ARCHITECTURE.md` | diff --git a/README.zh-CN.md b/README.zh-CN.md index f9541fa44..a2c849eb3 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -257,6 +257,7 @@ Antigravity 作为独立 Google 渠道管理，支持浏览器/导入 OAuth 凭 | [API 文档](docs/API.md) | 所有 API 端点、请求/响应示例、错误码说明 | `docs/API.md` | | [使用指南](docs/USAGE.md) | 客户端接入、SDK 示例、媒体任务与常见错误排查 | `docs/USAGE.md` | | [Antigravity 接入](docs/ANTIGRAVITY.md) | Google OAuth、实验性 API Key 渠道、模型、风险与协议验证状态 | `docs/ANTIGRAVITY.md` | +| [Codex Multi-Agent v2 兼容](docs/CODEX_MULTI_AGENT_V2_COMPAT.md) | 第三方子 Agent 明文字段兼容、风险、灰度与回滚 | `docs/CODEX_MULTI_AGENT_V2_COMPAT.md` | | [部署文档](docs/DEPLOYMENT.md) | 各种部署模式、升级指南、备份恢复 | `docs/DEPLOYMENT.md` | | [配置文档](docs/CONFIGURATION.md) | 环境变量、系统设置、配置优先级 | `docs/CONFIGURATION.md` | | [架构文档](docs/ARCHITECTURE.md) | 系统架构、调度算法、存储设计 | `docs/ARCHITECTURE.md` | diff --git a/docs/CODEX_MULTI_AGENT_V2_COMPAT.md b/docs/CODEX_MULTI_AGENT_V2_COMPAT.md new file mode 100644 index 000000000..a7f81e71b --- /dev/null +++ b/docs/CODEX_MULTI_AGENT_V2_COMPAT.md @@ -0,0 +1,240 @@ +# Codex Multi-Agent v2 第三方子 Agent 兼容 + +最后更新：2026-10-18 + +本文说明 Codex2API 的可选 Codex Multi-Agent v2 明文兼容层。它解决的是：GPT 父 Agent 在 v2 下把任务交给第三方子模型（例如 DeepSeek）时，子 Agent 能创建、能回消息，却反复表示“没有收到具体任务”。 + +## 结论与边界 + +这是默认关闭的实验性兼容层，需要在部署环境显式开启。 + +开启后，匹配的 v2 协作调用会改用明文参数。网关无法区分子 Agent 是原生 OpenAI 还是第三方模型：只要父请求匹配，父到子的任务消息都会以明文生成和传输。需要保留原生 OpenAI 子 Agent 加密时，不要开启，或只在隔离实例和账号池开启。 + +兼容层不解密已经生成的历史密文，也不修改 Codex 二进制。它改变的是后续请求的 schema 和参数生成方式。 + +它不改认证、账号选择、模型路由、计费、限流或 User-Agent，也不影响 v1、普通 Responses、Chat Completions、图片、reasoning、压缩、其他工具、relay 账号和非 Codex 客户端。 + +User-Agent 判断是兼容性启发式，不是安全边界。持有 API Key 的调用方可以伪造官方 UA；改写只作用于该调用方自己的请求，但部署方不应把“仅官方客户端”当作强访问控制。 + +## 风险与适用账号 + +- 上游看到的是非标准 `collaboration-optimize` schema。上游可能按 reserved/unknown schema 拒绝，也可能记录或触发账号风控。兼容层没有自动回退：上游拒绝时请求按上游错误返回。 +- 子任务正文会以明文进入第三方模型及其日志、缓存或转发路径。不要把敏感任务交给不受信的第三方。 +- 开关是进程级环境变量，不能按账号或模型排除。多实例处理同一流量时，所有实例必须保持一致。 +- 先在非主力账号或隔离实例灰度，验证 spawn、followup、send_message 和真实工具调用，观察上游 4xx 和账号状态，再扩大范围。 +- 如果账号风控不可接受，稳妥替代是等上游提供官方明文配置，或让子 Agent 走 v1。网关侧改 schema 一定会改变上游看到的请求。 + +## 为什么需要这一层 + +Codex Multi-Agent v2 的 `spawn_agent`、`send_message`、`followup_task` 把 `message` 参数标记为加密。官方 OpenAI 上游在收到该标记时生成密文，Codex 客户端把它存成 `agent_message` 的 `encrypted_content`，只留下 `Payload:` 标头是明文。第三方子模型读不到正文，于是回复“没有任务”。 + +Codex 客户端只有在同时满足以下条件时才走 `DirectPlaintextMessage`： + +1. `namespace` 精确等于 `collaboration`； +2. 工具名是 `spawn_agent`、`send_message` 或 `followup_task`； +3. `encrypted_function_args` 是显式空数组 `[]`。缺失或 `null` 不等价。 + +加密发生在父模型生成参数时，网关事后无法解密。兼容点必须在父请求的工具 schema，也就是父模型生成参数之前。 + +Codex 上游对 `collaboration.*` 保留工具要求 schema 逐字匹配；直接删改 `message.encrypted` 会被上游按 reserved schema 拒绝。兼容层因此使用临时命名空间 `collaboration-optimize`，并在响应返回客户端前还原。 + +源码依据： + +- `codex-rs/core/src/tools/router.rs`：`ToolCall::direct_source()` 的判定条件。 +- `codex-rs/core/src/tools/handlers/multi_agents_v2.rs`：非明文来源变成 `AgentMessage::Encrypted`。 +- `codex-rs/core/src/tools/handlers/multi_agents_spec.rs`：三个协作工具的 `message` 带 `.with_encrypted()`。 + +## 与 CLIProxyAPI 同类实现的关系 + +CLIProxyAPI 已经实现同类能力（MIT），本方案参考了它的核心思路： + +- 源码：[optimize_multi_agent_v2.go](https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go) + - [`OptimizeCodexMultiAgentV2Request`](https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go#L138-L155)：删除加密标记并临时改名。 + - [`PrepareCodexMultiAgentV2Tools`](https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go#L112-L136)：在 Responses 边界准备协作工具。 + - [`optimizeCodexCollaborationNamespace`](https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go#L695-L716)：把 `collaboration` 改为 `collaboration-optimize`。 + - [`RestoreCodexMultiAgentV2Response`](https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go#L718-L739)：在响应中还原命名空间。 + - [`rewriteCodexAgentMessageContent`](https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go#L834-L874)：为子侧转换 `agent_message`。 +- 配置：[`client.codex.optimize-multi-agent-v2`](https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/config.example.yaml#L327)，默认 `false`，OAuth 和 API-key 凭据都适用。 +- 许可：[MIT License](https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/LICENSE)。 + +本实现沿用核心变换，差异是：只在原生 Codex executor 生效；冲突、无效 JSON、嵌套或数量超限时 fail-closed；不把 `encrypted_content` 重标为 `input_text`；只处理 collaboration 工具族并按 namespace 限定。 + +相关 issue： + +- [CLIProxyAPI #4746](https://github.com/router-for-me/CLIProxyAPI/issues/4746)：`followup_task` 仍保持加密。 +- [CLIProxyAPI #4801](https://github.com/router-for-me/CLIProxyAPI/issues/4801)：第三方 Responses provider 不支持 `agent_message`。 +- [CLIProxyAPI #6233](https://github.com/router-for-me/CLIProxyAPI/issues/6233)：compat 模式的 `author` / `recipient` 处理。 + +## 启用与灰度 + +```bash +CODEX_MULTI_AGENT_V2_COMPAT=true +``` + +默认 `false`，接受 `1`、`true`、`yes`、`on`。变量按进程读取，修改后需要重启。 + +灰度步骤： + +1. 选一个隔离实例或独立账号池，不要让负载均衡把同一客户端随机分到开启和未开启的实例。 +2. 开启后让一个真实 Codex 会话执行 `spawn_agent`、`followup_task`、`send_message` 和一次只读工具调用。 +3. 成功判据是子 Agent 返回任务结果；`spawn_agent` 返回 id 只说明子线程建好了。 +4. 检查子请求的 `agent_message` 只包含 `input_text`，没有 `encrypted_content`。 +5. 观察上游 4xx、账号状态和兼容层日志。出现异常时关闭开关并重启所有相关实例。 + +## 精确生效条件 + +兼容层同时检查： + +- `CODEX_MULTI_AGENT_V2_COMPAT` 为真； +- 下游 `User-Agent` 是官方 Codex 客户端（`Codex Desktop/`、`codex-tui/`、`codex_cli_rs`、`codex_exec/`）； +- 实际选中原生 Codex 账号，relay 账号不会进入该执行器； +- 请求包含顶层 `collaboration` namespace 下的 `spawn_agent`、`send_message` 或 `followup_task`，或顶层的扁平 `collaboration.` / `collaboration__`； +- 这些工具中至少一个带 `parameters.properties.message.encrypted`； +- 请求没有已存在的 `collaboration-optimize` 命名空间冲突； +- 工具嵌套不超过 8 层，待改写工具不超过 16 个；超出时保持原请求并记录跳过日志。 + +任一条件不满足时不改写。 + +已验证的客户端版本：Codex CLI `0.162.0-alpha.17.2`。Codex Desktop 和其他版本没有单独验证；升级后需要重跑验收。 + +## 协议处理 + +请求侧： + +- 扫描顶层 `tools[]` 和 `input[].additional_tools[].tools[]`； +- 只识别顶层 `collaboration` namespace 下的三个 function，以及顶层扁平 `collaboration.` / `collaboration__`；其他 namespace 下的同名工具不改写； +- 删除这三个工具的 `parameters.properties.message.encrypted`； +- 把 `collaboration` namespace 改名为 `collaboration-optimize`；扁平工具改名为 `collaboration-optimize.` 或 `collaboration-optimize__`； +- 同步改写指向这些工具的 `tool_choice`，包括 `function`、`custom` 和 `allowed_tools` 形态； +- 其他工具和无关字段保持原样。 + +响应侧： + +- 在 `item`、`output`、`response.output`、`tools`、`response.tools`、`tool_choice`、`response.tool_choice` 中还原临时命名空间； +- 工具调用还原为 `namespace: collaboration` + `name: `；namespace 声明、回显的工具定义和 `tool_choice` 恢复原前缀； +- 上游省略 `encrypted_function_args` 或返回 `null` 时，如果 `arguments` 是合法 JSON、`message` 是字符串且不像已知密文，补全 `encrypted_function_args: []`； +- 兼容 `collaboration-optimize.` 和 `collaboration-optimize__`； +- 响应回显的扁平工具定义恢复为 `collaboration.`； +- `arguments`、`input`、`output` 等不透明字段不改写； +- 流式 SSE、非流式 JSON、HTTP 和原生 Responses WebSocket 共用同一还原逻辑； +- 数字保持原精度。 + +上游错误体不在还原范围。如果上游 4xx 文本包含 `collaboration-optimize`，客户端可能看到这个临时名；它是内部兼容细节，不包含任务正文。 + +## 安全失败策略 + +- 不解析任务正文，不解密，也不把 `encrypted_content` 重标为 `input_text`。 +- 补全 `encrypted_function_args: []` 前检查 `arguments` 是否为字符串、`message` 是否非空；缺字段、非法 JSON、空文本、非字符串和已知密文形态都保持原样。密文识别是启发式（`gAAAA`、JWE `eyJ`、`enc_`、长 base64url token），不是完整保证。 +- 请求已有 `collaboration-optimize` 冲突、JSON 无效、嵌套或数量超限时，保持原请求并记录跳过日志，不做半转换。 +- 响应还原只在本次请求真的执行过改名时生效；未改名的请求不触发任何响应改写。 +- 日志记录账号、模型、工具名、是否生效、是否疑似密文和跳过原因，不记录任务正文、认证信息或原始 SSE。 +- 上游拒绝改写后的 schema 时不会自动回退，表现为 4xx。看到生效日志和 4xx 同时出现时，先关闭开关并保留脱敏请求样本。 + +## 与典型接入方式的关系 + +默认路径只处理父请求的 schema 和响应 namespace 还原，不处理 `agent_message`。子请求通常由上游网关直接转发给第三方模型。 + +子侧是否可用取决于该网关：本仓自带的 Responses→Chat 转换目前不识别 `agent_message`，会返回 `cannot be represented by Chat Completions`；子网关需要原生支持全明文 `agent_message`，或做等价转换。早期验证中的 NewAPI 会忽略该项；后续子侧规范化修复已合入 [NewAPI fork PR #259](https://github.com/DeliciousBuding/new-api/pull/259)，配合本兼容层完成了端到端验收。其他子网关仍需检查自己的处理能力。 + +子侧转换只能处理全明文 `input_text`，不能把 `encrypted_content` 当作任务文本。本 PR 只负责父侧，不包含未接线的子侧适配器。 + +## 验证 + +截至 2026-10-18：单元测试、集成测试和真实 Codex CLI + mock 上游通过；父侧也已通过真实生产链路验证。 + +开启兼容层后，父模型的 `spawn_agent` 调用被补全为 `encrypted_function_args: []`，子请求的 `agent_message` 只包含明文 `input_text`，正文与父模型发出的任务一致。 + +子侧是独立依赖：第三方子 Agent 请求通常不经过本仓。早期验证确认，未转换 `agent_message` 时模型读不到任务；后续接入全明文规范化后，GPT v2 父 → DeepSeek 子的初始派发、同一子 Agent 的 `followup_task` 及两次只读工具调用通过，原生 GPT 子 Agent 回归通过。父调用的 `encrypted_function_args=[]`，子消息只包含 `input_text`。这次验收没有单独覆盖 `send_message`，也不覆盖第三方父模型强制 v2。 + +单元与集成测试覆盖： + +- 顶层和 `additional_tools`、嵌套 `collaboration` namespace、扁平 `collaboration.*` / `collaboration__*` 工具形态； +- 三个协作工具全部覆盖，其他 namespace 下的同名工具不受影响； +- 无 `spawn_agent`、只有 `send_message` / `followup_task` 的请求； +- 开关关闭、非 Codex 客户端、命名空间冲突、无效 JSON、嵌套或数量超限均保持原样； +- `tool_choice` 的 `function`、`custom`、`allowed_tools` 形态同步改写； +- SSE、非流式 JSON、HTTP 和 WebSocket 的还原； +- 扁平工具定义、双下划线、点号形式还原； +- 上游省略或返回 `null` 的 `encrypted_function_args`：明文字符串补全为 `[]`，密文、非法参数和缺字段保持原样； +- `arguments` 不被改写、数字精度保持、幂等； + +真实 Codex CLI + mock 上游： + +```bash +python scripts/repro-codex-multi-agent-v2.py --codex-exe /path/to/codex --child-model --mode native +python scripts/repro-codex-multi-agent-v2.py --codex-exe /path/to/codex --child-model --mode compat +``` + +`--child-model` 由调用者按本机 Codex catalog 选择；文档不固定具体模型名。 + +- native：子请求 `agent_message` = `input_text` 标头 + `encrypted_content`，复现失败形态； +- compat：子请求 `agent_message` 只包含 `input_text`。 + +脚本验证客户端判定规则；Go 集成测试验证网关还原；真实上游结果见下。脚本自身不读取或输出凭据，模型请求只发往本地 mock，Codex CLI 的 `CODEX_HOME` 指向本次运行的临时目录。 + +初次真实上游验证（2026-10-18，子侧修复前）： + +- 父请求已切换到明文协作工具，上游 `spawn_agent` 返回的调用被补全 `encrypted_function_args: []`； +- 子请求 `agent_message` 只包含 `input_text`，正文与父模型发出的任务一致； +- 子侧网关（NewAPI）未把 `agent_message` 转成普通 user message，模型仍读不到任务；同一条请求改成 `message` + `role=user` 后返回正确结果； +- 因此当时父侧边界已通过验证，子侧转换是独立阻塞项。 + +后续端到端验收（2026-10-18，子侧修复后）： + +- GPT v2 父 → DeepSeek 子的 `spawn_agent` 与同一子 Agent 的 `followup_task` 均返回正确任务结果； +- 子 Agent 完成两次只读工具调用，原生 GPT 子 Agent 回归通过； +- 父两次协作调用的 `encrypted_function_args=[]`，子两条 `agent_message` 均仅含 `input_text`； +- 上述结果依赖父侧兼容与子侧规范化共同生效，不表示任意子网关都已兼容。 + +额外验收清单： + +- `send_message` 的独立真实调用及其他客户端/子网关组合； +- 升级上游或客户端后，复验任务正文、后续派发、工具调用与上游错误。 + +## 已知限制 + +- 依赖 Codex 上游遵守修改后的 schema 并生成明文 `arguments`。上游省略 `encrypted_function_args` 时网关会补全；若 `message` 仍像密文，则不补并保留诊断。 +- 密文识别是启发式，不是完整保证；当前原生 Codex 上游使用 `gAAAA` 前缀，未知格式可能被误判或漏判。命中时选择不补，保持原生加密行为。 +- 无法区分原生 OpenAI 和第三方子 Agent；开启后所有匹配的 v2 协作调用都走明文。 +- 上游拒绝时不会自动回退，表现为 4xx。 +- 临时 namespace 是内部兼容细节；如果上游 reserved schema 或客户端判定规则变化，需要重新评估。 +- 只覆盖官方 Codex 客户端的 v2 协作工具；v1、普通工具和其他协议不受影响。 +- 上游错误体不还原；客户端可能看到临时命名空间。 +- 请求历史中的 `function_call` 记录保持原 namespace；真实账号已接受该历史，上游策略变化时需要复验。 +- User-Agent 判断不是安全边界。 +- 子侧网关需支持全明文 `agent_message` 或做等价转换。已修复的 NewAPI fork 组合验收通过，其他组合需自行验证。 +- 第三方父模型强制 v2 不在本兼容层覆盖范围；已验收的选择是第三方父模型使用 v1、GPT 父模型使用 v2。 + +## 回滚 + +把 `CODEX_MULTI_AGENT_V2_COMPAT` 设为 `false` 或删除变量，重启所有处理相关流量的实例。没有数据库迁移或持久状态。重启会断开在途的流式请求和 WebSocket 连接，客户端需要重连。 + +## 参考 + +Codex 客户端 `0.162.0-alpha.17.2`： + +- `codex-rs/core/src/tools/router.rs` +- `codex-rs/core/src/tools/handlers/multi_agents_v2.rs` +- `codex-rs/core/src/tools/handlers/multi_agents_spec.rs` + +OpenAI Codex： + +- [PR #26210](https://github.com/openai/codex/pull/26210)：Encrypt multi-agent v2 message payloads +- [PR #35845](https://github.com/openai/codex/pull/35845)：Support plaintext collaboration tool messages +- [Issue #36376](https://github.com/openai/codex/issues/36376)：OpenAI parent still sends encrypted V2 task to non-OpenAI child +- [Issue #36586](https://github.com/openai/codex/issues/36586)：Subagent task payload invisible to custom non-OpenAI providers +- [Issue #46939](https://github.com/openai/codex/issues/46939)：Add a config option to select plaintext delivery + +codex2api： + +- [Issue #342](https://github.com/james-6-23/codex2api/issues/342)：reserved `collaboration.spawn_agent` schema 报错 +- [Issue #741](https://github.com/james-6-23/codex2api/issues/741) / [PR #742](https://github.com/james-6-23/codex2api/pull/742)：Codex 客户端工具和历史兼容 + +CLIProxyAPI（MIT）： + +- [optimize_multi_agent_v2.go](https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go) +- [配置项 `client.codex.optimize-multi-agent-v2`](https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/config.example.yaml#L327)，默认 `false`，OAuth 和 API-key 凭据都适用 +- [MIT License](https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/LICENSE) +- [Issue #4746](https://github.com/router-for-me/CLIProxyAPI/issues/4746)：`followup_task` 仍保持加密 +- [Issue #4801](https://github.com/router-for-me/CLIProxyAPI/issues/4801)：第三方 Responses provider 不支持 `agent_message` +- [Issue #6233](https://github.com/router-for-me/CLIProxyAPI/issues/6233)：compat 模式的 `author` / `recipient` 处理 diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 7f07ee404..918e4f2a4 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -91,6 +91,7 @@ Codex2API 采用三层配置架构： | `CODEX_COMPACTION_AFFINITY_TTL` | 否 | `168h` | 加密压缩状态的来源亲和 TTL。缓存仅保存密文的 SHA-256 摘要、来源账号和兼容域；已知状态不会跨 Codex 官方、不同 Responses 中转或 Grok 上游流转 | | `CODEX_FINGERPRINT_DEBUG` | 否 | `false` | 输出脱敏指纹策略诊断日志，不记录 token | | `CODEX_REQUEST_COMPRESSION` | 否 | 跟随系统设置 | 覆盖系统设置「Codex HTTP 请求体压缩」。`zstd`/`on`/`true`/`1` 强制开启，`off`/`false`/`0` 强制关闭，未设置或取值无法识别时以系统设置为准。作为部署级逃生阀存在：DB 不可达或后台打不开时仍可整机切换 | +| `CODEX_MULTI_AGENT_V2_COMPAT` | 否 | `false` | 实验性：为官方 Codex 客户端的 Multi-Agent v2 协作工具启用明文参数兼容。仅原生 Codex 账号生效，重启生效；开启后匹配的 v2 子任务正文会以明文发送。详见 `docs/CODEX_MULTI_AGENT_V2_COMPAT.md` | | `CODEX_TELEMETRY_ENABLED` | 否 | 跟随系统设置 | 设为 `false` 时无视管理后台「客户端遥测」开关，部署层强制关闭模拟遥测外发 | | `CODEX_STATSIG_API_KEY` | 否 | 内置公开 key | 覆盖 Codex Desktop/CLI 共用的公开 Statsig SDK key，仅遥测开启时使用 | | `CODEX_SESSION_HEADER_MODE` | 否 | `native` | 出站会话头形态。`native` 发真实客户端的 `session-id` / `thread-id` / `x-client-request-id`；`legacy` 回退到旧的 `Session_id`（WS 另带 `Conversation_id`） | diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 9db2dd4b4..3564eb983 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -335,6 +335,10 @@ REDIS_ADDR=rediss://default:your-redis-password@your-redis-host:6379/0 ```bash # 快速调度器 FAST_SCHEDULER_ENABLED=true + +# 实验性：Codex Multi-Agent v2 第三方子 Agent 明文兼容（默认关闭） +# CODEX_MULTI_AGENT_V2_COMPAT=false +# 说明与灰度步骤见 docs/CODEX_MULTI_AGENT_V2_COMPAT.md ``` ### 2. 系统设置（通过管理后台） diff --git a/proxy/codex_multi_agent_v2.go b/proxy/codex_multi_agent_v2.go new file mode 100644 index 000000000..f45278527 --- /dev/null +++ b/proxy/codex_multi_agent_v2.go @@ -0,0 +1,675 @@ +package proxy + +import ( + "bytes" + "context" + "fmt" + "log" + "os" + "strings" + "sync" + + "github.com/tidwall/gjson" + "github.com/tidwall/sjson" +) + +// Codex multi-agent v2 兼容层。 +// +// 背景：Codex CLI/Desktop 的 v2 collaboration 工具把 spawn_agent / +// send_message / followup_task 的 message 参数标记为 encrypted。官方 OpenAI +// 上游在收到该标记时会生成密文，并由客户端存为 agent_message 的 +// encrypted_content；第三方子模型无法读取。Codex 只在 function_call 带有 +// 显式空数组 encrypted_function_args 且 namespace 为 collaboration 时走 +// DirectPlaintextMessage 分支。 +// +// 兼容层在父请求发往 Codex 上游前做两件事： +// 1. 删除三个协作工具 message schema 上的 encrypted 标记； +// 2. 把 collaboration 命名空间临时改名为 collaboration-optimize，避开上游 +// reserved schema 的逐字校验（issue #342）。 +// +// 上游响应中的 collaboration-optimize 再还原为 collaboration，使 Codex 按 +// 明文路径处理。无法安全判断时保持原生路径，不猜测密文内容。 +// +// 核心变换参考 CLIProxyAPI 的同类实现（MIT）： +// https://github.com/router-for-me/CLIProxyAPI/blob/911f49c/internal/client/codex/optimize-multi-agent-v2/optimize_multi_agent_v2.go +const ( + codexMultiAgentV2CompatEnv = "CODEX_MULTI_AGENT_V2_COMPAT" + + codexCollaborationNamespace = "collaboration" + codexNamespaceDotPrefix = codexCollaborationNamespace + "." + codexNamespaceDoublePrefix = codexCollaborationNamespace + "__" + codexOptimizedCollaborationNamespace = "collaboration-optimize" + codexOptimizedNamespaceDotPrefix = codexOptimizedCollaborationNamespace + "." + codexOptimizedNamespaceDoublePrefix = codexOptimizedCollaborationNamespace + "__" + + // 官方客户端的协作工具只有一层 namespace。限制深度和改写数量， + // 避免异常请求用深层嵌套或大量同名工具放大扫描/复制成本。 + codexMultiAgentV2MaxToolDepth = 8 + codexMultiAgentV2MaxToolPaths = 16 +) + +var codexCollaborationMessageTools = map[string]struct{}{ + "spawn_agent": {}, + "send_message": {}, + "followup_task": {}, +} + +var ( + codexCollaborationNamespaceBytes = []byte(codexCollaborationNamespace) + codexOptimizedNamespaceBytes = []byte(codexOptimizedCollaborationNamespace) + codexFunctionCallBytes = []byte("function_call") +) + +// codexMultiAgentV2State 记录本次请求是否真的执行了命名空间改写。 +// 只有改写过时才需要在响应侧还原，避免普通请求被二次处理。 +// 账号、模型和诊断去重使用互斥锁，因为 attempt 协程可能被提前遗弃， +// 而响应回调仍会读取状态。 +type codexMultiAgentV2State struct { + mu sync.Mutex + optimized bool + accountID int64 + model string + diagnosed map[string]struct{} +} + +func (s *codexMultiAgentV2State) markOptimized(accountID int64, model string) { + if s == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + s.optimized = true + s.accountID = accountID + s.model = model +} + +func (s *codexMultiAgentV2State) isOptimizedFor(accountID int64) bool { + if s == nil { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + return s.optimized && s.accountID == accountID +} + +// isOptimized 表示本次请求确实执行过命名空间改写。响应补全在函数内部再检查 +// 一次，避免未来新增响应路径忘记外层账号门槛时误补 marker。 +func (s *codexMultiAgentV2State) isOptimized() bool { + if s == nil { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + return s.optimized +} + +func (s *codexMultiAgentV2State) diagnosticContext() (int64, string) { + if s == nil { + return 0, "" + } + s.mu.Lock() + defer s.mu.Unlock() + return s.accountID, s.model +} + +// markDiagnosed 返回 true 表示同一 call_id 已经记录过，调用方应跳过日志。 +func (s *codexMultiAgentV2State) markDiagnosed(key string) bool { + if s == nil || key == "" { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + if s.diagnosed == nil { + s.diagnosed = make(map[string]struct{}) + } + if _, logged := s.diagnosed[key]; logged { + return true + } + s.diagnosed[key] = struct{}{} + return false +} + +type codexMultiAgentV2StateKey struct{} + +func withCodexMultiAgentV2State(ctx context.Context, state *codexMultiAgentV2State) context.Context { + if state == nil { + return ctx + } + return context.WithValue(ctx, codexMultiAgentV2StateKey{}, state) +} + +func codexMultiAgentV2StateFromContext(ctx context.Context) *codexMultiAgentV2State { + if ctx == nil { + return nil + } + state, _ := ctx.Value(codexMultiAgentV2StateKey{}).(*codexMultiAgentV2State) + return state +} + +// codexMultiAgentV2CompatEnabled 默认关闭，部署方通过环境变量显式启用。 +func codexMultiAgentV2CompatEnabled() bool { + switch strings.ToLower(strings.TrimSpace(os.Getenv(codexMultiAgentV2CompatEnv))) { + case "1", "true", "yes", "on": + return true + default: + return false + } +} + +// isCodexMultiAgentV2Client 只匹配官方 Codex 客户端身份，避免把普通 +// Responses/Chat 请求或第三方客户端纳入改写。 +func isCodexMultiAgentV2Client(userAgent string) bool { + userAgent = strings.TrimSpace(userAgent) + return strings.HasPrefix(userAgent, "Codex Desktop/") || + strings.HasPrefix(userAgent, "codex-tui/") || + userAgent == "codex_cli_rs" || + strings.HasPrefix(userAgent, "codex_cli_rs/") || + strings.HasPrefix(userAgent, "codex_exec/") +} + +type codexFlatToolRename struct { + suffix string + prefix string +} + +type codexMultiAgentV2ToolScan struct { + messagePaths []string + flatNames map[string]codexFlatToolRename // tool path -> 原名后缀与改名后的前缀 + namespacePaths []string + namespaceRecords map[string]struct{} + conflict bool + limitExceeded bool +} + +// prepareCodexMultiAgentV2Request 是纯函数：返回改写后的请求体以及是否执行了 +// 命名空间改名。任何不确定/冲突情况都返回原 body，让上游保持原生行为。 +func prepareCodexMultiAgentV2Request(body []byte) ([]byte, bool, error) { + if !bytes.Contains(body, codexCollaborationNamespaceBytes) { + return body, false, nil + } + if !gjson.ValidBytes(body) { + return body, false, fmt.Errorf("invalid JSON request body") + } + + scan := codexMultiAgentV2ToolScan{} + roots := gjson.GetManyBytes(body, "tools", "input") + if tools := roots[0]; tools.IsArray() { + scanCodexMultiAgentV2Tools(tools, "tools", "", "", true, 0, &scan) + } + if input := roots[1]; input.IsArray() { + for index, item := range input.Array() { + if !item.IsObject() || strings.TrimSpace(item.Get("type").String()) != "additional_tools" { + continue + } + scanCodexMultiAgentV2Tools(item.Get("tools"), fmt.Sprintf("input.%d.tools", index), "", "", true, 0, &scan) + } + } + + if scan.conflict { + return body, false, fmt.Errorf("request defines reserved %q namespace", codexOptimizedCollaborationNamespace) + } + if scan.limitExceeded { + return body, false, fmt.Errorf("collaboration tool scan exceeds nesting or count limit") + } + // 只处理带 message.encrypted 的 collaboration 工具：没有可删的加密标记时 + // 保持原生路径，避免无意义地改 schema 和命名空间。 + if len(scan.messagePaths) == 0 || (len(scan.namespacePaths) == 0 && len(scan.flatNames) == 0) { + return body, false, nil + } + + updated := body + for _, path := range scan.messagePaths { + var err error + updated, err = sjson.DeleteBytes(updated, path+".parameters.properties.message.encrypted") + if err != nil { + return body, false, err + } + } + for _, path := range scan.namespacePaths { + var err error + updated, err = sjson.SetBytes(updated, path+".name", codexOptimizedCollaborationNamespace) + if err != nil { + return body, false, err + } + } + for path, rename := range scan.flatNames { + var err error + updated, err = sjson.SetBytes(updated, path+".name", rename.prefix+rename.suffix) + if err != nil { + return body, false, err + } + } + updated, err := rewriteCodexMultiAgentV2ToolChoice(updated, &scan) + if err != nil { + return body, false, err + } + return updated, true, nil +} + +// rewriteCodexMultiAgentV2ToolChoice 同步改写指向协作工具的 tool_choice， +// 避免工具定义改名后 choice 仍指向旧名导致上游 400。只有本次确实改过 +// namespace 或对应扁平工具时才改写。 +func rewriteCodexMultiAgentV2ToolChoice(body []byte, scan *codexMultiAgentV2ToolScan) ([]byte, error) { + choice := gjson.GetBytes(body, "tool_choice") + if !choice.IsObject() { + return body, nil + } + switch strings.TrimSpace(choice.Get("type").String()) { + case "function", "custom": + return rewriteCodexMultiAgentV2ToolChoiceItem(body, "tool_choice", choice, scan) + case "allowed_tools": + tools := choice.Get("tools") + if !tools.IsArray() { + return body, nil + } + updated := body + for index, item := range tools.Array() { + var err error + updated, err = rewriteCodexMultiAgentV2ToolChoiceItem( + updated, fmt.Sprintf("tool_choice.tools.%d", index), item, scan, + ) + if err != nil { + return body, err + } + } + return updated, nil + default: + return body, nil + } +} + +func rewriteCodexMultiAgentV2ToolChoiceItem(body []byte, path string, item gjson.Result, scan *codexMultiAgentV2ToolScan) ([]byte, error) { + if !item.IsObject() { + return body, nil + } + namespace := strings.TrimSpace(item.Get("namespace").String()) + name := strings.TrimSpace(item.Get("name").String()) + updated := body + if scan != nil && len(scan.namespacePaths) > 0 && namespace == codexCollaborationNamespace { + var err error + updated, err = sjson.SetBytes(updated, path+".namespace", codexOptimizedCollaborationNamespace) + if err != nil { + return body, err + } + } + var suffix, renamePrefix string + switch { + case strings.HasPrefix(name, codexNamespaceDotPrefix): + suffix = strings.TrimPrefix(name, codexNamespaceDotPrefix) + renamePrefix = codexOptimizedNamespaceDotPrefix + case strings.HasPrefix(name, codexNamespaceDoublePrefix): + suffix = strings.TrimPrefix(name, codexNamespaceDoublePrefix) + renamePrefix = codexOptimizedNamespaceDoublePrefix + } + if suffix != "" && scan != nil && scan.flatToolRenamed(suffix) { + var err error + updated, err = sjson.SetBytes(updated, path+".name", renamePrefix+suffix) + if err != nil { + return body, err + } + } + return updated, nil +} + +func (s *codexMultiAgentV2ToolScan) flatToolRenamed(name string) bool { + if s == nil { + return false + } + for _, rename := range s.flatNames { + if rename.suffix == name { + return true + } + } + return false +} + +// scanCodexMultiAgentV2Tools 只识别协作工具族的两种正式形态： +// 顶层 collaboration namespace 下的三个 function，以及顶层扁平 +// collaboration.。其他 namespace 或裸工具名即使同名也不改写。 +// +// 只有出现 parameters.properties.message.encrypted 时才记录，避免对已经 +// 明文的 schema 做无意义改名。 +func scanCodexMultiAgentV2Tools(tools gjson.Result, path, namespace, namespacePath string, root bool, depth int, scan *codexMultiAgentV2ToolScan) { + if !tools.IsArray() { + return + } + if depth > codexMultiAgentV2MaxToolDepth { + scan.limitExceeded = true + return + } + for index, tool := range tools.Array() { + if !tool.IsObject() { + continue + } + toolType := strings.TrimSpace(tool.Get("type").String()) + name := strings.TrimSpace(tool.Get("name").String()) + + if isCodexOptimizedNamespaceName(name) { + scan.conflict = true + continue + } + + switch toolType { + case "function": + if namespace == codexCollaborationNamespace { + if _, ok := codexCollaborationMessageTools[name]; !ok { + continue + } + if !tool.Get("parameters.properties.message.encrypted").Exists() { + continue + } + if len(scan.messagePaths) >= codexMultiAgentV2MaxToolPaths { + scan.limitExceeded = true + return + } + scan.messagePaths = append(scan.messagePaths, fmt.Sprintf("%s.%d", path, index)) + if namespacePath != "" { + if _, recorded := scan.namespaceRecords[namespacePath]; !recorded { + if scan.namespaceRecords == nil { + scan.namespaceRecords = make(map[string]struct{}) + } + scan.namespaceRecords[namespacePath] = struct{}{} + scan.namespacePaths = append(scan.namespacePaths, namespacePath) + } + } + continue + } + if root { + var suffix, renamePrefix string + switch { + case strings.HasPrefix(name, codexNamespaceDotPrefix): + suffix = strings.TrimPrefix(name, codexNamespaceDotPrefix) + renamePrefix = codexOptimizedNamespaceDotPrefix + case strings.HasPrefix(name, codexNamespaceDoublePrefix): + suffix = strings.TrimPrefix(name, codexNamespaceDoublePrefix) + renamePrefix = codexOptimizedNamespaceDoublePrefix + } + if suffix == "" { + continue + } + if _, ok := codexCollaborationMessageTools[suffix]; !ok { + continue + } + if !tool.Get("parameters.properties.message.encrypted").Exists() { + continue + } + if len(scan.messagePaths) >= codexMultiAgentV2MaxToolPaths { + scan.limitExceeded = true + return + } + toolPath := fmt.Sprintf("%s.%d", path, index) + scan.messagePaths = append(scan.messagePaths, toolPath) + if scan.flatNames == nil { + scan.flatNames = make(map[string]codexFlatToolRename) + } + scan.flatNames[toolPath] = codexFlatToolRename{suffix: suffix, prefix: renamePrefix} + } + case "namespace": + childNamespace := "" + if root && name == codexCollaborationNamespace { + childNamespace = codexCollaborationNamespace + } + toolPath := fmt.Sprintf("%s.%d", path, index) + scanCodexMultiAgentV2Tools(tool.Get("tools"), toolPath+".tools", childNamespace, toolPath, false, depth+1, scan) + } + } +} + +func isCodexOptimizedNamespaceName(name string) bool { + return name == codexOptimizedCollaborationNamespace || + strings.HasPrefix(name, codexOptimizedNamespaceDotPrefix) || + strings.HasPrefix(name, codexOptimizedNamespaceDoublePrefix) +} + +// restoreCodexMultiAgentV2Response 把上游响应中的临时命名空间还原为 +// collaboration。只改写工具调用项和 namespace 声明，arguments/input/output +// 等不透明字段保持原样。 +// +// 这里按已知响应容器定向改写，而不是整包 decode 成 any 再 marshal： +// response.completed 会回显完整工具数组，整包重编码会放大延迟和垃圾。 +func restoreCodexMultiAgentV2Response(payload []byte, state *codexMultiAgentV2State) []byte { + if len(payload) == 0 { + return payload + } + if !bytes.Contains(payload, codexOptimizedNamespaceBytes) { + // 上游可能把 namespace 归一化回 collaboration；本次请求确实改写过时， + // 仍要检查 function_call 是否缺 encrypted_function_args。 + if state == nil || !state.isOptimized() || !bytes.Contains(payload, codexFunctionCallBytes) { + return payload + } + } + if !gjson.ValidBytes(payload) { + return payload + } + + paths := [...]string{"item", "output", "response.output", "tools", "response.tools", "tool_choice", "response.tool_choice"} + results := gjson.GetManyBytes(payload, paths[:]...) + updated := payload + for index, result := range results { + var err error + updated, err = restoreCodexMultiAgentV2Container(updated, paths[index], result, state) + if err != nil { + return payload + } + } + return updated +} + +func restoreCodexMultiAgentV2Container(updated []byte, path string, container gjson.Result, state *codexMultiAgentV2State) ([]byte, error) { + switch { + case container.IsObject(): + return restoreCodexMultiAgentV2Item(updated, path, container, state) + case container.IsArray(): + var err error + for index, item := range container.Array() { + updated, err = restoreCodexMultiAgentV2Item(updated, fmt.Sprintf("%s.%d", path, index), item, state) + if err != nil { + return updated, err + } + } + } + return updated, nil +} + +func restoreCodexMultiAgentV2Item(updated []byte, path string, item gjson.Result, state *codexMultiAgentV2State) ([]byte, error) { + if !item.IsObject() { + return updated, nil + } + itemType := strings.TrimSpace(item.Get("type").String()) + isToolCall := itemType == "function_call" || itemType == "custom_tool_call" + if isToolCall { + namespace := strings.TrimSpace(item.Get("namespace").String()) + name := strings.TrimSpace(item.Get("name").String()) + restoredName := name + restoreNamespace := namespace == codexOptimizedCollaborationNamespace + switch { + case strings.HasPrefix(name, codexOptimizedNamespaceDotPrefix): + restoredName = strings.TrimPrefix(name, codexOptimizedNamespaceDotPrefix) + restoreNamespace = true + case strings.HasPrefix(name, codexOptimizedNamespaceDoublePrefix): + restoredName = strings.TrimPrefix(name, codexOptimizedNamespaceDoublePrefix) + restoreNamespace = true + } + if restoreNamespace { + var err error + updated, err = sjson.SetBytes(updated, path+".namespace", codexCollaborationNamespace) + if err != nil { + return updated, err + } + } + // 上游可能直接返回 collaboration，而不是临时命名空间；只要本次请求 + // 确实改写过，且是三个协作工具之一，就检查并按需补全明文标记。 + if itemType == "function_call" && (restoreNamespace || namespace == codexCollaborationNamespace) { + var err error + var injected bool + updated, injected, err = ensureCodexPlaintextFunctionArgs(updated, path, restoredName, item, state) + if err != nil { + return updated, err + } + if !injected { + diagnoseCodexMultiAgentV2ToolCall(restoredName, item, state) + } + } + if restoredName != name { + var err error + updated, err = sjson.SetBytes(updated, path+".name", restoredName) + if err != nil { + return updated, err + } + } + return updated, nil + } + if itemType == "namespace" && strings.TrimSpace(item.Get("name").String()) == codexOptimizedCollaborationNamespace { + var err error + updated, err = sjson.SetBytes(updated, path+".name", codexCollaborationNamespace) + if err != nil { + return updated, err + } + return updated, nil + } + if itemType == "allowed_tools" { + tools := item.Get("tools") + if tools.IsArray() { + for index, child := range tools.Array() { + var err error + updated, err = restoreCodexMultiAgentV2Item( + updated, fmt.Sprintf("%s.tools.%d", path, index), child, state, + ) + if err != nil { + return updated, err + } + } + } + return updated, nil + } + // 部分响应会回显工具定义或 tool_choice；扁平工具在请求侧被改成 + // collaboration-optimize.，返回客户端前恢复原前缀。 + if itemType == "function" || itemType == "custom" { + if namespace := strings.TrimSpace(item.Get("namespace").String()); namespace == codexOptimizedCollaborationNamespace { + var err error + updated, err = sjson.SetBytes(updated, path+".namespace", codexCollaborationNamespace) + if err != nil { + return updated, err + } + } + name := strings.TrimSpace(item.Get("name").String()) + var restoredName string + switch { + case strings.HasPrefix(name, codexOptimizedNamespaceDotPrefix): + restoredName = codexNamespaceDotPrefix + strings.TrimPrefix(name, codexOptimizedNamespaceDotPrefix) + case strings.HasPrefix(name, codexOptimizedNamespaceDoublePrefix): + restoredName = codexCollaborationNamespace + "__" + strings.TrimPrefix(name, codexOptimizedNamespaceDoublePrefix) + } + if restoredName != "" && restoredName != name { + var err error + updated, err = sjson.SetBytes(updated, path+".name", restoredName) + if err != nil { + return updated, err + } + } + } + return updated, nil +} + +// ensureCodexPlaintextFunctionArgs 把上游省略的 encrypted_function_args 显式 +// 补成 []。客户端只在 encrypted_function_args == Some([]) 时才把协作工具参数 +// 当作明文；字段缺失或为 null 时仍按密文交付，父模型生成的明文 message 会被 +// 客户端重新包进子会话的 encrypted_content。 +// +// 补全前校验 arguments 是合法 JSON、message 是字符串，且不像已知密文前缀。 +// 上游仍返回密文时保持原样，把判断留给诊断日志和上层错误。 +func ensureCodexPlaintextFunctionArgs(updated []byte, path, name string, item gjson.Result, state *codexMultiAgentV2State) ([]byte, bool, error) { + if state == nil || !state.isOptimized() { + return updated, false, nil + } + if _, ok := codexCollaborationMessageTools[name]; !ok { + return updated, false, nil + } + marker := item.Get("encrypted_function_args") + if marker.Exists() && marker.Type != gjson.Null { + return updated, false, nil + } + argumentsResult := item.Get("arguments") + if argumentsResult.Type != gjson.String { + return updated, false, nil + } + arguments := strings.TrimSpace(argumentsResult.String()) + if arguments == "" || !gjson.Valid(arguments) { + return updated, false, nil + } + message := gjson.Get(arguments, "message") + if !message.Exists() || message.Type != gjson.String { + return updated, false, nil + } + messageText := strings.TrimSpace(message.String()) + if messageText == "" || looksLikeCodexCiphertext(messageText) { + return updated, false, nil + } + updated, err := sjson.SetBytes(updated, path+".encrypted_function_args", []string{}) + if err != nil { + return updated, false, err + } + accountID, model := state.diagnosticContext() + if callID := strings.TrimSpace(item.Get("call_id").String()); callID == "" || !state.markDiagnosed(callID+"\x00"+name+"\x00plaintext") { + log.Printf("[multi-agent-v2] 已为上游 %s 调用补全 encrypted_function_args: [] (account=%d model=%s)", + name, accountID, model) + } + return updated, true, nil +} + +// looksLikeCodexCiphertext 是响应补全前的 fail-closed 启发式，不是完整的 +// 密文识别。已知的 Fernet 前缀、JWE 紧凑格式和 enc_ 前缀直接拒绝；长度较长 +// 且只含 base64url 字符的单 token 也按密文处理。命中时保持原样，让客户端 +// 继续按加密消息处理，避免把密文伪装成明文任务。 +func looksLikeCodexCiphertext(value string) bool { + value = strings.TrimSpace(value) + if value == "" { + return false + } + if strings.HasPrefix(value, codexEncryptedContentPrefix) || + strings.HasPrefix(value, "eyJ") || + strings.HasPrefix(value, "enc_") { + return true + } + if len(value) < 64 || strings.ContainsAny(value, " \t\r\n") { + return false + } + for _, r := range value { + switch { + case r >= 'A' && r <= 'Z': + case r >= 'a' && r <= 'z': + case r >= '0' && r <= '9': + case r == '-', r == '_', r == '=': + default: + return false + } + } + return true +} + +// diagnoseCodexMultiAgentV2ToolCall 在还原响应时检查上游是否真的返回了 +// 明文标记。缺失 encrypted_function_args: [] 时客户端仍会按加密消息处理， +// 这是兼容层未生效的关键可诊断信号；只记录账号、模型、工具名和密文形态， +// 不记录任务正文。 +func diagnoseCodexMultiAgentV2ToolCall(name string, item gjson.Result, state *codexMultiAgentV2State) { + if _, ok := codexCollaborationMessageTools[name]; !ok { + return + } + arguments := item.Get("arguments").String() + if strings.TrimSpace(arguments) == "" { + // output_item.added 可能还没有 arguments；等到 done 再判断。 + return + } + if marker := item.Get("encrypted_function_args"); marker.IsArray() && len(marker.Array()) == 0 { + return + } + accountID, model := state.diagnosticContext() + if callID := strings.TrimSpace(item.Get("call_id").String()); callID != "" { + if state.markDiagnosed(callID + "\x00" + name) { + return + } + } + message := gjson.Get(arguments, "message").String() + log.Printf("[multi-agent-v2] 上游 %s 调用未返回 encrypted_function_args: []，Codex 仍会按加密消息处理 (account=%d model=%s ciphertext_like=%t)", + name, accountID, model, looksLikeCodexCiphertext(message)) +} diff --git a/proxy/codex_multi_agent_v2_integration_test.go b/proxy/codex_multi_agent_v2_integration_test.go new file mode 100644 index 000000000..d7e41d98f --- /dev/null +++ b/proxy/codex_multi_agent_v2_integration_test.go @@ -0,0 +1,349 @@ +package proxy + +import ( + "bytes" + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/codex2api/auth" + "github.com/codex2api/config" + "github.com/codex2api/database" + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" + "github.com/tidwall/gjson" +) + +const codexMultiAgentV2TestBody = `{ + "model":"gpt-5.5", + "stream":true, + "input":[ + {"type":"additional_tools","role":"developer","tools":[ + {"type":"namespace","name":"collaboration","tools":[ + {"type":"function","name":"followup_task","parameters":{"type":"object","properties":{"message":{"type":"string","encrypted":true}}}}, + {"type":"function","name":"send_message","parameters":{"type":"object","properties":{"message":{"type":"string","encrypted":true}}}}, + {"type":"function","name":"spawn_agent","parameters":{"type":"object","properties":{"task_name":{"type":"string"},"message":{"type":"string","encrypted":true}},"required":["task_name","message"]}} + ]}, + {"type":"function","name":"business_tool","parameters":{"type":"object","properties":{"message":{"type":"string","encrypted":true}}}} + ]}, + {"type":"message","role":"user","content":[{"type":"input_text","text":"delegate this task"}]} + ] +}` + +func newCodexMultiAgentV2TestHandler(t *testing.T, upstream http.Handler) *Handler { + t.Helper() + gin.SetMode(gin.TestMode) + server := httptest.NewServer(upstream) + t.Cleanup(server.Close) + previousResin := resinCfg.Load() + SetResinConfig(&ResinConfig{BaseURL: server.URL, PlatformName: "test"}) + t.Cleanup(func() { resinCfg.Store(previousResin) }) + previousSettings := CurrentRuntimeSettings() + nextSettings := previousSettings + nextSettings.CodexRequestCompression = false + ApplyRuntimeSettings(nextSettings) + t.Cleanup(func() { ApplyRuntimeSettings(previousSettings) }) + + store := auth.NewStore(nil, nil, &database.SystemSettings{MaxConcurrency: 1, TestConcurrency: 1, TestModel: "gpt-5.5"}) + account := &auth.Account{DBID: 1, AccessToken: "at-1", PlanType: "pro", AccountID: "acct-1"} + account.SetDispatchCountLimit(1) + store.AddAccount(account) + return NewHandler(store, nil, &config.Config{AllowAnonymousV1: true}, nil) +} + +func runCodexMultiAgentV2Responses(t *testing.T, handler *Handler, body []byte, userAgent string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + if userAgent != "" { + req.Header.Set("User-Agent", userAgent) + } + recorder := httptest.NewRecorder() + ctx, _ := gin.CreateTestContext(recorder) + ctx.Request = req + handler.Responses(ctx) + return recorder +} + +func codexMultiAgentV2UpstreamSSE(optimized bool) string { + namespace := codexCollaborationNamespace + if optimized { + namespace = codexOptimizedCollaborationNamespace + } + return `data: {"type":"response.created","response":{"id":"resp_ma"}}` + "\n\n" + + `data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_ma","namespace":"` + namespace + `","name":"spawn_agent","arguments":"{\"task_name\":\"probe\",\"message\":\"PLAINTEXT_TASK\"}","encrypted_function_args":[]}}` + "\n\n" + + `data: {"type":"response.completed","response":{"id":"resp_ma","status":"completed","usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}` + "\n\n" +} + +func TestResponsesCodexMultiAgentV2CompatRewritesParentAndRestoresResponse(t *testing.T) { + t.Setenv(codexMultiAgentV2CompatEnv, "1") + upstreamBodies := make(chan []byte, 1) + handler := newCodexMultiAgentV2TestHandler(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + upstreamBodies <- body + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(codexMultiAgentV2UpstreamSSE(true))) + })) + + recorder := runCodexMultiAgentV2Responses(t, handler, []byte(codexMultiAgentV2TestBody), "codex_exec/0.162.0-alpha.17.2") + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String()) + } + + upstreamBody := <-upstreamBodies + if ns := gjson.GetBytes(upstreamBody, "input.0.tools.0.name").String(); ns != codexOptimizedCollaborationNamespace { + t.Fatalf("upstream namespace = %q, want %q; body=%s", ns, codexOptimizedCollaborationNamespace, upstreamBody) + } + for _, path := range []string{ + "input.0.tools.0.tools.0.parameters.properties.message.encrypted", + "input.0.tools.0.tools.1.parameters.properties.message.encrypted", + "input.0.tools.0.tools.2.parameters.properties.message.encrypted", + } { + if gjson.GetBytes(upstreamBody, path).Exists() { + t.Fatalf("upstream still has %s: %s", path, upstreamBody) + } + } + if !gjson.GetBytes(upstreamBody, "input.0.tools.1.parameters.properties.message.encrypted").Exists() { + t.Fatalf("unrelated tool encrypted marker was removed: %s", upstreamBody) + } + + downstream := recorder.Body.String() + if strings.Contains(downstream, codexOptimizedCollaborationNamespace) { + t.Fatalf("downstream leaked optimized namespace: %s", downstream) + } + if !strings.Contains(downstream, `"namespace":"collaboration"`) { + t.Fatalf("downstream namespace was not restored: %s", downstream) + } + if !strings.Contains(downstream, `"encrypted_function_args":[]`) { + t.Fatalf("downstream plaintext marker was lost: %s", downstream) + } +} + +func codexMultiAgentV2UpstreamSSEWithoutMarker(namespace string) string { + return `data: {"type":"response.created","response":{"id":"resp_ma"}}` + "\n\n" + + `data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_ma","namespace":"` + namespace + `","name":"spawn_agent","arguments":"{\"task_name\":\"probe\",\"message\":\"PLAINTEXT_TASK\"}"}}` + "\n\n" + + `data: {"type":"response.completed","response":{"id":"resp_ma","status":"completed","usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}` + "\n\n" +} + +func TestResponsesCodexMultiAgentV2CompatInjectsMissingMarker(t *testing.T) { + t.Setenv(codexMultiAgentV2CompatEnv, "1") + handler := newCodexMultiAgentV2TestHandler(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(codexMultiAgentV2UpstreamSSEWithoutMarker(codexOptimizedCollaborationNamespace))) + })) + + recorder := runCodexMultiAgentV2Responses(t, handler, []byte(codexMultiAgentV2TestBody), "codex_exec/0.162.0-alpha.17.2") + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String()) + } + + downstream := recorder.Body.String() + if strings.Contains(downstream, codexOptimizedCollaborationNamespace) { + t.Fatalf("downstream leaked optimized namespace: %s", downstream) + } + if !strings.Contains(downstream, `"namespace":"collaboration"`) { + t.Fatalf("downstream namespace was not restored: %s", downstream) + } + if !strings.Contains(downstream, `"encrypted_function_args":[]`) { + t.Fatalf("downstream plaintext marker was not injected: %s", downstream) + } +} + +func TestResponsesCodexMultiAgentV2CompatInjectsMarkerWithCollaborationNamespace(t *testing.T) { + t.Setenv(codexMultiAgentV2CompatEnv, "1") + handler := newCodexMultiAgentV2TestHandler(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(codexMultiAgentV2UpstreamSSEWithoutMarker(codexCollaborationNamespace))) + })) + + recorder := runCodexMultiAgentV2Responses(t, handler, []byte(codexMultiAgentV2TestBody), "codex_exec/0.162.0-alpha.17.2") + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String()) + } + + downstream := recorder.Body.String() + if !strings.Contains(downstream, `"namespace":"collaboration"`) { + t.Fatalf("downstream namespace = %s", downstream) + } + if !strings.Contains(downstream, `"encrypted_function_args":[]`) { + t.Fatalf("downstream plaintext marker was not injected: %s", downstream) + } +} + +func TestResponsesCodexMultiAgentV2CompatDisabledLeavesRequestAndResponseUntouched(t *testing.T) { + t.Setenv(codexMultiAgentV2CompatEnv, "0") + upstreamBodies := make(chan []byte, 1) + handler := newCodexMultiAgentV2TestHandler(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + upstreamBodies <- body + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(codexMultiAgentV2UpstreamSSE(true))) + })) + + recorder := runCodexMultiAgentV2Responses(t, handler, []byte(codexMultiAgentV2TestBody), "codex_exec/0.162.0-alpha.17.2") + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String()) + } + + upstreamBody := <-upstreamBodies + if ns := gjson.GetBytes(upstreamBody, "input.0.tools.0.name").String(); ns != codexCollaborationNamespace { + t.Fatalf("disabled compat changed upstream namespace: %q", ns) + } + if !gjson.GetBytes(upstreamBody, "input.0.tools.0.tools.2.parameters.properties.message.encrypted").Exists() { + t.Fatalf("disabled compat removed encrypted marker: %s", upstreamBody) + } + if !strings.Contains(recorder.Body.String(), codexOptimizedCollaborationNamespace) { + t.Fatalf("disabled compat unexpectedly restored response: %s", recorder.Body.String()) + } +} + +func TestResponsesCodexMultiAgentV2CompatNonStreamRestoresResponse(t *testing.T) { + t.Setenv(codexMultiAgentV2CompatEnv, "1") + handler := newCodexMultiAgentV2TestHandler(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(codexMultiAgentV2UpstreamSSE(true))) + })) + + body := []byte(strings.Replace(codexMultiAgentV2TestBody, `"stream":true`, `"stream":false`, 1)) + recorder := runCodexMultiAgentV2Responses(t, handler, body, "codex_exec/0.162.0-alpha.17.2") + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String()) + } + response := recorder.Body.Bytes() + if strings.Contains(string(response), codexOptimizedCollaborationNamespace) { + t.Fatalf("non-stream response leaked optimized namespace: %s", response) + } + if ns := gjson.GetBytes(response, "output.0.namespace").String(); ns != codexCollaborationNamespace { + t.Fatalf("non-stream namespace = %q, want %q; body=%s", ns, codexCollaborationNamespace, response) + } + if marker := gjson.GetBytes(response, "output.0.encrypted_function_args"); !marker.IsArray() || marker.Raw != "[]" { + t.Fatalf("non-stream plaintext marker = %s, want []", marker.Raw) + } +} + +func TestResponsesCodexMultiAgentV2CompatConflictFailsClosed(t *testing.T) { + t.Setenv(codexMultiAgentV2CompatEnv, "1") + upstreamBodies := make(chan []byte, 1) + handler := newCodexMultiAgentV2TestHandler(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + upstreamBodies <- body + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(codexMultiAgentV2UpstreamSSE(false))) + })) + + conflict := strings.Replace(codexMultiAgentV2TestBody, + `{"type":"function","name":"business_tool"`, + `{"type":"namespace","name":"collaboration-optimize","tools":[]},{"type":"function","name":"business_tool"`, 1) + recorder := runCodexMultiAgentV2Responses(t, handler, []byte(conflict), "codex_exec/0.162.0-alpha.17.2") + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String()) + } + upstreamBody := <-upstreamBodies + if ns := gjson.GetBytes(upstreamBody, "input.0.tools.0.name").String(); ns != codexCollaborationNamespace { + t.Fatalf("conflict changed upstream namespace: %q", ns) + } + if !gjson.GetBytes(upstreamBody, "input.0.tools.0.tools.2.parameters.properties.message.encrypted").Exists() { + t.Fatalf("conflict removed encrypted marker: %s", upstreamBody) + } +} + +func TestResponsesCodexMultiAgentV2CompatIgnoresOtherClients(t *testing.T) { + t.Setenv(codexMultiAgentV2CompatEnv, "1") + upstreamBodies := make(chan []byte, 1) + handler := newCodexMultiAgentV2TestHandler(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + upstreamBodies <- body + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(codexMultiAgentV2UpstreamSSE(false))) + })) + + recorder := runCodexMultiAgentV2Responses(t, handler, []byte(codexMultiAgentV2TestBody), "curl/8.7.1") + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String()) + } + upstreamBody := <-upstreamBodies + if ns := gjson.GetBytes(upstreamBody, "input.0.tools.0.name").String(); ns != codexCollaborationNamespace { + t.Fatalf("non-Codex client was rewritten: %q", ns) + } +} + +func TestResponsesWebSocketCodexMultiAgentV2CompatRewritesAndRestores(t *testing.T) { + t.Setenv(codexMultiAgentV2CompatEnv, "1") + gin.SetMode(gin.TestMode) + + previousExec := WebsocketExecuteFunc + t.Cleanup(func() { WebsocketExecuteFunc = previousExec }) + bodyCh := make(chan []byte, 1) + WebsocketExecuteFunc = func(ctx context.Context, account *auth.Account, requestBody []byte, sessionID string, proxyOverride string, apiKey string, deviceCfg *DeviceProfileConfig, headers http.Header, poolRouteKey string) (*http.Response, error) { + bodyCh <- append([]byte(nil), requestBody...) + sse := codexMultiAgentV2UpstreamSSE(true) + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(sse)), + }, nil + } + + store := auth.NewStore(nil, nil, &database.SystemSettings{MaxConcurrency: 2, TestConcurrency: 1, TestModel: "gpt-5.5"}) + store.AddAccount(&auth.Account{DBID: 1, AccessToken: "at", PlanType: "plus", AccountID: "acct-1"}) + handler := NewHandler(store, nil, &config.Config{AllowAnonymousV1: true}, nil) + router := gin.New() + handler.RegisterRoutes(router) + server := httptest.NewServer(router) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/v1/responses" + headers := http.Header{"User-Agent": []string{"codex_exec/0.162.0-alpha.17.2"}} + conn, resp, err := websocket.DefaultDialer.Dial(wsURL, headers) + if err != nil { + if resp != nil { + t.Fatalf("dial websocket failed: %v status=%d", err, resp.StatusCode) + } + t.Fatalf("dial websocket failed: %v", err) + } + defer conn.Close() + + if err := conn.WriteMessage(websocket.TextMessage, []byte(`{"model":"gpt-5.5","stream":true,"input":[ + {"type":"additional_tools","role":"developer","tools":[ + {"type":"namespace","name":"collaboration","tools":[ + {"type":"function","name":"spawn_agent","parameters":{"type":"object","properties":{"task_name":{"type":"string"},"message":{"type":"string","encrypted":true}}}} + ]} + ]}, + {"type":"message","role":"user","content":[{"type":"input_text","text":"delegate"}]} + ]}`)); err != nil { + t.Fatalf("write request: %v", err) + } + + select { + case gotBody := <-bodyCh: + if ns := gjson.GetBytes(gotBody, "input.0.tools.0.name").String(); ns != codexOptimizedCollaborationNamespace { + t.Fatalf("upstream WS namespace = %q; body=%s", ns, gotBody) + } + if gjson.GetBytes(gotBody, "input.0.tools.0.tools.0.parameters.properties.message.encrypted").Exists() { + t.Fatalf("upstream WS encrypted marker was not removed: %s", gotBody) + } + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for upstream WS request") + } + + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + for { + _, message, err := conn.ReadMessage() + if err != nil { + t.Fatalf("read downstream message: %v", err) + } + if eventType := gjson.GetBytes(message, "type").String(); eventType == "response.output_item.done" { + if ns := gjson.GetBytes(message, "item.namespace").String(); ns != codexCollaborationNamespace { + t.Fatalf("downstream WS namespace = %q; message=%s", ns, message) + } + if marker := gjson.GetBytes(message, "item.encrypted_function_args"); !marker.IsArray() || marker.Raw != "[]" { + t.Fatalf("downstream WS plaintext marker = %s; message=%s", marker.Raw, message) + } + return + } + } +} diff --git a/proxy/codex_multi_agent_v2_test.go b/proxy/codex_multi_agent_v2_test.go new file mode 100644 index 000000000..27e0e3a76 --- /dev/null +++ b/proxy/codex_multi_agent_v2_test.go @@ -0,0 +1,718 @@ +package proxy + +import ( + "strings" + "testing" + + "github.com/tidwall/gjson" +) + +func TestIsCodexMultiAgentV2Client(t *testing.T) { + tests := []struct { + ua string + want bool + }{ + {"Codex Desktop/0.162.0-alpha.17.2", true}, + {"codex-tui/0.162.0", true}, + {"codex_cli_rs", true}, + {"codex_cli_rs/0.162.0", true}, + {"codex_exec/0.162.0-alpha.17.2 (Windows 10.0)", true}, + {"curl/8.7.1", false}, + {"OpenAI/Python 1.0", false}, + {"", false}, + } + for _, tt := range tests { + if got := isCodexMultiAgentV2Client(tt.ua); got != tt.want { + t.Fatalf("isCodexMultiAgentV2Client(%q) = %v, want %v", tt.ua, got, tt.want) + } + } +} + +func TestPrepareCodexMultiAgentV2RequestRewritesAdditionalTools(t *testing.T) { + payload := []byte(`{ + "model":"gpt-6.1-sol", + "input":[ + {"type":"additional_tools","role":"developer","tools":[ + {"type":"namespace","name":"functions","tools":[{"type":"custom","name":"exec"}]}, + {"type":"namespace","name":"collaboration","tools":[ + {"type":"function","name":"followup_task","parameters":{"type":"object","properties":{"message":{"type":"string","encrypted":true}}}}, + {"type":"function","name":"interrupt_agent","parameters":{"type":"object","properties":{}}}, + {"type":"function","name":"send_message","parameters":{"type":"object","properties":{"message":{"type":"string","encrypted":true}}}}, + {"type":"function","name":"spawn_agent","parameters":{"type":"object","properties":{"message":{"type":"string","encrypted":true}}}} + ]}, + {"type":"function","name":"business_tool","parameters":{"type":"object","properties":{"message":{"type":"string","encrypted":true}}}} + ]} + ] + }`) + + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !optimized { + t.Fatal("expected namespace to be optimized") + } + if ns := gjson.GetBytes(got, "input.0.tools.1.name").String(); ns != codexOptimizedCollaborationNamespace { + t.Fatalf("namespace = %q, want %q", ns, codexOptimizedCollaborationNamespace) + } + for _, path := range []string{ + "input.0.tools.1.tools.0.parameters.properties.message.encrypted", + "input.0.tools.1.tools.2.parameters.properties.message.encrypted", + "input.0.tools.1.tools.3.parameters.properties.message.encrypted", + } { + if gjson.GetBytes(got, path).Exists() { + t.Fatalf("%s was not removed: %s", path, got) + } + } + if !gjson.GetBytes(got, "input.0.tools.2.parameters.properties.message.encrypted").Exists() { + t.Fatalf("unrelated tool encrypted marker was removed: %s", got) + } + if ns := gjson.GetBytes(got, "input.0.tools.0.name").String(); ns != "functions" { + t.Fatalf("unrelated namespace = %q, want functions", ns) + } +} + +func TestPrepareCodexMultiAgentV2RequestTopLevelAndFlatTools(t *testing.T) { + payload := []byte(`{ + "tools":[ + {"type":"function","name":"collaboration.spawn_agent","parameters":{"type":"object","properties":{"message":{"type":"string","encrypted":true}}}}, + {"type":"function","name":"collaboration.send_message","parameters":{"type":"object","properties":{"message":{"type":"string","encrypted":true}}}}, + {"type":"function","name":"collaboration.wait_agent","parameters":{"type":"object","properties":{}}} + ] + }`) + + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !optimized { + t.Fatal("expected flat collaboration tools to be optimized") + } + if name := gjson.GetBytes(got, "tools.0.name").String(); name != "collaboration-optimize.spawn_agent" { + t.Fatalf("spawn name = %q", name) + } + if name := gjson.GetBytes(got, "tools.1.name").String(); name != "collaboration-optimize.send_message" { + t.Fatalf("send name = %q", name) + } + if name := gjson.GetBytes(got, "tools.2.name").String(); name != "collaboration.wait_agent" { + t.Fatalf("wait name = %q", name) + } + for _, path := range []string{ + "tools.0.parameters.properties.message.encrypted", + "tools.1.parameters.properties.message.encrypted", + } { + if gjson.GetBytes(got, path).Exists() { + t.Fatalf("%s was not removed: %s", path, got) + } + } +} + +func TestPrepareCodexMultiAgentV2RequestWithoutSpawnStillRewritesMessageTools(t *testing.T) { + payload := []byte(`{"input":[{"type":"additional_tools","tools":[{"type":"namespace","name":"collaboration","tools":[ + {"type":"function","name":"send_message","parameters":{"type":"object","properties":{"message":{"type":"string","encrypted":true}}}}, + {"type":"function","name":"followup_task","parameters":{"type":"object","properties":{"message":{"type":"string","encrypted":true}}}} + ]}]}]}`) + + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !optimized { + t.Fatal("expected send_message/followup_task to be optimized without spawn_agent") + } + if ns := gjson.GetBytes(got, "input.0.tools.0.name").String(); ns != codexOptimizedCollaborationNamespace { + t.Fatalf("namespace = %q, want %q", ns, codexOptimizedCollaborationNamespace) + } + for _, path := range []string{ + "input.0.tools.0.tools.0.parameters.properties.message.encrypted", + "input.0.tools.0.tools.1.parameters.properties.message.encrypted", + } { + if gjson.GetBytes(got, path).Exists() { + t.Fatalf("%s was not removed: %s", path, got) + } + } +} + +func TestPrepareCodexMultiAgentV2RequestFlatMessageToolWithoutSpawn(t *testing.T) { + payload := []byte(`{"tools":[ + {"type":"function","name":"collaboration.send_message","parameters":{"properties":{"message":{"encrypted":true}}}} + ]}`) + + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !optimized { + t.Fatal("expected flat collaboration.send_message to be optimized") + } + if name := gjson.GetBytes(got, "tools.0.name").String(); name != "collaboration-optimize.send_message" { + t.Fatalf("name = %q, want collaboration-optimize.send_message", name) + } + if gjson.GetBytes(got, "tools.0.parameters.properties.message.encrypted").Exists() { + t.Fatalf("flat message.encrypted was not removed: %s", got) + } +} + +func TestPrepareCodexMultiAgentV2RequestIgnoresSameNamedToolsOutsideCollaboration(t *testing.T) { + payload := []byte(`{"tools":[ + {"type":"namespace","name":"collaboration","tools":[ + {"type":"function","name":"spawn_agent","parameters":{"properties":{"message":{"encrypted":true}}}} + ]}, + {"type":"namespace","name":"business","tools":[ + {"type":"function","name":"send_message","parameters":{"properties":{"message":{"encrypted":true}}}} + ]}, + {"type":"function","name":"followup_task","parameters":{"properties":{"message":{"encrypted":true}}}} + ]}`) + + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !optimized { + t.Fatal("expected collaboration namespace to be optimized") + } + if gjson.GetBytes(got, "tools.0.tools.0.parameters.properties.message.encrypted").Exists() { + t.Fatalf("collaboration message.encrypted was not removed: %s", got) + } + for _, path := range []string{ + "tools.1.tools.0.parameters.properties.message.encrypted", + "tools.2.parameters.properties.message.encrypted", + } { + if !gjson.GetBytes(got, path).Exists() { + t.Fatalf("same-named tool outside collaboration was rewritten: %s", path) + } + } + if name := gjson.GetBytes(got, "tools.1.name").String(); name != "business" { + t.Fatalf("business namespace changed: %q", name) + } +} + +func TestCodexMultiAgentV2StateTracksAccountAndDiagnostics(t *testing.T) { + state := &codexMultiAgentV2State{} + if state.isOptimizedFor(1) { + t.Fatal("empty state must not be optimized") + } + state.markOptimized(1, "gpt-5.5") + if !state.isOptimizedFor(1) { + t.Fatal("optimized state must match its account") + } + if state.isOptimizedFor(2) { + t.Fatal("optimized state must not match a different account") + } + accountID, model := state.diagnosticContext() + if accountID != 1 || model != "gpt-5.5" { + t.Fatalf("diagnostic context = %d/%q", accountID, model) + } + if state.markDiagnosed("call-1") { + t.Fatal("first diagnosis must not be treated as duplicate") + } + if !state.markDiagnosed("call-1") { + t.Fatal("second diagnosis must be treated as duplicate") + } +} + +func TestPrepareCodexMultiAgentV2RequestHandlesEmptyAndNonArrayTools(t *testing.T) { + for _, payload := range [][]byte{ + nil, + {}, + []byte(`{}`), + []byte(`{"tools":null}`), + []byte(`{"tools":{}}`), + []byte(`{"tools":[]}`), + } { + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err != nil { + t.Fatalf("unexpected error for %q: %v", payload, err) + } + if optimized { + t.Fatalf("unexpected optimization for %q", payload) + } + if string(got) != string(payload) { + t.Fatalf("payload changed for %q: %s", payload, got) + } + } +} + +func TestPrepareCodexMultiAgentV2RequestRewritesToolChoice(t *testing.T) { + tests := []struct { + name string + payload string + wantPath string + wantString string + }{ + { + name: "flat", + payload: `{"tools":[{"type":"function","name":"collaboration.spawn_agent","parameters":{"properties":{"message":{"encrypted":true}}}}], + "tool_choice":{"type":"function","name":"collaboration.spawn_agent"}}`, + wantPath: "tool_choice.name", + wantString: "collaboration-optimize.spawn_agent", + }, + { + name: "namespace", + payload: `{"tools":[{"type":"namespace","name":"collaboration","tools":[{"type":"function","name":"spawn_agent","parameters":{"properties":{"message":{"encrypted":true}}}}]}], + "tool_choice":{"type":"function","namespace":"collaboration","name":"spawn_agent"}}`, + wantPath: "tool_choice.namespace", + wantString: "collaboration-optimize", + }, + { + name: "namespace non-message tool", + payload: `{"tools":[{"type":"namespace","name":"collaboration","tools":[ + {"type":"function","name":"spawn_agent","parameters":{"properties":{"message":{"encrypted":true}}}}, + {"type":"function","name":"interrupt_agent","parameters":{"properties":{}}} + ]}], + "tool_choice":{"type":"function","namespace":"collaboration","name":"interrupt_agent"}}`, + wantPath: "tool_choice.namespace", + wantString: "collaboration-optimize", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, optimized, err := prepareCodexMultiAgentV2Request([]byte(tt.payload)) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !optimized { + t.Fatal("expected tool_choice request to be optimized") + } + if value := gjson.GetBytes(got, tt.wantPath).String(); value != tt.wantString { + t.Fatalf("%s = %q, want %q", tt.wantPath, value, tt.wantString) + } + }) + } +} + +func TestPrepareCodexMultiAgentV2RequestFlatDoubleUnderscore(t *testing.T) { + payload := []byte(`{"tools":[ + {"type":"function","name":"collaboration__send_message","parameters":{"properties":{"message":{"encrypted":true}}}} + ]}`) + + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !optimized { + t.Fatal("expected flat collaboration__send_message to be optimized") + } + if name := gjson.GetBytes(got, "tools.0.name").String(); name != "collaboration-optimize__send_message" { + t.Fatalf("name = %q, want collaboration-optimize__send_message", name) + } +} + +func TestPrepareCodexMultiAgentV2RequestRewritesAllowedToolsChoice(t *testing.T) { + payload := []byte(`{"tools":[{"type":"namespace","name":"collaboration","tools":[ + {"type":"function","name":"spawn_agent","parameters":{"properties":{"message":{"encrypted":true}}}} + ]}], + "tool_choice":{"type":"allowed_tools","tools":[ + {"type":"function","namespace":"collaboration","name":"spawn_agent"} + ]}}`) + + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !optimized { + t.Fatal("expected allowed_tools request to be optimized") + } + if ns := gjson.GetBytes(got, "tool_choice.tools.0.namespace").String(); ns != codexOptimizedCollaborationNamespace { + t.Fatalf("tool_choice.tools.0.namespace = %q", ns) + } +} + +func TestRestoreCodexMultiAgentV2ResponseRestoresToolChoice(t *testing.T) { + payload := []byte(`{ + "tool_choice":{"type":"function","namespace":"collaboration-optimize","name":"spawn_agent"}, + "response":{"tool_choice":{"type":"allowed_tools","tools":[ + {"type":"function","namespace":"collaboration-optimize","name":"followup_task"} + ]}} + }`) + + got := restoreCodexMultiAgentV2Response(payload, nil) + if ns := gjson.GetBytes(got, "tool_choice.namespace").String(); ns != codexCollaborationNamespace { + t.Fatalf("tool_choice.namespace = %q", ns) + } + if ns := gjson.GetBytes(got, "response.tool_choice.tools.0.namespace").String(); ns != codexCollaborationNamespace { + t.Fatalf("response.tool_choice.tools.0.namespace = %q", ns) + } +} + +func TestPrepareCodexMultiAgentV2RequestAlreadyPlaintextUnchanged(t *testing.T) { + payload := []byte(`{"tools":[{"type":"namespace","name":"collaboration","tools":[ + {"type":"function","name":"spawn_agent","parameters":{"properties":{"message":{"type":"string"}}}} + ]}]}`) + + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if optimized { + t.Fatal("already plaintext schema must not be optimized") + } + if string(got) != string(payload) { + t.Fatalf("already plaintext schema changed: %s", got) + } +} + +func TestPrepareCodexMultiAgentV2RequestRejectsDeepNesting(t *testing.T) { + nested := `{"type":"function","name":"spawn_agent","parameters":{"properties":{"message":{"encrypted":true}}}}` + for i := 0; i <= codexMultiAgentV2MaxToolDepth; i++ { + nested = `{"type":"namespace","name":"layer","tools":[` + nested + `]}` + } + payload := []byte(`{"tools":[{"type":"namespace","name":"collaboration","tools":[` + nested + `]}]}`) + + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err == nil { + t.Fatal("expected nesting limit error") + } + if optimized { + t.Fatal("deeply nested request must not be optimized") + } + if string(got) != string(payload) { + t.Fatalf("deeply nested request changed: %s", got) + } +} + +func TestPrepareCodexMultiAgentV2RequestRejectsTooManyMessageTools(t *testing.T) { + var tools strings.Builder + for i := 0; i <= codexMultiAgentV2MaxToolPaths; i++ { + if i > 0 { + tools.WriteString(",") + } + tools.WriteString(`{"type":"function","name":"spawn_agent","parameters":{"properties":{"message":{"encrypted":true}}}}`) + } + payload := []byte(`{"tools":[{"type":"namespace","name":"collaboration","tools":[` + tools.String() + `]}]}`) + + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err == nil { + t.Fatal("expected tool count limit error") + } + if optimized { + t.Fatal("oversized request must not be optimized") + } + if string(got) != string(payload) { + t.Fatalf("oversized request changed: %s", got) + } +} + +func TestPrepareCodexMultiAgentV2RequestConflictUnchanged(t *testing.T) { + payload := []byte(`{"tools":[ + {"type":"namespace","name":"collaboration","tools":[{"type":"function","name":"spawn_agent","parameters":{"properties":{"message":{"encrypted":true}}}}]}, + {"type":"namespace","name":"collaboration-optimize","tools":[]} + ]}`) + + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err == nil { + t.Fatal("expected conflict error") + } + if optimized { + t.Fatal("conflict must not optimize") + } + if string(got) != string(payload) { + t.Fatalf("conflict changed payload: %s", got) + } +} + +func TestPrepareCodexMultiAgentV2RequestLeavesUnrelatedRequestsUnchanged(t *testing.T) { + payload := []byte(`{"model":"gpt-6.1-sol","tools":[{"type":"function","name":"business_tool","parameters":{"properties":{"message":{"encrypted":true}}}}]}`) + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if optimized { + t.Fatal("unrelated request must not be optimized") + } + if string(got) != string(payload) { + t.Fatalf("unrelated request changed: %s", got) + } +} + +func TestPrepareCodexMultiAgentV2RequestInvalidJSONUnchanged(t *testing.T) { + payload := []byte(`{"collaboration":`) + got, optimized, err := prepareCodexMultiAgentV2Request(payload) + if err == nil { + t.Fatal("expected invalid JSON error") + } + if optimized { + t.Fatal("invalid JSON must not optimize") + } + if string(got) != string(payload) { + t.Fatalf("invalid JSON changed: %s", got) + } +} + +func TestRestoreCodexMultiAgentV2Response(t *testing.T) { + payload := []byte(`{ + "type":"response.completed", + "response":{ + "output":[ + {"type":"function_call","name":"spawn_agent","namespace":"collaboration-optimize","arguments":{"namespace":"collaboration-optimize","name":"collaboration-optimize__opaque"}}, + {"type":"function_call","name":"collaboration-optimize__send_message","arguments":"{}","encrypted_function_args":[]}, + {"type":"function_call","name":"collaboration-optimize.spawn_agent","namespace":null,"arguments":"{}","encrypted_function_args":[]}, + {"type":"message","namespace":"collaboration-optimize","name":"collaboration-optimize__plain"}, + {"type":"function_call","name":"business_tool","namespace":"business","arguments":"{}"} + ], + "tools":[{"type":"namespace","name":"collaboration-optimize"}] + } + }`) + + got := restoreCodexMultiAgentV2Response(payload, nil) + if ns := gjson.GetBytes(got, "response.output.0.namespace").String(); ns != codexCollaborationNamespace { + t.Fatalf("output 0 namespace = %q", ns) + } + if name := gjson.GetBytes(got, "response.output.1.name").String(); name != "send_message" { + t.Fatalf("output 1 name = %q", name) + } + if ns := gjson.GetBytes(got, "response.output.1.namespace").String(); ns != codexCollaborationNamespace { + t.Fatalf("output 1 namespace = %q", ns) + } + if name := gjson.GetBytes(got, "response.output.2.name").String(); name != "spawn_agent" { + t.Fatalf("output 2 name = %q", name) + } + if ns := gjson.GetBytes(got, "response.output.2.namespace").String(); ns != codexCollaborationNamespace { + t.Fatalf("output 2 namespace = %q", ns) + } + // arguments 是不透明字段，不能被改写。 + if ns := gjson.GetBytes(got, "response.output.0.arguments.namespace").String(); ns != codexOptimizedCollaborationNamespace { + t.Fatalf("arguments namespace was rewritten: %q", ns) + } + // 非工具调用项的普通 namespace 字段保持原样。 + if ns := gjson.GetBytes(got, "response.output.3.namespace").String(); ns != codexOptimizedCollaborationNamespace { + t.Fatalf("ordinary namespace was rewritten: %q", ns) + } + if name := gjson.GetBytes(got, "response.output.3.name").String(); name != "collaboration-optimize__plain" { + t.Fatalf("ordinary name was rewritten: %q", name) + } + if name := gjson.GetBytes(got, "response.tools.0.name").String(); name != codexCollaborationNamespace { + t.Fatalf("tools namespace = %q", name) + } + if ns := gjson.GetBytes(got, "response.output.4.namespace").String(); ns != "business" { + t.Fatalf("business tool namespace changed: %q", ns) + } + + // 幂等：再次还原不应继续变化。 + again := restoreCodexMultiAgentV2Response(got, nil) + if string(again) != string(got) { + t.Fatalf("restore is not idempotent:\nfirst=%s\nsecond=%s", got, again) + } +} + +func TestRestoreCodexMultiAgentV2ResponseInjectsPlaintextFunctionArgs(t *testing.T) { + tests := []struct { + name string + payload string + path string + wantInjected bool + wantMarker string + wantNamespace string + stateNil bool + }{ + { + name: "missing marker", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent","call_id":"call_1","arguments":"{\"message\":\"do the task\"}"}}`, + path: "item", + wantInjected: true, + wantMarker: "[]", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "null marker", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"send_message","call_id":"call_2","arguments":"{\"message\":\"follow up\"}","encrypted_function_args":null}}`, + path: "item", + wantInjected: true, + wantMarker: "[]", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "existing empty marker", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"followup_task","call_id":"call_3","arguments":"{\"message\":\"again\"}","encrypted_function_args":[]}}`, + path: "item", + wantMarker: "[]", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "existing encrypted marker", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent","call_id":"call_4","arguments":"{\"message\":\"do the task\"}","encrypted_function_args":["message"]}}`, + path: "item", + wantMarker: `["message"]`, + wantNamespace: codexCollaborationNamespace, + }, + { + name: "ciphertext-like message", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent","call_id":"call_5","arguments":"{\"message\":\"gAAAAABciphertext\"}"}}`, + path: "item", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "invalid arguments", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent","call_id":"call_6","arguments":"not-json"}}`, + path: "item", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "arguments without message", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent","call_id":"call_7","arguments":"{\"task_name\":\"x\"}"}}`, + path: "item", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "empty message", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent","call_id":"call_7a","arguments":"{\"message\":\"\"}"}}`, + path: "item", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "whitespace message", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent","call_id":"call_7b","arguments":"{\"message\":\" \"}"}}`, + path: "item", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "arguments object", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent","call_id":"call_7c","arguments":{"message":"do the task"}}}`, + path: "item", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "long base64url message", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent","call_id":"call_7d","arguments":"{\"message\":\"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\"}"}}`, + path: "item", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "nil state does not inject", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent","call_id":"call_7e","arguments":"{\"message\":\"do the task\"}"}}`, + path: "item", + wantNamespace: codexCollaborationNamespace, + stateNil: true, + }, + { + name: "flat double underscore name", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","name":"collaboration-optimize__spawn_agent","call_id":"call_8","arguments":"{\"message\":\"do the task\"}"}}`, + path: "item", + wantInjected: true, + wantMarker: "[]", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "response output container", + payload: `{"type":"response.completed","response":{"output":[{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent","call_id":"call_9","arguments":"{\"message\":\"do the task\"}"}]}}`, + path: "response.output.0", + wantInjected: true, + wantMarker: "[]", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "non-collaboration tool untouched", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"business","name":"spawn_agent","call_id":"call_10","arguments":"{\"message\":\"do the task\"}"}}`, + path: "item", + wantNamespace: "business", + }, + { + name: "collaboration namespace missing marker", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration","name":"spawn_agent","call_id":"call_11","arguments":"{\"message\":\"do the task\"}"}}`, + path: "item", + wantInjected: true, + wantMarker: "[]", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "collaboration namespace ciphertext", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration","name":"spawn_agent","call_id":"call_12","arguments":"{\"message\":\"gAAAAABciphertext\"}"}}`, + path: "item", + wantNamespace: codexCollaborationNamespace, + }, + { + name: "collaboration namespace nil state", + payload: `{"type":"response.output_item.done","item":{"type":"function_call","namespace":"collaboration","name":"spawn_agent","call_id":"call_13","arguments":"{\"message\":\"do the task\"}"}}`, + path: "item", + wantNamespace: codexCollaborationNamespace, + stateNil: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + state := &codexMultiAgentV2State{} + state.markOptimized(1, "gpt-5.5") + if tc.stateNil { + state = nil + } + got := restoreCodexMultiAgentV2Response([]byte(tc.payload), state) + marker := gjson.GetBytes(got, tc.path+".encrypted_function_args") + if tc.wantMarker == "" { + if marker.Exists() { + t.Fatalf("encrypted_function_args should be absent, got %s", marker.Raw) + } + } else if marker.Raw != tc.wantMarker { + t.Fatalf("encrypted_function_args = %s, want %s", marker.Raw, tc.wantMarker) + } + if ns := gjson.GetBytes(got, tc.path+".namespace").String(); ns != tc.wantNamespace { + t.Fatalf("namespace = %q, want %q", ns, tc.wantNamespace) + } + if tc.wantInjected { + // arguments 是不透明字段，补标记不能改动正文。 + beforeArgs := gjson.GetBytes([]byte(tc.payload), tc.path+".arguments").Raw + afterArgs := gjson.GetBytes(got, tc.path+".arguments").Raw + if afterArgs != beforeArgs { + t.Fatalf("arguments changed: before=%s after=%s", beforeArgs, afterArgs) + } + again := restoreCodexMultiAgentV2Response(got, state) + if string(again) != string(got) { + t.Fatalf("injection is not idempotent:\nfirst=%s\nsecond=%s", got, again) + } + } + }) + } +} + +func TestRestoreCodexMultiAgentV2ResponseRestoresFlatToolDefinitions(t *testing.T) { + payload := []byte(`{ + "tools":[ + {"type":"function","name":"collaboration-optimize.spawn_agent"}, + {"type":"custom","name":"collaboration-optimize__send_message"}, + {"type":"function","name":"business_tool"} + ], + "response":{"tools":[{"type":"function","name":"collaboration-optimize.followup_task"}]} + }`) + + got := restoreCodexMultiAgentV2Response(payload, nil) + if name := gjson.GetBytes(got, "tools.0.name").String(); name != "collaboration.spawn_agent" { + t.Fatalf("tools.0.name = %q", name) + } + if name := gjson.GetBytes(got, "tools.1.name").String(); name != "collaboration__send_message" { + t.Fatalf("tools.1.name = %q", name) + } + if name := gjson.GetBytes(got, "tools.2.name").String(); name != "business_tool" { + t.Fatalf("business tool changed: %q", name) + } + if name := gjson.GetBytes(got, "response.tools.0.name").String(); name != "collaboration.followup_task" { + t.Fatalf("response.tools.0.name = %q", name) + } +} + +func TestRestoreCodexMultiAgentV2ResponseRejectsTrailingGarbage(t *testing.T) { + payload := []byte(`{"type":"response.completed","response":{"output":[{"type":"function_call","namespace":"collaboration-optimize","name":"spawn_agent"}]}}garbage`) + if got := restoreCodexMultiAgentV2Response(payload, nil); string(got) != string(payload) { + t.Fatalf("trailing garbage payload changed: %s", got) + } +} + +func TestRestoreCodexMultiAgentV2ResponsePreservesNumbers(t *testing.T) { + payload := []byte(`{"type":"response.completed","response":{"id":"resp_1","created_at":1792300000123,"output":[{"type":"function_call","name":"spawn_agent","namespace":"collaboration-optimize","arguments":"{}","encrypted_function_args":[]}]}}`) + got := restoreCodexMultiAgentV2Response(payload, nil) + if !strings.Contains(string(got), `"created_at":1792300000123`) { + t.Fatalf("created_at was re-encoded: %s", got) + } +} + +func TestRestoreCodexMultiAgentV2ResponseNoop(t *testing.T) { + for _, payload := range [][]byte{ + nil, + []byte(`{"type":"response.completed","response":{"output":[]}}`), + []byte(`{"type":"response.completed","response":{"output":[{"type":"function_call","namespace":"collaboration"}]}}`), + []byte(`not json collaboration-optimize`), + } { + if got := restoreCodexMultiAgentV2Response(payload, nil); string(got) != string(payload) { + t.Fatalf("noop restore changed payload: %s", got) + } + } +} diff --git a/proxy/executor.go b/proxy/executor.go index 430492425..ddec3f11f 100644 --- a/proxy/executor.go +++ b/proxy/executor.go @@ -574,6 +574,20 @@ func ExecuteRequest(ctx context.Context, account *auth.Account, requestBody []by requestBody = ApplyCodexFingerprintToBody(requestBody, account, headers) // 账号绑定时区：改写 environment_context 的时区/日期，与指纹收敛一样在分叉前统一处理。 requestBody = ApplyCodexTimezoneToBody(requestBody, account, time.Now()) + // Codex multi-agent v2 兼容层：仅在调用方显式安装状态、部署方开启开关、 + // 且下游确实是 Codex 客户端时改写 collaboration 工具 schema。冲突、无效 + // JSON 或结构超限的请求保持原生路径，绝不猜测任务正文。 + if state := codexMultiAgentV2StateFromContext(ctx); state != nil && + codexMultiAgentV2CompatEnabled() && isCodexMultiAgentV2Client(headers.Get("User-Agent")) { + model := gjson.GetBytes(requestBody, "model").String() + if updated, optimized, err := prepareCodexMultiAgentV2Request(requestBody); err != nil { + log.Printf("[multi-agent-v2] 兼容层跳过请求改写 (account=%d model=%s): %v", account.ID(), model, err) + } else if optimized { + requestBody = updated + state.markOptimized(account.ID(), model) + log.Printf("[multi-agent-v2] 父请求已切换为明文协作工具 (account=%d model=%s)", account.ID(), model) + } + } // lite 信号收敛：签名在 payload 规则改写后采集（规则可注入/删除 WS 标记，改写 // 前采集会让注入失效、删除被回填），模型也已被入口映射/规则定稿——已知不支持 // lite 的模型带信号上游必 400，发出前剥离。 diff --git a/proxy/handler.go b/proxy/handler.go index a0c7acb1a..56cbfafb7 100644 --- a/proxy/handler.go +++ b/proxy/handler.go @@ -4007,6 +4007,10 @@ func (h *Handler) Responses(c *gin.Context) { // OpenAI Responses relay body 仅在实际命中 relay 账号时惰性生成，避免 Codex 路径重复转换。 // previous_response_id 缓存按下游 API Key 隔离，防止跨用户注入他人对话历史。 respCacheOwner := responseCacheOwner(apiKeyID) + multiAgentV2State := &codexMultiAgentV2State{} + if codexMultiAgentV2CompatEnabled() { + c.Request = c.Request.WithContext(withCodexMultiAgentV2State(c.Request.Context(), multiAgentV2State)) + } bodyPreparation := prepareResponsesBodyForOwnerDetailed(rawBody, respCacheOwner) codexBody, expandedInputRaw := bodyPreparation.Body, bodyPreparation.ExpandedInputRaw continuationStatus, continuationReason, continuationUnavailable := responseCachePreparationFailure(bodyPreparation) @@ -5346,6 +5350,9 @@ func (h *Handler) Responses(c *gin.Context) { preflightPassthrough := continuousRetryPreflightPassthrough(preflightSettings) emptyIncomplete := &emptyIncompleteTracker{} forwardWithEvent := func(sseEvent string, data []byte) bool { + if multiAgentV2State.isOptimizedFor(account.ID()) { + data = restoreCodexMultiAgentV2Response(data, multiAgentV2State) + } streamDiag.markUpstreamFrame() if continuousRetryBuffersAttempts(continuousRetryPolicy) { compactionProvenancePayloads = append(compactionProvenancePayloads, bytes.Clone(data)) @@ -5605,6 +5612,9 @@ func (h *Handler) Responses(c *gin.Context) { seenImageOutputs := make(map[string]struct{}) emptyIncomplete := &emptyIncompleteTracker{} readErr = readSSEStreamWithContinuousRetryKeepalive(readCtx, resp.Body, func(sseEvent string, data []byte) bool { + if multiAgentV2State.isOptimizedFor(account.ID()) { + data = restoreCodexMultiAgentV2Response(data, multiAgentV2State) + } if continuousRetryBuffersAttempts(continuousRetryPolicy) { compactionProvenancePayloads = append(compactionProvenancePayloads, bytes.Clone(data)) } else { diff --git a/proxy/responses_ws.go b/proxy/responses_ws.go index 66314bdf0..bd9783497 100644 --- a/proxy/responses_ws.go +++ b/proxy/responses_ws.go @@ -323,6 +323,10 @@ func (h *Handler) forwardResponsesWebSocketTurn(c *gin.Context, conn *websocket. return writeResponsesWSErrorForContext(c, conn, apiKeyModelRequestError(err).apiErr) } defer func() { c.Request = quotaParentRequest }() + multiAgentV2State := &codexMultiAgentV2State{} + if codexMultiAgentV2CompatEnabled() { + c.Request = c.Request.WithContext(withCodexMultiAgentV2State(c.Request.Context(), multiAgentV2State)) + } c.Set(promptGuardPolicyEventIDContextKey, policyEventID) rawBody, model, apiErr := normalizeResponsesWebSocketClientPayload(rawPayload) if apiErr != nil { @@ -1178,6 +1182,7 @@ func (h *Handler) streamResponsesWSUpstream( emptyIncomplete := &emptyIncompleteTracker{} terminalFailureEventType := "" wroteAnyBody := false + multiAgentV2State := codexMultiAgentV2StateFromContext(c.Request.Context()) var wsReplay *continuousRetryWSReplay if continuousRetryBuffersAttempts(continuousRetryPolicy) { wsReplay = h.newContinuousRetryWSReplay() @@ -1228,6 +1233,9 @@ func (h *Handler) streamResponsesWSUpstream( } readErr = readSSEStreamWithContinuousRetryKeepalive(c.Request.Context(), resp.Body, func(sseEvent string, data []byte) bool { + if multiAgentV2State != nil && multiAgentV2State.isOptimizedFor(account.ID()) { + data = restoreCodexMultiAgentV2Response(data, multiAgentV2State) + } if wsReplay == nil { h.recordCompactionProvenanceFromPayload(context.Background(), account, data) } diff --git a/scripts/repro-codex-multi-agent-v2.py b/scripts/repro-codex-multi-agent-v2.py new file mode 100644 index 000000000..6eba6ba69 --- /dev/null +++ b/scripts/repro-codex-multi-agent-v2.py @@ -0,0 +1,350 @@ +#!/usr/bin/env python3 +"""复现 Codex Multi-Agent v2 子任务交付边界。 + +这是一个本地、无需凭据的协议实验脚本：启动 mock Responses 上游和一个可还原 +临时 collaboration namespace 的代理，再用真实 Codex CLI 请求该代理。脚本自身 +不读取或输出凭据，模型请求只发往本地 mock；Codex CLI 的 CODEX_HOME 指向本次 +运行的临时目录，不会读写调用者日常使用的配置和会话。 + +示例： + python scripts/repro-codex-multi-agent-v2.py --codex-exe C:/path/to/codex.exe --child-model --mode compat + python scripts/repro-codex-multi-agent-v2.py --codex-exe C:/path/to/codex.exe --child-model --mode native + +compat：代理把 collaboration-optimize 还原为 collaboration，mock 返回 + encrypted_function_args: []。预期子请求中的 agent_message 只包含明文 + input_text。 +native：不做还原，也不返回 encrypted_function_args。预期子请求中的 + agent_message 包含 encrypted_content（即用户报告的失败形态）。 +""" + +import argparse +import json +import os +import subprocess +import sys +import tempfile +import threading +import time +from http.client import HTTPConnection +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +# Windows 控制台默认可能是 GBK；统一按 UTF-8 输出，避免中文提示乱码。 +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + + +def sse(events): + out = b"" + for name, data in events: + out += f"event: {name}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n".encode() + return out + + +def completed(response_id): + return ( + "response.completed", + { + "type": "response.completed", + "response": { + "id": response_id, + "status": "completed", + "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + }, + }, + ) + + +class Capture: + def __init__(self, out_dir): + self.out_dir = Path(out_dir) + self.out_dir.mkdir(parents=True, exist_ok=True) + self.lock = threading.Lock() + self.count = 0 + + def save(self, body): + with self.lock: + self.count += 1 + index = self.count + (self.out_dir / f"request-{index}.json").write_bytes(body) + return index + + +def make_mock_handler(capture, mode, child_model): + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, fmt, *args): + pass + + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length) + index = capture.save(body) + try: + data = json.loads(body) + except Exception: + data = {} + input_items = data.get("input") or [] + call_names = [ + item.get("name") + for item in input_items + if isinstance(item, dict) and item.get("type") == "function_call" + ] + is_child = any( + isinstance(item, dict) and item.get("type") == "agent_message" + for item in input_items + ) + + if is_child: + events = [ + ("response.created", {"type": "response.created", "response": {"id": f"resp_child_{index}"}}), + ( + "response.output_item.done", + { + "type": "response.output_item.done", + "item": { + "type": "message", + "id": f"msg_child_{index}", + "role": "assistant", + "content": [{"type": "output_text", "text": "REPRO_CHILD_OK result=323"}], + }, + }, + ), + completed(f"resp_child_{index}"), + ] + elif "wait_agent" in call_names: + events = [ + ("response.created", {"type": "response.created", "response": {"id": f"resp_done_{index}"}}), + ( + "response.output_item.done", + { + "type": "response.output_item.done", + "item": { + "type": "message", + "id": f"msg_done_{index}", + "role": "assistant", + "content": [{"type": "output_text", "text": "REPRO_PARENT_DONE"}], + }, + }, + ), + completed(f"resp_done_{index}"), + ] + elif "spawn_agent" in call_names: + events = [ + ("response.created", {"type": "response.created", "response": {"id": f"resp_wait_{index}"}}), + ( + "response.output_item.done", + { + "type": "response.output_item.done", + "item": { + "type": "function_call", + "call_id": f"call_wait_{index}", + "namespace": "collaboration", + "name": "wait_agent", + "arguments": json.dumps({"timeout_ms": 30000}), + "encrypted_function_args": [], + }, + }, + ), + completed(f"resp_wait_{index}"), + ] + else: + item = { + "type": "function_call", + "call_id": f"call_spawn_{index}", + "namespace": "collaboration-optimize" if mode == "compat" else "collaboration", + "name": "spawn_agent", + "arguments": json.dumps( + { + "task_name": "repro_child", + "model": child_model, + "reasoning_effort": "low", + "fork_turns": "none", + "message": "REPRO_TASK_323 compute 17*19 and reply REPRO_CHILD_OK result=323", + }, + ensure_ascii=False, + ), + } + if mode == "compat": + item["encrypted_function_args"] = [] + events = [ + ("response.created", {"type": "response.created", "response": {"id": f"resp_spawn_{index}"}}), + ("response.output_item.done", {"type": "response.output_item.done", "item": item}), + completed(f"resp_spawn_{index}"), + ] + + payload = sse(events) + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + return Handler + + +def make_proxy_handler(mock_port, rewrite): + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, fmt, *args): + pass + + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(length) + conn = HTTPConnection("127.0.0.1", mock_port, timeout=120) + conn.request("POST", self.path, body=body, headers={"Content-Type": "application/json"}) + response = conn.getresponse() + payload = response.read() + if rewrite: + payload = payload.replace(b'"collaboration-optimize"', b'"collaboration"') + payload = payload.replace(b"collaboration-optimize.", b"collaboration.") + payload = payload.replace(b"collaboration-optimize__", b"collaboration__") + self.send_response(response.status) + self.send_header("Content-Type", response.getheader("Content-Type", "text/event-stream")) + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + return Handler + + +def start_server(handler): + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server + + +def find_child_agent_message(request_path, child_model): + data = json.loads(request_path.read_text(encoding="utf-8")) + if data.get("model") != child_model: + return None + for item in data.get("input") or []: + if isinstance(item, dict) and item.get("type") == "agent_message": + return item + return None + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--codex-exe", required=True, help="官方 Codex CLI 可执行文件路径") + parser.add_argument("--mode", choices=("compat", "native"), default="compat") + parser.add_argument("--parent-model", default="gpt-5.5") + parser.add_argument( + "--child-model", + required=True, + help="spawn_agent 可用的子模型名，由调用者按本机 Codex catalog 选择", + ) + parser.add_argument("--provider", default="custom") + parser.add_argument("--timeout", type=int, default=120) + args = parser.parse_args() + + work_dir = Path(tempfile.mkdtemp(prefix="codex-mav2-repro-")) + capture = Capture(work_dir / "requests") + mock = start_server(make_mock_handler(capture, args.mode, args.child_model)) + proxy = start_server(make_proxy_handler(mock.server_port, args.mode == "compat")) + workspace = work_dir / "workspace" + workspace.mkdir() + # 隔离 Codex 的配置与会话目录，避免读写调用者日常使用的 CODEX_HOME。 + codex_home = work_dir / "codex-home" + codex_home.mkdir() + child_env = os.environ.copy() + child_env["CODEX_HOME"] = str(codex_home) + + prompt = ( + "Spawn exactly one child with task_name repro_child, model " + f"{args.child_model}, and message REPRO_TASK_323 compute 17*19. " + "Then wait for it and report its reply." + ) + command = [ + args.codex_exe, + "exec", + "--json", + "--sandbox", + "read-only", + "--skip-git-repo-check", + "-C", + str(workspace), + "-m", + args.parent_model, + "-c", + f'model_provider="{args.provider}"', + "-c", + 'model_reasoning_effort="low"', + "-c", + "features.multi_agent=true", + "-c", + "features.multi_agent_v2=true", + "-c", + "agents.max_threads=2", + "-c", + "agents.max_depth=1", + "-c", + f'model_providers.{args.provider}.name="repro"', + "-c", + f'model_providers.{args.provider}.base_url="http://127.0.0.1:{proxy.server_port}/v1"', + "-c", + f'model_providers.{args.provider}.wire_api="responses"', + "-c", + f"model_providers.{args.provider}.requires_openai_auth=false", + "-c", + f"model_providers.{args.provider}.request_max_retries=0", + "-c", + f"model_providers.{args.provider}.stream_max_retries=0", + prompt, + ] + + print(f"工作目录: {work_dir}") + print("正在用真实 Codex CLI 请求本地 mock 上游...") + try: + result = subprocess.run( + command, capture_output=True, text=True, timeout=args.timeout, env=child_env + ) + except subprocess.TimeoutExpired: + print("FAIL: Codex CLI 超时") + return 2 + finally: + mock.shutdown() + proxy.shutdown() + + print(f"codex 退出码: {result.returncode}") + child_item = None + for request_path in sorted(capture.out_dir.glob("request-*.json")): + item = find_child_agent_message(request_path, args.child_model) + if item is not None: + child_item = (request_path, item) + break + + if child_item is None: + print("FAIL: 未捕获到包含 agent_message 的子请求") + print(result.stdout[-2000:]) + print(result.stderr[-2000:]) + return 2 + + request_path, item = child_item + parts = [ + (part.get("type"), len(part.get("text") or part.get("encrypted_content") or "")) + for part in item.get("content") or [] + if isinstance(part, dict) + ] + print(f"子请求: {request_path.name}") + print(f"agent_message content 组成: {parts}") + + has_encrypted = any(part_type == "encrypted_content" for part_type, _ in parts) + has_plaintext = any(part_type == "input_text" for part_type, _ in parts) + if args.mode == "compat": + ok = has_plaintext and not has_encrypted + expectation = "纯明文 input_text" + else: + ok = has_encrypted + expectation = "不透明 encrypted_content" + print(f"预期: {expectation}") + print("PASS" if ok else "FAIL") + return 0 if ok else 1 + + +if __name__ == "__main__": + sys.exit(main())