Verify Dispatcher restart does not repeat recorded call
This commit is contained in:
@@ -410,4 +410,78 @@ func TestCurrentDispatcherCommandStartsWithIsolatedMQHTTPAndAgent(t *testing.T)
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("isolated Dispatcher did not stop after cancellation")
|
||||
}
|
||||
|
||||
previousSIP, previousDiscovery := sipReads.Load(), discoveryReads.Load()
|
||||
restartContext, restartCancel := context.WithCancel(context.Background())
|
||||
defer restartCancel()
|
||||
restartCommand := newRootCommand()
|
||||
restartCommand.SetContext(restartContext)
|
||||
restartCommand.SetOut(io.Discard)
|
||||
restartCommand.SetErr(io.Discard)
|
||||
restartCommand.SetArgs([]string{"dispatcher", "--mode", "mock"})
|
||||
restarted := make(chan error, 1)
|
||||
go func() { restarted <- restartCommand.Execute() }()
|
||||
restartDeadline := time.After(8 * time.Second)
|
||||
for sipReads.Load() <= previousSIP || discoveryReads.Load() <= previousDiscovery {
|
||||
select {
|
||||
case err := <-restarted:
|
||||
t.Fatalf("isolated Dispatcher could not reopen its original SQLite and session: %v", err)
|
||||
case <-restartDeadline:
|
||||
t.Fatal("isolated Dispatcher did not restart its approved HTTP snapshot")
|
||||
case <-time.After(20 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
for {
|
||||
queue, err := admin.QueueInspect(taskRoute.Queue)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if queue.Consumers > 0 {
|
||||
break
|
||||
}
|
||||
select {
|
||||
case err := <-restarted:
|
||||
t.Fatalf("isolated Dispatcher stopped before restoring its task queue: %v", err)
|
||||
case <-restartDeadline:
|
||||
t.Fatal("isolated Dispatcher did not restore its task consumer")
|
||||
case <-time.After(20 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
if err := admin.PublishWithContext(restartContext, taskRoute.Exchange, taskRoute.BindingKey, true, false, amqp.Publishing{ContentType: "application/json", DeliveryMode: amqp.Persistent, Body: commandBody}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for {
|
||||
queue, err := admin.QueueInspect(taskRoute.Queue)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if queue.Messages == 0 {
|
||||
break
|
||||
}
|
||||
select {
|
||||
case err := <-restarted:
|
||||
t.Fatalf("isolated Dispatcher stopped before acknowledging the duplicate call: %v", err)
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("duplicate call was not consumed after restart")
|
||||
case <-time.After(20 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
time.Sleep(250 * time.Millisecond)
|
||||
if putCount.Load() != 1 {
|
||||
t.Fatalf("restart repeated an already uploaded call: PUT count=%d", putCount.Load())
|
||||
}
|
||||
select {
|
||||
case err := <-restarted:
|
||||
t.Fatalf("isolated Dispatcher stopped after duplicate delivery: %v", err)
|
||||
default:
|
||||
}
|
||||
restartCancel()
|
||||
select {
|
||||
case err := <-restarted:
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
t.Fatalf("restarted Dispatcher shutdown failed: %v", err)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("restarted Dispatcher did not stop after cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -79,19 +79,20 @@
|
||||
|
||||
- 已新增 Agent 内存录音单次 PUT 组件 `UploadClient.UploadBytes`：复用受限授权、大小和 SHA-256 核对，成功路径不写临时录音或通话结果文件。OSS 明确拒绝保留状态码且不自动重试；传输结果不明时返回专用错误、停止自动重试并隐藏带签名的 URL。空授权头拒绝而非使服务崩溃。这里只验证组件，尚未连接实际录音或最终结果。
|
||||
- 内部 OSS 授权增加 Dispatcher 原始 bucket，官方 SDK 签发时返回获批 bucket；Agent 收到缺少 bucket 的授权会在 PUT 前拒绝，失败恢复的显式重申请若返回了不同 bucket，也在再次 PUT 前拒绝。两条红灯测试证明先前会错误上传；修复后全包测试及 Agent/RPC/OSS race 测试通过。这里只核验本地授权载体，不代表新主入口已完成签发或真实 OSS 已验证。
|
||||
- Agent 失败恢复隔离组件：仅在 OSS PUT 明确失败后，把原 bucket/object_key 对应的录音和通话信息两文件写入私有目录并同步落盘；双文件缺失或损坏明确报错、不伪造结果。完成保存后固定 48 小时窗口，按 1 分钟递增至最长 1 小时重试；到期保留原文件。PUT 前持久写入 in-flight，结果不明或进程重启不会二次 PUT；确认上传后先持久记录成功，再经注入的 Mock 回报通话结果，回报失败/重启仅重发原结果。启动扫描识别遗漏文件并提供不暴露原路径的稳定摘要；主程序录音授权与回报已接线，但尚未通过含任务的端到端联测。
|
||||
- Agent 失败恢复隔离组件：仅在 OSS PUT 明确失败后，把原 bucket/object_key 对应的录音和通话信息两文件写入私有目录并同步落盘；双文件缺失或损坏明确报错、不伪造结果。完成保存后固定 48 小时窗口，按 1 分钟递增至最长 1 小时重试；到期保留原文件。PUT 前持久写入 in-flight，结果不明或进程重启不会二次 PUT；确认上传后先持久记录成功，再经注入的 Mock 回报通话结果，回报失败/重启仅重发原结果。启动扫描识别遗漏文件并提供不暴露原路径的稳定摘要；根 Dispatcher 已通过含任务的本机隔离正向联测；失败和 48 小时边界分别由本段的分层故障测试核验。
|
||||
- Dispatcher 的无录音最终结果隔离组件：`CurrentStore.RecordCallResult` 仅在确认通话结束后，按持久任务快照校验任务、被叫、主叫和已选线路，并以源执行事件固定生成唯一最终结果身份；消息通过严格 MQ Schema 校验后与 outbox 在同一事务写入。同内容重投/重启只恢复原消息，冲突结果和 SQLite 写入失败均不会产生第二份结果。此条事务测试为隔离组件；下述根命令隔离联测已连通批准的 Mock Agent 回报，但不代表外部应用收讫。
|
||||
- Dispatcher 的原始 OSS 目标及已上传结果隔离组件：新 SQLite 布局把一次通话的 upload_id、bucket、object_key、录音格式/时长/大小和 SHA-256 唯一绑定到已保留的执行；不保存临时 URL 或 TOKEN。旧布局拒绝启动并原样保留待交付 outbox，不自动迁移或清理。已签发录音目标不能通过空录音结果绕过上传；已有空录音最终结果不能再签发录音目标。`RecordUploadedCallResult` 仅接受与持久绑定完全一致的录音事实及 Agent 所报告的成功 PUT 状态，录音确认与唯一最终结果 outbox 同事务提交；丢失回报或 MQ 投递时重用原消息，已确认后拒绝再次签发 PUT 授权。Mock 证明的是本地状态约束，不是独立 OSS 校验或真实 Agent 身份验证。
|
||||
- Dispatcher 的终结与外呼回执顺序竞争隔离修复：原流程在 Agent 接受执行的 RPC 返回后才写入外呼回执，快速结束或 RPC 超时可能先到；现以一次 SQLite 事务在确认通话已结束后补齐原回执并释放占用，未知执行仅在确认结束后释放。迟到的执行响应、超时和重复结束不会产生第二份回执；注入 outbox 写入失败保留原占用。并发竞争及结束后立即生成唯一最终结果有单元测试；根 Dispatcher 的 Agent 会话、合成 Mock 呼叫、上传事实和 MQ 最终结果现已在本机隔离联测；MQ 失败后的根命令重启恢复仍待验证。
|
||||
- Dispatcher 的终结与外呼回执顺序竞争隔离修复：原流程在 Agent 接受执行的 RPC 返回后才写入外呼回执，快速结束或 RPC 超时可能先到；现以一次 SQLite 事务在确认通话已结束后补齐原回执并释放占用，未知执行仅在确认结束后释放。迟到的执行响应、超时和重复结束不会产生第二份回执；注入 outbox 写入失败保留原占用。并发竞争及结束后立即生成唯一最终结果有单元测试；根 Dispatcher 的 Agent 会话、合成 Mock 呼叫、上传事实和 MQ 最终结果现已在本机隔离联测；根命令遭遇 MQ 持续故障后的完整重启恢复仍待 P08 注入验证。
|
||||
- Dispatcher 录音事实 Unary RPC 隔离服务：`RequestRecordingUpload`、`ReportCallEnded`、`ReportCallResult` 均要求已配置本 D、核验 mTLS 指纹及当前 Agent 会话、数字租户和已保留的执行；复用官方 SDK 仅对原始录音签发固定 15 分钟授权，显式重申请仍用相同 bucket/object_key。结束事实可先于外呼响应而持久化原回执；录音结果核对 D 已存目标和 Agent 报告的成功 PUT，再与唯一结果 outbox 同事务提交。Mock 覆盖会话/租户拒绝、同资产重申请、上传前结果拒绝、坏 JSON、已上传与无录音结果及重复回报。隔离测试还通过本地双向 TLS 的 gRPC 实际传输：Agent `RecordingClient` 每次读取并克隆当前会话元数据，经受控 D 客户端领取授权、上报结束和唯一结果；未配置客户端明确拒绝。其分层测试不包含 OSS PUT；下述本地隔离链路另验证直传，根 Dispatcher 与批准的 Mock Agent 服务已在本机双向 TLS 链路完成合成媒体录音、录音 Unary、单次 OSS PUT 和共享 MQ 最终结果；真实 Agent CLI 与外部服务未在该测试中启动。
|
||||
- 内存录音隔离组件：`RecordingSession` 仅复制共享通话流程实际读到和成功发送的 16-kHz PCM16，`EncodeMonoWAV` 直接在内存生成有界单声道 WAV；空音频、奇数字节、超过上限及未成功发送的音频都不能伪造成可上传录音。单元与 race 测试未产生业务文件。批准执行的合成 Mock runner 已接入该组件；Mock 中观测到的帧不等于真实 Asterisk 通话的全量媒体验收。
|
||||
- Agent 录音交付隔离组件：`RecordingDelivery` 先确认结束，再依照录音是否实际生成分别上报唯一空录音结果或请求原授权并直传内存 WAV；录音生成失败保留通话真实结果、空录音对象及明确原因，不虚构上传事实。隔离测试通过本地 HTTP PUT 和假 Dispatcher RPC 覆盖成功无业务文件、OSS 明确失败后私有文件保存、恢复写入失败、未知 PUT 隔离、重启重领原目标、上传已确认后只重发原结果。再次调用不会隐式重新 PUT；正常已确认上传但尚未被 D 持久收讫的跨进程间隙仍受 K16 边界约束。此条失败/恢复测试是分层证据；根 Dispatcher 通过临时 RabbitMQ、批准 Agent 服务与本地 HTTPS OSS 的正向联测另见下述记录。
|
||||
- 最终结果隔离组件：`ApprovedExecution` 已保留获批任务的 `caller_profile_id`；`FinalResultPayload` 只从批准的任务身份、确认时间及实际收到的用户媒体窗口生成当前唯一结果；最终 ASR 文本与字面关键词拒联写入完整转写，未播放的助手候选内容不冒充转写，未知 SIP 响应码保留 `null`，无录音保留 `{}`。单元测试直接核验当前 MQ Schema、字段缺失/乱序时窗与无应答负例；开场白发送失败不得计入已播放音频。此组件已接入 Agent 合成 runner，但未证明真实 RTP 转写时间精度。
|
||||
- 本地隔离链路：`RunApprovedCall` 显式使用合成 `ApprovedMockPipeline`，让共享批准通话流程按 ASR-only 限制消费实际读出的 PCM Mock 媒体帧；`RecordingSession` 生成内存 WAV，`FinalResultPayload` 仅用最终模拟识别及实测采集时窗构造结果。Agent 经双向 TLS gRPC 向 Dispatcher 确认结束、领取原始资产签名授权，再对本地 HTTPS OSS Mock 单次 PUT；Dispatcher 将原上传事实与唯一最终结果 outbox 同事务提交。测试确认一次 PUT、一次结果、无业务文件，使用官方 SDK 生成的路径；Mock ASR 与 OSS 服务不验证真实供应商协议或签名。这不是主入口执行、RabbitMQ 投递或外部验收。
|
||||
- `ApprovedRecordedMockCall` 将显式批准的 ASR-only AI 快照、合成 PCM 媒体、每通话独立脚本、实际采集的有界 WAV、用户最终识别时窗和 `RecordingDelivery` 组合到一个可供 Agent 工人调用的 Mock runner；隔离测试验证先结束事实、一次本地 OSS PUT、唯一最终结果及正常路径无业务文件。无实际读入媒体时明确报告失败并以空录音、生成失败原因收口，不伪造上传或转写；缺失每通话交付器、交付器与签发 D/数字租户/事件不一致、私有恢复目录并非 `0700` 或模拟脚本不满足已批准 AI 时，均在发外呼接受回执前拒绝。此 runner 已接到隔离 Mock 主 CLI 的 `ApprovedCallWorker`，该合成 runner 已在根 Dispatcher 驱动的本机隔离流中完成 D→A 执行、Agent→D 录音、HTTPS OSS 单次 PUT 和 MQ 出站；这不代表真实线路或供应商通过。
|
||||
- 根 Dispatcher 正向隔离链路：`bash scripts/check-current-mq-mock.sh` 启用临时 RabbitMQ、五类本机 Mock HTTP 资源和预建任务队列；根 Dispatcher 激活实际批准的 Mock Agent 服务并处理一条 `call.execute`。Agent 以合成 16-kHz PCM 生成 WAV，经双向 TLS 领取授权后单次 PUT 至本机 HTTPS OSS；共享结果队列收到 `answered`/`uploaded` 的唯一最终事实，WAV 字节数和 SHA-256 与 PUT 一致，正常路径私有恢复目录没有业务文件。Agent 使用与根命令相同的服务组装函数，但该测试未启动 Agent CLI；MQ 确认不是 SaaS 应用收讫，真实外部及失败重启仍需分别验证。
|
||||
- 根 Dispatcher 正向隔离链路：`bash scripts/check-current-mq-mock.sh` 启用临时 RabbitMQ、五类本机 Mock HTTP 资源和预建任务队列；根 Dispatcher 激活实际批准的 Mock Agent 服务并处理一条 `call.execute`。Agent 以合成 16-kHz PCM 生成 WAV，经双向 TLS 领取授权后单次 PUT 至本机 HTTPS OSS；共享结果队列收到 `answered`/`uploaded` 的唯一最终事实，WAV 字节数和 SHA-256 与 PUT 一致，正常路径私有恢复目录没有业务文件。Agent 使用与根命令相同的服务组装函数，但该测试未启动 Agent CLI；原 SQLite 关闭重启并重投同一指令后恢复消费，OSS PUT 仍为一次；MQ 确认不是 SaaS 应用收讫，根命令持续 MQ 故障及真实外部仍待验证。
|
||||
- 分层故障与恢复：Agent 测试覆盖空录音、录音生成失败、OSS 明确失败后私有落盘、落盘失败保留原音频、未知 PUT 停止自动重传、确认上传后仅重试回报、以及到期恰好 48 小时保留供人工而不 PUT；Dispatcher/SQLite 测试覆盖录音事实与唯一结果 outbox 的事务回滚、重启后保留原授权目标及同一出站身份；隔离 RabbitMQ 测试证明无路由时 mandatory/confirm 不谎报送达。根命令另在原 SQLite/Agent 会话上重启并重投相同 `call.execute`，确认没有第二次 PUT。分层测试不等于真实 SaaS 应用收讫或长时 MQ 故障后的主入口完整恢复。
|
||||
- 本机 HTTPS Mock 信任：Agent 上传及 Dispatcher 配置读取使用现有 mTLS CA 信任束校验证书，不放宽主机名或证书验证，仍拒绝非本机目标及跨域重定向。测试确认已批准的本机 HTTPS 证书可访问，未知证书与无效信任束均失败；未使用 `InsecureSkipVerify`。这不是对真实 OSS 的验证。
|
||||
- 已验证：`go test ./... -count=1`、`go test -race ./... -count=1`、`go vet ./...`、`go build ./...`、`PATH=/tmp/sip-go-agent-tools/bin:$PATH bash scripts/check-current-contracts.sh`、`PATH=/tmp/sip-go-agent-tools/bin:$PATH bash scripts/check-proto.sh`、`git diff --check`。根 Dispatcher 的本机 Mock 正向执行、录音与共享 MQ 结果已通过；主入口 MQ 失败及重启恢复尚未整体联测，不能据此宣称 P06 完成或真实外部验收通过。
|
||||
- 已验证：`go test ./... -count=1`、`go test -race ./... -count=1`、`go vet ./...`、`go build ./...`、`PATH=/tmp/sip-go-agent-tools/bin:$PATH bash scripts/check-current-contracts.sh`、`PATH=/tmp/sip-go-agent-tools/bin:$PATH bash scripts/check-proto.sh`、`git diff --check`。P06 的本机隔离正向、失败注入和重启防重复场景已分别通过；根命令在 MQ 长时故障后的完整恢复仍需 P08 验证，真实外部服务、SaaS 应用收讫与线路均未验收，不冒充生产通过。
|
||||
|
||||
## 验收台账
|
||||
|
||||
|
||||
Reference in New Issue
Block a user