From ddd4de20fac1d224ad5afb8d710bcc52e7fcd821 Mon Sep 17 00:00:00 2001 From: Rogee Date: Fri, 9 Oct 2026 17:14:33 +0800 Subject: [PATCH] fix(ai): ignore duplicate params in favor of task and session fields --- AGENTS.md | 4 +- contracts/schema | 2 +- .../ai-params-task-precedence-20261009.md | 39 +++++++++++++++++ .../dispatcher-sip-startup-load-20261009.md | 2 + internal/ai/asr_sdk_test.go | 8 +++- internal/ai/binding.go | 20 --------- internal/ai/llm_params_test.go | 6 ++- internal/ai/params_precedence_test.go | 43 +++++++++++++++++++ internal/ai/pipeline.go | 8 +++- internal/ai/protocol_runtime_test.go | 4 +- internal/ai/schema_params_test.go | 6 +-- internal/dispatcher/ai_test.go | 17 ++++---- internal/rpc/approved_snapshot_test.go | 23 +++++++--- 13 files changed, 136 insertions(+), 46 deletions(-) create mode 100644 docs/evidence/ai-params-task-precedence-20261009.md create mode 100644 internal/ai/params_precedence_test.go diff --git a/AGENTS.md b/AGENTS.md index be975f6..c7af5b3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -79,7 +79,7 @@ ## 当前范围与权威入口 -- **2026-10-09 最新固定环境事实**：使用者确认修复正式 Dispatcher 的 SIP 加载遗漏并重部署；源码 `b37d9f3` 已推送、安装，Agent/Asterisk 的 revision `16`、三条数字 ID endpoint/AOR/主叫/contact/UDP/PCMA 已从原生运行态逐条核对，Asterisk 未重启，零活动通话，未清数据、未独立拨号。Dispatcher 仍因真实 SaaS 任务 `4`（revision `1`）同时包含 `agent.tts.voice` 与 `agent.tts.params.voice` 而退出重启；同值重复也触发现行参数保护规则，不静默删除或放宽检查。需 SaaS 移除 params 中的重复字段并增加修订后再核对，当前业务尚未 ready，不得把 SIP 加载成功或状态中的 `saas_pending` 标签当作全部配置/真实通话/生产签收。证据见 [`docs/evidence/dispatcher-sip-startup-load-20261009.md`](docs/evidence/dispatcher-sip-startup-load-20261009.md)。 +- **2026-10-09 最新固定环境事实**：使用者确认修复正式 Dispatcher 的 SIP 加载遗漏并重部署；源码 `b37d9f3` 已推送、安装，Agent/Asterisk 的 revision `16`、三条数字 ID endpoint/AOR/主叫/contact/UDP/PCMA 已从原生运行态逐条核对，Asterisk 未重启，零活动通话，未清数据、未独立拨号。Dispatcher 仍因真实 SaaS 任务 `4`（revision `1`）同时包含 `agent.tts.voice` 与 `agent.tts.params.voice` 而退出重启；该失败是当时采用的冲突拒绝规则造成；随后使用者确认改为任务/session 字段优先、忽略 params 中重名键，共享合同已更新为 `f490afa`，SaaS 无需删除原有重复 voice。新规则部署及最新业务状态见 [`docs/evidence/ai-params-task-precedence-20261009.md`](docs/evidence/ai-params-task-precedence-20261009.md)；未完成现场核对前，不得把 SIP 加载成功或状态中的 `saas_pending` 标签当作全部配置/真实通话/生产签收。证据见 [`docs/evidence/dispatcher-sip-startup-load-20261009.md`](docs/evidence/dispatcher-sip-startup-load-20261009.md)。 - **当前正式部署目标授权（使用非正式配置，`muzr4gzv-jzx8pg`）**：先核验、合入 main 并推送，再完善一键部署脚本，在登记测试机 `server.sip` 使用根目录 `saas.env`、`rabbitmq.env`、`aliyun-oss.env` 和真实 SaaS 配置，以现有 `nonprod-real` 启动固定的 `go-sip-agent.service`、`go-sip-dispatcher.service`；原生 `go-sip-asterisk.service` 保留。配置固定在 `~/.config/go-sip/`，业务数据固定在 `~/.local/share/go-sip/`，制品在 `~/.local/opt/go-sip/releases/`，不再用每次试验的新业务目录或 Mock 代替正式部署验证。正式部署入口为 `deploys/production/production.py`；当前使用非正式配置，仍保留 `nonprod-real`、登记测试机及原有授权边界，命名不代表生产验收。下方日期记录是历史事实，其旧授权限制不覆盖本条新授权。 - **本轮外呼授权**：真实 SaaS 下发的全部获批外呼任务允许按正式规则执行，不再逐通询问；仍核对归属、号码、时段、额度、并发、有效授权与幂等，不自行生成外呼消息、不盲重拨或换线重拨。独立工具试拨及超出上述真实 SaaS 任务范围的操作仍需另行确认。没有任务或未接通时如实标记未验证，不以 Mock 代签真实通话或生产。 @@ -110,7 +110,7 @@ - 非生产真实 Agent 的 Asterisk `res_hep`/`res_hep_pjsip` 仅镜像至本机 UDP：在 ARI Dial 前读取本次 SIP Call-ID 并绑定执行，按 Call-ID 和原始 INVITE transaction 只采集真实最终响应的状态码、状态行与完整 `raw`；丢失/无法关联以 `sip_capture_error` 显式报告，不从 ARI、号码或挂断原因推测。镜像缺失不阻止已证实终结的额度释放；未经确认的占用仍保留。配置及回滚见 [`deploys/cell/README.md`](deploys/cell/README.md)，不得将完整 SIP 报文写入普通日志、提交或聊天；本地测试通过不等于测试机已部署 HEP 或真实接通。 - 2026-10-08 用户批准本地配置规则调整及新模型调用支持：SaaS SIP `revision` 可省略，完整读取覆盖旧配置；`transport/auth_mode` 判定忽略大小写，但不补其它缺失线路参数。Dispatcher 持久分配内部 SIP 加载代次并核对 Agent/Asterisk，旧通话排空与未知占用保留规则不变。任务列表仅忽略 `schema_version/control_seq`。任务用唯一的 `provider_ref` 对应连接的 `provider_code`，`provider_id` 仅为连接记录标识；一份连接可供多用途复用；任务决定厂商/模型/ASR–LLM–TTS 用途，provider 只提供连接信息，不要求 `role/adapter/enabled`，未知供应商/协议及不可表达的连接参数显式拒绝。2026-10-08 追加用户确认：移除精确型号/音色业务硬编码；同协议模型及可表达参数从任务取得。当前 TTS 固定使用 task-based WebSocket，不按 model 名或地址猜协议、不失败回退 HTTP。模型实际可用性由供应商调用返回事实决定，不以本地模拟代签。新增本地调用代码及模拟测试不授权部署、拨号或真实 AI 请求。SQLite 当前布局因新增持久 SIP 快照变为 3；旧布局只读拒绝，不自动迁移或改写旧文件。 - 2026-10-08 用户确认按共享合同 `b502ad2` 本地适配任务线路并发：`allowed_trunk_ids` 只接受包含 `trunk_id/concurrency` 的对象数组，重复线路拒绝；非负整数 `concurrency` 的 0 表示该任务不使用该线路。选线及 SQLite 原子占用同时遵守租户、任务总量、线路总量、任务在线路上的上限，未知执行不释放；任务修订在选线后变化时拒绝原选择，不用旧快照发出新呼叫。占用仍由现有 inbox 与冻结任务快照确定，不新增业务表、不转换旧快照或迁移旧库。ASR/TTS 使用连接的 `ws_endpoint`，LLM 使用 `api_endpoint`，缺失不互相回退。本轮仅本地代码、测试和说明，不部署、不拨号、不请求真实 AI。 -- AI 使用任务内不可变授权快照：当前使用火山 ASR、OpenAI 兼容 LLM、百炼 task-based ASR/TTS，TTS 固定 WebSocket。任务的 `params` 为必填、可空的平坦对象，值为字符串、数字、布尔值或 null；原样透传，不校验厂商参数值、不补 SDK 默认值，不限制精确型号/音色。任务/session 的 model、voice、消息及事务身份不得被 params 覆盖，冲突明确拒绝。不设或检查 `prompt.max_bytes`，提示词不截断。ASR 输入格式与内部媒体格式分离：任务 params.sample_rate 声明所选服务支持的目标采样率，Agent 须在对接边界转换并保证实际音频与声明一致，不得忽略配置、仅重标参数或失败换率。当前内部媒体仍为单声道 16 kHz PCM16，ASR 边界转换尚未实现，已列入 docs/TODO.md；合同或 Schema 接受其它采样率不代表当前运行已经支持，不能以本次更新声称媒体兼容。ASR-only 不启动 LLM/TTS，完整 AI 不借旧语音测试的授权或参数。只有最终用户 ASR 文本的明确字面关键词可触发拒联/挂断；`agent.conversation.hangup_keywords` 仅接受必填非空 `name`/`triggers`/`closingRemark` 的对象数组，多组命中按配置顺序取第一组，经获批 TTS 完整播放该组结束语后主动挂断，不调用 LLM、不开始新一轮对话；合成/播放失败须明确报错并尝试结束，不重播，旧字符串数组直接拒绝；不由 SDK 默认值、环境、CLI、metadata 或宽松 Schema 改写业务参数，不因 SDK 重试产生第二次发起/收费或重播。日志只存脱敏版本/摘要/计数，不存密钥、prompt、完整对话或音频。 +- AI 使用任务内不可变授权快照：当前使用火山 ASR、OpenAI 兼容 LLM、百炼 task-based ASR/TTS，TTS 固定 WebSocket。任务的 `params` 为必填、可空的平坦对象，值为字符串、数字、布尔值或 null；原样透传，不校验厂商参数值、不补 SDK 默认值，不限制精确型号/音色。任务/session 的 model、voice、消息及事务身份具有优先权；params 中与生成字段重名的键忽略（包括同值、不同值或 null），不因此拒绝任务，不修改获批原始快照，其余参数继续原值透传。不设或检查 `prompt.max_bytes`，提示词不截断。ASR 输入格式与内部媒体格式分离：任务 params.sample_rate 声明所选服务支持的目标采样率，Agent 须在对接边界转换并保证实际音频与声明一致，不得忽略配置、仅重标参数或失败换率。当前内部媒体仍为单声道 16 kHz PCM16，ASR 边界转换尚未实现，已列入 docs/TODO.md；合同或 Schema 接受其它采样率不代表当前运行已经支持，不能以本次更新声称媒体兼容。ASR-only 不启动 LLM/TTS，完整 AI 不借旧语音测试的授权或参数。只有最终用户 ASR 文本的明确字面关键词可触发拒联/挂断；`agent.conversation.hangup_keywords` 仅接受必填非空 `name`/`triggers`/`closingRemark` 的对象数组，多组命中按配置顺序取第一组，经获批 TTS 完整播放该组结束语后主动挂断，不调用 LLM、不开始新一轮对话；合成/播放失败须明确报错并尝试结束，不重播，旧字符串数组直接拒绝；不由 SDK 默认值、环境、CLI、metadata 或宽松 Schema 改写业务参数，不因 SDK 重试产生第二次发起/收费或重播。日志只存脱敏版本/摘要/计数，不存密钥、prompt、完整对话或音频。 - **私有配置位置（本机路径相对本仓库根目录，只读，绝不提交）**：`.local/provider-ai.env` 是 `0600` 的 `KEY=VALUE` 文件；字段名为 `BAILIAN_API_KEY`、`BAILIAN_BASE_URL`、`BAILIAN_WSS_BASE_URL`、`BAILIAN_TTS_VOICE`、`VOLC_ASR_APP_NAME`、`VOLC_ASR_APP_KEY`、`VOLCENGINE_ACCESS_KEY`、`VOLCENGINE_SECRET_KEY`、`VOLCENGINE_REGION`、`VOLCENGINE_DISABLE_SSL`。根目录 `aliyun-oss.env` 也是 `0600`，**不是 shell env 文件**；它以冒号分隔，字段名准确为 `bucket`、`Endpoint`、`Region`，以及 `RAM` 下的 `username`、`accessKeyId`、`accessKeySecret`（大小写须保持原样）。测试机 `rogee` 用户的现行 ARI 文件位于 `~/.config/go-sip-asterisk/{ari.conf,http.conf,ari-secret}`，不是旧 `.local/asterisk-*/ari.conf`；访问测试机前先核对已登记的 SSH 主机指纹，不展示 `ari-secret`。 - **下次安全读取步骤**：先确认工作目录是本仓库，用 `stat` 仅检查本机两份文件是否存在、所有者与权限 `0600`；不满足即停止。按各自格式在受限本机进程中解析所需字段到内存，不执行 `source`、不打印全文/字段值、不写临时明文副本，不把密钥、签名 URL、音频或完整对话带入聊天、日志、提交及长期证据。AI 的历史文件只可作为**获准凭据来源**，模型/voice/速度等仍由当前获批的 task/providers 快照固定，不能用环境变量覆盖。OSS 历史文件也不能直接传给 `DISPATCHER_OSS_CONFIG_FILE`：该运行配置要求私有 JSON、`dispatcher_id` 和 `oss` 字段，并以环境变量**名称引用**密钥；需按现行合同构造并核验授权后才能使用。普通构建和测试不读取这些私有文件；真实服务测试必须显式启用对应 opt-in 并受现行门禁约束。 - **授权边界**：本轮验收目标是三条已登记线路的真实接通及 LLM 正常应答；旧两个号码已分别在三条线路试拨，六通均为 SIP 480，零接通。新增号码的历史逐次授权不等于本次代码变更获准部署或拨号；本次全局审查本地业务硬编码并以 SaaS 配置快照决定任务、线路和额度，**不部署、不拨号**。上述 AI 与 OSS 私有配置仍仅用于另经明确授权的非生产测试，下次任务须重新确认范围和真实服务调用授权，不能沿用本轮或历史一次性授权。不得把历史配置直接当 SaaS 快照、任务授权或真实呼叫准入，不覆盖/清理旧 OSS 对象。 diff --git a/contracts/schema b/contracts/schema index 4f51579..f490afa 160000 --- a/contracts/schema +++ b/contracts/schema @@ -1 +1 @@ -Subproject commit 4f515796e30c0cfd6f86b39ec2324fa9a8b2b964 +Subproject commit f490afab764a702dda919c718dca1a05e5bceb0a diff --git a/docs/evidence/ai-params-task-precedence-20261009.md b/docs/evidence/ai-params-task-precedence-20261009.md new file mode 100644 index 0000000..0e0d3a4 --- /dev/null +++ b/docs/evidence/ai-params-task-precedence-20261009.md @@ -0,0 +1,39 @@ +# AI 参数重名优先级调整（2026-10-09） + +## 使用者确认的规则与范围 + +此前契约和绑定检查拒绝 params 中的协议重名字段，真实 SaaS 任务 4 同时给出 `agent.tts.voice` 与 `agent.tts.params.voice`，导致 Dispatcher 退出。 + +使用者明确确认变更：**任务/会话生成字段优先，忽略 params 中重名键，其他参数按原值传递**。同值、不同值或 null 均使用该优先级；不修改获批任务原始快照，不要求 SaaS 删除原有重复字段。参数仍是必填、可空的平坦对象，本轮不改变其他结构和业务规则。 + +同步共享契约、实现、测试和说明，验证后提交并推送、重新部署 `server.sip`。保留固定配置、数据和既有 SaaS 任务授权，不清理或迁移数据、不操作真实 MQ 拓扑、不重启 Asterisk、不独立试拨或主动投递任务。 + +## 共享契约及 SaaS 通知 + +共享仓库 `sip-contracts` 提交 `f490afab764a702dda919c718dca1a05e5bceb0a` 已推送，本项目同步固定该提交。规范位置为 `contracts/schema/saas-dispatcher.md` 的 AI 参数段落。Schema 已接受这类平坦对象，无需改变数据结构。 + +提供给 SaaS 开发人员的变更通知：重名参数不再导致任务拒绝，任务/session 的 model、voice、消息及事务身份具有优先权。现有任务中重复 voice 不需要删除或更改任务内容，其他参数继续原值传递；其余现行契约不变。 + +## 实际改动 + +- 删除绑定阶段 ASR/LLM/TTS 重名参数拒绝和废弃的 `reservedParams` 路径，不保留旧拒绝兼容层。 +- LLM 和百炼 TTS 已有“复制 params，再写入任务/会话字段”的组装逻辑，分别保证 model/messages、voice 优先；补完整任务绑定到实际请求的回归，不只是手工构造绑定对象。 +- 火山 ASR 在对接 SDK 前复制 params，并移除与 SDK 生成字段重名的 `reqid`、`sequence`、`model_name`。请求使用任务 model 和真实 session 身份，其他参数保持原值，原快照及绑定中的 params 不改。 +- SDK 自身保护生成身份的规则不变，本次是业务优先级调整，不是 SDK 缺陷，不改依赖版本或引入本地 replace。 +- 更新现行 `AGENTS.md`，旧部署证据标明当时拒绝规则及要求 SaaS 删除 voice 的建议已被取代，历史事实仍保留。 + +## 验证 + +- 测试先加入重名键，旧代码在 ASR、LLM、TTS 绑定阶段明确失败；修复后通过。 +- 同值、不同值、null 三组重名值均接受，任务字段不被改写，原始任务快照不变。 +- ASR WebSocket 实际请求：model_name 采用任务值，reqid/sequence 由 session 生成，原绑定中重名值仍保留；其他字段以及 false、null、大整数保持原值。 +- LLM HTTP 实际请求：model/messages 采用任务/会话值，其他字段原值传递。 +- 百炼 TTS WebSocket 实际请求：voice 采用任务值，其他字段原值传递。 +- Agent 正式快照与重放保护测试接受重名键、不改冻结绑定；缺失/null/非对象/嵌套参数等原先无效结构仍明确拒绝，拒绝不得持久接纳或拨号。更新旧的“重名即无效”测试为真正的结构错误测试，不削弱基础校验。 +- `make proto-check acceptance-local`、`make release-check-local`、`go vet ./...`、`go test -race ./...`、构建及隔离 RabbitMQ/HTTPS/双向 TLS 端到端全部通过。 +- 业务单元覆盖率 **70.8%**。 +- 本地 mock 请求不代表供应商实际可用、真实通话、AI/录音/OSS 或生产验收。 + +## 部署验证 + +安装后补固定源码、制品校验、真实 SaaS 任务是否继续拒绝、服务就绪与剩余问题；未验证前不声称恢复。 diff --git a/docs/evidence/dispatcher-sip-startup-load-20261009.md b/docs/evidence/dispatcher-sip-startup-load-20261009.md index bfe7bc8..e3f435b 100644 --- a/docs/evidence/dispatcher-sip-startup-load-20261009.md +++ b/docs/evidence/dispatcher-sip-startup-load-20261009.md @@ -1,5 +1,7 @@ # 正式 Dispatcher SIP 加载遗漏修复（2026-10-09） +> 本文记录当时的冲突拒绝规则和部署事实。随后使用者确认 params 重名键应忽略、任务/session 字段优先；该旧拒绝规则及“要求 SaaS 删除重复 voice”的建议已被新契约取代，不作为现行要求。新规则与后续部署结果见 [参数优先级变更记录](ai-params-task-precedence-20261009.md)。 + ## 授权与完成标准 使用者在只读根因定位后确认按已讨论方案修复、验证并重新部署 `server.sip`：正式入口接入现有 SIP 加载能力，保留占用检查与实际版本核对，补启动和更新回归测试。 diff --git a/internal/ai/asr_sdk_test.go b/internal/ai/asr_sdk_test.go index 28ff169..4cca526 100644 --- a/internal/ai/asr_sdk_test.go +++ b/internal/ai/asr_sdk_test.go @@ -42,7 +42,7 @@ func TestASRParametersReachDoubaoSDKWebSocket(t *testing.T) { defer server.Close() task, providers := currentFixture(t, tc.mode) task = changeCurrentAgent(t, task, func(a map[string]any) { - a["asr"].(map[string]any)["params"] = map[string]any{"result_type": tc.resultType, "enable_nonstream": tc.nonstream, "unknown_flag": false, "nullable": nil, "opaque": "vendor-only", "large_number": json.Number("9007199254740993")} + a["asr"].(map[string]any)["params"] = map[string]any{"reqid": "ignored-params-reqid", "sequence": 99, "model_name": "ignored-params-model", "result_type": tc.resultType, "enable_nonstream": tc.nonstream, "unknown_flag": false, "nullable": nil, "opaque": "vendor-only", "large_number": json.Number("9007199254740993")} }) asrProvider := providers["volcengine"] asrProvider.WSEndpoint = "ws" + strings.TrimPrefix(server.URL, "http") @@ -76,6 +76,12 @@ func TestASRParametersReachDoubaoSDKWebSocket(t *testing.T) { } audio := payload["audio"].(map[string]any) request := payload["request"].(map[string]any) + if request["model_name"] != bound.ASR.Model || request["reqid"] == "ignored-params-reqid" || request["reqid"] == "" || request["sequence"] != float64(1) { + t.Fatal("params overrode the ASR task model or session identity") + } + if string(bound.ASR.Params["model_name"]) != `"ignored-params-model"` || string(bound.ASR.Params["reqid"]) != `"ignored-params-reqid"` || string(bound.ASR.Params["sequence"]) != "99" { + t.Fatal("ASR request construction changed the original params") + } if len(audio) != 4 || len(request) != 9 || audio["format"] != "pcm_s16le" || audio["sample_rate"] != float64(16000) || audio["channel"] != float64(1) || audio["bits"] != float64(16) || request["result_type"] != tc.resultType || request["enable_nonstream"] != tc.nonstream || request["unknown_flag"] != false || request["nullable"] != nil || request["opaque"] != "vendor-only" { t.Fatalf("SDK ASR request omitted approved input/interim fields: audio=%v request=%v", audio, request) } diff --git a/internal/ai/binding.go b/internal/ai/binding.go index b98a495..af86520 100644 --- a/internal/ai/binding.go +++ b/internal/ai/binding.go @@ -101,9 +101,6 @@ func Bind(task configread.Task, providers map[string]configread.Provider) (Bindi if err != nil { return Binding{}, err } - if err := reservedParams(s.ASR.Params, "reqid", "sequence", "model_name"); err != nil { - return Binding{}, err - } b.ASR = ASRConfig{Provider: p, Model: s.ASR.Model, Params: s.ASR.Params} if s.Mode == "asr_only" { return b, nil @@ -115,17 +112,11 @@ func Bind(task configread.Task, providers map[string]configread.Provider) (Bindi if err != nil { return Binding{}, err } - if err := reservedParams(s.LLM.Params, "model", "messages"); err != nil { - return Binding{}, err - } b.LLM = &LLMConfig{Provider: p, Model: s.LLM.Model, Params: s.LLM.Params} p, err = currentProvider(providers, s.TTS.ProviderRef, "tts") if err != nil { return Binding{}, err } - if err := reservedParams(s.TTS.Params, "voice"); err != nil { - return Binding{}, err - } b.TTS = &TTSConfig{Provider: p, Protocol: TTSProtocolDashScopeTask, Model: s.TTS.Model, Voice: s.TTS.Voice, Params: s.TTS.Params} b.Prompt = s.Prompt.Text b.AllowedVariables = append([]string(nil), s.Prompt.AllowedVariables...) @@ -150,17 +141,6 @@ func Bind(task configread.Task, providers map[string]configread.Provider) (Bindi return b, nil } -// Protocol identity belongs to the task/session, not vendor business parameters. -// Reject collisions explicitly instead of overwriting or dropping supplied params. -func reservedParams(params map[string]json.RawMessage, keys ...string) error { - for _, key := range keys { - if _, found := params[key]; found { - return fmt.Errorf("params conflicts with protocol field %q", key) - } - } - return nil -} - func currentProvider(providers map[string]configread.Provider, ref, role string) (configread.Provider, error) { p, ok := providers[ref] if !ok || p.Code != ref || p.Credential == "" { diff --git a/internal/ai/llm_params_test.go b/internal/ai/llm_params_test.go index 07a6f33..15b8dac 100644 --- a/internal/ai/llm_params_test.go +++ b/internal/ai/llm_params_test.go @@ -13,7 +13,7 @@ import ( func TestLLMWirePreservesOpaqueScalarParamsWithoutDefaults(t *testing.T) { task, ps := currentFixture(t, "full_ai") task = changeCurrentAgent(t, task, func(a map[string]any) { - a["llm"].(map[string]any)["params"] = map[string]any{"opaque": "vendor-value", "unknown_flag": false, "nullable": nil, "large_number": json.Number("9007199254740993"), "stream": false} + a["llm"].(map[string]any)["params"] = map[string]any{"model": "ignored-params-model", "messages": "ignored-params-messages", "opaque": "vendor-value", "unknown_flag": false, "nullable": nil, "large_number": json.Number("9007199254740993"), "stream": false} }) b, e := Bind(task, ps) if e != nil { @@ -36,6 +36,10 @@ func TestLLMWirePreservesOpaqueScalarParamsWithoutDefaults(t *testing.T) { if len(body) != 7 || body["stream"] != false || body["model"] != "example-chat" || body["opaque"] != "vendor-value" || body["unknown_flag"] != false || body["nullable"] != nil { t.Error("params dropped, rewritten or defaulted") } + messages, ok := body["messages"].([]any) + if !ok || len(messages) != 2 || messages[1].(map[string]any)["content"] != "Final user text" { + t.Error("params overrode the task/session messages") + } w.Header().Set("Content-Type", "application/json") io.WriteString(w, `{"choices":[{"message":{"role":"assistant","content":"ok"}}]}`) })) diff --git a/internal/ai/params_precedence_test.go b/internal/ai/params_precedence_test.go new file mode 100644 index 0000000..fd32023 --- /dev/null +++ b/internal/ai/params_precedence_test.go @@ -0,0 +1,43 @@ +package ai + +import ( + "bytes" + "testing" +) + +func TestTaskFieldsWinOverSameDifferentAndNullParamValues(t *testing.T) { + for _, name := range []string{"same", "different", "null"} { + t.Run(name, func(t *testing.T) { + task, providers := currentFixture(t, "full_ai") + canonical, err := Bind(task, providers) + if err != nil { + t.Fatal(err) + } + task = changeCurrentAgent(t, task, func(agent map[string]any) { + asrModel, llmModel, voice := any("ignored-model"), any("ignored-model"), any("ignored-voice") + if name == "same" { + asrModel, llmModel, voice = canonical.ASR.Model, canonical.LLM.Model, canonical.TTS.Voice + } else if name == "null" { + asrModel, llmModel, voice = nil, nil, nil + } + agent["asr"].(map[string]any)["params"] = map[string]any{"model_name": asrModel, "reqid": "ignored-id", "sequence": 99, "opaque": false} + agent["llm"].(map[string]any)["params"] = map[string]any{"model": llmModel, "messages": "ignored-messages", "opaque": false} + agent["tts"].(map[string]any)["params"] = map[string]any{"voice": voice, "opaque": false} + }) + original := append([]byte(nil), task.Raw...) + bound, err := Bind(task, providers) + if err != nil { + t.Fatalf("duplicate opaque params must not reject the task: %v", err) + } + if bound.ASR.Model != canonical.ASR.Model || bound.LLM.Model != canonical.LLM.Model || bound.TTS.Voice != canonical.TTS.Voice { + t.Fatal("params changed canonical task fields") + } + if string(bound.ASR.Params["opaque"]) != "false" || string(bound.LLM.Params["opaque"]) != "false" || string(bound.TTS.Params["opaque"]) != "false" { + t.Fatal("nonconflicting opaque params were changed") + } + if !bytes.Equal(original, task.Raw) { + t.Fatal("binding changed the approved task snapshot") + } + }) + } +} diff --git a/internal/ai/pipeline.go b/internal/ai/pipeline.go index 281a5bd..8baec1a 100644 --- a/internal/ai/pipeline.go +++ b/internal/ai/pipeline.go @@ -149,10 +149,16 @@ func (b Binding) Recognize(ctx context.Context, pcm16 []byte) (string, error) { doubaospeech.WithResourceID(doubaospeech.ResourceASRStreamV2), doubaospeech.WithWebSocketURL(u.String()), ) + // Task/session identity wins over opaque params. Keep the approved + // snapshot unchanged and remove only collisions from the SDK extras. + parameters := cloneParams(b.ASR.Params) + for _, key := range []string{"reqid", "sequence", "model_name"} { + delete(parameters, key) + } request := doubaospeech.ASRV2Config{ Format: doubaospeech.FormatPCMS16LE, SampleRate: 16000, Channel: 1, Bits: 16, Request: &doubaospeech.ASRV2RequestConfig{ModelName: b.ASR.Model}, - Parameters: b.ASR.Params, + Parameters: parameters, } session, err := client.ASRV2.OpenStreamSession(ctx, &request) if err != nil { diff --git a/internal/ai/protocol_runtime_test.go b/internal/ai/protocol_runtime_test.go index afacd52..4ded167 100644 --- a/internal/ai/protocol_runtime_test.go +++ b/internal/ai/protocol_runtime_test.go @@ -14,11 +14,13 @@ import ( func TestTaskTTSProtocolForwardsOpaqueModelVoiceAndParams(t *testing.T) { task, ps := protocolTask(t, TTSProtocolDashScopeTask) + task = changeCurrentAgent(t, task, func(agent map[string]any) { + agent["tts"].(map[string]any)["params"] = map[string]any{"voice": "ignored-params-voice", "opaque": "vendor-only", "rate": 9.5, "unknown_flag": false, "nullable": nil, "large_number": json.Number("9007199254740993")} + }) b, err := Bind(task, ps) if err != nil { t.Fatal(err) } - b.TTS.Params = map[string]json.RawMessage{"opaque": json.RawMessage(`"vendor-only"`), "rate": json.RawMessage("9.5"), "unknown_flag": json.RawMessage("false"), "nullable": json.RawMessage("null"), "large_number": json.RawMessage("9007199254740993")} expected := []byte{1, 0, 2, 0} s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { c, e := websocket.Accept(w, r, nil) diff --git a/internal/ai/schema_params_test.go b/internal/ai/schema_params_test.go index b9bd15e..f325130 100644 --- a/internal/ai/schema_params_test.go +++ b/internal/ai/schema_params_test.go @@ -29,12 +29,12 @@ func TestNewSchemaBindsCodeReferenceAndOpaqueParams(t *testing.T) { } } -func TestParamsRejectOnlyMalformedShapeAndReservedProtocolFields(t *testing.T) { - for _, value := range []any{nil, []any{}, map[string]any{"nested": map[string]any{}}, map[string]any{"messages": "override"}} { +func TestParamsRejectOnlyMalformedShape(t *testing.T) { + for _, value := range []any{nil, []any{}, map[string]any{"nested": map[string]any{}}} { task, providers := currentFixture(t, "full_ai") task = changeCurrentAgent(t, task, func(a map[string]any) { a["llm"].(map[string]any)["params"] = value }) if _, err := Bind(task, providers); err == nil { - t.Fatal("invalid parameter shape or protocol identity override accepted") + t.Fatal("invalid parameter shape accepted") } } } diff --git a/internal/dispatcher/ai_test.go b/internal/dispatcher/ai_test.go index 55a11b2..abd1155 100644 --- a/internal/dispatcher/ai_test.go +++ b/internal/dispatcher/ai_test.go @@ -6,7 +6,6 @@ import ( "net/http" "net/http/httptest" "path/filepath" - "strings" "testing" "time" @@ -14,17 +13,17 @@ import ( "git.ipao.vip/rogee/go-sip/internal/store" ) -func unsupportedASRTask(t *testing.T) []byte { +func malformedASRTask(t *testing.T) []byte { t.Helper() original := configExample(t, "config-read-task-asr") - modified := bytes.Replace(original, []byte(`"params":{}`), []byte(`"params":{"reqid":"override"}`), 1) + modified := bytes.Replace(original, []byte(`"params":{}`), []byte(`"params":{"nested":{"unexpected":true}}`), 1) if bytes.Equal(original, modified) { t.Fatal("ASR fixture did not contain params") } return modified } -func TestBootstrapRejectsProtocolOverrideBeforeOpeningAdmission(t *testing.T) { +func TestBootstrapRejectsMalformedParamsBeforeOpeningAdmission(t *testing.T) { id := "c046b893-8628-4589-ae50-619d049248a6" server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { var body []byte @@ -40,7 +39,7 @@ func TestBootstrapRejectsProtocolOverrideBeforeOpeningAdmission(t *testing.T) { case "/internal/v1/dispatcher/ai-providers": body = configExample(t, "config-read-providers") case "/internal/v1/dispatcher/task/task-asr": - body = unsupportedASRTask(t) + body = malformedASRTask(t) case "/internal/v1/dispatcher/tenant/1001/quota": body = configExample(t, "config-read-quota") default: @@ -67,8 +66,8 @@ func TestBootstrapRejectsProtocolOverrideBeforeOpeningAdmission(t *testing.T) { DrainControls: func(context.Context) error { drained = true; return nil }, } err = b.Run(context.Background()) - if err == nil || !strings.Contains(err.Error(), "protocol field") || drained { - t.Fatalf("SDK-unsupported task must not be admitted or drained: %v, drained=%t", err, drained) + if err == nil || drained { + t.Fatalf("malformed task must not be admitted or drained: %v, drained=%t", err, drained) } if admitted, err := db.CanAdmit(id, 1001, "task-asr"); err != nil || admitted { t.Fatalf("invalid AI task opened admission: admitted=%t err=%v", admitted, err) @@ -85,7 +84,7 @@ func TestExecuteCorruptAISnapshotFailsClosedBeforeReservation(t *testing.T) { } defer db.Close() snapshot := policySnapshot(t) - if err := snapshot.Task.UnmarshalJSON(unsupportedASRTask(t)); err != nil { + if err := snapshot.Task.UnmarshalJSON(malformedASRTask(t)); err != nil { t.Fatal(err) } id := snapshot.Task.DispatcherID @@ -104,7 +103,7 @@ func TestExecuteCorruptAISnapshotFailsClosedBeforeReservation(t *testing.T) { Now: func() time.Time { return monday(9, 30) }, } err = controller.ProcessExecute(context.Background(), executeBody(t, "bad-ai", "15003164745")) - if err == nil || !strings.Contains(err.Error(), "protocol field") || len(originator.calls) != 0 { + if err == nil || len(originator.calls) != 0 { t.Fatalf("corrupt AI config must not reach Agent: err=%v calls=%d", err, len(originator.calls)) } if admitted, err := db.CanAdmit(id, 1001, "task-asr"); err != nil || admitted { diff --git a/internal/rpc/approved_snapshot_test.go b/internal/rpc/approved_snapshot_test.go index ece638c..126c34e 100644 --- a/internal/rpc/approved_snapshot_test.go +++ b/internal/rpc/approved_snapshot_test.go @@ -17,6 +17,12 @@ import ( func TestApprovedAgentReceivesFrozenASROnlySnapshot(t *testing.T) { now := time.Date(2026, 9, 20, 10, 0, 0, 0, time.UTC) req := approvedTestRequest(t, now) + req.TaskConfigJson = bytes.Replace(req.TaskConfigJson, []byte(`"params":{}`), []byte(`"params":{"reqid":"ignored-id","sequence":99,"model_name":"ignored-model","opaque":false}`), 1) + hash, err := configread.ExecutionBindingSHA256(req.TaskConfigJson, req.ProvidersJson, req.SipRevision) + if err != nil { + t.Fatal(err) + } + req.BindingSha256 = hash attempts := 0 s := activatedApprovedServer(t, now, filepath.Join(t.TempDir(), "session.json"), req.DispatcherId, 1, func(_ context.Context, call ApprovedExecution) error { @@ -27,6 +33,9 @@ func TestApprovedAgentReceivesFrozenASROnlySnapshot(t *testing.T) { if call.AI.Mode != "asr_only" || call.AI.LLM != nil || call.AI.TTS != nil || call.AI.ASR.Provider.Credential != "sk-******" { t.Fatal("Agent did not receive the exact approved ASR-only provider snapshot") } + if string(call.AI.ASR.Params["reqid"]) != `"ignored-id"` || string(call.AI.ASR.Params["opaque"]) != "false" { + t.Fatal("Agent changed the original approved params before request construction") + } return nil }) result, err := s.ExecuteApproved(context.Background(), req) @@ -38,12 +47,12 @@ func TestApprovedAgentReceivesFrozenASROnlySnapshot(t *testing.T) { } } -func TestApprovedAgentRejectsProtocolOverrideWithoutPersistingAttempt(t *testing.T) { +func TestApprovedAgentRejectsMalformedParamsWithoutPersistingAttempt(t *testing.T) { now := time.Date(2026, 9, 20, 10, 0, 0, 0, time.UTC) req := approvedTestRequest(t, now) - req.TaskConfigJson = bytes.Replace(req.TaskConfigJson, []byte(`"params":{}`), []byte(`"params":{"reqid":"override"}`), 1) - if !strings.Contains(string(req.TaskConfigJson), `"reqid":"override"`) { - t.Fatal("protocol override was not inserted into fixture") + req.TaskConfigJson = bytes.Replace(req.TaskConfigJson, []byte(`"params":{}`), []byte(`"params":{"nested":{"unexpected":true}}`), 1) + if !strings.Contains(string(req.TaskConfigJson), `"nested":{"unexpected":true}`) { + t.Fatal("malformed params were not inserted into fixture") } hash, err := configread.ExecutionBindingSHA256(req.TaskConfigJson, req.ProvidersJson, req.SipRevision) if err != nil { @@ -55,10 +64,10 @@ func TestApprovedAgentRejectsProtocolOverrideWithoutPersistingAttempt(t *testing s := activatedApprovedServer(t, now, state, req.DispatcherId, 1, func(context.Context, ApprovedExecution) error { attempts++; return nil }) _, err = s.ExecuteApproved(context.Background(), req) - if status.Code(err) != codes.FailedPrecondition || attempts != 0 || strings.Contains(err.Error(), "example-only-not-a-real-secret") { - t.Fatalf("unsupported AI must fail before dialing or leaking credentials: attempts=%d err=%v", attempts, err) + if status.Code(err) != codes.InvalidArgument || attempts != 0 || strings.Contains(err.Error(), "example-only-not-a-real-secret") { + t.Fatalf("malformed AI must fail before dialing or leaking credentials: attempts=%d err=%v", attempts, err) } if _, statErr := os.Stat(state + ".approved"); !os.IsNotExist(statErr) { - t.Fatalf("no attempt should be durable before SDK capability validation: %v", statErr) + t.Fatalf("no attempt should be durable before AI snapshot validation: %v", statErr) } }