diff --git a/AGENTS.md b/AGENTS.md index 24f6378..40da88e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -104,14 +104,16 @@ ## SaaS、Dispatcher 与 Agent 的现行边界 +- 2026-10-09 使用者确认保留 SaaS 的 revision 字段与现行 Schema，仅移除外部修订号的强接纳检查：收到队列事件后重读并采用完整当前配置，不因任务/额度修订号不增长、降低或同号内容变化拒绝。启动时列表与随后详情的来源修订号/状态变化采用详情，持久 pause/stop 不被 HTTP running 清除；同租户未来呼叫绑定最新额度，已接纳执行和未知占用保留。选线到占用之间比较实际业务配置摘要（不含 task_revision/status），内部 SIP 加载代次、归属、幂等及原生加载核验不变。共享规范仅调整文字、不改 Schema；SaaS 开发须同步理解，不再要求状态变化递增修订号。本次获准本地 Chrome/browser-harness 操作租户任务与管理员线路页面、提交推送及无重置部署，并仅通过专用测试任务导入已核对平台白名单的号码；最多两次单号码导入/两通外呼，前通确认结束后才可下一通，白名单不可核对或结果未知即停止。原线路地址/主叫/鉴权、白名单与现存数据不得改变，不生成 MQ 外呼或做 500 通压力测试。 + - SaaS→Dispatcher 的**五类只读配置**为 `GET /internal/v1/dispatcher/sip`、`/task/:task_id`、`/tasks`、`/tenant/:tenant_id/quota`、`/ai-providers`；路径前缀固定，均须校验 Dispatcher UUID/资源归属、数字 `tenant_id`、完整快照、来源、有效授权和版本。配置读失败、过期、矛盾或不确定时关新准入；没有旧 MQ 配置回退、通用业务 HTTP、ETag 兜底或偷偷启用旧执行字段。已接纳任务持久绑定原快照。 - 呼叫、控制、必要回执和**每通话唯一最终结果**经固定 `v1` RabbitMQ Topic/队列，任务与控制队列由 SaaS 预建，Dispatcher 不可自行建/删/绑定；stop 时先停该任务消费者并关闭持久准入，再 purge 仅该任务队列的待投递消息，失败不回成功；SaaS 停止继续投递已 stop 任务，后续误投递不执行。结果进入指定共享 durable 队列。独立 D UUID 和接收队列不能广播后正文过滤；`tenant_key` 原值保留，任务只能由归属 D 执行。入站先校验和持久 inbox 后 ACK，状态/outbox 同事务；persistent、mandatory、无 return、publisher confirm 成功才记交付，confirm **不是** SaaS 应用收讫。失败/确认丢失与重启只重发同身份消息，不重复拨号或捏造结果。 -- `call.execute` 只带获批 `task_id/callee`，调用线路、AI 和时限由该任务快照固定；主叫由绑定线路的 SIP 快照唯一 `caller_id` 决定，不再由任务挑选主叫档案；`task.control` 的 start/pause/resume/stop/edit 无旧 CAS/版本字段，控制回 task/action/status 处理结果，stop 不可恢复。任务启动/重启完整读取一次列表，运行中不定时轮询任务列表或任务配置；新建任务由 start 读取任务配置和租户额度，resume 重读并解除人工暂停，edit 只重读、校验并持久更新已有任务的配置和额度，不改变人工暂停/停止状态、不向 Agent 发动作，也不影响已发出的通话快照；旧修订号及停止后 edit 拒绝。全局 AI 服务商列表每次启动只读取一次，本进程全部任务复用，变更须重启后才生效。启动时全量读取并核验 SIP，运行中只由 `sip.config` 通知触发全量 SIP 读取，无常规定时 SIP 轮询；任务 start/resume/edit 使用已生效 SIP 快照，不自行拉取或向 Agent 核验 SIP。SIP 修订变化待旧呼叫结束且 Agent 已加载后，仅在本地重新绑定任务快照，不重读任务列表；待处理期间新准入关闭。发现分页同快照先全部校验后提交，MQ 控制持久状态高于偶发 HTTP 状态；冲突关准入、resume 须重新核验。历史命令不因消息年龄过期，但实际呼出前仍检查任务归属/事件号码格式/时段/授权/额度和有效通话上限；任务结束不删未确认结果、恢复、幂等或未知占用。 +- `call.execute` 只带获批 `task_id/callee`，调用线路、AI 和时限由该任务快照固定；主叫由绑定线路的 SIP 快照唯一 `caller_id` 决定，不再由任务挑选主叫档案；`task.control` 的 start/pause/resume/stop/edit 无旧 CAS/版本字段，控制回 task/action/status 处理结果，stop 不可恢复。任务启动/重启完整读取一次列表，运行中不定时轮询任务列表或任务配置；新建任务由 start 读取任务配置和租户额度，resume 重读并解除人工暂停，edit 只重读、校验并持久更新已有任务的配置和额度，不改变人工暂停/停止状态、不向 Agent 发动作，也不影响已发出的通话快照；停止后的 edit 拒绝；外部修订号接纳规则以 2026-10-09 的更新为准。全局 AI 服务商列表每次启动只读取一次，本进程全部任务复用，变更须重启后才生效。启动时全量读取并核验 SIP，运行中只由 `sip.config` 通知触发全量 SIP 读取，无常规定时 SIP 轮询；任务 start/resume/edit 使用已生效 SIP 快照，不自行拉取或向 Agent 核验 SIP。SIP 修订变化待旧呼叫结束且 Agent 已加载后，仅在本地重新绑定任务快照，不重读任务列表；待处理期间新准入关闭。发现分页同快照先全部校验后提交，MQ 控制持久状态高于偶发 HTTP 状态；冲突关准入、resume 须重新核验。历史命令不因消息年龄过期，但实际呼出前仍检查任务归属/事件号码格式/时段/授权/额度和有效通话上限；任务结束不删未确认结果、恢复、幂等或未知占用。 - 独立 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 用户确认按共享合同 `b502ad2` 本地适配任务线路并发：`allowed_trunk_ids` 只接受包含 `trunk_id/concurrency` 的对象数组，重复线路拒绝；非负整数 `concurrency` 的 0 表示该任务不使用该线路。选线及 SQLite 原子占用同时遵守租户、任务总量、线路总量、任务在线路上的上限，未知执行不释放；任务修订在选线后变化时拒绝原选择，不用旧快照发出新呼叫。占用仍由现有 inbox 与冻结任务快照确定，不新增业务表、不转换旧快照或迁移旧库。ASR/TTS 使用连接的 `ws_endpoint`，LLM 使用 `api_endpoint`，缺失不互相回退。本轮仅本地代码、测试和说明，不部署、不拨号、不请求真实 AI。 +- 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`。 - **下次安全读取步骤**：先确认工作目录是本仓库，用 `stat` 仅检查本机两份文件是否存在、所有者与权限 `0600`；不满足即停止。按各自格式在受限本机进程中解析所需字段到内存，不执行 `source`、不打印全文/字段值、不写临时明文副本，不把密钥、签名 URL、音频或完整对话带入聊天、日志、提交及长期证据。AI 的历史文件只可作为**获准凭据来源**，模型/voice/速度等仍由当前获批的 task/providers 快照固定，不能用环境变量覆盖。OSS 历史文件也不能直接传给 `DISPATCHER_OSS_CONFIG_FILE`：该运行配置要求私有 JSON、`dispatcher_id` 和 `oss` 字段，并以环境变量**名称引用**密钥；需按现行合同构造并核验授权后才能使用。普通构建和测试不读取这些私有文件；真实服务测试必须显式启用对应 opt-in 并受现行门禁约束。 diff --git a/contracts/schema b/contracts/schema index f490afa..ff74afe 160000 --- a/contracts/schema +++ b/contracts/schema @@ -1 +1 @@ -Subproject commit f490afab764a702dda919c718dca1a05e5bceb0a +Subproject commit ff74afe396eb811745a20cd88d167585d6593132 diff --git a/docs/evidence/saas-queue-refresh-20261009.md b/docs/evidence/saas-queue-refresh-20261009.md new file mode 100644 index 0000000..bde7fbf --- /dev/null +++ b/docs/evidence/saas-queue-refresh-20261009.md @@ -0,0 +1,25 @@ +# SaaS 队列驱动配置刷新（2026-10-09） + +## 根因与改动 + +固定环境的 Dispatcher 在启动时拒绝任务 3：`revision 2 changed content`。来源任务修订号未变，状态由 paused 改为 running；旧逻辑把当前任务配置误当作不可修改的历史执行快照。 + +使用者确认保留 revision 字段与现行 Schema，但不以 SaaS 修订号增减、相等或同号内容变化阻止更新。共享规范文字提交为 `ff74afe396eb811745a20cd88d167585d6593132`；未改 Schema。 + +- 启动列表之后读取的任务详情为当前配置；详情状态变化不再导致版本冲突。 +- 任务刷新、发现、edit 及租户额度采用当次完整读取；最新额度同时更新同租户的未来任务绑定。 +- 已持久暂停/停止不被 HTTP running 清除。停止后的 edit、归属、时段、幂等与未知占用规则保留。 +- 选线到占用之间按真实任务配置摘要核验，不以 task_revision/status 元数据判断变化；内部 SIP 加载代次及 Agent/Asterisk 实际加载核验不变。 +- 历史已接纳执行的冻结快照不被当前配置或额度更新改写。 + +## 本地验证 + +`make check`、`make release-check-local`、`make coverage` 通过；业务语句覆盖率 70.8%。包括同修订状态/内容变化、降低任务/额度来源修订号、MQ 暂停保留、同修订选线期间业务变化拒绝、来源元数据单独变化不拒绝、跨任务额度更新，以及冻结执行/未知占用保留的回归测试。RabbitMQ/HTTPS/mTLS 隔离端到端运行通过，不代表真实通话验收。 + +## 真实环境与测试边界 + +本地 Chrome 已分别登录租户与隔离管理员会话。隐藏标签页会使页面交互停滞；确认此现象后激活标签页，新建任务对话框可正常打开，尚未提交新任务或导入号码。 + +只读检查：固定数据库 inbox 为空；Agent/Asterisk active，Dispatcher 仍 auto-restart。平台现有任务 3 显示“开启中”，包含 981 个号码且没有关闭按钮。本轮最多两通，不能在未核实关闭该任务前任意恢复业务准入。使用者已批准暂时关闭原已开启任务、保留号码和数据，且不自动重新开启。 + +当前记录尚未证明新版部署、平台白名单、任务实际回执、管理员线路热加载或真实通话。未重置数据库、清理业务队列、改线路地址/主叫/鉴权、导入号码、外呼、请求 AI/OSS 或重启 Asterisk。 diff --git a/internal/configread/task_identity.go b/internal/configread/task_identity.go new file mode 100644 index 0000000..4467be0 --- /dev/null +++ b/internal/configread/task_identity.go @@ -0,0 +1,28 @@ +package configread + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" +) + +// TaskConfigurationSHA256 protects selection-to-originate freshness without +// treating SaaS revision/status metadata as immutable business configuration. +func TaskConfigurationSHA256(task Task) (string, error) { + decoder := json.NewDecoder(bytes.NewReader(task.Raw)) + decoder.UseNumber() + var value map[string]any + if err := decoder.Decode(&value); err != nil { + return "", fmt.Errorf("decode selected task configuration: %w", err) + } + delete(value, "task_revision") + delete(value, "status") + body, err := json.Marshal(value) + if err != nil { + return "", fmt.Errorf("encode selected task configuration: %w", err) + } + sum := sha256.Sum256(body) + return hex.EncodeToString(sum[:]), nil +} diff --git a/internal/configread/task_identity_test.go b/internal/configread/task_identity_test.go new file mode 100644 index 0000000..be07bb8 --- /dev/null +++ b/internal/configread/task_identity_test.go @@ -0,0 +1,34 @@ +package configread + +import ( + "encoding/json" + "testing" +) + +func TestTaskConfigurationIdentityIgnoresOnlyReceptionMetadata(t *testing.T) { + task := Task{Raw: json.RawMessage(`{"task_revision":9,"status":"running","name":"test","agent":{"asr":{"params":{"sample_rate":16000}}}}`)} + original, err := TaskConfigurationSHA256(task) + if err != nil { + t.Fatal(err) + } + for _, body := range []string{ + `{"name":"test","status":"paused","task_revision":1,"agent":{"asr":{"params":{"sample_rate":16000}}}}`, + `{"task_revision":9,"status":"running","name":"test","agent":{"asr":{"params":{"sample_rate":8000}}}}`, + } { + current, err := TaskConfigurationSHA256(Task{Raw: json.RawMessage(body)}) + if err != nil { + t.Fatal(err) + } + var value map[string]any + if err := json.Unmarshal([]byte(body), &value); err != nil { + t.Fatal(err) + } + metadataOnly := value["status"] == "paused" + if (original == current) != metadataOnly { + t.Fatalf("configuration identity mismatch: metadataOnly=%v", metadataOnly) + } + } + if _, err := TaskConfigurationSHA256(Task{Raw: json.RawMessage(`{`)}); err == nil { + t.Fatal("accepted malformed task JSON") + } +} diff --git a/internal/dispatcher/config.go b/internal/dispatcher/config.go index c3284fa..2c105c3 100644 --- a/internal/dispatcher/config.go +++ b/internal/dispatcher/config.go @@ -89,10 +89,7 @@ func (b Bootstrap) Run(ctx context.Context) error { if err != nil { return fmt.Errorf("retrieve complete assigned task list: %w", err) } - if err := b.Store.ApplyDiscoverySnapshot(b.DispatcherID, tasks); err != nil { - return fmt.Errorf("persist complete task discovery: %w", err) - } - for _, task := range tasks { + for i, task := range tasks { if task.Status == "stopped" { continue } @@ -100,16 +97,18 @@ func (b Bootstrap) Run(ctx context.Context) error { if err != nil { return fmt.Errorf("read assigned task %q: %w", task.TaskID, err) } - if snapshot.Task.TaskRevision != task.TaskRevision || snapshot.Task.Status != task.Status { - return fmt.Errorf("task %q configuration differs from approved discovery snapshot", task.TaskID) - } + tasks[i].TaskRevision = snapshot.Task.TaskRevision + tasks[i].Status = snapshot.Task.Status if err := validateAISnapshot(snapshot); err != nil { return err } if err := b.Store.SaveSnapshot(snapshot); err != nil { - return fmt.Errorf("persist immutable task %q: %w", task.TaskID, err) + return fmt.Errorf("persist current task %q: %w", task.TaskID, err) } } + if err := b.Store.ApplyDiscoverySnapshot(b.DispatcherID, tasks); err != nil { + return fmt.Errorf("persist complete current task discovery: %w", err) + } if b.SIP != nil { *b.SIP = sip } diff --git a/internal/dispatcher/config_test.go b/internal/dispatcher/config_test.go index 64724ab..2ff0bbc 100644 --- a/internal/dispatcher/config_test.go +++ b/internal/dispatcher/config_test.go @@ -3,6 +3,7 @@ package dispatcher import ( "bytes" "context" + "encoding/json" "errors" "log/slog" "net/http" @@ -29,6 +30,7 @@ func configExample(t *testing.T, name string) []byte { func TestBootstrapRequiresSIPLoadingAndControlDrain(t *testing.T) { id := "c046b893-8628-4589-ae50-619d049248a6" called := make([]string, 0) + currentTaskStatus := "running" server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { called = append(called, r.URL.Path) var response []byte @@ -44,7 +46,18 @@ func TestBootstrapRequiresSIPLoadingAndControlDrain(t *testing.T) { case "/internal/v1/dispatcher/ai-providers": response = configExample(t, "config-read-providers") case "/internal/v1/dispatcher/task/task-asr": - response = configExample(t, "config-read-task-asr") + var task map[string]any + if err := json.Unmarshal(configExample(t, "config-read-task-asr"), &task); err != nil { + t.Fatal(err) + } + task["status"] = currentTaskStatus + var err error + response, err = json.Marshal(task) + if err != nil { + t.Error(err) + w.WriteHeader(http.StatusInternalServerError) + return + } case "/internal/v1/dispatcher/tenant/1001/quota": response = configExample(t, "config-read-quota") default: @@ -112,6 +125,18 @@ func TestBootstrapRequiresSIPLoadingAndControlDrain(t *testing.T) { t.Fatalf("missing SIP trace %q", message) } } + currentTaskStatus = "paused" + if err := bootstrap.Run(context.Background()); err != nil { + t.Fatalf("same-revision status change/detail newer than startup list rejected: %v", err) + } + latest, err := db.ReadSnapshot(id, 1001, "task-asr") + if err != nil || latest.Task.Status != "paused" { + t.Fatalf("latest detail not adopted: %s %v", latest.Task.Status, err) + } + if admitted, err := db.CanAdmit(id, 1001, "task-asr"); err != nil || admitted { + t.Fatalf("paused detail opened admission: %v %v", admitted, err) + } + currentTaskStatus = "running" bootstrap.DrainControls = func(context.Context) error { return db.NoteSIPChange(id, 9) } if err := bootstrap.Run(context.Background()); err != nil { t.Fatalf("notification delivered during startup must remain pending for runtime: %v", err) diff --git a/internal/dispatcher/execute.go b/internal/dispatcher/execute.go index cdd2480..5499593 100644 --- a/internal/dispatcher/execute.go +++ b/internal/dispatcher/execute.go @@ -178,8 +178,12 @@ func (c *ExecuteController) dispatchPending(ctx context.Context, cmd store.Execu if err != nil { return fmt.Errorf("task %q rule validation: %w", cmd.TaskID, err) } + taskSHA, err := configread.TaskConfigurationSHA256(snapshot.Task) + if err != nil { + return err + } reservation := store.CallReservation{ - TrunkID: choice.TrunkID, SIPRevision: snapshot.SIP.Revision, TaskRevision: snapshot.Task.TaskRevision, + TrunkID: choice.TrunkID, SIPRevision: snapshot.SIP.Revision, TaskRevision: snapshot.Task.TaskRevision, TaskConfigSHA256: taskSHA, CallerID: choice.CallerID, DialedCallee: choice.DialedCallee, Deadline: choice.Deadline, } diff --git a/internal/rpc/recording_server_flow_test.go b/internal/rpc/recording_server_flow_test.go index c0b77c3..805f809 100644 --- a/internal/rpc/recording_server_flow_test.go +++ b/internal/rpc/recording_server_flow_test.go @@ -71,8 +71,12 @@ func recordingRPCFixture(t *testing.T) (*RecordingServer, *store.Store, context. t.Fatal(err) } now := time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC) + taskSHA, err := configread.TaskConfigurationSHA256(snapshot.Task) + if err != nil { + t.Fatal(err) + } if err := database.ReserveExecute(recordingDispatcherID, command.EventID, store.CallReservation{ - TrunkID: "trunk-mock", SIPRevision: 8, TaskRevision: 1, CallerID: "BD00000000", DialedCallee: command.Callee, Deadline: now.Add(2 * time.Minute), + TrunkID: "trunk-mock", SIPRevision: 8, TaskRevision: 1, TaskConfigSHA256: taskSHA, CallerID: "BD00000000", DialedCallee: command.Callee, Deadline: now.Add(2 * time.Minute), }, now); err != nil { t.Fatal(err) } diff --git a/internal/store/calls.go b/internal/store/calls.go index 6a7bde4..07166b2 100644 --- a/internal/store/calls.go +++ b/internal/store/calls.go @@ -35,12 +35,13 @@ type Execution struct { } type CallReservation struct { - TrunkID string - SIPRevision int64 - TaskRevision int64 - CallerID string - DialedCallee string - Deadline time.Time + TrunkID string + SIPRevision int64 + TaskRevision int64 + TaskConfigSHA256 string + CallerID string + DialedCallee string + Deadline time.Time } type OutboxEvent struct { @@ -155,8 +156,12 @@ func (s *Store) ReserveExecute(dispatcherID, eventID string, selected CallReserv if snapshot.Task.DispatcherID != dispatcherID || snapshot.Task.TenantID != cmd.TenantID || snapshot.Task.TaskID != cmd.TaskID || snapshot.SIP.Revision != revision || snapshot.Quota.TenantID != cmd.TenantID { return errors.New("admission snapshot identity mismatch") } - if snapshot.Task.TaskRevision != selected.TaskRevision { - return fmt.Errorf("%w: task revision changed after trunk selection", ErrNotReady) + currentTaskSHA, err := configread.TaskConfigurationSHA256(snapshot.Task) + if err != nil { + return err + } + if selected.TaskConfigSHA256 == "" || currentTaskSHA != selected.TaskConfigSHA256 { + return fmt.Errorf("%w: task configuration changed after trunk selection", ErrNotReady) } if snapshot.Task.MaxConcurrentCalls <= 0 || snapshot.Quota.MaxConcurrentCalls <= 0 { return fmt.Errorf("%w: missing quota", ErrNotReady) diff --git a/internal/store/calls_test.go b/internal/store/calls_test.go index db0217c..9149f6c 100644 --- a/internal/store/calls_test.go +++ b/internal/store/calls_test.go @@ -2,6 +2,7 @@ package store import ( "errors" + "os" "path/filepath" "strings" "testing" @@ -35,7 +36,19 @@ func currentCall(id string) ExecuteCommand { return ExecuteCommand{DispatcherID: currentDispatcherID, EventID: id, TenantID: 1001, TaskID: "task-asr", Callee: "15003164745", IssuedAt: "2026-09-21T01:30:00Z"} } func currentReservation() CallReservation { - return CallReservation{TrunkID: "trunk-mock", SIPRevision: 8, TaskRevision: 1, CallerID: "BD00000000", DialedCallee: "15003164745", Deadline: time.Date(2026, 9, 21, 1, 32, 0, 0, time.UTC)} + body, err := os.ReadFile("../../contracts/schema/examples/config-read-task-asr.json") + if err != nil { + panic(err) + } + var task configread.Task + if err := task.UnmarshalJSON(body); err != nil { + panic(err) + } + hash, err := configread.TaskConfigurationSHA256(task) + if err != nil { + panic(err) + } + return CallReservation{TrunkID: "trunk-mock", SIPRevision: 8, TaskRevision: 1, TaskConfigSHA256: hash, CallerID: "BD00000000", DialedCallee: "15003164745", Deadline: time.Date(2026, 9, 21, 1, 32, 0, 0, time.UTC)} } func TestExecuteInboxPreventsDuplicateOriginationAfterRestart(t *testing.T) { diff --git a/internal/store/control_flow.go b/internal/store/control_flow.go index e6288ca..d66b282 100644 --- a/internal/store/control_flow.go +++ b/internal/store/control_flow.go @@ -46,9 +46,8 @@ func (s *Store) CompleteStart(snapshot configread.Snapshot, eventID string) erro return nil } -// CompleteEdit advances only a previously assigned task's configuration -// revision and its acknowledgment. Pause/stop barriers and running calls stay -// untouched; they are never inferred from a configuration edit. +// CompleteEdit replaces the assigned task's current configuration and acknowledgment. +// Pause/stop barriers and running calls are never inferred from an edit. func (s *Store) CompleteEdit(snapshot configread.Snapshot, eventID string) (string, error) { task := snapshot.Task tx, err := s.db.Begin() @@ -57,10 +56,9 @@ func (s *Store) CompleteEdit(snapshot configread.Snapshot, eventID string) (stri } defer tx.Rollback() var state, status string - var revision int64 var present int - err = tx.QueryRow(`SELECT control_state,status,task_revision,present FROM dispatcher_tasks - WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, task.DispatcherID, task.TenantID, task.TaskID).Scan(&state, &status, &revision, &present) + err = tx.QueryRow(`SELECT control_state,status,present FROM dispatcher_tasks + WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, task.DispatcherID, task.TenantID, task.TaskID).Scan(&state, &status, &present) if err != nil && !errors.Is(err, sql.ErrNoRows) { return "", fmt.Errorf("read edit task state: %w", err) } @@ -73,9 +71,6 @@ func (s *Store) CompleteEdit(snapshot configread.Snapshot, eventID string) (stri } return "rejected", nil } - if task.TaskRevision < revision { - return "", fmt.Errorf("edit task %q revision regressed from %d to %d", task.TaskID, revision, task.TaskRevision) - } if err := saveSnapshot(tx, snapshot); err != nil { return "", fmt.Errorf("bind edited task snapshot: %w", err) } diff --git a/internal/store/control_test.go b/internal/store/control_test.go index 08b0829..bae224f 100644 --- a/internal/store/control_test.go +++ b/internal/store/control_test.go @@ -66,18 +66,20 @@ func TestEditCommitsSnapshotRevisionAndAckTogether(t *testing.T) { if err := s.db.QueryRow(`SELECT task_revision FROM dispatcher_tasks WHERE task_id='task-asr'`).Scan(&revision); err != nil || revision != 2 { t.Fatalf("task and snapshot revisions differ: revision=%d err=%v", revision, err) } - stale := currentStoreSnapshot(t) - if _, err := s.CompleteEdit(stale, "edit-stale"); err == nil { - t.Fatal("stale edit downgraded the task") + stale := updated + stale.Task.Raw = []byte(strings.Replace(string(updated.Task.Raw), `"task_revision":2`, `"task_revision":1`, 1)) + stale.Task.TaskRevision = 1 + if status, err := s.CompleteEdit(stale, "edit-lower-revision"); err != nil || status != "applied" { + t.Fatalf("latest event with lower source revision was rejected: %s %v", status, err) } conflicting := updated conflicting.Task.Raw = []byte(strings.Replace(string(updated.Task.Raw), `"status":"running"`, `"status":"paused"`, 1)) - if _, err := s.CompleteEdit(conflicting, "edit-conflicting"); err == nil { - t.Fatal("same revision changed task contents") + if status, err := s.CompleteEdit(conflicting, "edit-same-revision"); err != nil || status != "applied" { + t.Fatalf("latest event with same source revision was rejected: %s %v", status, err) } persisted, err = s.ReadSnapshot(currentDispatcherID, 1001, "task-asr") - if err != nil || persisted.Task.TaskRevision != 2 || string(persisted.Task.Raw) != string(updated.Task.Raw) { - t.Fatalf("failed edit changed bound snapshot: revision=%d err=%v", persisted.Task.TaskRevision, err) + if err != nil || persisted.Task.TaskRevision != 2 || string(persisted.Task.Raw) != string(conflicting.Task.Raw) { + t.Fatalf("latest edit was not bound: revision=%d err=%v", persisted.Task.TaskRevision, err) } if err := s.ApplyControl(currentDispatcherID, 1001, "task-asr", "stop"); err != nil { t.Fatal(err) diff --git a/internal/store/global_revision_test.go b/internal/store/global_revision_test.go index fb243a4..03ce662 100644 --- a/internal/store/global_revision_test.go +++ b/internal/store/global_revision_test.go @@ -10,13 +10,14 @@ import ( "git.ipao.vip/rogee/go-sip/internal/configread" ) -func TestGlobalSIPAndQuotaRevisionCannotChangeContent(t *testing.T) { +func TestInternalSIPGenerationRemainsVerifiedWhileQuotaRevisionIsMetadata(t *testing.T) { s, err := Open(filepath.Join(t.TempDir(), "state.db")) if err != nil { t.Fatal(err) } defer s.Close() base := currentStoreSnapshot(t) + base.Quota.QuotaRevision = 3 if err := s.SaveSnapshot(base); err != nil { t.Fatal(err) } @@ -30,8 +31,8 @@ func TestGlobalSIPAndQuotaRevisionCannotChangeContent(t *testing.T) { } changedQuota := base changedQuota.Quota.MaxConcurrentCalls++ - if err := s.SaveSnapshot(changedQuota); err == nil { - t.Fatal("same quota revision accepted different content") + if err := s.SaveSnapshot(changedQuota); err != nil { + t.Fatalf("current quota with unchanged source revision rejected: %v", err) } regressed := base regressed.SIP.Revision-- @@ -40,8 +41,8 @@ func TestGlobalSIPAndQuotaRevisionCannotChangeContent(t *testing.T) { } regressed = base regressed.Quota.QuotaRevision-- - if err := s.SaveSnapshot(regressed); err == nil { - t.Fatal("older tenant quota revision was installed") + if err := s.SaveSnapshot(regressed); err != nil { + t.Fatalf("current quota with lower source revision rejected: %v", err) } loaded, err := s.ReadSnapshot(base.Task.DispatcherID, base.Task.TenantID, base.Task.TaskID) if err != nil { @@ -60,7 +61,7 @@ func TestGlobalSIPAndQuotaRevisionCannotChangeContent(t *testing.T) { } } -func TestSharedGlobalRevisionConflictAcrossTasks(t *testing.T) { +func TestTenantQuotaRefreshUpdatesAllTaskBindings(t *testing.T) { s, err := Open(filepath.Join(t.TempDir(), "state.db")) if err != nil { t.Fatal(err) @@ -87,7 +88,11 @@ func TestSharedGlobalRevisionConflictAcrossTasks(t *testing.T) { second = first second.Task = task second.Quota.MaxConcurrentCalls++ - if err := s.SaveSnapshot(second); err == nil { - t.Fatal("two tasks accepted different tenant quotas under same revision") + if err := s.SaveSnapshot(second); err != nil { + t.Fatalf("latest quota rejected on another task under the same source revision: %v", err) + } + current, err := s.ReadSnapshot(currentDispatcherID, first.Task.TenantID, first.Task.TaskID) + if err != nil || current.Quota != second.Quota { + t.Fatalf("another task retained stale tenant quota: %+v %v", current.Quota, err) } } diff --git a/internal/store/source_revision_call_test.go b/internal/store/source_revision_call_test.go new file mode 100644 index 0000000..ec949f6 --- /dev/null +++ b/internal/store/source_revision_call_test.go @@ -0,0 +1,82 @@ +package store + +import ( + "encoding/json" + "testing" + "time" +) + +func TestSourceMetadataRefreshPreservesAcceptedCallAndUnknownOccupancy(t *testing.T) { + s := preparedCurrentCallStore(t) + first := recordTrunkCall(t, s, "task-asr", "call-before-refresh") + at := time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC) + selection := currentTaskReservation(t, s, "task-asr") + if err := s.ReserveExecute(first.DispatcherID, first.EventID, selection, at); err != nil { + t.Fatal(err) + } + if err := s.MarkExecuteUnknown(first.DispatcherID, first.EventID); err != nil { + t.Fatal(err) + } + var frozen string + if err := s.db.QueryRow(`SELECT snapshot_json FROM dispatcher_inbox WHERE event_id=?`, first.EventID).Scan(&frozen); err != nil { + t.Fatal(err) + } + snapshot, err := s.ReadSnapshot(currentDispatcherID, 1001, "task-asr") + if err != nil { + t.Fatal(err) + } + var task map[string]any + if err := json.Unmarshal(snapshot.Task.Raw, &task); err != nil { + t.Fatal(err) + } + task["name"] = "current configuration" + body, err := json.Marshal(task) + if err != nil { + t.Fatal(err) + } + if err := snapshot.Task.UnmarshalJSON(body); err != nil { + t.Fatal(err) + } + snapshot.Quota.MaxConcurrentCalls = 1 + if _, err := s.CompleteEdit(snapshot, "edit-same-source-revision"); err != nil { + t.Fatal(err) + } + var after, status string + if err := s.db.QueryRow(`SELECT snapshot_json,status FROM dispatcher_inbox WHERE event_id=?`, first.EventID).Scan(&after, &status); err != nil { + t.Fatal(err) + } + if frozen != after || status != "unknown" { + t.Fatal("current configuration refresh rewrote an accepted execution") + } + if occupied, err := s.OccupiedCalls(currentDispatcherID); err != nil || occupied != 1 { + t.Fatalf("refresh released unknown occupancy: %d %v", occupied, err) + } +} + +func TestSelectionIgnoresSourceRevisionButStillFreezesCurrentMetadata(t *testing.T) { + s := preparedCurrentCallStore(t) + selection := currentTaskReservation(t, s, "task-asr") + snapshot, err := s.ReadSnapshot(currentDispatcherID, 1001, "task-asr") + if err != nil { + t.Fatal(err) + } + var task map[string]any + if err := json.Unmarshal(snapshot.Task.Raw, &task); err != nil { + t.Fatal(err) + } + task["task_revision"] = 8 + body, err := json.Marshal(task) + if err != nil { + t.Fatal(err) + } + if err := snapshot.Task.UnmarshalJSON(body); err != nil { + t.Fatal(err) + } + if _, err := s.CompleteEdit(snapshot, "edit-metadata-only"); err != nil { + t.Fatal(err) + } + cmd := recordTrunkCall(t, s, "task-asr", "call-after-metadata") + if err := s.ReserveExecute(cmd.DispatcherID, cmd.EventID, selection, time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC)); err != nil { + t.Fatalf("source revision alone fenced the selection: %v", err) + } +} diff --git a/internal/store/source_revision_test.go b/internal/store/source_revision_test.go new file mode 100644 index 0000000..2ea4c17 --- /dev/null +++ b/internal/store/source_revision_test.go @@ -0,0 +1,68 @@ +package store + +import ( + "encoding/json" + "path/filepath" + "testing" + + "git.ipao.vip/rogee/go-sip/internal/configread" +) + +func TestQueueRefreshAdoptsTaskAndQuotaWithoutSourceRevisionOrdering(t *testing.T) { + s, err := Open(filepath.Join(t.TempDir(), "current.db")) + if err != nil { + t.Fatal(err) + } + defer s.Close() + snapshot := currentStoreSnapshot(t) + if err := s.SaveSnapshot(snapshot); err != nil { + t.Fatal(err) + } + for _, revision := range []int64{snapshot.Task.TaskRevision, 1, 9, 1} { + var task map[string]any + if err := json.Unmarshal(snapshot.Task.Raw, &task); err != nil { + t.Fatal(err) + } + task["task_revision"] = revision + task["status"] = "paused" + task["name"] = "current event content" + body, err := json.Marshal(task) + if err != nil { + t.Fatal(err) + } + if err := snapshot.Task.UnmarshalJSON(body); err != nil { + t.Fatal(err) + } + snapshot.Quota.QuotaRevision = 1 + snapshot.Quota.MaxConcurrentCalls = 1 + if err := s.SaveSnapshot(snapshot); err != nil { + t.Fatalf("current task/quota rejected at source revision %d: %v", revision, err) + } + stored, err := s.ReadSnapshot(currentDispatcherID, 1001, "task-asr") + if err != nil || stored.Task.TaskRevision != revision || stored.Task.Status != "paused" || stored.Quota.QuotaRevision != 1 || stored.Quota.MaxConcurrentCalls != 1 { + t.Fatalf("latest received snapshot was not adopted: %+v %v", stored, err) + } + } +} + +func TestDiscoveryAdoptsCurrentRevisionWithoutErasingMQPause(t *testing.T) { + s, err := Open(filepath.Join(t.TempDir(), "current.db")) + if err != nil { + t.Fatal(err) + } + defer s.Close() + for _, revision := range []int64{9, 1, 1} { + task := configread.DiscoveredTask{TaskID: "task-asr", TenantID: 1001, TaskRevision: revision, Status: "running"} + if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.DiscoveredTask{task}); err != nil { + t.Fatalf("current discovery revision %d rejected: %v", revision, err) + } + if err := s.ApplyControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil { + t.Fatal(err) + } + var state string + var storedRevision int64 + if err := s.db.QueryRow(`SELECT control_state,task_revision FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, currentDispatcherID, 1001, "task-asr").Scan(&state, &storedRevision); err != nil || state != "paused" || storedRevision != revision { + t.Fatalf("revision refresh erased persisted control: %s %d %v", state, storedRevision, err) + } + } +} diff --git a/internal/store/store.go b/internal/store/store.go index 3dfbb1d..a654a13 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -295,14 +295,6 @@ func applyDiscoveredTasks(tx *sql.Tx, dispatcherID string, tasks []configread.Di if !errors.Is(err, sql.ErrNoRows) { return fmt.Errorf("check task %q tenant ownership: %w", task.TaskID, err) } - var oldRevision int64 - err = tx.QueryRow(`SELECT task_revision FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, dispatcherID, task.TenantID, task.TaskID).Scan(&oldRevision) - if err != nil && !errors.Is(err, sql.ErrNoRows) { - return fmt.Errorf("check task %q discovery revision: %w", task.TaskID, err) - } - if err == nil && task.TaskRevision < oldRevision { - return fmt.Errorf("task %q discovery revision regressed", task.TaskID) - } initialControl := "" if task.Status == "paused" || task.Status == "stopped" { initialControl = task.Status @@ -432,50 +424,32 @@ func saveSnapshot(tx *sql.Tx, snapshot configread.Snapshot) error { } sum := sha256.Sum256(binding) digest := hex.EncodeToString(sum[:]) - if err := validateGlobalRevisions(tx, snapshot); err != nil { + if err := validateInternalSIPGeneration(tx, snapshot); err != nil { return err } - var oldRevision int64 - var oldDigest string - err = tx.QueryRow(`SELECT task_revision,content_sha256 FROM dispatcher_configs WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, task.DispatcherID, task.TenantID, task.TaskID).Scan(&oldRevision, &oldDigest) - if err != nil && !errors.Is(err, sql.ErrNoRows) { - return fmt.Errorf("read immutable snapshot version: %w", err) - } - if err == nil { - if oldRevision > task.TaskRevision { - return fmt.Errorf("task %q revision regressed", task.TaskID) - } - if oldRevision == task.TaskRevision && oldDigest != digest { - return fmt.Errorf("task %q revision %d changed content", task.TaskID, task.TaskRevision) - } - if oldRevision == task.TaskRevision { - if _, err := tx.Exec(`UPDATE dispatcher_configs SET sip_revision=?,quota_revision=?,snapshot_json=? WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, snapshot.SIP.Revision, snapshot.Quota.QuotaRevision, full, task.DispatcherID, task.TenantID, task.TaskID); err != nil { - return fmt.Errorf("refresh global SIP/quota for task %q: %w", task.TaskID, err) - } - return nil - } - } if _, err = tx.Exec(`INSERT INTO dispatcher_configs(dispatcher_id,tenant_id,task_id,task_revision,sip_revision,quota_revision,content_sha256,snapshot_json) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(dispatcher_id,tenant_id,task_id) DO UPDATE SET task_revision=excluded.task_revision,sip_revision=excluded.sip_revision,quota_revision=excluded.quota_revision, content_sha256=excluded.content_sha256,snapshot_json=excluded.snapshot_json`, task.DispatcherID, task.TenantID, task.TaskID, task.TaskRevision, snapshot.SIP.Revision, snapshot.Quota.QuotaRevision, digest, full); err != nil { return fmt.Errorf("persist task snapshot %q: %w", task.TaskID, err) } + quota, err := json.Marshal(snapshot.Quota) + if err != nil { + return fmt.Errorf("encode current tenant quota: %w", err) + } + if _, err := tx.Exec(`UPDATE dispatcher_configs SET quota_revision=?,snapshot_json=json_set(snapshot_json,'$.quota',json(?)) WHERE dispatcher_id=? AND tenant_id=?`, snapshot.Quota.QuotaRevision, string(quota), task.DispatcherID, task.TenantID); err != nil { + return fmt.Errorf("refresh current tenant quota bindings: %w", err) + } return nil } -// validateGlobalRevisions prevents two task bindings from silently -// disagreeing about the same approved Dispatcher SIP or tenant quota revision. -// Older revisions also cannot replace a newer binding across tasks. -func validateGlobalRevisions(tx *sql.Tx, snapshot configread.Snapshot) error { +// Internal SIP generations prove the configuration loaded by Agent/Asterisk. +// External task/quota revisions are metadata, not configuration acceptance gates. +func validateInternalSIPGeneration(tx *sql.Tx, snapshot configread.Snapshot) error { newSIP, err := json.Marshal(snapshot.SIP) if err != nil { return fmt.Errorf("encode approved SIP revision: %w", err) } - newQuota, err := json.Marshal(snapshot.Quota) - if err != nil { - return fmt.Errorf("encode approved quota revision: %w", err) - } rows, err := tx.Query(`SELECT tenant_id,task_id,sip_revision,quota_revision,snapshot_json FROM dispatcher_configs WHERE dispatcher_id=?`, snapshot.Task.DispatcherID) if err != nil { return fmt.Errorf("inspect current global revision bindings: %w", err) @@ -510,21 +484,6 @@ func validateGlobalRevisions(tx *sql.Tx, snapshot configread.Snapshot) error { return fmt.Errorf("SIP revision %d changed approved content across task %q", sipRevision, taskID) } } - if tenantID != snapshot.Task.TenantID { - continue - } - if quotaRevision > snapshot.Quota.QuotaRevision { - return fmt.Errorf("tenant quota revision regressed below task %q", taskID) - } - if quotaRevision == snapshot.Quota.QuotaRevision { - previous, err := json.Marshal(old.Quota) - if err != nil { - return fmt.Errorf("encode existing tenant quota revision: %w", err) - } - if !bytes.Equal(previous, newQuota) { - return fmt.Errorf("tenant quota revision %d changed approved content across task %q", quotaRevision, taskID) - } - } } if err := rows.Err(); err != nil { return fmt.Errorf("read global revision bindings: %w", err) diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 2f13aae..cd28221 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -94,7 +94,7 @@ func TestStoreFreshSnapshotAndControlSurviveRestart(t *testing.T) { } } -func TestStoreRejectsSameRevisionDifferentContent(t *testing.T) { +func TestStoreAcceptsSameRevisionCurrentContent(t *testing.T) { s, err := Open(filepath.Join(t.TempDir(), "current.db")) if err != nil { t.Fatal(err) @@ -106,8 +106,12 @@ func TestStoreRejectsSameRevisionDifferentContent(t *testing.T) { } snapshot.Task.Name = "changed without changing revision" snapshot.Task.Raw = []byte(strings.Replace(string(snapshot.Task.Raw), `"ASR example"`, `"changed without changing revision"`, 1)) - if err := s.SaveSnapshot(snapshot); err == nil { - t.Fatal("accepted different content under same immutable revision") + if err := s.SaveSnapshot(snapshot); err != nil { + t.Fatalf("queue refresh rejected current content under the same source revision: %v", err) + } + stored, err := s.ReadSnapshot(currentDispatcherID, 1001, "task-asr") + if err != nil || stored.Task.Name != snapshot.Task.Name { + t.Fatalf("latest task content was not persisted: %q %v", stored.Task.Name, err) } } diff --git a/internal/store/task_trunk_concurrency_test.go b/internal/store/task_trunk_concurrency_test.go index b247299..0e4304f 100644 --- a/internal/store/task_trunk_concurrency_test.go +++ b/internal/store/task_trunk_concurrency_test.go @@ -70,6 +70,21 @@ func taskTrunkStore(t *testing.T, snapshots ...configread.Snapshot) (*Store, str return s, path } +func currentTaskReservation(t *testing.T, s *Store, taskID string) CallReservation { + t.Helper() + snapshot, err := s.ReadSnapshot(currentDispatcherID, 1001, taskID) + if err != nil { + t.Fatal(err) + } + selection := currentReservation() + selection.TaskRevision = snapshot.Task.TaskRevision + selection.TaskConfigSHA256, err = configread.TaskConfigurationSHA256(snapshot.Task) + if err != nil { + t.Fatal(err) + } + return selection +} + func recordTrunkCall(t *testing.T, s *Store, taskID, eventID string) ExecuteCommand { t.Helper() cmd := currentCall(eventID) @@ -91,7 +106,7 @@ func TestTaskTrunkReservationRejectsDisabledAndUnlistedLines(t *testing.T) { t.Run(tc.name, func(t *testing.T) { s, _ := taskTrunkStore(t, trunkLimitSnapshot(t, "task-asr", 1, tc.limit)) cmd := recordTrunkCall(t, s, "task-asr", "call-disabled") - selection := currentReservation() + selection := currentTaskReservation(t, s, "task-asr") selection.TrunkID = tc.selected at := time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC) if err := s.ReserveExecute(cmd.DispatcherID, cmd.EventID, selection, at); !errors.Is(err, ErrNotReady) { @@ -105,11 +120,11 @@ func TestTaskTrunkLimitIsScopedToTaskAndUnknownSurvivesRestart(t *testing.T) { s, path := taskTrunkStore(t, trunkLimitSnapshot(t, "task-asr", 1, 1), trunkLimitSnapshot(t, "task-other", 1, 1)) at := time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC) first := recordTrunkCall(t, s, "task-asr", "call-first") - if err := s.ReserveExecute(first.DispatcherID, first.EventID, currentReservation(), at); err != nil { + if err := s.ReserveExecute(first.DispatcherID, first.EventID, currentTaskReservation(t, s, "task-asr"), at); err != nil { t.Fatal(err) } other := recordTrunkCall(t, s, "task-other", "call-other") - if err := s.ReserveExecute(other.DispatcherID, other.EventID, currentReservation(), at); err != nil { + if err := s.ReserveExecute(other.DispatcherID, other.EventID, currentTaskReservation(t, s, "task-other"), at); err != nil { t.Fatalf("another task lost its independent per-line capacity: %v", err) } if err := s.MarkExecuteUnknown(first.DispatcherID, first.EventID); err != nil { @@ -131,13 +146,13 @@ func TestTaskTrunkLimitIsScopedToTaskAndUnknownSurvivesRestart(t *testing.T) { t.Fatalf("task trunk occupancy after restart=%v err=%v", occupancy, err) } second := recordTrunkCall(t, s, "task-asr", "call-second") - if err := s.ReserveExecute(second.DispatcherID, second.EventID, currentReservation(), at); !errors.Is(err, ErrCapacity) { + if err := s.ReserveExecute(second.DispatcherID, second.EventID, currentTaskReservation(t, s, "task-asr"), at); !errors.Is(err, ErrCapacity) { t.Fatalf("unknown call released task-line capacity on restart: %v", err) } if err := s.FinishExecute(first.DispatcherID, first.EventID); err != nil { t.Fatal(err) } - if err := s.ReserveExecute(second.DispatcherID, second.EventID, currentReservation(), at); err != nil { + if err := s.ReserveExecute(second.DispatcherID, second.EventID, currentTaskReservation(t, s, "task-asr"), at); err != nil { t.Fatalf("confirmed end did not release task-line capacity: %v", err) } } @@ -149,11 +164,12 @@ func TestTaskTrunkConcurrentReservationsCannotExceedLimit(t *testing.T) { for i := range commands { commands[i] = recordTrunkCall(t, s, "task-asr", fmt.Sprintf("call-%d", i)) } + selection := currentTaskReservation(t, s, "task-asr") var accepted atomic.Int64 var wg sync.WaitGroup for _, cmd := range commands { wg.Go(func() { - err := s.ReserveExecute(cmd.DispatcherID, cmd.EventID, currentReservation(), at) + err := s.ReserveExecute(cmd.DispatcherID, cmd.EventID, selection, at) if err == nil { accepted.Add(1) } else if !errors.Is(err, ErrCapacity) { @@ -167,22 +183,22 @@ func TestTaskTrunkConcurrentReservationsCannotExceedLimit(t *testing.T) { } } -func TestTaskTrunkEditRechecksLimitAndSelectionRevision(t *testing.T) { +func TestTaskTrunkEditSameRevisionRechecksSelectionConfiguration(t *testing.T) { s, _ := taskTrunkStore(t, trunkLimitSnapshot(t, "task-asr", 1, 2)) at := time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC) first := recordTrunkCall(t, s, "task-asr", "call-before-edit") - if err := s.ReserveExecute(first.DispatcherID, first.EventID, currentReservation(), at); err != nil { + oldSelection := currentTaskReservation(t, s, "task-asr") + if err := s.ReserveExecute(first.DispatcherID, first.EventID, oldSelection, at); err != nil { t.Fatal(err) } - if state, err := s.CompleteEdit(trunkLimitSnapshot(t, "task-asr", 2, 1), "edit-line-limit"); err != nil || state != "applied" { + if state, err := s.CompleteEdit(trunkLimitSnapshot(t, "task-asr", 1, 1), "edit-line-limit"); err != nil || state != "applied" { t.Fatalf("edit=%q err=%v", state, err) } second := recordTrunkCall(t, s, "task-asr", "call-after-edit") - if err := s.ReserveExecute(second.DispatcherID, second.EventID, currentReservation(), at); !errors.Is(err, ErrNotReady) { + if err := s.ReserveExecute(second.DispatcherID, second.EventID, oldSelection, at); !errors.Is(err, ErrNotReady) { t.Fatalf("old selection revision was accepted after edit: %v", err) } - selection := currentReservation() - selection.TaskRevision = 2 + selection := currentTaskReservation(t, s, "task-asr") if err := s.ReserveExecute(second.DispatcherID, second.EventID, selection, at); !errors.Is(err, ErrCapacity) { t.Fatalf("new lower task-line limit was ignored: %v", err) }