From e779fe1c571d7b135ebd49a64031565523206fb9 Mon Sep 17 00:00:00 2001 From: Rogee Date: Tue, 13 Oct 2026 14:56:50 +0800 Subject: [PATCH] fix(media): use static PCMA for Asterisk RTP audio --- AGENTS.md | 2 + cmd/sip-call/README.md | 16 +- cmd/sip-call/sip.env.example | 3 +- cmd/sip-go-agent/agent_real.go | 2 +- deploys/cell/README.md | 8 + deploys/test/sip-call-home/README.md | 20 +- deploys/test/sip-call-home/results/README.md | 3 +- deploys/test/sip-call-home/sip/README.md | 2 +- deploys/test/sip-call-home/task/README.md | 14 +- .../evidence/sip-call-capture-asr-20261013.md | 83 ++++++++ internal/ai/asr_sample_rate.go | 2 + internal/ai/protocol_pcma_test.go | 51 +++++ internal/ai/protocol_turn_test.go | 48 ++++- internal/asterisk/call.go | 11 +- internal/asterisk/call_test.go | 18 +- internal/asterisk/pcma_media_test.go | 92 +++++++++ internal/callflow/flow.go | 4 + internal/media/pcma_stream_test.go | 146 ++++++++++++++ internal/media/rtp_test.go | 12 +- internal/media/stream.go | 178 ++++++++---------- internal/rpc/approved_debug_test.go | 2 +- internal/rpc/approved_recorded_real.go | 23 ++- internal/rpc/approved_recorded_real_test.go | 22 +-- internal/sipcall/run_ai.go | 1 - 24 files changed, 590 insertions(+), 173 deletions(-) create mode 100644 docs/evidence/sip-call-capture-asr-20261013.md create mode 100644 internal/ai/protocol_pcma_test.go create mode 100644 internal/asterisk/pcma_media_test.go create mode 100644 internal/media/pcma_stream_test.go diff --git a/AGENTS.md b/AGENTS.md index 6c07e3d..697b54f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -79,6 +79,8 @@ ## 当前范围与权威入口 +- **2026-10-13 原生媒体 PCMA 修复（本地，未部署）**：使用者确认只修改程序与 Asterisk 之间的媒体连接，线路与 Asterisk 全局配置不变。共享原生 ExternalMedia/RTP 固定 `alaw`、静态 payload 8、8 kHz、20 ms；内部仍为单声道 little-endian PCM16/16 kHz，供应商 ASR/TTS 按 task 配置转换。移除不受该 RTP 接口支持的 SLIN16/动态 payload 路径及调用方配置，不新增回退。采用现有 Pion/G.711/重采样能力；错误 codec、半个 PCM 样本和已取消播放显式失败，关键阶段只记格式、采样率、字节数、时长和统计。已知正负 PCMA 样本、开场白在用户开口前发出及用户回话解码、16 种 ASR×TTS 配置的实际本机 RTP 完整回合通过。共享代码影响本地 Agent 与独立 sip-call 构建，但未安装任何新版、未再拨号或请求真实 AI；原 8 kHz 实测失败及 Dispatcher task 6 缺采样率的现场事实不被本地通过覆盖。来源、授权和验证记录见 [`docs/evidence/sip-call-capture-asr-20261013.md`](docs/evidence/sip-call-capture-asr-20261013.md)。部署与真实复测须另行确认。 + - **SaaS Server v0.1（独立测试服务，已部署登记测试机）**：`cmd/saas-server` 使用 Fiber/v3，从指定私有 JSON 文件只读加载五类配置；完整快照更新仅内存生效。HTTP/curl 显式发送单条外呼、start/resume/edit 和 SIP 刷新，不自动投递、重发或重启服务；源 `event_id` 原样保留。共用现行 vhost，租户 ID 固定 `90000+原ID`、任务 ID 固定 `test-原ID`，仅在内存映射，保留原 Dispatcher 编号；只创建测试任务/反馈队列，正式控制/反馈队列不创建、绑定或消费。非测试反馈副本只在测试私有队列确认；测试反馈文件同步保存后 ACK，查询只覆盖本次运行，无历史恢复或发送记录持久化。Dispatcher/Agent、共享业务代码、合同和现有部署不变；使用者已确认本工具合入 main、推送并部署 `server.sip`：二进制 `~/.local/bin/saas-server`，配置 `~/.saas/`，新用户服务 enabled+active，端口8089，共用现行 `/waihu` vhost。配置从当前 SaaS 只读生成，MQ端点从测试机现行 Dispatcher 配置只读取得。本工具不改动原 Agent/Dispatcher/Asterisk，**现行 Dispatcher 未接入新服务，测试任务队列无消费者，不能声称测试机外呼反馈已完成**；更新时观察到原 Dispatcher 因正式任务6缺 `ASR params.sample_rate` 在重启，另行排查，不补默认值或改业务代码掩盖；不授权真实拨号或 AI/OSS。使用见 [`cmd/saas-server/README.md`](cmd/saas-server/README.md)，本地检查见 [`docs/evidence/saas-server-v0.1-local.md`](docs/evidence/saas-server-v0.1-local.md)，部署事实见 [`docs/evidence/saas-server-server-sip-deployment.md`](docs/evidence/saas-server-server-sip-deployment.md)。 - **2026-10-12 接通时长字段已合入并部署登记测试机**：共享契约 `bfa3380` 已推送，本项目业务源码 `8340efd` 已快进合入、推送 `main` 并在 `server.sip` 固定部署。`call.execute.result` 增加 `answered_at` 与 `connected_duration_ms`，原 `duration_ms` 不变；未接通且确认结束为 0，原通道 Up/Destroyed 时间证据缺失为 null，SaaS 不应从 outcome 推断接通时长。重新部署未重置数据、不重启 Asterisk、不抓包/主动试拨；Agent/Dispatcher 均 Debug 开、Capture 关，Asterisk PID 777461 不变，零活动通道，服务均 enabled+active、零重启；实际制品 SHA-256 独立复现。SaaS 探测 `saas_pending`、`business_ready=false`，未观测新字段真实到达、接通通话或生产签收，仍需 SaaS 开发对接与确认。文档后续补记不改变已安装的业务源码版本。详见 [`docs/evidence/connected-call-duration-local.md`](docs/evidence/connected-call-duration-local.md)。 diff --git a/cmd/sip-call/README.md b/cmd/sip-call/README.md index 54ff03b..72b781d 100644 --- a/cmd/sip-call/README.md +++ b/cmd/sip-call/README.md @@ -22,18 +22,18 @@ chmod 600 sip-xx.env ## 标准用户目录 -测试机配置统一放在 `~/.sip-call/`，目录和每项用途见 [`deploys/test/sip-call-home/README.md`](../../deploys/test/sip-call-home/README.md)：`ai/providers.json` 存完整 SaaS `ai_providers` 资源，`task/.json` 存完整 `task_config`，`sip/` 存线路 ENV，`results/` 存每次运行输出。配置目录为 `0700`，含凭据的 JSON/ENV 为 `0600`；不要放入 Git、聊天或普通日志。 +测试机配置统一放在 `~/.sip-call/`，目录和每项用途见 [`deploys/test/sip-call-home/README.md`](../../deploys/test/sip-call-home/README.md)：`ai/providers.json` 存完整 SaaS `ai_providers` 资源，`task/-task-asr-tts.json` 存完整 `task_config`，`sip/` 存线路 ENV；每次运行输出单独存于非隐藏目录 `~/sip-call-results/`。ASR-only 文件名省略 TTS 部分；文件名仅用于识别，正文不变。配置目录为 `0700`，含凭据的 JSON/ENV 为 `0600`；不要放入 Git、聊天或普通日志。 示例调用： ```sh ~/.local/bin/sip-call --ai \ - --task "$HOME/.sip-call/task/.json" \ + --task "$HOME/.sip-call/task/full_ai-task7-asr16000-tts16000.json" \ --provider "$HOME/.sip-call/ai/providers.json" \ "$HOME/.sip-call/sip/.env" <原始号码> ``` -线路 ENV 的 `OUTPUT_DIR` 应指向 `$HOME/.sip-call/results` 的绝对路径。此命令会实际拨号并请求 AI；部署验收只检查文件/权限及 `--help`，不执行该命令。 +线路 ENV 的 `OUTPUT_DIR` 应指向 `$HOME/sip-call-results` 的绝对路径，目录保留 `0700`。此命令会实际拨号并请求 AI；部署验收只检查文件/权限及 `--help`，不执行该命令。 ## 运行条件 @@ -55,7 +55,7 @@ sudo setcap cap_net_raw,cap_net_admin=eip /usr/bin/tcpdump /usr/sbin/getcap /usr/bin/tcpdump ``` -先确认该路径是主机实际安装的 tcpdump；不对 `sip-call`、tshark 或整个目录设置能力。普通用户仍使用前述不带 sudo 的 `-D` 命令。权限不足时不拨号、不回退到无抓包模式，输出“已请求，但未启动”；具体启动错误保留在本次私有 `tcpdump.log`。 +先确认该路径是主机实际安装的 tcpdump；不对 `sip-call`、tshark 或整个目录设置能力。普通用户仍使用前述不带 sudo 的 `-D` 命令。权限不足时不拨号、不回退到无抓包模式，输出“已请求，但未启动”；具体启动错误保留在本次私有 `tcpdump.log`。抓包能力正确仍可能被 AppArmor 拒绝：Debian 的 tcpdump 规则禁止写用户隐藏目录，内核日志会记录 `apparmor="DENIED" profile="tcpdump"`。应使用上述非隐藏结果目录，不停用 AppArmor，不放宽私有目录权限；`any` 不支持混杂模式的警告本身不是退出原因。 ## 自动配置与退出 @@ -67,9 +67,15 @@ sudo setcap cap_net_raw,cap_net_admin=eip /usr/bin/tcpdump `~/sip-call/tool-backups/` 是安装时保存的旧版工具和来源清单，不是 SIP 配置备份，也不是拨号必需目录。本次修改不删除这些旧工具或任何历史证据。 +## AI 媒体连接 + +AI 程序与 Asterisk 的 ExternalMedia/RTP 连接固定使用 PCMA（`alaw`、静态 payload 8、8 kHz、20 ms），不使用该接口无法协商的 SLIN16 动态 payload，也不设置可选回退。程序内部继续使用单声道 little-endian PCM16/16 kHz；task 指定的 ASR/TTS 采样率在各供应商接口边界实际转换，不靠改标签冒充。此规则只改变程序与 Asterisk 的连接，不修改 SIP 线路、Asterisk 全局配置，也不要求重启 Asterisk。 + +Full AI 有开场白时先合成并提交到媒体连接，再采集用户声音。关键阶段日志仅记录格式、采样率、字节数、时长和收发统计；不记录音频或完整对话。错误 payload、损坏数据及已取消播放明确失败，不隐藏成 ASR 无文本。 + ## 输出与证据 -归属与输入检查通过后创建独立私有目录，此后的成功或失败均有 `result.json`（检查未通过时不创建结果）：线路、服务地址、主叫、原始/线路被叫、SIP Call-ID、拨号前绑定时间、接通时间、结束确认时间、AI 模式、本地录音路径、最终状态及失败原因。AI 模式将实际录音保存为 `recording.wav`；ASR 转写和 LLM 回复只输出到当前终端，不写入 `result.json` 或普通日志。终端可能包含通话内容，勿转存到普通日志或聊天。 +归属与输入检查通过后创建独立私有目录，此后的成功或失败均有 `result.json`（检查未通过时不创建结果）：线路、服务地址、主叫、原始/线路被叫、SIP Call-ID、拨号前绑定时间、接通时间、结束确认时间、AI 模式、本地录音路径、最终状态及失败原因。AI 模式将实际录音保存为 `recording.wav`；它合并本地提交的合成声音和收到的声音，听到开场白不证明 Asterisk 已向对端播放，须结合真实出站 RTP 和接听者确认；ASR 转写和 LLM 回复只输出到当前终端，不写入 `result.json` 或普通日志。终端可能包含通话内容，勿转存到普通日志或聊天。 开启调试后另有： diff --git a/cmd/sip-call/sip.env.example b/cmd/sip-call/sip.env.example index 8d1b8ea..cb56398 100644 --- a/cmd/sip-call/sip.env.example +++ b/cmd/sip-call/sip.env.example @@ -17,7 +17,8 @@ SIP_CODEC=alaw ASTERISK_CONFIG_DIR=/home/rogee/.config/go-sip-asterisk ASTERISK_BIN=/usr/local/sbin/asterisk ASTERISK_LIBRARY_DIR=/usr/local/lib -OUTPUT_DIR=/home/rogee/.sip-call/results +# Non-hidden private directory: tcpdump AppArmor denies user dot-directories. +OUTPUT_DIR=/home/rogee/sip-call-results # Ring: 1-120 seconds. Connected hold: 1-30 seconds. No redial. RING_SECONDS=20 HOLD_SECONDS=3 diff --git a/cmd/sip-go-agent/agent_real.go b/cmd/sip-go-agent/agent_real.go index 1e16c47..890fa6d 100644 --- a/cmd/sip-go-agent/agent_real.go +++ b/cmd/sip-go-agent/agent_real.go @@ -88,7 +88,7 @@ func newRealAgentServer(ctx context.Context, settings config.AgentEnvironment, d Journal: &agent.ResultJournal{Root: settings.RecoveryRoot}, } return (&rpc.ApprovedRecordedRealCall{ - Loader: loader, MediaPayloadType: 118, EvidenceRoot: settings.EvidenceRoot, Capture: settings.Capture, + Loader: loader, EvidenceRoot: settings.EvidenceRoot, Capture: settings.Capture, MaxWAVBytes: 64 << 20, ReportTimeout: 15 * time.Minute, Delivery: delivery, Mirror: mirror, }).Prepare(execution) }, diff --git a/deploys/cell/README.md b/deploys/cell/README.md index b87e42b..5c7ac28 100644 --- a/deploys/cell/README.md +++ b/deploys/cell/README.md @@ -56,6 +56,14 @@ before any call. Do not substitute another Asterisk version or a container image The Go Agent package only consumes the resulting approved Cell artifact and reports its applied revision. Local Mock fixtures do not prove real services. +The Go media adapter uses the existing ARI RTP ExternalMedia interface with +`alaw` (static RTP payload 8, 8 kHz, 20 ms). It converts this wire format to/from +canonical mono little-endian PCM16/16 kHz; task-selected ASR/TTS sample rates +remain separate vendor boundaries. This changes no PJSIP trunk or global +Asterisk configuration and requires no Asterisk restart. SLIN16/dynamic RTP +payload selection is removed: RTP ExternalMedia has no SDP negotiation for it. +Application media-profile identity is not an ExternalMedia wire-codec selector. + Before every real outbound attempt, the operator must obtain a fresh user confirmation in the current conversation that names the SIP channel, raw target number and, when `agent --debug/-D` is enabled, capture plan. External tcpdump diff --git a/deploys/test/sip-call-home/README.md b/deploys/test/sip-call-home/README.md index 4bff6cb..ce3a176 100644 --- a/deploys/test/sip-call-home/README.md +++ b/deploys/test/sip-call-home/README.md @@ -1,30 +1,34 @@ # `~/.sip-call` 配置目录 -此目录中的文档会随独立 `sip-call` 工具部署到 `rogee` 用户家目录；仓库不保存真实线路、task 或 provider 配置。 +此目录中的文档随独立 `sip-call` 工具部署到 `rogee` 用户家目录；仓库不保存真实线路、task 或 provider 配置。 ## 目录 ```text ~/.sip-call/ ├── ai/ # providers.json：完整 ai_providers 资源 -├── task/ # .json：完整 task_config 资源 -├── results/ # 本地运行结果和录音 -└── sip/ # 线路 ENV；由旧 ~/.sip-providers 备份后迁入 +├── task/ # -task-asr-tts.json：完整 task_config +└── sip/ # 线路 ENV +~/sip-call-results/ # 本地运行结果和录音；不是隐藏目录 ``` -每个子目录有自己的 `README.md`。`~/.sip-call` 和所有子目录使用 `0700`；SaaS 返回的 task/provider JSON、线路 ENV 使用 `0600`，文件归 `rogee` 所有。不要将真实配置放入 Git、聊天、普通日志或 shell history。迁移前须在远端做私密完整备份及清单，校验新路径后才移除旧路径；备份保留供恢复。 +配置及结果目录均为 `0700`；JSON、ENV 和结果文件为 `0600`，归 `rogee` 所有。不要将真实配置、录音或正文放入 Git、聊天、普通日志或 shell history。修改前须私密备份并校验；既有结果不移动、不删除。 ## SaaS 配置 -只拉取经确认的生产测试 task 和 provider 资源，分别保存为 `task/.json` 与 `ai/providers.json`，保留完整合同结构，不裁剪字段、不编辑业务内容。文件可能含 API 凭据、提示词或业务配置；写入时不打印正文。失败、来源不明或完整性校验不通过时停止，不使用 Mock 或旧快照替代。 +只取得经确认用于测试的完整 task/provider 资源，保留合同结构，不裁剪字段、不编辑业务内容。文件可能含凭据、提示词或业务配置；不打印正文。失败、来源不明或完整性校验不通过时停止，不使用 Mock 或旧快照替代。 + +Task 文件名标明模式、task ID 和配置的采样率，例如 `full_ai-task7-asr16000-tts16000.json`。ASR-only 用 `asr_only-task-asr.json`；无明确 TTS 采样率时用 `tts-auto`，不猜测厂商实际采样率。文件名仅方便识别，正文和执行规则不变。 ## 调用 ```sh ~/.local/bin/sip-call --ai \ - --task "$HOME/.sip-call/task/.json" \ + --task "$HOME/.sip-call/task/full_ai-task7-asr16000-tts16000.json" \ --provider "$HOME/.sip-call/ai/providers.json" \ "$HOME/.sip-call/sip/.env" <原始号码> ``` -每个线路 ENV 的 `OUTPUT_DIR` 必须是 `/home/rogee/.sip-call/results`。此命令会真实拨号并请求 AI；部署只检查制品、帮助、目录及权限，不运行该命令。真实执行仍须另获明确授权，并满足线路及任务规定。 +线路 ENV 使用 `OUTPUT_DIR=/home/rogee/sip-call-results`。Debian 的 tcpdump AppArmor 规则禁止写入用户隐藏目录，因此结果不能放在 `~/.sip-call/results`；不要停用规则、放宽目录权限或用 sudo 执行 sip-call。 + +命令会真实拨号并请求 AI；部署检查不执行该命令。真实执行须另获明确授权，并满足线路及任务规定。每个配置子目录有自己的 `README.md`；本仓库 `results/README.md` 描述外部结果目录。 diff --git a/deploys/test/sip-call-home/results/README.md b/deploys/test/sip-call-home/results/README.md index 204d789..93b0682 100644 --- a/deploys/test/sip-call-home/results/README.md +++ b/deploys/test/sip-call-home/results/README.md @@ -1,9 +1,10 @@ # 本地结果与录音 -每条线路 ENV 的 `OUTPUT_DIR` 指向 `/home/rogee/.sip-call/results`。每次 `sip-call` 运行在此创建独立私有结果目录，保存本次 `result.json`、成功生成的 `recording.wav`，以及显式 `--debug/-D` 时的本通抓包证据。 +每条线路 ENV 的 `OUTPUT_DIR` 指向 `/home/rogee/sip-call-results`（独立于隐藏配置目录）。每次 `sip-call` 运行在此创建独立私有结果目录，保存本次 `result.json`、成功生成的 `recording.wav`，以及显式 `--debug/-D` 时的本通抓包证据。 - 目录及结果子目录权限：`0700` - 结果/录音/抓包文件权限：`0600` - 转写和 LLM 回复只呈现在当前终端，不写入 `result.json` 或普通日志。 - 录音和抓包可能含敏感通话数据；不上传 OSS，不复制到 Git、聊天或普通日志。 +- tcpdump AppArmor 禁止写入用户隐藏目录；不要改回 `~/.sip-call/results`。旧目录的结果原样保留，不迁移、不删除。 - 程序不自动清理结果。只有确认保留要求和备份范围后，才可人工处理旧结果。 diff --git a/deploys/test/sip-call-home/sip/README.md b/deploys/test/sip-call-home/sip/README.md index 77ecf9f..dfe605c 100644 --- a/deploys/test/sip-call-home/sip/README.md +++ b/deploys/test/sip-call-home/sip/README.md @@ -6,7 +6,7 @@ - 每个线路 ENV：`rogee:rogee`、`0600` 普通文件 - 仅放 `KEY=VALUE`；不 `source`，不通过 shell 环境变量覆盖线路值。 - 保持原 SIP 地址、主叫、鉴权和前缀；不得为了测试自行改值。 -- 将每个 ENV 的 `OUTPUT_DIR` 设置为绝对路径 `/home/rogee/.sip-call/results`。 +- 将每个 ENV 的 `OUTPUT_DIR` 设置为绝对路径 `/home/rogee/sip-call-results`（非隐藏目录，避免 tcpdump AppArmor 禁止写入隐藏路径；保留 `0700`）。 - 命令中传入的 ENV 直接决定线路，不做 task trunk 匹配、自动选线、失败换线或重拨。 运行 `sip-call` 会真实呼叫；AI 模式还会请求 provider。安装、迁移或查看帮助都不代表已授权拨号或 AI 请求。 diff --git a/deploys/test/sip-call-home/task/README.md b/deploys/test/sip-call-home/task/README.md index 83a6861..11b859c 100644 --- a/deploys/test/sip-call-home/task/README.md +++ b/deploys/test/sip-call-home/task/README.md @@ -1,11 +1,11 @@ # Task 配置 -每个 `.json` 保存 SaaS 返回的完整 `task_config` 资源，作为 `sip-call --task` 输入。文件名使用资源中的 task ID；正文不删改，工作模式从 task 读取，不另传 `--mode`。 +每个 JSON 保存 SaaS 返回的完整 `task_config` 资源，作为 `sip-call --task` 输入。文件名区分任务模式和采样率： -- 目录权限：`0700` -- task JSON：`rogee:rogee`、`0600` -- 仅保存经确认用于测试的 task；不要批量复制未审阅任务。 -- task 可能包含提示词、业务文本或 provider 引用；不输出正文到日志或聊天。 -- direct `sip-call` 的线路取自本次传入的 SIP ENV，不据 `allowed_trunk_ids` 自动选线。 +- Full AI：`full_ai-task-asr-tts.json` +- ASR-only：`asr_only-task-asr.json` +- TTS 未显式配置采样率：用 `tts-auto`，不猜测厂商实际采样率。 -通过授权的 SaaS 只读接口取得完整资源。Task 文件只作为本地测试输入，不接入 Dispatcher、队列、任务反馈或 OSS。 +例如 `full_ai-task7-asr16000-tts16000.json`。只改文件名，不修改正文、task ID 或配置；模式从正文读取，不另传 `--mode`。不同厂商或格式但上述标签相同时，可追加可读的厂商/格式标签区分。 + +目录为 `0700`，JSON 为 `rogee:rogee`、`0600`。仅保存经确认用于测试的完整资源，不批量复制未审阅任务，不输出提示词、业务正文或凭据。direct sip-call 使用本次传入的 SIP ENV，不按 task 自动选线，不接入 Dispatcher、队列、任务反馈或 OSS。真实拨号及 AI 请求需另获授权。 diff --git a/docs/evidence/sip-call-capture-asr-20261013.md b/docs/evidence/sip-call-capture-asr-20261013.md new file mode 100644 index 0000000..87d5650 --- /dev/null +++ b/docs/evidence/sip-call-capture-asr-20261013.md @@ -0,0 +1,83 @@ +# sip-call 抓包及 ASR 定位（2026-10-13） + +## 当前结果：PCMA 本地修复完成，未部署 + +使用者随后确认采用 PCMA 方案：只修改程序与 Asterisk 之间的媒体连接，不改线路或 Asterisk 全局配置；安装与真实复测另行确认。以下“仅定位”是此前阶段的授权，不阻止这次已确认的本地修复，但不授权部署或拨号。 + +- 本地共享原生媒体路径现固定 `alaw` / RTP payload 8 / 8 kHz / 20 ms；内部仍为单声道 little-endian PCM16/16 kHz。现有 Pion RTP、G.711 库和重采样器完成转换，不新增依赖。 +- 移除旧 SLIN16/动态 payload 路径、可选 wire format 及 NativeDial/录音执行器的 payload 配置；独立 sip-call 与业务 Agent 的本地构建都使用同一修复，不保留回退。 +- 非 PCMA、损坏数据、非完整 PCM16 样本及已取消播放明确失败；不伪装为 ASR 静音。日志记录开场白合成/提交、收到音频解码、输入采集、ASR 实际准备音频和媒体收发统计，仅格式/采样率/长度/时长，无音频或完整对话。 +- 修复前新增回归先失败；修复后正负及近零 PCMA 标准样本、20 ms 帧/8 kHz RTP 时间戳、取消/损坏输入、开场白在收到用户包之前发送及用户回话解码均通过。 +- 8/16/24/48 kHz ASR×TTS 共 16 种完整 WebSocket 回合，额外将开场白、用户声音和 AI 回复分别通过**实际本机 UDP/Pion RTP/PCMA**传输，独立检查可表示样本、长度、时长和双向 packet/byte 统计。供应商端仍为隔离 Mock，不代表真实识别成功。 +- 最终 `make check`（含完整 race/vet/build、43 JSON 合同、Proto、Python 和实际隔离 RabbitMQ/HTTPS/mTLS）、独立工具构建、`make release-check-local` 通过；业务覆盖率 **73.1%**，`git diff --check` 通过。 +- 本地独立 `dist/sip-call` SHA-256 为 `f93bcc5c9d1720865f58126d583d31636c8f0902e82937b44b9f2e9a957b9364`；仅本地制品，未安装，不冒充旧工具的现场版本。 +- 本轮基线 `528093cdf3c7efde35c21103772dd8e329f65f45`，共享合同 `940d9224fd2270791bf2be8d28046f65f7a853f3`；未改共享合同。修改仍未提交，发布检查明确 dirty / production_approval=false。发布包中的主 Agent artifact hash 不能代替独立 sip-call 制品来源。 +- **没有安装新版、再次拨号、真实 AI 请求、SaaS 改写或业务数据重置。** 8 kHz 旧实测失败、未执行 16 kHz 及 task 6 缺少 ASR 采样率导致 Dispatcher 重启的事实仍有效，待新的现场复核；本地修复不能覆盖这些事实。 + +## 原定位阶段的范围及边界 + +使用者要求定位首次未拨号的抓包失败；随后批准调整结果目录、可辨识 task 文件名、本地采样率组合测试及最多两通真实 AI 验证。8 kHz/16 kHz 两份测试配置以只读取得的真实 SaaS task 4 为基线；16 kHz 使用 `fun-asr-realtime`，经使用者单独批准，不修改 SaaS 原任务。 + +8 kHz 测试后使用者要求查清开场白与 ASR 问题，最后明确选择 **仅定位，不安装或拨号**。因此没有安装修复版、没有进行 16 kHz 呼叫，也没有追加真实 AI 请求。下述本地缺陷复现不是已部署修复或真实 AI 签收。 + +## 首次失败：已定位并调整配置 + +- 原运行 `sipai-b4bdb080c22e36303cbec862` 的 tcpdump 在就绪前退出，未拨号、未启动 AI。 +- 私有 `tcpdump.log` 报告无法创建 pcap；内核记录 `apparmor="DENIED" profile="tcpdump"`，目标位于用户隐藏目录 `.sip-call/`。 +- 用户、属主、目录权限及 tcpdump 的 `cap_net_admin,cap_net_raw` 正确。`any` 不支持混杂模式的提示不是退出原因。 +- 已备份三份线路 ENV 和原 task JSON，三份 ENV 只改 `OUTPUT_DIR=/home/rogee/sip-call-results`；目录仍为 `0700`，配置/抓包为 `0600`。原结果目录和字节保留。 +- 原 `task/7.json` 改名 `full_ai-task7-asr16000-tts16000.json`，正文未变。 +- 非隐藏目录的 tcpdump 启动和 TShark 读取自检通过，过滤文档地址 `192.0.2.1`，零包、未产生流量。未停用 AppArmor，未给 sip-call 提权，未放宽权限。 + +## 配置及本地验证 + +- 旧 task 7 使用 fixture 模型/音色；真实 SaaS task 7 返回 404。没有用占位模型请求真实供应商。 +- 只读取得真实任务列表、task 4/6 及 providers，完整资源只保存在主机私有目录。SaaS 任务及线路未编辑。 +- 真实 task 4 为 Full AI，ASR `fun-asr-flash-8k-realtime`、LLM `qwen3.8-flash`、TTS `cosyvoice-v3-flash`，音色 `longanyang`。8 kHz ASR 模型不能仅改参数用于 16 kHz。 +- 新建 `full_ai-task4-asr8000-tts8000.json` 和 `full_ai-task4-asr16000-tts16000.json`。除两角色采样率和已批准的 16 kHz ASR 模型外，其余 task 内容保持来源值。 +- 本地新增 8/16/24/48 kHz 的 ASR×TTS 16 种 WebSocket 完整回合测试，验证声明采样率、实际 PCM 样本及 100 ms 时长，不用生产重采样器生成期望值。 +- 首轮 `make check`、独立工具构建、`make release-check-local` 及业务覆盖率 73.0% 通过；同步到本地 `cb0c944` 后最终 `make check` 和 16 组合测试再次通过。这些不代表真实媒体正确。 + +## 8 kHz 真实测试 + +执行前确认零活动通道、零未完成/未知 inbox、零未确认 outbox、空 Agent 恢复；停止 Dispatcher、Agent，保留原数据。只执行一通获批百应测试，未重拨、换线、接 MQ/OSS 或重启 Asterisk。 + +结果目录：`/home/rogee/sip-call-results/sipai-dcf515aefb69894b8b93d2a0/`。 + +- SIP 200、实际接通、结束已确认。 +- 收到 712 个 PCMA RTP 包，共 156640 RTP 字节；同一服务端地址用于 SDP 媒体。抓包中向服务端发送的 RTP 为 0。 +- Full AI 流程启动，但 ASR 报告 `Bailian task ASR finished without final user text`，未完成 AI 回答。 +- `recording.wav` 为单声道 16 kHz，约 12.1 秒。**该录音合并本地发送的合成声音和收到的声音，不证明开场白已被 Asterisk 接纳或被对方听见。** +- 恢复服务时 Agent、Dispatcher 均短暂显示 active；**最终复核 Agent active，而 Dispatcher 已进入 activating/auto-restart**。最新结构化错误明确为真实 task 6 缺少必填 `agent.asr.params.sample_rate`；此事没有修复、没有改写 SaaS 配置，不能声称 Dispatcher 持续正常。Asterisk PID 777461 未变、零活动通道，PJSIP 原配置字节、属主及权限保持不变，临时配置已移除。 +- 已安装 sip-call SHA-256：`76ecafa3c747a2aa116dd714824328fa01cf8fed10a51bf421f1c1dcbbc67c13`。可读取 Go 1.27.1 构建信息，但未取得 VCS 来源，不把当前本地源码自动视为该制品的完整来源证明。 + +## 新媒体问题的定位结论 + +### 1. 开场白配置及程序顺序正常，但当前 RTP 接法不满足 Asterisk 接收约束 + +真实 task 4 的开场白非空（36 字）。`internal/callflow/flow.go` 的 Full AI 分支先合成并发送开场白，然后才采集用户声音，不存在故意等待用户先说话的规则。 + +`internal/asterisk/call.go` 使用 ExternalMedia/RTP、`format=slin16`，调用方使用动态 payload 118。Asterisk 维护者说明：ExternalMedia RTP 没有 SDP 映射协商，不能因为 Asterisk 发出 118 就认为它也会接纳 118；不支持的动态输入会被丢弃。公开说明与现场“接到声音、没有向被叫发送 RTP”的现象一致，但本次抓包不包含本机 ExternalMedia 回环，尚无这一跳的逐包现场证据。 + +来源： +- https://issues.asterisk.org/jira/browse/ASTERISK-28751 +- https://community.asterisk.org/t/ari-external-media-code-issue/110111 + +项目已经有 A-law 编解码及明确的媒体格式接口；不需要自己重写协议或更换 Asterisk。这里仅记录根因线索和可用能力，没有切换媒体接法或部署。 + +### 2. 本地 SLIN16 接收/发送缺少字节序转换，已独立复现 + +修复前的 `internal/media/stream.go` 在 SLIN16 路径直接把网络字节交作内部 PCM16，并直接发送内部 PCM16。RTP L16 使用网络字节序，项目内部 PCM16 使用 little endian。 + +仅用本机 UDP 和既有 Pion RTP 库的私有诊断程序复现，无 Asterisk、SIP、AI 或外部服务调用： + +- 已知 L16 样本 4660（网络 `12 34`），当前接收后按内部 PCM16 解读为 13330。 +- 已知内部 PCM16 样本 4660（`34 12`），当前发送后按 L16 解读为 13330。 + +该复现证明修复前源码的音频边界有真实错误；它会影响声音、语音检测和 ASR 输入。**尚未证明它是本次 ASR 无文本的唯一原因**：录音包含本地合成声音，且真实回环包/实际上传音频没有保存；独立 PCMA 解码与录音的波形比较没有得到足以证明输入一致的结果。不能用音量、字节交换后能量或本地 mock 通过冒充真实 ASR 根因签收。 + +## 未完成及后续所需证据 + +- 未安装媒体修复、未做 16 kHz 真实测试、未验证真实 ASR–LLM–TTS 全链路或生产。 +- 本地已知 PCM 样本和完整回合验证已完成，关键阶段统计已补充。如后续获批部署及真实复测，仍需验证 Asterisk 实际接纳并向线路发送、真实用户输入及 ASR 最终文本，不能仅用本地提交日志或混合录音代签。 +- 真实复测仍需符合使用者最新授权；不能因此前批准两通而越过此次“仅定位”的明确限制。 diff --git a/internal/ai/asr_sample_rate.go b/internal/ai/asr_sample_rate.go index a928723..84c2d48 100644 --- a/internal/ai/asr_sample_rate.go +++ b/internal/ai/asr_sample_rate.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "math" "git.ipao.vip/rogee/go-sip/internal/audio" @@ -48,5 +49,6 @@ func prepareASRPCM16(pcm16 []byte, targetRate int) ([]byte, error) { if len(resampled) == 0 || len(resampled)%2 != 0 { return nil, errors.New("ASR resampling produced no complete PCM16 samples") } + slog.Info("ASR audio prepared", "input_sample_rate", internalPCM16SampleRate, "input_bytes", len(pcm16), "asr_sample_rate", targetRate, "asr_bytes", len(resampled), "audio_duration_ms", len(resampled)*1000/(targetRate*2)) return resampled, nil } diff --git a/internal/ai/protocol_pcma_test.go b/internal/ai/protocol_pcma_test.go new file mode 100644 index 0000000..218e5ff --- /dev/null +++ b/internal/ai/protocol_pcma_test.go @@ -0,0 +1,51 @@ +package ai + +import ( + "context" + "testing" + + "git.ipao.vip/rogee/go-sip/internal/media" +) + +// Use actual local UDP/Pion RTP and G.711, not a media mock, for the opening, +// user input and AI reply in every ASR/TTS rate combination. +func protocolPCMAViaRTP(t *testing.T, ctx context.Context, pcm []byte) []byte { + t.Helper() + receiver, err := media.ListenRTP("127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer func() { + if err := receiver.Close(); err != nil { + t.Error(err) + } + }() + sender, err := media.ListenRTP("127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer func() { + if err := sender.Close(); err != nil { + t.Error(err) + } + }() + if err := sender.SetPeer(receiver.LocalAddr().String()); err != nil { + t.Fatal(err) + } + if err := sender.SendPCM16(ctx, pcm, 16000); err != nil { + t.Fatal(err) + } + var received []byte + for len(received) < len(pcm) { + frame, err := receiver.ReadPayload(ctx) + if err != nil { + t.Fatal(err) + } + received = append(received, frame...) + } + sentStats, receivedStats := sender.Stats(), receiver.Stats() + if len(received) != len(pcm) || sentStats.SentPackets != receivedStats.ReceivedPackets || sentStats.SentBytes != receivedStats.ReceivedBytes || sentStats.SentBytes != uint64(len(pcm)/4) { + t.Fatalf("PCMA duration/packet/byte mismatch: canonical=%d received=%d sent=%+v receivedStats=%+v", len(pcm), len(received), sentStats, receivedStats) + } + return received +} diff --git a/internal/ai/protocol_turn_test.go b/internal/ai/protocol_turn_test.go index 7fcfe1d..74c238d 100644 --- a/internal/ai/protocol_turn_test.go +++ b/internal/ai/protocol_turn_test.go @@ -16,17 +16,39 @@ import ( ) func TestProtocolModelsCompleteTurnWithoutPerModelCode(t *testing.T) { + testProtocolResampledTurn(t, 16000, 16000) +} + +func TestTaskASRTTSSampleRateCombinations(t *testing.T) { + for _, asrRate := range []int{8000, 16000, 24000, 48000} { + for _, ttsRate := range []int{8000, 16000, 24000, 48000} { + t.Run(fmt.Sprintf("full_ai_asr%d_tts%d", asrRate, ttsRate), func(t *testing.T) { + testProtocolResampledTurn(t, asrRate, ttsRate) + }) + } + } +} + +func testProtocolResampledTurn(t *testing.T, asrRate, ttsRate int) { + t.Helper() task, providers := protocolTask(t, TTSProtocolDashScopeTask) task = changeCurrentAgent(t, task, func(a map[string]any) { a["conversation"].(map[string]any)["opening"] = "Hello." a["conversation"].(map[string]any)["hangup_keywords"] = []any{} + a["asr"].(map[string]any)["params"].(map[string]any)["sample_rate"] = asrRate + a["tts"].(map[string]any)["params"].(map[string]any)["sample_rate"] = ttsRate }) bound, err := Bind(task, providers) if err != nil { t.Fatal(err) } var asrCalls, ttsCalls, llmCalls atomic.Int32 - expected := []byte{1, 0, 2, 0} + // A 100-ms constant PCM16 signal independently verifies sample values and + // duration, rather than calculating expectations with the production resampler. + // 1008 is exactly representable in PCMA, so both rates and values can be + // checked independently through the real RTP codec boundary. + expected := bytes.Repeat([]byte{0xf0, 0x03}, 1600) + ttsAudio := bytes.Repeat([]byte{0xf0, 0x03}, ttsRate/10) speech := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { c, err := websocket.Accept(w, r, nil) if err != nil { @@ -44,8 +66,12 @@ func TestProtocolModelsCompleteTurnWithoutPerModelCode(t *testing.T) { if cmd.Payload["model"] != "unlisted-asr-model-2029" { t.Error("ASR model replaced") } + if got := cmd.Payload["parameters"].(map[string]any)["sample_rate"]; got != float64(asrRate) { + t.Errorf("ASR declared rate = %v, want %d", got, asrRate) + } + var received []byte for { - kind, _, err := c.Read(ctx) + kind, chunk, err := c.Read(ctx) if err != nil { t.Error(err) return @@ -53,6 +79,10 @@ func TestProtocolModelsCompleteTurnWithoutPerModelCode(t *testing.T) { if kind == websocket.MessageText { break } + received = append(received, chunk...) + } + if want := bytes.Repeat([]byte{0xf0, 0x03}, asrRate/10); !bytes.Equal(received, want) { + t.Errorf("ASR actual PCM: %d bytes, want %d with unchanged sample values", len(received), len(want)) } mockEvent(c, ctx, cmd.Header.TaskID, "result-generated", map[string]any{"output": map[string]any{"sentence": map[string]any{"text": "Tell me more.", "sentence_end": true, "begin_time": 0, "end_time": 100}}}) case "tts": @@ -60,12 +90,15 @@ func TestProtocolModelsCompleteTurnWithoutPerModelCode(t *testing.T) { if cmd.Payload["model"] != "unlisted-tts-model-2029" || cmd.Payload["parameters"].(map[string]any)["voice"] != "unlisted-voice" { t.Error("TTS model/voice replaced") } + if got := cmd.Payload["parameters"].(map[string]any)["sample_rate"]; got != float64(ttsRate) { + t.Errorf("TTS declared rate = %v, want %d", got, ttsRate) + } mockRead(t, c, ctx) finish := mockRead(t, c, ctx) if finish.Header.Action != "finish-task" { t.Error("TTS lifecycle changed") } - c.Write(ctx, websocket.MessageBinary, expected) + c.Write(ctx, websocket.MessageBinary, ttsAudio) default: t.Error("unknown role") return @@ -96,10 +129,17 @@ func TestProtocolModelsCompleteTurnWithoutPerModelCode(t *testing.T) { if err != nil || !bytes.Equal(opening, expected) { t.Fatalf("opening: %v", err) } - result, err := call.RunTurn(ctx, make([]byte, 3200)) + if played := protocolPCMAViaRTP(t, ctx, opening); !bytes.Equal(played, expected) { + t.Fatal("opening changed across Asterisk PCMA media boundary") + } + userPCM := protocolPCMAViaRTP(t, ctx, expected) + result, err := call.RunTurn(ctx, userPCM) if err != nil || result.Transcript != "Tell me more." || result.Reply != "Generic response." || !bytes.Equal(result.AudioPCM16, expected) || result.EndedByKeyword { t.Fatalf("generic model turn: %+v %v", result, err) } + if played := protocolPCMAViaRTP(t, ctx, result.AudioPCM16); !bytes.Equal(played, expected) { + t.Fatal("AI reply changed across Asterisk PCMA media boundary") + } if asrCalls.Load() != 1 || llmCalls.Load() != 1 || ttsCalls.Load() != 2 { t.Fatalf("unexpected/retried requests: ASR=%d LLM=%d TTS=%d", asrCalls.Load(), llmCalls.Load(), ttsCalls.Load()) } diff --git a/internal/asterisk/call.go b/internal/asterisk/call.go index 3b9fc25..0e61832 100644 --- a/internal/asterisk/call.go +++ b/internal/asterisk/call.go @@ -18,13 +18,12 @@ import ( "github.com/CyCoreSystems/ari/v5/client/native" ) -// NativeDial contains explicit SIP identity and the business Agent's SLIN16 -// profile. Business calls use Dispatcher-selected values; isolated line tests -// use their declared ENV identity and omit media resources. +// NativeDial contains explicit SIP identity. Business calls use Dispatcher- +// selected values; isolated line tests use their declared ENV identity. Media +// calls use fixed PCMA RTP, independent of the application's PCM16 profile. type NativeDial struct { ExecutionID, TrunkID, DialedCallee, CallerID string AnswerTimeout time.Duration - MediaPayloadType uint8 BindSIPCall func(executionID, callID string) error // before Dial; no number/time matching } @@ -154,7 +153,7 @@ func dialWithOptions(ctx context.Context, client ari.Client, request NativeDial, if lineTest { originate.App = "sip-call" } else { - call.Media, err = media.ListenRTPWithFormat("127.0.0.1:0", request.MediaPayloadType, media.FormatSLIN16, 16000) + call.Media, err = media.ListenRTP("127.0.0.1:0") if err != nil { return nil, &NativeCallFailure{Phase: "rtp_listen", Cause: err} } @@ -214,7 +213,7 @@ func dialWithOptions(ctx context.Context, client ari.Client, request NativeDial, mediaKey := ari.NewKey(ari.ChannelKey, request.ExecutionID+"-media") call.external, err = client.Channel().ExternalMedia(mediaKey, ari.ExternalMediaOptions{ App: "go-sip-agent", ExternalHost: net.JoinHostPort("127.0.0.1", strconv.Itoa(call.Media.LocalAddr().(*net.UDPAddr).Port)), - Format: "slin16", Encapsulation: "rtp", Transport: "udp", ConnectionType: "client", Direction: "both", + Format: media.WireFormat, Encapsulation: "rtp", Transport: "udp", ConnectionType: "client", Direction: "both", }) if err != nil { // Use the same identity to stop an ExternalMedia channel created diff --git a/internal/asterisk/call_test.go b/internal/asterisk/call_test.go index 2233025..268acdc 100644 --- a/internal/asterisk/call_test.go +++ b/internal/asterisk/call_test.go @@ -48,6 +48,7 @@ type testChannels struct { dataChecks int hungup []string mediaPeerIP string + mediaPeerPort string callerID string noAnswer bool endAfterHangup bool @@ -104,6 +105,9 @@ func (c *testChannels) GetVariable(_ *ari.Key, name string) (string, error) { case "UNICASTRTP_LOCAL_ADDRESS": return c.mediaPeerIP, nil case "UNICASTRTP_LOCAL_PORT": + if c.mediaPeerPort != "" { + return c.mediaPeerPort, nil + } return "17000", nil } return "", errors.New("unexpected ARI variable") @@ -187,7 +191,7 @@ func TestNativeCallBindsSIPCallIDBeforeOneDialAndActualRTPBridge(t *testing.T) { boundBeforeDial := false call, err := dialWithClient(ctx, client, NativeDial{ ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", - AnswerTimeout: time.Second, MediaPayloadType: 118, + AnswerTimeout: time.Second, BindSIPCall: func(executionID, callID string) error { boundBeforeDial = executionID == "exec-1" && callID == "isolated-sip-call-id" && client.channels.issued == 0 return nil @@ -199,7 +203,7 @@ func TestNativeCallBindsSIPCallIDBeforeOneDialAndActualRTPBridge(t *testing.T) { if call.Media == nil || !boundBeforeDial || client.channels.issued != 1 || client.channels.create.Endpoint != "PJSIP/708915000000001@shuqi" || client.channels.callerID != "BD1234" || client.channels.create.ChannelID != "exec-1" || client.channels.createRef != nil || client.channels.create.Originator != "" || - client.channels.media.Format != "slin16" || client.channels.media.App != "go-sip-agent" || + client.channels.media.Format != "alaw" || client.channels.media.App != "go-sip-agent" || len(client.bridges.attached) != 2 || client.bridges.attached[0] != "exec-1" { t.Fatal("native channel/ExternalMedia was not attached exactly once to the approved bridge") } @@ -236,7 +240,7 @@ func TestNativeCallStopsMediaOnMatchingCarrierHangup(t *testing.T) { client := &testARIClient{channels: &testChannels{events: events, mediaPeerIP: "127.0.0.1"}, bridges: &testBridges{}, bus: &testBus{sub: answerEvents{events: events}}} ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() - call, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, MediaPayloadType: 118, BindSIPCall: testBindSIPCall}) + call, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, BindSIPCall: testBindSIPCall}) if err != nil { t.Fatal(err) } @@ -269,7 +273,7 @@ func TestNativeCallUnknownOriginationMustAttemptHangupWithoutRetry(t *testing.T) } ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() - if _, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, MediaPayloadType: 118, BindSIPCall: testBindSIPCall}); err == nil || !strings.Contains(err.Error(), "outcome unknown") || NativeCallPhase(err) != "dial" { + if _, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, BindSIPCall: testBindSIPCall}); err == nil || !strings.Contains(err.Error(), "outcome unknown") || NativeCallPhase(err) != "dial" { t.Fatalf("unknown originate outcome must never be reported as a completed call: %v", err) } if client.channels.issued != 1 || len(client.channels.hungup) != 1 || client.channels.hungup[0] != "exec-1" || client.closed != 1 { @@ -291,7 +295,7 @@ func TestNativeCallUnknownBridgeOrMediaCreationCleansKnownIdentities(t *testing. client := &testARIClient{channels: channels, bridges: bridges, bus: &testBus{sub: answerEvents{events: events}}} ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() - if _, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, MediaPayloadType: 118, BindSIPCall: testBindSIPCall}); err == nil || NativeCallPhase(err) != map[string]string{"bridge": "bridge_create", "external-media": "external_media_create"}[step] { + if _, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, BindSIPCall: testBindSIPCall}); err == nil || NativeCallPhase(err) != map[string]string{"bridge": "bridge_create", "external-media": "external_media_create"}[step] { t.Fatalf("unknown bridge/media outcome needs its exact failure phase: %v", err) } wantHungup := 1 @@ -332,7 +336,7 @@ func TestNativeCallPreAnswerReleaseRequiresMissingOriginatedChannel(t *testing.T client := &testARIClient{channels: channels, bridges: &testBridges{}, bus: &testBus{sub: answerEvents{events: make(chan ari.Event)}}} ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) defer cancel() - _, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, MediaPayloadType: 118, BindSIPCall: testBindSIPCall}) + _, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, BindSIPCall: testBindSIPCall}) if err == nil || NativeCallPhase(err) != "answer_wait" || NativeCallConfirmedPreAnswerEnd(err) != tc.confirmed || channels.dataChecks < 1 || (tc.dataErr != nil && channels.dataChecks != 1) || channels.issued != 1 || client.closed != 1 { t.Fatalf("only the confirmed missing channel can release capacity: err=%v confirmed=%v checks=%d issued=%d closed=%d", err, NativeCallConfirmedPreAnswerEnd(err), channels.dataChecks, channels.issued, client.closed) } @@ -348,7 +352,7 @@ func TestNativeCallAnswerTimeoutNeverCreatesFakeMedia(t *testing.T) { } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) defer cancel() - if _, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, MediaPayloadType: 118, BindSIPCall: testBindSIPCall}); err == nil || !strings.Contains(err.Error(), "deadline") { + if _, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "exec-1", TrunkID: "shuqi", DialedCallee: "708915000000001", CallerID: "BD1234", AnswerTimeout: time.Second, BindSIPCall: testBindSIPCall}); err == nil || !strings.Contains(err.Error(), "deadline") { t.Fatalf("unanswered channel must fail rather than synthesize media: %v", err) } if client.channels.issued != 1 || len(client.channels.hungup) == 0 || client.closed != 1 { diff --git a/internal/asterisk/pcma_media_test.go b/internal/asterisk/pcma_media_test.go new file mode 100644 index 0000000..417beae --- /dev/null +++ b/internal/asterisk/pcma_media_test.go @@ -0,0 +1,92 @@ +package asterisk + +import ( + "bytes" + "context" + "encoding/binary" + "net" + "strconv" + "testing" + "time" + + "github.com/CyCoreSystems/ari/v5" + "github.com/pion/rtp/v2" +) + +func TestNativeMediaUsesPCMABeforeUserSpeaksAndDecodesUserReply(t *testing.T) { + peer, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + defer peer.Close() + events := make(chan ari.Event, 2) + client := &testARIClient{ + channels: &testChannels{events: events, mediaPeerIP: "127.0.0.1", mediaPeerPort: strconv.Itoa(peer.LocalAddr().(*net.UDPAddr).Port), endAfterHangup: true}, + bridges: &testBridges{}, + bus: &testBus{sub: answerEvents{events: events}}, + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + call, err := dialWithClient(ctx, client, NativeDial{ExecutionID: "pcma-native", TrunkID: "test", DialedCallee: "123", CallerID: "caller", AnswerTimeout: time.Second, BindSIPCall: testBindSIPCall}) + if err != nil { + t.Fatal(err) + } + defer func() { + if err := call.Close(); err != nil { + t.Error(err) + } + }() + if client.channels.media.Format != "alaw" { + t.Fatalf("Asterisk media format=%q", client.channels.media.Format) + } + // The opening must reach Asterisk's configured peer without any inbound RTP. + opening := bytes.Repeat([]byte{0xe8, 0x03}, 320) + if err := call.Media.SendPCM16(ctx, opening, 16000); err != nil { + t.Fatal(err) + } + if err := peer.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + buf := make([]byte, 2048) + n, _, err := peer.ReadFromUDP(buf) + if err != nil { + t.Fatal(err) + } + var received rtp.Packet + if err := received.Unmarshal(buf[:n]); err != nil { + t.Fatal(err) + } + if received.PayloadType != 8 || !bytes.Equal(received.Payload, bytes.Repeat([]byte{0xfa}, 160)) { + t.Fatalf("opening was not PCMA: pt=%d bytes=%d", received.PayloadType, len(received.Payload)) + } + // A known negative PCMA user sample must become signed little-endian PCM16, + // not network-endian bytes or a falsely relabeled 8-kHz payload. + user := rtp.Packet{Header: rtp.Header{Version: 2, PayloadType: 8, SequenceNumber: 1, SSRC: 99}, Payload: bytes.Repeat([]byte{0x7a}, 160)} + raw, err := user.Marshal() + if err != nil { + t.Fatal(err) + } + target, err := net.ResolveUDPAddr("udp", client.channels.media.ExternalHost) + if err != nil { + t.Fatal(err) + } + if _, err := peer.WriteToUDP(raw, target); err != nil { + t.Fatal(err) + } + reply, err := call.Media.ReadPayload(ctx) + if err != nil { + t.Fatal(err) + } + if len(reply) != 640 { + t.Fatalf("decoded user frame=%d bytes, want 20 ms at 16 kHz", len(reply)) + } + for i := 0; i < len(reply); i += 2 { + if got := int16(binary.LittleEndian.Uint16(reply[i:])); got != -1008 { + t.Fatalf("user sample %d=%d want -1008", i/2, got) + } + } + stats := call.Media.Stats() + if stats.SentPackets != 1 || stats.ReceivedPackets != 1 || stats.SentBytes != 160 || stats.ReceivedBytes != 160 { + t.Fatalf("both media directions must be evidenced: %+v", stats) + } +} diff --git a/internal/callflow/flow.go b/internal/callflow/flow.go index 00d5ad1..a84ffd7 100644 --- a/internal/callflow/flow.go +++ b/internal/callflow/flow.go @@ -52,11 +52,13 @@ func executeFlow(ctx context.Context, session MediaSession, mode ai.Mode, open f if err != nil { return Result{}, err } + slog.Info("AI opening synthesized", "pcm_sample_rate", media.PCM16SampleRate, "audio_bytes", len(openingPCM)) if len(openingPCM) > 0 { if err := session.SendPCM16(ctx, openingPCM, 16000); err != nil { return result, err } result.OutboundTurns = append(result.OutboundTurns, clonePCM(openingPCM)) + slog.Info("AI opening submitted to media", "audio_bytes", len(openingPCM), "pcm_sample_rate", media.PCM16SampleRate) } } for turnIndex := 0; turnIndex < maxTurns; turnIndex++ { @@ -68,6 +70,7 @@ func executeFlow(ctx context.Context, session MediaSession, mode ai.Mode, open f result.Inbound = inbound result.InboundTurns = append(result.InboundTurns, clonePCM(inbound)) result.CaptureWindows = append(result.CaptureWindows, captured.Window) + slog.Info("user audio captured", "turn", turnIndex+1, "pcm_sample_rate", media.PCM16SampleRate, "audio_bytes", len(inbound), "audio_duration_ms", len(inbound)*1000/(media.PCM16SampleRate*2)) if len(inbound) < 3200 { return result, fmt.Errorf("captured audio turn %d is too short", turnIndex+1) } @@ -118,6 +121,7 @@ func executeFlow(ctx context.Context, session MediaSession, mode ai.Mode, open f return result, fmt.Errorf("send AI reply turn %d: %w", turnIndex+1, err) } result.OutboundTurns = append(result.OutboundTurns, clonePCM(turn.AudioPCM16)) + slog.Info("AI reply submitted to media", "turn", turnIndex+1, "pcm_sample_rate", media.PCM16SampleRate, "audio_bytes", len(turn.AudioPCM16)) result.RTP = session.Stats() } result.RTP = session.Stats() diff --git a/internal/media/pcma_stream_test.go b/internal/media/pcma_stream_test.go new file mode 100644 index 0000000..935a9a1 --- /dev/null +++ b/internal/media/pcma_stream_test.go @@ -0,0 +1,146 @@ +package media + +import ( + "bytes" + "context" + "encoding/binary" + "net" + "strings" + "testing" + "time" + + "github.com/pion/rtp/v2" +) + +func pcmaTestStream(t *testing.T) (*RTPStream, *net.UDPConn) { + t.Helper() + stream, err := ListenRTP("127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = stream.Close() }) + peer, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = peer.Close() }) + if err := stream.SetPeer(peer.LocalAddr().String()); err != nil { + t.Fatal(err) + } + return stream, peer +} + +func TestDefaultRTPStreamSendsPCMAGoldenSamplesAnd8KClock(t *testing.T) { + stream, peer := pcmaTestStream(t) + // Positive PCM16 1000 is encoded as PCMA 0xfa. A-law is lossy: decode is 1008. + pcm := bytes.Repeat([]byte{0xe8, 0x03}, 640) // 40 ms of canonical 16 kHz PCM. + if err := stream.SendPCM16(context.Background(), pcm, 16000); err != nil { + t.Fatal(err) + } + if err := peer.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + var firstTimestamp uint32 + for i := 0; i < 2; i++ { + buf := make([]byte, 2048) + n, _, err := peer.ReadFromUDP(buf) + if err != nil { + t.Fatal(err) + } + var packet rtp.Packet + if err := packet.Unmarshal(buf[:n]); err != nil { + t.Fatal(err) + } + if packet.PayloadType != 8 || !bytes.Equal(packet.Payload, bytes.Repeat([]byte{0xfa}, 160)) { + t.Fatalf("frame %d: payload type=%d, bytes=%d; want PCMA/8 with 160 golden samples", i, packet.PayloadType, len(packet.Payload)) + } + if i == 0 { + firstTimestamp = packet.Timestamp + } + if packet.Timestamp != firstTimestamp+uint32(i*160) { + t.Fatalf("RTP timestamp=%d: wire clock must be 8 kHz", packet.Timestamp) + } + } + stats := stream.Stats() + if stats.SentPackets != 2 || stats.SentBytes != 320 { + t.Fatalf("wrong wire stats: %+v", stats) + } +} + +func TestDefaultRTPStreamDecodesPCMABothSignsWithoutByteSwap(t *testing.T) { + for _, tc := range []struct { + name string + wire byte + sample int16 + }{ + {"positive", 0xfa, 1008}, {"negative", 0x7a, -1008}, {"near_zero", 0xd5, 8}, + } { + t.Run(tc.name, func(t *testing.T) { + stream, peer := pcmaTestStream(t) + packet := rtp.Packet{Header: rtp.Header{Version: 2, PayloadType: 8, SequenceNumber: 1, Timestamp: 0, SSRC: 9}, Payload: bytes.Repeat([]byte{tc.wire}, 160)} + raw, err := packet.Marshal() + if err != nil { + t.Fatal(err) + } + if _, err := peer.WriteToUDP(raw, stream.LocalAddr().(*net.UDPAddr)); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + pcm, err := stream.ReadPayload(ctx) + if err != nil { + t.Fatal(err) + } + if len(pcm) != 640 { + t.Fatalf("canonical bytes=%d, want 640 (20 ms at 16 kHz)", len(pcm)) + } + for i := 0; i < len(pcm); i += 2 { + if sample := int16(binary.LittleEndian.Uint16(pcm[i:])); sample != tc.sample { + t.Fatalf("sample %d=%d want %d", i/2, sample, tc.sample) + } + } + stats := stream.Stats() + if stats.ReceivedPackets != 1 || stats.ReceivedBytes != 160 { + t.Fatalf("wrong wire stats: %+v", stats) + } + }) + } +} + +func TestRTPStreamRejectsWrongCodecInsteadOfASRTimeout(t *testing.T) { + stream, peer := pcmaTestStream(t) + packet := rtp.Packet{Header: rtp.Header{Version: 2, PayloadType: 118, SSRC: 9}, Payload: []byte{0x12, 0x34}} + raw, err := packet.Marshal() + if err != nil { + t.Fatal(err) + } + if _, err := peer.WriteToUDP(raw, stream.LocalAddr().(*net.UDPAddr)); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if _, err := stream.ReadPayload(ctx); err == nil || !strings.Contains(err.Error(), "payload type") { + t.Fatalf("wrong codec must be explicit, got %v", err) + } +} + +func TestRTPStreamRejectsPartialPCMAndCancelledPlaybackBeforeSending(t *testing.T) { + for _, name := range []string{"odd_sample", "cancelled"} { + t.Run(name, func(t *testing.T) { + stream, _ := pcmaTestStream(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + pcm := []byte{0xe8, 0x03, 0xe8} + if name == "cancelled" { + cancel() + pcm = bytes.Repeat([]byte{0xe8, 0x03}, 320) + } + if err := stream.SendPCM16(ctx, pcm, 16000); err == nil { + t.Fatal("invalid/cancelled audio was accepted") + } + if stats := stream.Stats(); stats.SentPackets != 0 { + t.Fatalf("audio was sent before validation: %+v", stats) + } + }) + } +} diff --git a/internal/media/rtp_test.go b/internal/media/rtp_test.go index 3a6b5c9..8d34c11 100644 --- a/internal/media/rtp_test.go +++ b/internal/media/rtp_test.go @@ -52,7 +52,7 @@ func TestRTPStreamSendsAfterExplicitPeerBeforeInbound(t *testing.T) { t.Fatal(err) } defer receiver.Close() - stream, err := ListenRTP("127.0.0.1:0", 118) + stream, err := ListenRTP("127.0.0.1:0") if err != nil { t.Fatal(err) } @@ -60,7 +60,7 @@ func TestRTPStreamSendsAfterExplicitPeerBeforeInbound(t *testing.T) { if err := stream.SetPeer(receiver.LocalAddr().String()); err != nil { t.Fatal(err) } - pcm := make([]byte, pcm16FrameSamples*2) + pcm := make([]byte, PCM16SampleRate/50*2) if err := stream.SendPCM16(context.Background(), pcm, 16000); err != nil { t.Fatal(err) } @@ -76,18 +76,18 @@ func TestRTPStreamSendsAfterExplicitPeerBeforeInbound(t *testing.T) { if err := packet.Unmarshal(buf[:n]); err != nil { t.Fatal(err) } - if packet.PayloadType != 118 || len(packet.Payload) != len(pcm) { + if packet.PayloadType != WirePayloadType || len(packet.Payload) != WireSampleRate/50 { t.Fatalf("unexpected packet: payload_type=%d payload_bytes=%d", packet.PayloadType, len(packet.Payload)) } } func TestALAWProfileNormalizesWireAudioToCanonicalPCM16(t *testing.T) { - receiver, err := ListenRTPWithFormat("127.0.0.1:0", 8, FormatALAW, 8000) + receiver, err := ListenRTP("127.0.0.1:0") if err != nil { t.Fatal(err) } defer receiver.Close() - sender, err := ListenRTPWithFormat("127.0.0.1:0", 8, FormatALAW, 8000) + sender, err := ListenRTP("127.0.0.1:0") if err != nil { t.Fatal(err) } @@ -95,7 +95,7 @@ func TestALAWProfileNormalizesWireAudioToCanonicalPCM16(t *testing.T) { if err := sender.SetPeer(receiver.LocalAddr().String()); err != nil { t.Fatal(err) } - pcm := make([]byte, pcm16FrameSamples*2) + pcm := make([]byte, PCM16SampleRate/50*2) if err := sender.SendPCM16(context.Background(), pcm, 16000); err != nil { t.Fatal(err) } diff --git a/internal/media/stream.go b/internal/media/stream.go index 990f3ec..a086d98 100644 --- a/internal/media/stream.go +++ b/internal/media/stream.go @@ -6,7 +6,9 @@ import ( "encoding/binary" "errors" "fmt" + "log/slog" "net" + "sync" "time" "git.ipao.vip/rogee/go-sip/internal/audio" @@ -16,31 +18,24 @@ import ( const ( maxRTPPacketBytes = 2048 - pcm16FrameSamples = 320 // 20 ms at 16 kHz, the canonical AI frame size + WireFormat = "alaw" + WireSampleRate = 8000 + WirePayloadType = 8 + PCM16SampleRate = 16000 ) -// Format is the wire codec used by an Asterisk ExternalMedia channel. The -// CallFlow always consumes and produces canonical mono PCM16/16 kHz. -type Format string - -const ( - FormatSLIN16 Format = "slin16" - FormatALAW Format = "alaw" -) - -// RTPStream is a bounded UDP/RTP adapter for an Asterisk ExternalMedia channel. -// Pion owns RTP framing; this package only binds the project media contract and -// exposes signed-linear PCM16 payloads to the AI runtime. +// RTPStream uses the static PCMA profile supported by Asterisk RTP ExternalMedia, +// which has no SDP negotiation. Its application boundary is mono little-endian +// PCM16 at 16 kHz. Pion owns RTP framing; g711 owns PCMA coding. +// One playback writer is used by CallFlow; receiving and reading stats are safe +// while it plays audio. type RTPStream struct { conn *net.UDPConn + mu sync.Mutex peer *net.UDPAddr - payloadType uint8 sequence uint16 timestamp uint32 ssrc uint32 - format Format - wireSampleRate int - wireFrameBytes int receivedPackets uint64 receivedBytes uint64 sentPackets uint64 @@ -54,34 +49,7 @@ type RTPStats struct { SentBytes uint64 } -func ListenRTP(address string, payloadType uint8) (*RTPStream, error) { - return ListenRTPWithFormat(address, payloadType, FormatSLIN16, 16000) -} - -func ValidateFormat(format Format, payloadType uint8, sampleRate int) error { - return validateFormat(format, payloadType, sampleRate) -} - -func validateFormat(format Format, payloadType uint8, sampleRate int) error { - switch format { - case FormatSLIN16: - if sampleRate != 16000 || payloadType < 96 || payloadType > 127 { - return fmt.Errorf("invalid slin16 RTP profile: sample_rate=%d payload_type=%d", sampleRate, payloadType) - } - case FormatALAW: - if sampleRate != 8000 || payloadType != 8 { - return fmt.Errorf("invalid alaw RTP profile: sample_rate=%d payload_type=%d", sampleRate, payloadType) - } - default: - return fmt.Errorf("unsupported RTP format %q", format) - } - return nil -} - -func ListenRTPWithFormat(address string, payloadType uint8, format Format, sampleRate int) (*RTPStream, error) { - if err := validateFormat(format, payloadType, sampleRate); err != nil { - return nil, err - } +func ListenRTP(address string) (*RTPStream, error) { addr, err := net.ResolveUDPAddr("udp", address) if err != nil { return nil, err @@ -95,15 +63,7 @@ func ListenRTPWithFormat(address string, payloadType uint8, format Format, sampl _ = conn.Close() return nil, err } - return &RTPStream{ - conn: conn, - payloadType: payloadType, - sequence: 1, - ssrc: binary.BigEndian.Uint32(ssrcBytes[:]), - format: format, - wireSampleRate: sampleRate, - wireFrameBytes: sampleRate / 50 * 2, - }, nil + return &RTPStream{conn: conn, sequence: 1, ssrc: binary.BigEndian.Uint32(ssrcBytes[:])}, nil } func (s *RTPStream) LocalAddr() net.Addr { return s.conn.LocalAddr() } @@ -112,12 +72,14 @@ func (s *RTPStream) Close() error { if s == nil || s.conn == nil { return nil } + stats := s.Stats() + slog.Info("RTP media closing", "wire_format", WireFormat, "wire_sample_rate", WireSampleRate, "payload_type", WirePayloadType, + "received_packets", stats.ReceivedPackets, "received_bytes", stats.ReceivedBytes, "sent_packets", stats.SentPackets, "sent_bytes", stats.SentBytes) return s.conn.Close() } -// SetPeer configures the Asterisk ExternalMedia RTP destination before the -// first inbound packet arrives. A received packet may still refine it to the -// actual source address used by the connected RTP socket. +// SetPeer allows opening playback before the first inbound packet, using +// Asterisk's UNICASTRTP_LOCAL_* values. Inbound RTP may refine its source address. func (s *RTPStream) SetPeer(address string) error { if s == nil || s.conn == nil { return errors.New("RTP stream is closed") @@ -126,49 +88,62 @@ func (s *RTPStream) SetPeer(address string) error { if err != nil { return err } + s.mu.Lock() s.peer = peer + s.mu.Unlock() return nil } -// ReadPayload reads one RTP packet, validates version/payload type and remembers -// the Asterisk peer for subsequent outbound audio. +// ReadPayload decodes one PCMA RTP payload and returns canonical PCM16/16 kHz. +// Wrong codecs or malformed packets are errors, not hidden as ASR silence. func (s *RTPStream) ReadPayload(ctx context.Context) ([]byte, error) { if s == nil || s.conn == nil { return nil, errors.New("RTP stream is closed") } buf := make([]byte, maxRTPPacketBytes) for { + if err := ctx.Err(); err != nil { + return nil, err + } deadline := time.Now().Add(250 * time.Millisecond) + if d, ok := ctx.Deadline(); ok && d.Before(deadline) { + deadline = d + } if err := s.conn.SetReadDeadline(deadline); err != nil { return nil, err } n, peer, err := s.conn.ReadFromUDP(buf) if err != nil { if ne, ok := err.(net.Error); ok && ne.Timeout() { - select { - case <-ctx.Done(): - return nil, ctx.Err() - default: - continue - } + continue } return nil, err } var packet rtp.Packet if err := packet.Unmarshal(buf[:n]); err != nil { - continue + return nil, fmt.Errorf("invalid RTP packet: %w", err) } - if packet.Version != 2 || packet.PayloadType != s.payloadType { - continue + if packet.Version != 2 { + return nil, fmt.Errorf("unsupported RTP version %d", packet.Version) } + if packet.PayloadType != WirePayloadType { + return nil, fmt.Errorf("RTP payload type %d does not match required PCMA payload type %d", packet.PayloadType, WirePayloadType) + } + if len(packet.Payload) == 0 { + return nil, errors.New("empty PCMA RTP payload") + } + pcm := audio.ResamplePCM16(g711.DecodeAlaw(packet.Payload), WireSampleRate, PCM16SampleRate) + s.mu.Lock() s.peer = peer s.receivedPackets++ s.receivedBytes += uint64(len(packet.Payload)) - payload := append([]byte(nil), packet.Payload...) - if s.format == FormatALAW { - payload = audio.ResamplePCM16(g711.DecodeAlaw(payload), s.wireSampleRate, 16000) + first := s.receivedPackets == 1 + s.mu.Unlock() + if first { + slog.Info("RTP input decoded", "wire_format", WireFormat, "wire_sample_rate", WireSampleRate, "payload_type", packet.PayloadType, + "wire_bytes", len(packet.Payload), "pcm_sample_rate", PCM16SampleRate, "pcm_bytes", len(pcm)) } - return payload, nil + return pcm, nil } } @@ -176,62 +151,62 @@ func (s *RTPStream) Stats() RTPStats { if s == nil { return RTPStats{} } + s.mu.Lock() + defer s.mu.Unlock() return RTPStats{ReceivedPackets: s.receivedPackets, ReceivedBytes: s.receivedBytes, SentPackets: s.sentPackets, SentBytes: s.sentBytes} } -// SendPCM16 sends signed-linear mono PCM16 in 20 ms RTP frames. The peer may -// be configured from Asterisk's UNICASTRTP_LOCAL_* variables before capture. +// SendPCM16 converts canonical PCM16 to PCMA and paces 20-ms RTP frames. +// Encoded sizes and timestamps use the 8-kHz wire clock, not the PCM clock. func (s *RTPStream) SendPCM16(ctx context.Context, pcm []byte, sampleRate int) error { if s == nil || s.conn == nil { return errors.New("RTP stream is closed") } - if s.peer == nil { - return errors.New("RTP peer is unknown; configure the ExternalMedia peer first") - } - if sampleRate != 16000 { + if sampleRate != PCM16SampleRate { return fmt.Errorf("unsupported PCM sample rate %d", sampleRate) } - wirePCM := audio.ResamplePCM16(pcm, sampleRate, s.wireSampleRate) + if len(pcm)%2 != 0 { + return errors.New("playback PCM16 is not aligned to complete samples") + } + if err := ctx.Err(); err != nil { + return err + } + s.mu.Lock() + peer := s.peer + s.mu.Unlock() + if peer == nil { + return errors.New("RTP peer is unknown; configure the ExternalMedia peer first") + } + wirePCM := audio.ResamplePCM16(pcm, sampleRate, WireSampleRate) + slog.Info("RTP playback prepared", "pcm_sample_rate", sampleRate, "pcm_bytes", len(pcm), "wire_format", WireFormat, + "wire_sample_rate", WireSampleRate, "payload_type", WirePayloadType, "wire_samples", len(wirePCM)/2) for offset := 0; offset < len(wirePCM); { - end := offset + s.wireFrameBytes - if s.format == FormatALAW { - end = offset + s.wireSampleRate/50*2 + if err := ctx.Err(); err != nil { + return err } + end := offset + WireSampleRate/50*2 if end > len(wirePCM) { end = len(wirePCM) } - if end-offset < 2 { - break - } frame := wirePCM[offset:end] - payload := frame - if s.format == FormatALAW { - payload = g711.EncodeAlaw(frame) - } - packet := &rtp.Packet{ - Header: rtp.Header{ - Version: 2, - PayloadType: s.payloadType, - SequenceNumber: s.sequence, - Timestamp: s.timestamp, - SSRC: s.ssrc, - }, - Payload: append([]byte(nil), payload...), - } + payload := g711.EncodeAlaw(frame) + packet := &rtp.Packet{Header: rtp.Header{Version: 2, PayloadType: WirePayloadType, SequenceNumber: s.sequence, Timestamp: s.timestamp, SSRC: s.ssrc}, Payload: payload} encoded, err := packet.Marshal() if err != nil { return err } - if err := s.conn.SetWriteDeadline(time.Now().Add(1 * time.Second)); err != nil { + if err := s.conn.SetWriteDeadline(time.Now().Add(time.Second)); err != nil { return err } - if _, err := s.conn.WriteToUDP(encoded, s.peer); err != nil { + if _, err := s.conn.WriteToUDP(encoded, peer); err != nil { return err } s.sequence++ s.timestamp += uint32(len(frame) / 2) + s.mu.Lock() s.sentPackets++ s.sentBytes += uint64(len(payload)) + s.mu.Unlock() offset = end select { case <-ctx.Done(): @@ -239,5 +214,6 @@ func (s *RTPStream) SendPCM16(ctx context.Context, pcm []byte, sampleRate int) e case <-time.After(20 * time.Millisecond): } } + slog.Info("RTP playback submitted", "pcm_bytes", len(pcm), "wire_samples", len(wirePCM)/2) return nil } diff --git a/internal/rpc/approved_debug_test.go b/internal/rpc/approved_debug_test.go index 6e13be4..6c1c50b 100644 --- a/internal/rpc/approved_debug_test.go +++ b/internal/rpc/approved_debug_test.go @@ -12,7 +12,7 @@ func TestRealCallCaptureControlsExternalEvidenceOnly(t *testing.T) { t.Run(map[bool]string{false: "normal", true: "debug"}[debug], func(t *testing.T) { fixture, _, _, _, approved := recordedMockFixture(t, nil, time.Second) approved.CallerID, approved.DialedCallee, approved.RingTimeout = "BD93205882", "7089"+approved.Callee, time.Second - r := &ApprovedRecordedRealCall{Capture: debug, Mirror: asterisk.NewHEPMirror(), MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, Delivery: fixture.Delivery, originator: func(ctx context.Context, _ asterisk.NativeDial) (realMedia, error) { + r := &ApprovedRecordedRealCall{Capture: debug, Mirror: asterisk.NewHEPMirror(), MaxWAVBytes: 4096, ReportTimeout: time.Second, Delivery: fixture.Delivery, originator: func(ctx context.Context, _ asterisk.NativeDial) (realMedia, error) { t.Fatal("prepare must not dial") return realMedia{}, nil }} diff --git a/internal/rpc/approved_recorded_real.go b/internal/rpc/approved_recorded_real.go index 2e875de..c8802b9 100644 --- a/internal/rpc/approved_recorded_real.go +++ b/internal/rpc/approved_recorded_real.go @@ -22,15 +22,14 @@ import ( // exactly one native ARI call. Unlike the isolated Mock it never supplies a // synthetic transcript, media frame, outcome, or recording. type ApprovedRecordedRealCall struct { - Loader asterisk.Loader - MediaPayloadType uint8 - EvidenceRoot string - Capture bool - MaxWAVBytes int64 - ReportTimeout time.Duration - Delivery *agent.RecordingDelivery - Mirror *asterisk.HEPMirror - originator func(context.Context, asterisk.NativeDial) (realMedia, error) // isolated tests only + Loader asterisk.Loader + EvidenceRoot string + Capture bool + MaxWAVBytes int64 + ReportTimeout time.Duration + Delivery *agent.RecordingDelivery + Mirror *asterisk.HEPMirror + originator func(context.Context, asterisk.NativeDial) (realMedia, error) // isolated tests only } type realMedia struct { @@ -44,7 +43,7 @@ type realMedia struct { // Prepare rejects missing signed identity and recovery capacity before the // Agent acknowledges the call. No SIP channel is created during preparation. func (r *ApprovedRecordedRealCall) Prepare(approved ApprovedExecution) (func(context.Context) error, error) { - if r == nil || r.Delivery == nil || r.Mirror == nil || r.ReportTimeout <= 0 || r.MaxWAVBytes <= 44 || r.MediaPayloadType < 96 || r.MediaPayloadType > 127 { + if r == nil || r.Delivery == nil || r.Mirror == nil || r.ReportTimeout <= 0 || r.MaxWAVBytes <= 44 { return nil, errors.New("real call requires bounded media and per-call delivery") } if approved.DispatcherID == "" || approved.TenantID <= 0 || approved.SourceEventID == "" || approved.CallID != approved.SourceEventID || approved.TaskID == "" || @@ -87,8 +86,8 @@ func (r *ApprovedRecordedRealCall) Prepare(approved ApprovedExecution) (func(con dial := asterisk.NativeDial{ ExecutionID: approved.SourceEventID, TrunkID: approved.SelectedTrunkID, DialedCallee: approved.DialedCallee, CallerID: approved.CallerID, - AnswerTimeout: approved.RingTimeout, MediaPayloadType: r.MediaPayloadType, - BindSIPCall: r.Mirror.Bind, + AnswerTimeout: approved.RingTimeout, + BindSIPCall: r.Mirror.Bind, } addSIPResponse := func(ctx context.Context, facts *callflow.FinalCallFacts) { mirrorCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), time.Second) diff --git a/internal/rpc/approved_recorded_real_test.go b/internal/rpc/approved_recorded_real_test.go index c7b05ee..4c518cb 100644 --- a/internal/rpc/approved_recorded_real_test.go +++ b/internal/rpc/approved_recorded_real_test.go @@ -76,12 +76,12 @@ func TestApprovedRecordedRealCallReportsOnlyEndedObservedCall(t *testing.T) { answered := time.Now().Add(-2500 * time.Millisecond).UTC() var nativeEnd time.Time runner := &ApprovedRecordedRealCall{Mirror: asterisk.NewHEPMirror(), - MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, + MaxWAVBytes: 4096, ReportTimeout: time.Second, Delivery: fixture.Delivery, originator: func(ctx context.Context, dial asterisk.NativeDial) (realMedia, error) { originate++ - if dial.ExecutionID != approved.SourceEventID || dial.DialedCallee != approved.DialedCallee || dial.CallerID != approved.CallerID || dial.TrunkID != approved.SelectedTrunkID || dial.MediaPayloadType != 118 { - t.Errorf("signed SIP identity or media profile changed: %+v", dial) + if dial.ExecutionID != approved.SourceEventID || dial.DialedCallee != approved.DialedCallee || dial.CallerID != approved.CallerID || dial.TrunkID != approved.SelectedTrunkID { + t.Errorf("signed SIP identity changed: %+v", dial) } return realMedia{ctx: ctx, session: endedRealMedia{}, close: func() error { hangup++; nativeEnd = time.Now().UTC(); return nil }, answerAt: answered, endAt: func() time.Time { return nativeEnd }}, nil }, @@ -106,7 +106,7 @@ func TestApprovedRecordedRealCallReportsConfirmedMediaSetupFailure(t *testing.T) approved.CallerID, approved.DialedCallee, approved.RingTimeout = "BD93205882", "7089"+approved.Callee, time.Second var originate, hangup int runner := &ApprovedRecordedRealCall{Mirror: asterisk.NewHEPMirror(), - MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, + MaxWAVBytes: 4096, ReportTimeout: time.Second, Delivery: fixture.Delivery, originator: func(ctx context.Context, _ asterisk.NativeDial) (realMedia, error) { originate++ @@ -131,7 +131,7 @@ func TestApprovedRecordedRealCallMissingMediaWithUnknownHangupDoesNotReport(t *t fixture, stub, _, _, approved := recordedMockFixture(t, nil, time.Second) approved.CallerID, approved.DialedCallee, approved.RingTimeout = "BD93205882", "7089"+approved.Callee, time.Second runner := &ApprovedRecordedRealCall{Mirror: asterisk.NewHEPMirror(), - MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, + MaxWAVBytes: 4096, ReportTimeout: time.Second, Delivery: fixture.Delivery, originator: func(ctx context.Context, _ asterisk.NativeDial) (realMedia, error) { return realMedia{ctx: ctx, close: func() error { return errors.New("hangup response missing") }}, nil @@ -147,7 +147,7 @@ func TestApprovedRecordedRealCallDoesNotReportUnknownHangup(t *testing.T) { approved.CallerID, approved.DialedCallee, approved.RingTimeout = "BD93205882", "7089"+approved.Callee, time.Second var originate, hangup int runner := &ApprovedRecordedRealCall{Mirror: asterisk.NewHEPMirror(), - MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, + MaxWAVBytes: 4096, ReportTimeout: time.Second, Delivery: fixture.Delivery, originator: func(ctx context.Context, _ asterisk.NativeDial) (realMedia, error) { originate++ @@ -184,7 +184,7 @@ func TestApprovedRecordedRealCallReportsExactSIPResponseBeforeAnswer(t *testing. invite := "INVITE sip:b@localhost SIP/2.0\r\n" + common raw := "SIP/2.0 480 Temporarily Unavailable\r\n" + common runner := &ApprovedRecordedRealCall{ - MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, + MaxWAVBytes: 4096, ReportTimeout: time.Second, Mirror: mirror, Delivery: fixture.Delivery, originator: func(_ context.Context, dial asterisk.NativeDial) (realMedia, error) { if dial.BindSIPCall == nil || dial.BindSIPCall(dial.ExecutionID, callID) != nil { @@ -216,7 +216,7 @@ func TestApprovedRecordedRealCallReportsVerifiedPreAnswerEndWithoutRecording(t * approved.CallerID, approved.DialedCallee, approved.RingTimeout = "BD93205882", "7089"+approved.Callee, time.Second originate := 0 runner := &ApprovedRecordedRealCall{Mirror: asterisk.NewHEPMirror(), - MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, Delivery: fixture.Delivery, + MaxWAVBytes: 4096, ReportTimeout: time.Second, Delivery: fixture.Delivery, originator: func(context.Context, asterisk.NativeDial) (realMedia, error) { originate++ return realMedia{}, &asterisk.NativeCallFailure{Phase: "answer_wait", ConfirmedEnd: true, Cause: errors.New("no answer")} @@ -242,7 +242,7 @@ func TestApprovedRecordedRealCallDoesNotReportUnverifiedPreAnswerEnd(t *testing. fixture, stub, _, _, approved := recordedMockFixture(t, nil, time.Second) approved.CallerID, approved.DialedCallee, approved.RingTimeout = "BD93205882", "7089"+approved.Callee, time.Second runner := &ApprovedRecordedRealCall{Mirror: asterisk.NewHEPMirror(), - MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, Delivery: fixture.Delivery, + MaxWAVBytes: 4096, ReportTimeout: time.Second, Delivery: fixture.Delivery, originator: func(context.Context, asterisk.NativeDial) (realMedia, error) { return realMedia{}, &asterisk.NativeCallFailure{Phase: "answer_wait", Cause: errors.New("channel state unavailable")} }, @@ -257,7 +257,7 @@ func TestApprovedRecordedRealCallCannotOriginateTwiceOrReportUnknownOrigination( approved.CallerID, approved.DialedCallee, approved.RingTimeout = "BD93205882", "7089"+approved.Callee, time.Second originate := 0 runner := &ApprovedRecordedRealCall{Mirror: asterisk.NewHEPMirror(), - MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, + MaxWAVBytes: 4096, ReportTimeout: time.Second, Delivery: fixture.Delivery, originator: func(context.Context, asterisk.NativeDial) (realMedia, error) { originate++ @@ -280,7 +280,7 @@ func TestApprovedRecordedRealCallRejectsMissingDeliveryBeforeOrigination(t *test _, _, _, _, approved := recordedMockFixture(t, nil, time.Second) called := false runner := &ApprovedRecordedRealCall{ - MaxWAVBytes: 4096, MediaPayloadType: 118, ReportTimeout: time.Second, + MaxWAVBytes: 4096, ReportTimeout: time.Second, originator: func(context.Context, asterisk.NativeDial) (realMedia, error) { called = true return realMedia{session: callflow.NewMemorySession(nil)}, nil diff --git a/internal/sipcall/run_ai.go b/internal/sipcall/run_ai.go index e9cc77e..971ba2d 100644 --- a/internal/sipcall/run_ai.go +++ b/internal/sipcall/run_ai.go @@ -168,7 +168,6 @@ func runAI(ctx context.Context, c Config, number string, debug bool, config AICo request := asterisk.NativeDial{ ExecutionID: id, TrunkID: c.Endpoint(), DialedCallee: route, CallerID: c.CallerID, AnswerTimeout: time.Duration(config.Task.RingTimeoutMS) * time.Millisecond, - MediaPayloadType: 118, BindSIPCall: func(_ string, callID string) error { if callID == "" { return errors.New("missing native SIP Call-ID")