From c2ee8a5a4bd590568af1225b64adbbd71066fec6 Mon Sep 17 00:00:00 2001 From: Rogee Date: Sat, 10 Oct 2026 15:02:30 +0800 Subject: [PATCH] fix: preserve source event IDs across all feedback --- AGENTS.md | 4 +- .../dispatcher_integration_test.go | 15 ++ contracts/schema | 2 +- deploys/test/saas-mock/publish_test.go | 6 +- deploys/test/saas-mock/result_wait.go | 7 +- docs/contracts.md | 2 + docs/evidence/event-id-lifecycle-20261010.md | 82 ++++++ internal/contract/event_identity_test.go | 33 +++ internal/dispatcher/control_test.go | 2 +- internal/dispatcher/event_identity_test.go | 74 ++++++ internal/dispatcher/execute.go | 6 +- .../dispatcher/runtime_integration_test.go | 58 ++++- internal/mq/broker_integration_test.go | 6 +- internal/store/calls.go | 6 +- internal/store/calls_test.go | 2 +- internal/store/control_ack_test.go | 2 +- internal/store/control_flow.go | 4 +- internal/store/control_receipt_test.go | 4 +- internal/store/event_identity_test.go | 245 ++++++++++++++++++ internal/store/rejected_data.go | 4 +- internal/store/rejection_identity.go | 4 +- internal/store/result.go | 12 +- internal/store/result_test.go | 10 +- internal/store/store.go | 13 +- internal/store/upload.go | 2 +- internal/store/upload_test.go | 2 +- 26 files changed, 553 insertions(+), 54 deletions(-) create mode 100644 docs/evidence/event-id-lifecycle-20261010.md create mode 100644 internal/contract/event_identity_test.go create mode 100644 internal/dispatcher/event_identity_test.go create mode 100644 internal/store/event_identity_test.go diff --git a/AGENTS.md b/AGENTS.md index 7ec27ef..098af8f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -79,6 +79,8 @@ ## 当前范围与权威入口 +- **2026-10-10 event_id 全链路原值规则（本地修复，未部署）**：针对已接收事件的回执、拒绝、执行反馈及 `call.execute.result` 必须原样保留源 `event_id`，包括其中的 numberID，不哈希、拼接或清洗。共享合同 `ef3009d` 已推送；本项目同步固定引用及实现/测试。Dispatcher 外发保存、查询、发送确认及恢复统一以 `(dispatcher_id,event_type,event_id)` 区分，同 ID 的派发回执与最终结果独立处理；录音授权只检查实际最终结果。移除测试专用 SaaS Mock 的旧派生 ID 判定，不改变真实 SaaS 代码。SQLite 当前布局为 **4**，旧布局只读拒绝，不自动迁移、重置或绕过；现存业务数据与运行服务未改。源码文件名、录音存储键、上传及 RPC 动作编号不是源 event_id，保留各自独立身份。核验及 SaaS 开发同步说明见 [`docs/evidence/event-id-lifecycle-20261010.md`](docs/evidence/event-id-lifecycle-20261010.md)。 + - **2026-10-10 SaaS 就绪延期授权**：此次任务回执更新部署中，真实任务接口 HTTP 200 返回 `code=500` 错误正文而非任务列表；使用者明确批准暂不等待 SaaS 修复、继续部署。新增显式 `deploy --defer-saas-readiness`，只延期已通过 MQ 独立登录/关闭、HTTP 200 合同验证失败的 SaaS 就绪检查，默认关闭；不修改运行 Schema、不启用旧配置、Mock 或业务准入。探测保留安全路径/原因与延期标记，Dispatcher 必须实际 auto-restart 且最新错误确认为 SaaS 配置校验，不能把 failed、数据库/证书/未知故障算完成。业务始终标记未 ready，原生 Asterisk、数据、真实呼叫授权与 Debug/Capture 边界不变。规则见 [`deploys/production/README.md`](deploys/production/README.md)。 - **2026-10-10 任务控制回执恢复**：共享合同 `e750fe4` 已推送，项目合并提交 `60fe840` 已进入 main。成功回执必须携带任务编号、动作、处理结果、原因代码和实际任务状态；结构拒绝仍逐字段返回有效身份与动作，不因整条 Schema 失败丢失有效字段，未知字段省略、不填 null，不信请求自带状态。状态读取故障须显式报错；补齐字段后的出站回执仍禁止作为入站命令执行。根因修复测试日志缓冲的并发读写在 `7357eae`，不改业务日志行为。完整检查、实际隔离 MQ/HTTPS/mTLS、72.6% 覆盖率及 Python21通过。使用者已批准 main 推送后的固定环境普通重部署，保持 Debug 开启、Capture 关闭，不 reset、不重启 Asterisk、不主动投递/试拨；现场事实以 `.local/task-control-deployment/` 的受控记录为准，不以本地通过代签 SaaS 收到或生产。合同与本地证据见 [`docs/evidence/task-control-receipt-fields-20261010.md`](docs/evidence/task-control-receipt-fields-20261010.md)。 @@ -146,7 +148,7 @@ - 独立 Dispatcher 的 SQLite 是任务、额度、inbox/outbox 的权威数据；Agent 无业务数据库，录音、执行与上传恢复只写受控私有文件。额度包含未知占用，新 boot/租约到期不得自动清除未知执行；明确 Agent 在拨号前拒绝时须持久拒绝并释放，已接纳但证实 ARI originate 从未提交的失败由原 Agent 经正式终结回执释放，RPC 交付未知时仍占用。2026-10-04 修复版 `d2c4a99` 已安装并通过非呼出时段主机诊断，新隔离测试租户额度 2/revision 2 已启用并由 Dispatcher 只读拉取；原快照保留；该条旧执行仅按前述经使用者授权的人工核实记录解除 Dispatcher 占用，绝非新 boot 或超时自动清除未知执行。仅 Agent 用机器可读标记证实本次未发起的拒绝可自动释放；旧执行的同错误码不算证明。不实现双活数据库、自动跨机热备、多 D 共享额度或第二租户公平。本轮不借本机 D1/D2 隔离夹具宣称多 D 运行。不得建立旧表/旧消息/旧 HTTP 执行兼容通道。 - Dispatcher↔Agent 复用 Unary gRPC 和受控 Endpoint；Agent 预绑定 D UUID 与服务端证书指纹，激活/会话代际、peer mTLS/SAN/SNI 和已签发期限须核对，新 boot 不清未知占用。Agent 不自行向 SaaS 取任务/AI/OSS 授权；Dispatcher 只用已经核验的 Agent `GetLoadedSIP` revision 开执行准入。本机 Mock 的加载回报不证明 Asterisk 已实际加载；真实 Agent（`nonprod-real` 与隔离 `sip-only`）在配置原子写入、PJSIP reload 和运行态 endpoint/AOR/UDP transport 一致后持久标记 revision，每次加载查询重新核验。正式业务 Dispatcher 在启动及 SIP 更新时必须通过现有 Agent 会话加载获批快照后再核对版本，不得漏接加载步骤；旧通话与未知占用保留、排空前关新准入。隔离 Mock 仍只核对预加载夹具，不发起原生 SIP reload。只支持明确的 UDP、IP/none 鉴权、无需 REGISTER 的 IPv4/PCMA 线路；未知字段和其他传输/鉴权/注册方式拒绝，不热更静态 transport。每条 SIP trunk 只允许一个原值 `caller_id`，Agent 把该值同时配置为 Asterisk endpoint 的 `from_user` 和 `callerid`，并在 `GetLoadedSIP` 前核对实际运行值；旧任务 `caller_profile_id` 与线路 `caller_profiles` 必须直接拒绝，不保留兼容路径。SIP 配置的唯一编辑/审批面仍是 management。 - 非生产真实 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 用户批准本地配置规则调整及新模型调用支持：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-10 event_id 原值修复后的当前布局为 4，旧布局只读拒绝，不自动迁移或改写旧文件。 - 2026-10-08 用户确认按共享合同 `b502ad2` 本地适配任务线路并发：`allowed_trunk_ids` 只接受包含 `trunk_id/concurrency` 的对象数组，重复线路拒绝；非负整数 `concurrency` 的 0 表示该任务不使用该线路。选线及 SQLite 原子占用同时遵守租户、任务总量、线路总量、任务在线路上的上限，未知执行不释放；选线后的业务配置变化拒绝原选择；2026-10-09 起以实际配置摘要而非外部任务修订号判断，来源修订号/状态元数据变化不单独拒绝。占用仍由现有 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 中与生成字段重名的键忽略（包括同值、不同值或 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`。 diff --git a/cmd/sip-go-agent/dispatcher_integration_test.go b/cmd/sip-go-agent/dispatcher_integration_test.go index ddbcf0f..045e2b5 100644 --- a/cmd/sip-go-agent/dispatcher_integration_test.go +++ b/cmd/sip-go-agent/dispatcher_integration_test.go @@ -333,11 +333,15 @@ func TestDispatcherCommandStartsWithIsolatedMQHTTPAndAgent(t *testing.T) { cancel() t.Fatal(err) } + sourceID := "numberID-command-0000123/Case" + commandBody = []byte(strings.ReplaceAll(string(commandBody), `"call-example"`, fmt.Sprintf("%q", sourceID))) if err := admin.PublishWithContext(process, taskRoute.Exchange, taskRoute.BindingKey, true, false, amqp.Publishing{ContentType: "application/json", DeliveryMode: amqp.Persistent, Body: commandBody}); err != nil { cancel() t.Fatal(err) } resultDeadline := time.After(12 * time.Second) + sawExecuteReceipt := false + sawFinalResult := false for { message, ok, err := admin.Get(result.Queue, false) if err != nil { @@ -346,6 +350,7 @@ func TestDispatcherCommandStartsWithIsolatedMQHTTPAndAgent(t *testing.T) { } if ok { var event struct { + EventID string `json:"event_id"` EventType string `json:"event_type"` Payload struct { Outcome string `json:"outcome"` @@ -362,6 +367,13 @@ func TestDispatcherCommandStartsWithIsolatedMQHTTPAndAgent(t *testing.T) { cancel() t.Fatalf("isolated result is invalid JSON: %v", err) } + if event.EventID != sourceID || message.MessageId != sourceID { + cancel() + t.Fatalf("isolated Agent/Dispatcher feedback changed source event identity: type=%q id=%q message_id=%q", event.EventType, event.EventID, message.MessageId) + } + if event.EventType == "call.execute" { + sawExecuteReceipt = true + } if err := message.Ack(false); err != nil { cancel() t.Fatalf("isolated result cannot be ACKed: %v", err) @@ -375,6 +387,9 @@ func TestDispatcherCommandStartsWithIsolatedMQHTTPAndAgent(t *testing.T) { cancel() t.Fatalf("isolated result WAV metadata does not match the single Mock PUT: size=%d format=%q", event.Payload.Recording.SizeBytes, event.Payload.Recording.Format) } + sawFinalResult = true + } + if sawExecuteReceipt && sawFinalResult { break } } diff --git a/contracts/schema b/contracts/schema index e750fe4..ef3009d 160000 --- a/contracts/schema +++ b/contracts/schema @@ -1 +1 @@ -Subproject commit e750fe4618e6472821ceaa74f6c6227d11f998e3 +Subproject commit ef3009df3a90f6e941ca0b24e46134aef17a374d diff --git a/deploys/test/saas-mock/publish_test.go b/deploys/test/saas-mock/publish_test.go index 667d413..90b356d 100644 --- a/deploys/test/saas-mock/publish_test.go +++ b/deploys/test/saas-mock/publish_test.go @@ -98,8 +98,8 @@ func TestOneShotResultMustMatchSourceIdentityAndPersistBeforeReceipt(t *testing. if err != nil { t.Fatal(err) } - sourceID := "event-once-1" - resultID := expectedResultEventID(data.dispatcherID, sourceID) + sourceID := "numberID-once-0000123/Case" + resultID := sourceID result := map[string]any{} example, err := os.ReadFile("../../../contracts/schema/examples/mq-result-uploaded.json") if err != nil { @@ -169,7 +169,7 @@ func TestSaaSMockAwaitsExactResultBeforeStoppingCapture(t *testing.T) { if err := json.Unmarshal(result, &message); err != nil { t.Fatal(err) } - message["event_id"] = expectedResultEventID(data.dispatcherID, eventID) + message["event_id"] = eventID message["dispatcher_id"] = data.dispatcherID payload := message["payload"].(map[string]any) payload["task_id"], payload["callee"], payload["trunk_id"] = "task-full", "15003164745", "trunk-mock" diff --git a/deploys/test/saas-mock/result_wait.go b/deploys/test/saas-mock/result_wait.go index 2217917..e3d228c 100644 --- a/deploys/test/saas-mock/result_wait.go +++ b/deploys/test/saas-mock/result_wait.go @@ -24,11 +24,6 @@ type oneShotResult struct { BodySHA256 string } -func expectedResultEventID(dispatcherID, sourceID string) string { - identity := sha256.Sum256([]byte("call.execute.result\x00" + dispatcherID + "\x00" + sourceID)) - return fmt.Sprintf("result-%x", identity[:]) -} - func verifyOneShotReceipt(body []byte, dispatcherID string, tenantID int64, sourceID string) (string, error) { if err := contract.ValidateCurrent("mq", body); err != nil { return "", errors.New("call dispatch receipt violates the current MQ contract") @@ -67,7 +62,7 @@ func summarizeOneShotResult(body []byte, dispatcherID, sourceID, taskID, callee, } `json:"recording"` } `json:"payload"` } - if err := json.Unmarshal(body, &event); err != nil || event.EventType != "call.execute.result" || event.EventID != expectedResultEventID(dispatcherID, sourceID) || event.DispatcherID != dispatcherID || event.Payload.TaskID != taskID || event.Payload.Callee != callee || event.Payload.TrunkID != trunkID { + if err := json.Unmarshal(body, &event); err != nil || event.EventType != "call.execute.result" || event.EventID != sourceID || event.DispatcherID != dispatcherID || event.Payload.TaskID != taskID || event.Payload.Callee != callee || event.Payload.TrunkID != trunkID { return oneShotResult{}, errors.New("final MQ result does not belong to the selected one-shot call") } sha := sha256.Sum256(body) diff --git a/docs/contracts.md b/docs/contracts.md index 536a09a..3546807 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -7,6 +7,8 @@ SaaS 开发人员与本项目共同维护业务规范、Schema、MQ 拓扑和正 本项目的 `contracts/schema` 是 Git submodule，只记录消费的确定提交，不维护另一份本地定义。 原 `contracts/local` 已移除；`docs/thirds/saas-dispatcher.md` 只提供导航。 内部 `proto/agent/agent.proto` 仍属于本项目，不移入 SaaS 业务契约。 + +2026-10-10 原 `event_id` 全链路反馈规则已同步至共享提交 `ef3009d`；本项目变更、核验和 SaaS 开发同步说明见 [event_id 生命周期核验](evidence/event-id-lifecycle-20261010.md)。具体消息定义仍只以共享合同为准；此次本地改动未部署，不代表线上反馈已切换。 任务控制回执及逐字段拒绝反馈规则统一见共享仓库的 `mq-task-control-receipt.schema.json`、`saas-dispatcher.md` 和 `examples/mq-control-{ack,rejected}.json`；SaaS 须同步采用这些入口，不以历史归档提案或旧的仅 status 回执作为现行结构。 | 位置 | 用途 | diff --git a/docs/evidence/event-id-lifecycle-20261010.md b/docs/evidence/event-id-lifecycle-20261010.md new file mode 100644 index 0000000..1a93b93 --- /dev/null +++ b/docs/evidence/event-id-lifecycle-20261010.md @@ -0,0 +1,82 @@ +# event_id 生命周期一致性：本地修复与核验 + +## 范围与来源 + +使用者确认：所有有源事件的反馈原样保留下发 `event_id`，包括其中的 numberID；其他事件同步检查。不修改真实 SaaS 代码，不部署、不拨号、不调用真实 AI/OSS，不迁移、重置或清理现存业务数据。 + +- 项目基线：`b41dc76`；独立分支：`fix/event-id-lifecycle`。 +- 开始前工作区干净，已 fetch 并快进核对；合同更新由 `make contracts-update` 获取共享 `main`。 +- 共享合同：`ef3009df3a90f6e941ca0b24e46134aef17a374d`，已推送并核对远端 `main`；本项目固定对应 submodule 指针。 +- 唯一业务定义仍在 [`contracts/schema/saas-dispatcher.md`](../../contracts/schema/saas-dispatcher.md) 及其 Schema。本记录只说明改动与证据，不维护另一份业务规范。 + +## 全链路逐项核对 + +| 路径 | 核对结论与处理 | +| --- | --- | +| MQ `call.execute` 接收、inbox、选线、冻结执行与 Agent `CallSpec` | 已保留原 ID，无需改号；重复或重启不重新发起呼叫。带 numberID 的特殊大小写、前缀、标点、Unicode 和前后空格原值均通过测试。 | +| `call.execute` 派发/拒绝回执 | 已保留原 ID；回执查询限定 `call.execute`，不再可能读到相同 ID 的最终结果。 | +| `call.execute.result` 无录音及已上传录音 | 原实现将源 ID 哈希为 `result-…`；已删除该生成函数，结果原样使用源 ID，首次结果正文保持不可变。 | +| 外发保存、查询、确认、重发与恢复 | 保存身份改为 Dispatcher、事件类型及原 ID；两种反馈分别确认，一种反馈的确认不会吞掉另一种。最终结果重发和重启恢复保持原 ID 与原正文，矛盾内容仍拒绝。 | +| 录音授权及上传结果 | 录音查询只判断对应 `call.execute.result` 是否存在；已发送派发回执不会阻断录音授权，真正最终结果存在后仍拒绝新建录音。原上传身份、目标与未知 PUT 屏障未变。 | +| 任务 start/pause/resume/stop/edit 回执 | 已保留原 ID；保存和重复回执查询、重新排队明确限定 `task.control`，保留原处理状态。 | +| `call.execute` / `task.control` / `sip.config` 结构拒绝及拒绝重放 | 已保留可核实的原 ID；按事件类型查询及重新排队，不重新排队相同 ID 的其他反馈。缺少有效身份时仍不伪造 ID。 | +| SIP 通知持久来源与应用完成 | 原通知身份已原样保存；成功加载的完成标记不额外生成业务反馈事件，不新增未定义的成功回执。 | +| Agent 结束回报、录音、结果日记及重启恢复 | `SourceEventID` 已保持原值；结束、上传、结果等 RPC 动作编号仍独立，动作编号不替换源事件 ID。 | +| 本项目测试专用 SaaS Mock 单通结果等待 | 发现旧 `expectedResultEventID` 哈希判定；已删除，直接核对原 ID。它是隔离测试辅助工具，不是修改真实 SaaS 业务代码。 | +| 文件名、录音对象键、上传身份、会话和请求身份 | 不等同于业务 event_id；其独立身份与必要的文件名摘要保留，不改变源事件字段，也不做无关重构。 | + +活动代码中的两个旧派生 ID 函数及其算法引用均已删除。历史证据、归档及独立负例保留原意，不作为运行回退或兼容规则。 + +## 保存结构与现存数据 + +SQLite 布局由 3 改为 **4**，`dispatcher_outbox` 主键为 `(dispatcher_id,event_type,event_id)`。启动先验证版本及这三个主键字段；旧布局明确报错且拒绝打开，不自动迁移、清库或用新空目录绕过。 + +`TestSourceEventIdentityRejectsVersionThreeWithoutWriting` 核实拒绝后旧版本号、表结构和既有反馈正文没有变化。本次仅在临时测试数据库中验证；未打开或改写线上数据库、Agent 恢复文件、录音或其他业务数据。**当前运行环境尚未切换此版本；不能直接用旧布局启动新版。** 后续若需要使用现存数据，必须另行确认数据处置方案，不能把本地通过当作迁移授权。 + +## TDD 与验证 + +原实现上先运行 `TestSourceEventIdentity*`：源 ID 被哈希、相同 ID 的多类型反馈冲突及旧布局误接纳均实际失败。测试专用 Mock 的原 ID 结果测试亦先失败，再修复验证器。日志留在未提交的 `.local/event-id-lifecycle/{red-tests,mock-red-tests}.log`。 + +| 项目 | 本次结果 | +| --- | --- | +| 原 ID、不同类型同 ID、独立确认、反序确认、错误类型确认 | `TestSourceEventIdentitySurvivesResultAndRestart`、`TestSourceEventIdentityAllowsIndependentFeedback`、`TestSourceEventIdentityConfirmationCannotConsumeAnotherType` 通过。 | +| 录音、最终结果和拒绝重放隔离 | `TestSourceEventIdentityUploadedResultRetainsOriginal`、`TestSourceEventIdentityUploadUsesOnlyFinalResult`、`TestSourceEventIdentityRejectionReplayOnlyRequeuesItsType` 通过。 | +| 确认丢失、不重复执行、原正文重发 | `TestOriginalEventIDSurvivesExecuteResultAndLostPublishConfirmation`、`TestExecuteTimeoutAndMQFailureDoNotRedial`、`TestFinalResultRestartResumesSameOutboxBody` 通过。 | +| 结束早于派发回执、矛盾内容、事务回滚 | `TestFinalResultBeforeOriginateAckDoesNotCreateSuccessReceipt`、`TestFinalResultRejectsChangedFactsAndUnboundRecording`、`TestFinalResultOutboxFailureRollsBackAndCanResume` 通过。 | +| 合同正常示例关联 | `TestCurrentCallExamplesRetainOriginalEventIdentity` 验证命令、派发回执及两种最终结果使用相同 ID；全部正反例通过。 | +| 实际隔离 RabbitMQ 结果收件 | `TestRuntimeIsolatedControlBacklogExecuteAndSharedResult` 实际核对同 ID 的回执/结果、AMQP MessageId 与正文；重复命令不重拨。 | +| 实际隔离 Agent/Dispatcher、双向 TLS 与 HTTPS 上传 | `TestDispatcherCommandStartsWithIsolatedMQHTTPAndAgent` 使用带 numberID 的原 ID，实际从共享队列读取回执及上传后的最终结果；一次模拟 HTTPS PUT，重启后重复命令不执行第二通。允许两种反馈乱序，分别核验收件。 | +| 全量本地检查 | `make check` 通过：格式、合同与 Proto 来源检查、`go vet ./...`、`go test -race ./...`、构建和实际隔离 MQ/HTTPS/mTLS 链路。Python 26 项通过。 | +| 业务单元覆盖率 | `make coverage` 通过；排除生成代码后的覆盖率 **72.8%**，不以集成测试提高该单元覆盖率。 | + +最终检查日志：`.local/event-id-lifecycle/make-check-final.log`；覆盖率日志与原始 profile：同目录 `coverage-final.log`、`coverage.out`、`coverage-business.out`。这些是本地测试材料，不是线上证据。 + +## A01–A12 / K01–K16 本次对照 + +历史清单见 [P08 本地验收](saas-dispatcher-p08-acceptance.md)；其旧合同位置、旧修订接纳规则和历史结果不代替当前共享合同。 + +| 编号 | 本次影响及核对 | +| --- | --- | +| A01/A04 | 直接影响事件身份语义与共享示例；共享提交、submodule、Schema 正反例与 Proto 检查通过，未改变 Proto。 | +| A02/A03 | 不新增运行入口或通信协议代次，MQ/HTTP 名称与拓扑不变；当前拓扑测试及实际隔离 MQ 回归通过。 | +| A05/A07 | 配置来源、关准入、任务控制和 SIP 屏障不放宽；既有全量测试通过，重复控制回执按类型独立保存。 | +| A06/A08 | 直接影响反馈确认；原消息身份重发和重复命令不重拨验证通过，未知执行及占用规则不变。 | +| A09/A10 | 直接影响最终结果、录音查询及重启恢复；上述结果/录音测试通过，旧布局拒绝且不改数据。 | +| A11/A12 | TDD、完整检查与 72.8% 覆盖率通过；仅本地，不扩大部署或真实业务授权。 | +| K01 | 拓扑与队列所有权不变，实际隔离共享队列收件通过。 | +| K02/K06 | 直接影响身份关联；原 ID 从执行到不同类型反馈一致，派发回执不等于通话结果。 | +| K03/K04/K05/K07 | 分页、停止不可恢复、暂停处理和 SIP 加载屏障不变；全量测试通过。 | +| K08/K09 | AI 授权与关键词行为未变；全量测试通过，未发真实供应商请求。 | +| K10 | 恢复间隔、48 小时起点与到期保留不变；录音失败/未知 PUT、确认丢失及重启测试通过。 | +| K11/K12/K13 | SIP 事实、外呼规则及单号拒绝语义不变；全量测试通过，不编造 SIP 码或放开授权。 | +| K14/K15/K16 | 通话结果、录音生成失败及双文件保存失败语义不变；相关测试通过，不伪造结果、不自动释放未知占用。 | + +## SaaS 开发同步说明 + +共享合同 `main` 已包含本次变更，入口是 `ef3009d` 及正常 MQ 示例。最终结果返回下发的原 `event_id`，因此源 ID 中的 numberID 无需另加字段或解析哈希即可保留；派发回执与最终结果仍是不同事件类型。字段范围没有增加旧执行字段或兼容通道。 + +本说明用于交给 SaaS 开发人员同步现行合同；**尚无对方已读取、应用已修改或线上已收到新结果的事实，不代签这些事项。** 当前项目源码仅在本地分支提交，不推送本项目远端、不合入 main。 + +## 发布检查补充 + +固定源码提交后的本地发布检查结果另行记录；不代表部署或生产批准。 diff --git a/internal/contract/event_identity_test.go b/internal/contract/event_identity_test.go new file mode 100644 index 0000000..aa69906 --- /dev/null +++ b/internal/contract/event_identity_test.go @@ -0,0 +1,33 @@ +package contract + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" +) + +func TestCurrentCallExamplesRetainOriginalEventIdentity(t *testing.T) { + var originalID string + for _, name := range []string{"mq-execute", "mq-execute-ack", "mq-result-no-recording", "mq-result-uploaded"} { + raw, err := os.ReadFile(filepath.Join("..", "..", "contracts", "schema", "examples", name+".json")) + if err != nil { + t.Fatal(err) + } + if err := ValidateCurrent("mq", raw); err != nil { + t.Fatalf("%s: %v", name, err) + } + var event struct { + EventID string `json:"event_id"` + } + if err := json.Unmarshal(raw, &event); err != nil { + t.Fatal(err) + } + if originalID == "" { + originalID = event.EventID + } + if event.EventID != originalID { + t.Fatalf("%s changed source identity: want=%q got=%q", name, originalID, event.EventID) + } + } +} diff --git a/internal/dispatcher/control_test.go b/internal/dispatcher/control_test.go index ea5ce7e..e6e326a 100644 --- a/internal/dispatcher/control_test.go +++ b/internal/dispatcher/control_test.go @@ -123,7 +123,7 @@ func TestControlPauseDefaultHangupAckAfterAgentDispatch(t *testing.T) { if err != nil || len(outbox) != 1 || outbox[0].EventType != "task.control" || !strings.Contains(string(outbox[0].Body), `"status":"applied"`) { t.Fatalf("no durable applied ack: %+v %v", outbox, err) } - if err := s.MarkOutboxConfirmed(controller.DispatcherID, "pause-1"); err != nil { + if err := s.MarkOutboxConfirmed(controller.DispatcherID, "task.control", "pause-1"); err != nil { t.Fatal(err) } agent.before = nil diff --git a/internal/dispatcher/event_identity_test.go b/internal/dispatcher/event_identity_test.go new file mode 100644 index 0000000..4883ed8 --- /dev/null +++ b/internal/dispatcher/event_identity_test.go @@ -0,0 +1,74 @@ +package dispatcher + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "testing" +) + +func TestOriginalEventIDSurvivesExecuteResultAndLostPublishConfirmation(t *testing.T) { + controller, originator, publisher, s := newExecuteFixture(t) + id := "numberID-0000123/Case" + command := executeBody(t, id, "15003164745") + if err := controller.ProcessExecute(context.Background(), command); err != nil { + t.Fatal(err) + } + if len(originator.calls) != 1 || originator.calls[0].EventID != id { + t.Fatalf("execute changed source identity: %+v", originator.calls) + } + if err := controller.FlushOutbox(context.Background()); err != nil { + t.Fatal(err) + } + if err := s.FinishExecute(controller.DispatcherID, id); err != nil { + t.Fatal(err) + } + var example struct { + Payload json.RawMessage `json:"payload"` + } + if err := json.Unmarshal(configExample(t, "mq-result-no-recording"), &example); err != nil { + t.Fatal(err) + } + result, created, err := s.RecordCallResult(controller.DispatcherID, id, example.Payload) + if err != nil || !created || result.EventID != id { + t.Fatalf("result lost source identity: %+v created=%t err=%v", result, created, err) + } + publisher.err = errors.New("injected confirm loss after publishing original result") + if err := controller.FlushOutbox(context.Background()); err == nil { + t.Fatal("confirmation loss hidden") + } + pending, err := s.ListPendingOutbox(controller.DispatcherID) + if err != nil || len(pending) != 1 || pending[0].EventType != "call.execute.result" || pending[0].EventID != id { + t.Fatalf("result lost or execute confirmation undone: %+v %v", pending, err) + } + if err := controller.ProcessExecute(context.Background(), command); err != nil || len(originator.calls) != 1 { + t.Fatalf("MQ redelivery redialed: %d %v", len(originator.calls), err) + } + publisher.err = nil + if err := controller.FlushOutbox(context.Background()); err != nil { + t.Fatal(err) + } + if len(publisher.bodies) != 3 || !bytes.Equal(publisher.bodies[1], publisher.bodies[2]) { + t.Fatalf("result retry changed message: %d publishes", len(publisher.bodies)) + } + for i, body := range publisher.bodies { + var event struct { + EventID string `json:"event_id"` + EventType string `json:"event_type"` + } + if err := json.Unmarshal(body, &event); err != nil || event.EventID != id { + t.Fatalf("publish %d changed source ID: %+v %v", i, event, err) + } + want := "call.execute.result" + if i == 0 { + want = "call.execute" + } + if event.EventType != want { + t.Fatalf("publish %d confused feedback type: %+v", i, event) + } + } + if pending, err := s.ListPendingOutbox(controller.DispatcherID); err != nil || len(pending) != 0 { + t.Fatalf("confirmed result still pending: %+v %v", pending, err) + } +} diff --git a/internal/dispatcher/execute.go b/internal/dispatcher/execute.go index 9486c95..58a75a5 100644 --- a/internal/dispatcher/execute.go +++ b/internal/dispatcher/execute.go @@ -314,11 +314,11 @@ func (c *ExecuteController) FlushOutbox(ctx context.Context) error { var failures []error for _, event := range pending { if err := c.Publisher.Publish(ctx, "agent-call.saas.v1", event.RoutingKey, event.Body); err != nil { - failures = append(failures, fmt.Errorf("publish event %q to SaaS queue: %w", event.EventID, err)) + failures = append(failures, fmt.Errorf("publish event %q/%q to SaaS queue: %w", event.EventType, event.EventID, err)) continue } - if err := c.Store.MarkOutboxConfirmed(c.DispatcherID, event.EventID); err != nil { - failures = append(failures, fmt.Errorf("persist queue confirmation for %q: %w", event.EventID, err)) + if err := c.Store.MarkOutboxConfirmed(c.DispatcherID, event.EventType, event.EventID); err != nil { + failures = append(failures, fmt.Errorf("persist queue confirmation for %q/%q: %w", event.EventType, event.EventID, err)) } } return errors.Join(failures...) diff --git a/internal/dispatcher/runtime_integration_test.go b/internal/dispatcher/runtime_integration_test.go index 2a960ad..d4804db 100644 --- a/internal/dispatcher/runtime_integration_test.go +++ b/internal/dispatcher/runtime_integration_test.go @@ -114,7 +114,8 @@ func TestRuntimeIsolatedControlBacklogExecuteAndSharedResult(t *testing.T) { } } publish(controlRoute, controlBody(t, "control-example", "pause", "drain")) - publish(taskRoute, executeBody(t, "call-example", "15803300952")) + sourceID := "numberID-runtime-0000123/Case" + publish(taskRoute, executeBody(t, sourceID, "15803300952")) snapshot := policySnapshot(t) sipJSON, err := json.Marshal(snapshot.SIP) @@ -239,7 +240,7 @@ func TestRuntimeIsolatedControlBacklogExecuteAndSharedResult(t *testing.T) { } select { case spec := <-agent.calls: - if spec.EventID != "call-example" || spec.TrunkID != "trunk-mock" || spec.Callee != "15803300952" { + if spec.EventID != sourceID || spec.TrunkID != "trunk-mock" || spec.Callee != "15803300952" { t.Fatalf("wrong SaaS event callee or dispatch: %+v", spec) } case <-finished: @@ -278,10 +279,59 @@ func TestRuntimeIsolatedControlBacklogExecuteAndSharedResult(t *testing.T) { t.Fatal(err) } } - if seen["control-example"] != "applied" || seen["resume-integration"] != "applied" || seen["call-example"] != "dispatched" { + if seen["control-example"] != "applied" || seen["resume-integration"] != "applied" || seen[sourceID] != "dispatched" { t.Fatalf("wrong shared SaaS results: %+v", seen) } - publish(taskRoute, executeBody(t, "call-example", "15803300952")) + // Complete this isolated Mock call and inspect the actual final result in + // the shared RabbitMQ queue: the already-confirmed dispatch receipt and + // final result must have exactly the same source identity. + if err := db.FinishExecute(id, sourceID); err != nil { + t.Fatal(err) + } + var resultExample struct { + Payload map[string]any `json:"payload"` + } + if err := json.Unmarshal(configExample(t, "mq-result-no-recording"), &resultExample); err != nil { + t.Fatal(err) + } + resultExample.Payload["callee"] = "15803300952" + resultPayload, err := json.Marshal(resultExample.Payload) + if err != nil { + t.Fatal(err) + } + originalResult, created, err := db.RecordCallResult(id, sourceID, resultPayload) + if err != nil || !created || originalResult.EventID != sourceID { + t.Fatalf("final result changed original identity: %+v %t %v", originalResult, created, err) + } + resultDeadline := time.After(5 * time.Second) + for { + msg, ok, err := admin.Get(shared, false) + if err != nil { + t.Fatal(err) + } + if ok { + var event struct { + EventID string `json:"event_id"` + EventType string `json:"event_type"` + } + if err := json.Unmarshal(msg.Body, &event); err != nil || event.EventID != sourceID || event.EventType != "call.execute.result" || msg.MessageId != sourceID || msg.DeliveryMode != amqp.Persistent || string(msg.Body) != string(originalResult.Body) { + t.Fatalf("actual MQ result did not retain original identity/body: %+v %v", event, err) + } + if err := msg.Ack(false); err != nil { + t.Fatal(err) + } + break + } + select { + case <-resultDeadline: + t.Fatal("same-ID final result never reached shared queue") + case <-time.After(20 * time.Millisecond): + } + } + if _, created, err := db.RecordCallResult(id, sourceID, resultPayload); err != nil || created { + t.Fatalf("result duplicate became new output: %t %v", created, err) + } + publish(taskRoute, executeBody(t, sourceID, "15803300952")) select { case spec := <-agent.calls: t.Fatalf("redelivery reoriginated call %s", spec.EventID) diff --git a/internal/mq/broker_integration_test.go b/internal/mq/broker_integration_test.go index 47488a7..1f4c6bf 100644 --- a/internal/mq/broker_integration_test.go +++ b/internal/mq/broker_integration_test.go @@ -243,7 +243,7 @@ func TestBrokerSharedResultQueueAndNoConfigure(t *testing.T) { // A dead connection must not claim delivery. An explicit new session can // publish the unchanged pending identity to the SaaS-owned shared queue. route, _ := tenant.ResultRoute(ids[0]) - pending := []byte(strings.Replace(string(resultBody), `"event_id":"call-result-example"`, `"event_id":"call-result-reopened"`, 1)) + pending := []byte(strings.Replace(string(resultBody), `"event_id":"call-example"`, `"event_id":"numberID-reopened-0000123"`, 1)) if err := brokers[0].Close(); err != nil { t.Fatal(err) } @@ -260,7 +260,7 @@ func TestBrokerSharedResultQueueAndNoConfigure(t *testing.T) { t.Fatalf("original pending identity was not delivered after reconnect: %v", err) } delivery, ok, err := admin.Get(shared, false) - if err != nil || !ok || delivery.MessageId != "call-result-reopened" || string(delivery.Body) != string(pending) || delivery.DeliveryMode != amqp.Persistent { + if err != nil || !ok || delivery.MessageId != "numberID-reopened-0000123" || string(delivery.Body) != string(pending) || delivery.DeliveryMode != amqp.Persistent { t.Fatalf("reconnected result was not received by the SaaS-owned queue with its original identity: ok=%t err=%v", ok, err) } if err := delivery.Ack(false); err != nil { @@ -365,7 +365,7 @@ func TestBrokerSharedResultQueueAndNoConfigure(t *testing.T) { if err := applied.Ack(false); err != nil { t.Fatal(err) } - if err := s.MarkOutboxConfirmed(ids[0], "control-applied-fields"); err != nil { + if err := s.MarkOutboxConfirmed(ids[0], "task.control", "control-applied-fields"); err != nil { t.Fatal(err) } for _, id := range ids { diff --git a/internal/store/calls.go b/internal/store/calls.go index 642996a..e55680a 100644 --- a/internal/store/calls.go +++ b/internal/store/calls.go @@ -418,7 +418,7 @@ func insertExecuteAck(tx *sql.Tx, dispatcherID, eventID string, tenantID int64, func requireExecuteAck(tx *sql.Tx, dispatcherID, eventID string) error { var eventType string - if err := tx.QueryRow(`SELECT event_type FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&eventType); err != nil { + if err := tx.QueryRow(`SELECT event_type FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='call.execute' AND event_id=?`, dispatcherID, eventID).Scan(&eventType); err != nil { return fmt.Errorf("confirmed call %q has no durable acknowledgment: %w", eventID, err) } if eventType != "call.execute" { @@ -520,8 +520,8 @@ func (s *Store) ListPendingOutbox(dispatcherID string) ([]OutboxEvent, error) { // Confirm means delivery to a bound queue, not application receipt. The row // remains durable and is never deleted without separate explicit evidence. -func (s *Store) MarkOutboxConfirmed(dispatcherID, eventID string) error { - result, err := s.db.Exec(`UPDATE dispatcher_outbox SET confirmed=1,confirmed_at=? WHERE dispatcher_id=? AND event_id=? AND confirmed=0`, time.Now().UTC().Format(time.RFC3339Nano), dispatcherID, eventID) +func (s *Store) MarkOutboxConfirmed(dispatcherID, eventType, eventID string) error { + result, err := s.db.Exec(`UPDATE dispatcher_outbox SET confirmed=1,confirmed_at=? WHERE dispatcher_id=? AND event_type=? AND event_id=? AND confirmed=0`, time.Now().UTC().Format(time.RFC3339Nano), dispatcherID, eventType, eventID) if err != nil { return fmt.Errorf("persist MQ queue confirm: %w", err) } diff --git a/internal/store/calls_test.go b/internal/store/calls_test.go index 9149f6c..f1c21d4 100644 --- a/internal/store/calls_test.go +++ b/internal/store/calls_test.go @@ -102,7 +102,7 @@ func TestExecuteRejectsOnlyOnceWithoutFinalResult(t *testing.T) { if err := s.RejectExecute(cmd.DispatcherID, cmd.EventID, "callee not permitted"); err == nil { t.Fatal("duplicate rejection created another outbox event") } - if err := s.MarkOutboxConfirmed(currentDispatcherID, outbox[0].EventID); err != nil { + if err := s.MarkOutboxConfirmed(currentDispatcherID, outbox[0].EventType, outbox[0].EventID); err != nil { t.Fatal(err) } if pending, err := s.ListPendingOutbox(currentDispatcherID); err != nil || len(pending) != 0 { diff --git a/internal/store/control_ack_test.go b/internal/store/control_ack_test.go index 567bdeb..ec69b36 100644 --- a/internal/store/control_ack_test.go +++ b/internal/store/control_ack_test.go @@ -23,7 +23,7 @@ func TestControlAckOnlyAfterDispatchAndInSameTransaction(t *testing.T) { if err != nil || len(outbox) != 1 || outbox[0].EventType != "task.control" || !strings.Contains(string(outbox[0].Body), `"status":"applied"`) { t.Fatalf("missing applied control ack: %+v %v", outbox, err) } - if err := s.MarkOutboxConfirmed(currentDispatcherID, "pause-event-1"); err != nil { + if err := s.MarkOutboxConfirmed(currentDispatcherID, "task.control", "pause-event-1"); err != nil { t.Fatal(err) } // A repeated control must be dispatched again, not suppressed by event ID. diff --git a/internal/store/control_flow.go b/internal/store/control_flow.go index f2cabf2..be0b5ad 100644 --- a/internal/store/control_flow.go +++ b/internal/store/control_flow.go @@ -231,7 +231,7 @@ func enqueueControlAck(tx *sql.Tx, dispatcherID string, tenantID int64, eventID, // A retransmission reports the original state, not a later task state. var oldType string var oldBody []byte - err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&oldType, &oldBody) + err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='task.control' AND event_id=?`, dispatcherID, eventID).Scan(&oldType, &oldBody) if err != nil && !errors.Is(err, sql.ErrNoRows) { return fmt.Errorf("inspect existing task control outbox: %w", err) } @@ -266,7 +266,7 @@ func enqueueControlAck(tx *sql.Tx, dispatcherID string, tenantID int64, eventID, if oldType != "task.control" || string(oldBody) != string(body) { return fmt.Errorf("control event identity %q conflicts with another outbound result", eventID) } - if _, err := tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID); err != nil { + if _, err := tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_type='task.control' AND event_id=?`, dispatcherID, eventID); err != nil { return fmt.Errorf("requeue repeated control acknowledgment: %w", err) } return nil diff --git a/internal/store/control_receipt_test.go b/internal/store/control_receipt_test.go index ddbcaeb..c3ba2dc 100644 --- a/internal/store/control_receipt_test.go +++ b/internal/store/control_receipt_test.go @@ -180,7 +180,7 @@ func TestRejectedControlReplayPreservesOriginalTaskState(t *testing.T) { if err := s.db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE event_id='rejected-repeat'`).Scan(&original); err != nil { t.Fatal(err) } - if err := s.MarkOutboxConfirmed(currentDispatcherID, "rejected-repeat"); err != nil { + if err := s.MarkOutboxConfirmed(currentDispatcherID, "task.control", "rejected-repeat"); err != nil { t.Fatal(err) } if _, err := s.db.Exec(`UPDATE dispatcher_tasks SET control_state='stopped' WHERE dispatcher_id=? AND tenant_id=1001 AND task_id='task-asr'`, currentDispatcherID); err != nil { @@ -221,7 +221,7 @@ func TestEditReceiptRetransmitsOriginalStateAfterTaskChanges(t *testing.T) { if err := s.db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE event_id='edit-repeat'`).Scan(&original); err != nil { t.Fatal(err) } - if err := s.MarkOutboxConfirmed(currentDispatcherID, "edit-repeat"); err != nil { + if err := s.MarkOutboxConfirmed(currentDispatcherID, "task.control", "edit-repeat"); err != nil { t.Fatal(err) } if err := s.PrepareControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil { diff --git a/internal/store/event_identity_test.go b/internal/store/event_identity_test.go new file mode 100644 index 0000000..2033333 --- /dev/null +++ b/internal/store/event_identity_test.go @@ -0,0 +1,245 @@ +package store + +import ( + "bytes" + "database/sql" + "encoding/json" + "errors" + "testing" + + "git.ipao.vip/rogee/go-sip/internal/contract" +) + +func TestSourceEventIdentitySurvivesResultAndRestart(t *testing.T) { + for _, id := range []string{"call-numberID-0000123", "result-numberID-0000123", "NumberID:0000123/Case", " 接收-numberID-0000123 "} { + t.Run(id, func(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentResultCall(t, s, id, true) + result, created, err := s.RecordCallResult(cmd.DispatcherID, id, currentResultPayload(t)) + if err != nil || !created || result.EventID != id { + t.Fatalf("source event ID changed: want=%q got=%q created=%t err=%v", id, result.EventID, created, err) + } + pending, err := s.ListPendingOutbox(cmd.DispatcherID) + if err != nil || len(pending) != 2 { + t.Fatalf("ack/result missing: %+v %v", pending, err) + } + for _, event := range pending { + var envelope struct { + EventID string `json:"event_id"` + } + if err := json.Unmarshal(event.Body, &envelope); err != nil || envelope.EventID != id || event.EventID != id { + t.Fatalf("feedback changed source ID: %+v %v", event, err) + } + } + path := s.path + if err := s.Close(); err != nil { + t.Fatal(err) + } + restarted, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer restarted.Close() + again, created, err := restarted.RecordCallResult(cmd.DispatcherID, id, currentResultPayload(t)) + if err != nil || created || again.EventID != id || !bytes.Equal(again.Body, result.Body) { + t.Fatalf("restart changed original result: %+v created=%t err=%v", again, created, err) + } + if _, created, err := restarted.RecordExecute(cmd); err != nil || created { + t.Fatalf("duplicate command became a new execution: created=%t err=%v", created, err) + } + }) + } +} + +func TestSourceEventIdentityAllowsIndependentFeedback(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentResultCall(t, s, "numberID-shared-0000123", true) + // Reproduce the required result identity independently of the generator so + // the old two-column primary key also fails before implementation changes. + if _, err := s.db.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body) VALUES(?,?,?,?,?)`, cmd.DispatcherID, cmd.EventID, "call.execute.result", "mock.route", []byte(`{"source":"original"}`)); err != nil { + t.Fatalf("different feedback types cannot share source ID: %v", err) + } + if err := s.MarkOutboxConfirmed(cmd.DispatcherID, "call.execute", cmd.EventID); err != nil { + t.Fatal(err) + } + pending, err := s.ListPendingOutbox(cmd.DispatcherID) + if err != nil || len(pending) != 1 || pending[0].EventType != "call.execute.result" { + t.Fatalf("ack confirmation consumed a different feedback: %+v %v", pending, err) + } +} + +func TestSourceEventIdentityConfirmationCannotConsumeAnotherType(t *testing.T) { + for _, first := range []string{"call.execute", "call.execute.result"} { + t.Run(first, func(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentResultCall(t, s, "numberID-confirm-0000123", true) + if _, _, err := s.RecordCallResult(cmd.DispatcherID, cmd.EventID, currentResultPayload(t)); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(cmd.DispatcherID, "task.control", cmd.EventID); err == nil { + t.Fatal("nonexistent feedback falsely confirmed") + } + if pending, err := s.ListPendingOutbox(cmd.DispatcherID); err != nil || len(pending) != 2 { + t.Fatalf("wrong type changed confirmations: %+v %v", pending, err) + } + if err := s.MarkOutboxConfirmed(cmd.DispatcherID, first, cmd.EventID); err != nil { + t.Fatal(err) + } + pending, err := s.ListPendingOutbox(cmd.DispatcherID) + if err != nil || len(pending) != 1 || pending[0].EventType == first || pending[0].EventID != cmd.EventID { + t.Fatalf("one confirmation consumed both feedback types: %+v %v", pending, err) + } + if err := s.MarkOutboxConfirmed(cmd.DispatcherID, first, cmd.EventID); err == nil { + t.Fatal("already-confirmed feedback falsely reported another transition") + } + if err := s.MarkExecuteDispatched(cmd.DispatcherID, cmd.EventID); err != nil { + t.Fatalf("late ack confused final result with execute receipt: %v", err) + } + if pending, err := s.ListPendingOutbox(cmd.DispatcherID); err != nil || len(pending) != 1 { + t.Fatalf("late ack recreated feedback: %+v %v", pending, err) + } + }) + } +} + +func TestSourceEventIdentityUploadedResultRetainsOriginal(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentResultCall(t, s, "numberID-uploaded-0000123", true) + grant := currentUploadBinding(cmd) + if _, _, err := s.BindRecordingUpload(grant); err != nil { + t.Fatal(err) + } + proof := UploadProof{UploadID: grant.UploadID, RecordingID: grant.RecordingID, StatusCode: 200, SizeBytes: grant.SizeBytes, SHA256: grant.ChecksumSHA256} + result, created, err := s.RecordUploadedCallResult(cmd.DispatcherID, cmd.EventID, currentUploadedResultPayload(t, grant), proof) + if err != nil || !created || result.EventID != cmd.EventID { + t.Fatalf("uploaded result changed source identity: %+v created=%t err=%v", result, created, err) + } + if err := s.MarkOutboxConfirmed(cmd.DispatcherID, "call.execute.result", cmd.EventID); err != nil { + t.Fatal(err) + } + again, created, err := s.RecordUploadedCallResult(cmd.DispatcherID, cmd.EventID, currentUploadedResultPayload(t, grant), proof) + if err != nil || created || again.EventID != cmd.EventID || !bytes.Equal(result.Body, again.Body) { + t.Fatalf("upload retry changed original result: %+v created=%t err=%v", again, created, err) + } + if pending, err := s.ListPendingOutbox(cmd.DispatcherID); err != nil || len(pending) != 1 || pending[0].EventType != "call.execute" { + t.Fatalf("uploaded result confirmation consumed execute receipt: %+v %v", pending, err) + } +} + +func TestSourceEventIdentityUploadUsesOnlyFinalResult(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentResultCall(t, s, "numberID-recording-0000123", true) + // A pending or confirmed execute receipt must not deny the first recording. + if err := s.MarkOutboxConfirmed(cmd.DispatcherID, "call.execute", cmd.EventID); err != nil { + t.Fatal(err) + } + grant := currentUploadBinding(cmd) + if _, created, err := s.BindRecordingUpload(grant); err != nil || !created { + t.Fatalf("execute receipt was mistaken for final result: created=%t err=%v", created, err) + } + withoutRecording := currentResultCall(t, s, "numberID-no-recording-0000456", true) + result, _, err := s.RecordCallResult(withoutRecording.DispatcherID, withoutRecording.EventID, currentResultPayload(t)) + if err != nil || result.EventID != withoutRecording.EventID { + t.Fatalf("final source ID changed: %+v %v", result, err) + } + if _, _, err := s.BindRecordingUpload(currentUploadBinding(withoutRecording)); !errors.Is(err, ErrResultConflict) { + t.Fatalf("new recording accepted after final result: %v", err) + } + if _, err := s.LoadRecordingUpload(withoutRecording.DispatcherID, withoutRecording.EventID); !errors.Is(err, ErrUploadNotFound) { + t.Fatalf("refused grant left partial recording: %v", err) + } +} + +func TestSourceEventIdentityRejectionReplayOnlyRequeuesItsType(t *testing.T) { + for _, kind := range []string{"call.execute", "task.control", "sip.config"} { + t.Run(kind, func(t *testing.T) { + s := preparedCurrentCallStore(t) + id := "numberID-rejected-0000123" + raw, err := json.Marshal(map[string]any{"event_id": id, "event_type": kind, "dispatcher_id": currentDispatcherID, "tenant_id": 1001, "payload": map[string]any{}}) + if err != nil { + t.Fatal(err) + } + cause := contract.ValidateCurrent("mq", raw) + if cause == nil { + t.Fatal("fixture unexpectedly valid") + } + if err := s.RejectData(currentDispatcherID, raw, cause); err != nil { + t.Fatal(err) + } + pending, err := s.ListPendingOutbox(currentDispatcherID) + if err != nil || len(pending) != 1 || pending[0].EventID != id || pending[0].EventType != kind { + t.Fatalf("rejection changed source identity/type: %+v %v", pending, err) + } + original := bytes.Clone(pending[0].Body) + if err := s.MarkOutboxConfirmed(currentDispatcherID, kind, id); err != nil { + t.Fatal(err) + } + // A different confirmed feedback sharing the identity must stay + // confirmed while the rejected receipt is requested again. + if _, err := s.db.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body,confirmed) VALUES(?,?,?,?,?,1)`, currentDispatcherID, id, "call.execute.result", "mock.route", []byte(`{}`)); err != nil { + t.Fatal(err) + } + for _, replay := range []bool{false, true} { + if replay { + tenantID := int64(1001) + if kind == "sip.config" { + tenantID = 0 + } + if rejected, err := s.ReplayRejectedEvent(currentDispatcherID, id, kind, tenantID); err != nil || !rejected { + t.Fatalf("original rejection not replayed: %t %v", rejected, err) + } + } else if err := s.RejectData(currentDispatcherID, raw, cause); err != nil { + t.Fatal(err) + } + pending, err := s.ListPendingOutbox(currentDispatcherID) + if err != nil || len(pending) != 1 || pending[0].EventID != id || pending[0].EventType != kind || !bytes.Equal(original, pending[0].Body) { + t.Fatalf("rejection requeued another feedback or changed original: %+v %v", pending, err) + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, kind, id); err != nil { + t.Fatal(err) + } + } + }) + } +} + +func TestSourceEventIdentityRejectsVersionThreeWithoutWriting(t *testing.T) { + s := preparedCurrentCallStore(t) + cmd := currentResultCall(t, s, "numberID-old-layout", true) + if _, err := s.db.Exec(`PRAGMA user_version=3`); err != nil { + t.Fatal(err) + } + var schemaBefore string + if err := s.db.QueryRow(`SELECT sql FROM sqlite_master WHERE name='dispatcher_outbox'`).Scan(&schemaBefore); err != nil { + t.Fatal(err) + } + pendingBefore, err := s.ListPendingOutbox(cmd.DispatcherID) + if err != nil { + t.Fatal(err) + } + path := s.path + if err := s.Close(); err != nil { + t.Fatal(err) + } + if reopened, err := Open(path); err == nil { + reopened.Close() + t.Fatal("old layout admitted or silently migrated") + } + db, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + var version int + var schemaAfter string + var body []byte + if err := db.QueryRow(`PRAGMA user_version`).Scan(&version); err != nil || version != 3 { + t.Fatalf("old version changed: %d %v", version, err) + } + if err := db.QueryRow(`SELECT sql FROM sqlite_master WHERE name='dispatcher_outbox'`).Scan(&schemaAfter); err != nil || schemaBefore != schemaAfter { + t.Fatalf("old schema changed: %v", err) + } + if err := db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='call.execute' AND event_id=?`, cmd.DispatcherID, cmd.EventID).Scan(&body); err != nil || !bytes.Equal(body, pendingBefore[0].Body) { + t.Fatalf("old feedback changed: %v", err) + } +} diff --git a/internal/store/rejected_data.go b/internal/store/rejected_data.go index 2399c15..f58c87d 100644 --- a/internal/store/rejected_data.go +++ b/internal/store/rejected_data.go @@ -125,14 +125,14 @@ func (s *Store) RejectData(dispatcherID string, original []byte, cause error) er } var oldType string var old []byte - err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, id.EventID).Scan(&oldType, &old) + err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type=? AND event_id=?`, dispatcherID, id.EventType, id.EventID).Scan(&oldType, &old) if errors.Is(err, sql.ErrNoRows) { _, err = tx.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body) VALUES(?,?,?,?,?)`, dispatcherID, id.EventID, id.EventType, route.BindingKey, body) } else if err == nil { if oldType != id.EventType || !bytes.Equal(old, body) { return errors.New("rejection conflicts with original event result") } - _, err = tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_id=?`, dispatcherID, id.EventID) + _, err = tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_type=? AND event_id=?`, dispatcherID, id.EventType, id.EventID) } if err != nil { return fmt.Errorf("persist rejected-data feedback: %w", err) diff --git a/internal/store/rejection_identity.go b/internal/store/rejection_identity.go index 427bcbb..a4a01ea 100644 --- a/internal/store/rejection_identity.go +++ b/internal/store/rejection_identity.go @@ -28,7 +28,7 @@ func (s *Store) ReplayRejectedEvent(dispatcherID, eventID, eventType string, ten func replayRejectedEvent(tx *sql.Tx, dispatcherID, eventID, eventType string, tenantID int64) (bool, error) { var kind string var body []byte - err := tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&kind, &body) + err := tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type=? AND event_id=?`, dispatcherID, eventType, eventID).Scan(&kind, &body) if errors.Is(err, sql.ErrNoRows) { return false, nil } @@ -53,6 +53,6 @@ func replayRejectedEvent(tx *sql.Tx, dispatcherID, eventID, eventType string, te if receipt.Payload.Status != "rejected" { return false, nil } - _, err = tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID) + _, err = tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_type=? AND event_id=?`, dispatcherID, eventType, eventID) return err == nil, err } diff --git a/internal/store/result.go b/internal/store/result.go index d815808..0406068 100644 --- a/internal/store/result.go +++ b/internal/store/result.go @@ -2,7 +2,6 @@ package store import ( "bytes" - "crypto/sha256" "database/sql" "encoding/json" "errors" @@ -19,11 +18,6 @@ var ErrResultConflict = errors.New("call already has a different final result") var ErrEndUnconfirmed = errors.New("final result requires confirmed call end and its frozen snapshot") var ErrResultInvalid = errors.New("final result input violates the approved contract") -func currentResultEventID(dispatcherID, sourceEventID string) string { - identity := sha256.Sum256([]byte("call.execute.result\x00" + dispatcherID + "\x00" + sourceEventID)) - return fmt.Sprintf("result-%x", identity[:]) -} - // RecordCallResult stores one no-recording final result only after confirmed // call end. A durable upload grant cannot be bypassed with an empty recording. func (s *Store) RecordCallResult(dispatcherID, sourceEventID string, payload []byte) (OutboxEvent, bool, error) { @@ -118,7 +112,7 @@ func (s *Store) recordCallResult(dispatcherID, sourceEventID string, payload []b if err != nil { return OutboxEvent{}, false, err } - eventID := currentResultEventID(dispatcherID, sourceEventID) + eventID := sourceEventID body, err := json.Marshal(struct { EventID string `json:"event_id"` EventType string `json:"event_type"` @@ -133,7 +127,7 @@ func (s *Store) recordCallResult(dispatcherID, sourceEventID string, payload []b if err := contract.ValidateCurrent("mq", body); err != nil { return OutboxEvent{}, false, fmt.Errorf("%w: MQ contract: %v", ErrResultInvalid, err) } - inserted, err := tx.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body) VALUES(?,?,?,?,?) ON CONFLICT(dispatcher_id,event_id) DO NOTHING`, dispatcherID, eventID, "call.execute.result", route.BindingKey, body) + inserted, err := tx.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body) VALUES(?,?,?,?,?) ON CONFLICT(dispatcher_id,event_type,event_id) DO NOTHING`, dispatcherID, eventID, "call.execute.result", route.BindingKey, body) if err != nil { return OutboxEvent{}, false, fmt.Errorf("persist final result outbox: %w", err) } @@ -143,7 +137,7 @@ func (s *Store) recordCallResult(dispatcherID, sourceEventID string, payload []b } var stored OutboxEvent stored.EventID = eventID - if err := tx.QueryRow(`SELECT event_type,routing_key,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&stored.EventType, &stored.RoutingKey, &stored.Body); err != nil { + if err := tx.QueryRow(`SELECT event_type,routing_key,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='call.execute.result' AND event_id=?`, dispatcherID, eventID).Scan(&stored.EventType, &stored.RoutingKey, &stored.Body); err != nil { return OutboxEvent{}, false, fmt.Errorf("inspect existing final result: %w", err) } if stored.EventType != "call.execute.result" || stored.RoutingKey != route.BindingKey || !bytes.Equal(stored.Body, body) { diff --git a/internal/store/result_test.go b/internal/store/result_test.go index 2dc6fc6..84b300f 100644 --- a/internal/store/result_test.go +++ b/internal/store/result_test.go @@ -63,7 +63,7 @@ func TestFinalResultNeedsConfirmedEndAndRemainsExactlyOne(t *testing.T) { t.Fatalf("confirmed end did not release the call before OSS/result delivery: %+v err=%v", occupancy, err) } event, created, err := s.RecordCallResult(cmd.DispatcherID, cmd.EventID, payload) - if err != nil || !created || event.EventType != "call.execute.result" || event.EventID == cmd.EventID { + if err != nil || !created || event.EventType != "call.execute.result" || event.EventID != cmd.EventID { t.Fatalf("exact final result not persisted: event=%+v created=%t err=%v", event, created, err) } route, err := tenant.ResultRoute(cmd.DispatcherID) @@ -78,8 +78,8 @@ func TestFinalResultNeedsConfirmedEndAndRemainsExactlyOne(t *testing.T) { if err != nil || created || !bytes.Equal(repeated.Body, event.Body) { t.Fatalf("redelivered identical result inserted twice: created=%t err=%v", created, err) } - for _, id := range []string{cmd.EventID, event.EventID} { - if err := s.MarkOutboxConfirmed(cmd.DispatcherID, id); err != nil { + for _, feedback := range outbox { + if err := s.MarkOutboxConfirmed(cmd.DispatcherID, feedback.EventType, feedback.EventID); err != nil { t.Fatal(err) } } @@ -124,7 +124,7 @@ func TestFinalResultAfterEndRacedOriginateAck(t *testing.T) { t.Fatalf("late originate response conflicted with the confirmed result: %v", err) } outbox, err := s.ListPendingOutbox(cmd.DispatcherID) - if err != nil || len(outbox) != 2 || outbox[0].EventID != cmd.EventID || outbox[1].EventID != result.EventID { + if err != nil || len(outbox) != 2 || outbox[0].EventID != cmd.EventID || outbox[0].EventType != "call.execute" || outbox[1].EventID != cmd.EventID || outbox[1].EventType != "call.execute.result" || result.EventID != cmd.EventID { t.Fatalf("early end lost its acknowledgment or produced duplicate results: outbox=%+v err=%v", outbox, err) } } @@ -220,7 +220,7 @@ func TestFinalResultRestartResumesSameOutboxBody(t *testing.T) { if err != nil || created || repeated.EventID != original.EventID || !bytes.Equal(repeated.Body, original.Body) { t.Fatalf("retry created a second result after restart: created=%t err=%v", created, err) } - if err := restarted.MarkOutboxConfirmed(cmd.DispatcherID, original.EventID); err != nil { + if err := restarted.MarkOutboxConfirmed(cmd.DispatcherID, original.EventType, original.EventID); err != nil { t.Fatal(err) } pending, err = restarted.ListPendingOutbox(cmd.DispatcherID) diff --git a/internal/store/store.go b/internal/store/store.go index 55a2561..be3837d 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -83,7 +83,7 @@ CREATE TABLE IF NOT EXISTS dispatcher_outbox ( body BLOB NOT NULL, confirmed INTEGER NOT NULL DEFAULT 0 CHECK(confirmed IN (0,1)), confirmed_at TEXT, - PRIMARY KEY(dispatcher_id,event_id) + PRIMARY KEY(dispatcher_id,event_type,event_id) ); CREATE TABLE IF NOT EXISTS dispatcher_recordings ( dispatcher_id TEXT NOT NULL, @@ -155,9 +155,16 @@ func Open(path string) (_ *Store, err error) { return nil, fmt.Errorf("unknown SQLite layout version %d without tables: preserve database before admission", version) } } else { - if len(seen) != len(allowed) || version != 3 { + if len(seen) != len(allowed) || version != 4 { return nil, fmt.Errorf("existing SQLite layout is incomplete or obsolete (tables=%d, version=%d): preserve data before admission", len(seen), version) } + var outboxIdentityColumns int + if err := db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('dispatcher_outbox') WHERE (name='dispatcher_id' AND pk=1) OR (name='event_type' AND pk=2) OR (name='event_id' AND pk=3)`).Scan(&outboxIdentityColumns); err != nil { + return nil, fmt.Errorf("inspect typed feedback identity layout: %w", err) + } + if outboxIdentityColumns != 3 { + return nil, errors.New("existing SQLite lacks typed feedback identities: preserve data before admission") + } var requiredColumns int if err := db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('dispatcher_state') WHERE name IN ('discovery_ready','applied_sip_revision','pending_sip_revision')`).Scan(&requiredColumns); err != nil { return nil, fmt.Errorf("inspect durable SIP state layout: %w", err) @@ -194,7 +201,7 @@ func Open(path string) (_ *Store, err error) { if _, err := tx.Exec(schema); err != nil { return nil, fmt.Errorf("create current SQLite schema: %w", err) } - if _, err := tx.Exec(`PRAGMA user_version=3`); err != nil { + if _, err := tx.Exec(`PRAGMA user_version=4`); err != nil { return nil, fmt.Errorf("set current SQLite layout version: %w", err) } if err := tx.Commit(); err != nil { diff --git a/internal/store/upload.go b/internal/store/upload.go index e6035ea..299e327 100644 --- a/internal/store/upload.go +++ b/internal/store/upload.go @@ -107,7 +107,7 @@ func (s *Store) BindRecordingUpload(proposed RecordingGrant) (RecordingGrant, bo return RecordingGrant{}, false, ErrUploadConflict } var finalized int - err = tx.QueryRow(`SELECT 1 FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, proposed.DispatcherID, currentResultEventID(proposed.DispatcherID, proposed.SourceEventID)).Scan(&finalized) + err = tx.QueryRow(`SELECT 1 FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='call.execute.result' AND event_id=?`, proposed.DispatcherID, proposed.SourceEventID).Scan(&finalized) if err == nil { return RecordingGrant{}, false, ErrResultConflict } diff --git a/internal/store/upload_test.go b/internal/store/upload_test.go index af3ab2e..88c6edc 100644 --- a/internal/store/upload_test.go +++ b/internal/store/upload_test.go @@ -55,7 +55,7 @@ func TestUploadBindingPersistsOnlyOneOriginalAssetAcrossRestart(t *testing.T) { } var version, seq int var databaseName, databasePath string - if err := s.db.QueryRow(`PRAGMA user_version`).Scan(&version); err != nil || version != 3 { + if err := s.db.QueryRow(`PRAGMA user_version`).Scan(&version); err != nil || version != 4 { t.Fatalf("current schema is not the new, reviewed recording layout: version=%d err=%v", version, err) } if err := s.db.QueryRow(`PRAGMA database_list`).Scan(&seq, &databaseName, &databasePath); err != nil || databasePath == "" {