feat(test): publish one authorized SaaS command over confirmed MQ

AGENTS.md

@@ -81,7 +81,7 @@
 - 唯一现行 SaaS↔Dispatcher 业务规范是 [`docs/thirds/saas-dispatcher.md`](docs/thirds/saas-dispatcher.md)；当前项目内 Schema、拓扑、正反例及来源/hash 在 [`contracts/local/`](contracts/local/)；内部 Agent RPC 在 [`proto/agent/agent.proto`](proto/agent/agent.proto)。本地验收与外部缺口见 [`docs/evidence/saas-dispatcher-p08-acceptance.md`](docs/evidence/saas-dispatcher-p08-acceptance.md)。Markdown 不代替机器合同或外部签收，也不另外维护一份平行字段定义。
 - 已由当前合同来源清单固定哈希的历史提案与计划保留**原字节**于 [`docs/archive/sources/`](docs/archive/sources/README.md)，使用者原有未提交的两份旧对接文档也按原字节归档；旧上游 v1 在 [`docs/archive/upstream/`](docs/archive/upstream/README.md) 可离线校验，但不嵌入运行合同。旧 F/W 工作包、旧 MQ-only 合同及归档不作为当前运行入口。固定 MQ `v1`、HTTP `/internal/v1/dispatcher/...` 和业务 revision 是现行通信规则，不是自有实现代次；不得为历史路径新建兼容或回退。
 - 当前业务范围仍仅**单节点、单 Dispatcher、单 Agent、单 Cell、单租户**。根命令只接受显式 `agent`/`dispatcher`；`mixed` 和裸 `real` 仍拒绝；隔离 `mock`、只读 `sip-only` 和需完整非生产证据/正式获批指令的 `nonprod-real` 为彼此独立的入口。新增非生产真实入口尚未在测试机完成全链路核验或拨号；不得以编译/本机 Mock 通过宣称可用。另有严格隔离的 `--mode sip-only`：只允许 Dispatcher 读完整 SIP、持久接纳归属 `sip.config`、经已激活双向 TLS Agent 会话将完整快照应用到原生 Asterisk，并从运行态核对版本；不发现任务、不启动业务呼叫、不开放准入、不处理其他业务控制。测试机已凭使用者批准，将三条历史登记地址以**测试快照显式声明的** UDP/IP 鉴权、无需 REGISTER 配置写入并核对 Asterisk 运行态；供应商尚未确认这些实际线路是否满足上述参数，无拨号，不证明线路可用或生产签收。没有真实 SaaS、management、真实通话或生产签收；真实百炼 LLM 与 OSS 已各做一次**不拨号、独立的最小连接/写入诊断**（见 [`docs/evidence/real-ai-oss-one-shot-20261003.md`](docs/evidence/real-ai-oss-one-shot-20261003.md)），这不是获批任务的 ASR/LLM/TTS、录音和上传全链路签收。生产发布包仍为 `production_approval=false`。任何本机 Mock 或 SIP-only 核验均不授权真实呼叫。
-- 使用者批准独立的测试专用 `deploys/test/saas-mock/` 仅模拟缺失的 SaaS：从 0600 的静态快照提供五类正式 HTTP 配置，在专用 RabbitMQ vhost 中由模拟 SaaS 预建现行拓扑；不在 Dispatcher 内注入快照；目前不发布外呼消息，也不替代 Agent/Asterisk/AI/OSS。测试用正式入口必须同时具备只读配置、专用 MQ、真实 Agent/Asterisk/AI/录音/OSS、逐通确认和拨号前活跃抓包证据，缺一项不得试拨。此模拟不构成真实 SaaS 签收。
+- 使用者批准独立的测试专用 `deploys/test/saas-mock/` 仅模拟缺失的 SaaS：从 0600 的静态快照提供五类正式 HTTP 配置，在专用 RabbitMQ vhost 中由模拟 SaaS 预建现行拓扑；不在 Dispatcher 内注入快照；默认不发布外呼消息，只有受限脚本已为同一 `event_id`/线路/原始号码启动抓包后，才可显式单次 MQ 投递；不替代 Agent/Asterisk/AI/OSS。测试用正式入口必须同时具备只读配置、专用 MQ、真实 Agent/Asterisk/AI/录音/OSS、逐通确认和拨号前活跃抓包证据，缺一项不得试拨。此模拟不构成真实 SaaS 签收。
 - 开发按 TDD 分批，小步提交；不得覆盖使用者现存修改/未跟踪文件，不自动清理、迁移或覆盖任何现存 SQLite、spool、outbox 和 Agent 恢复文件。旧 `.executions` 及恢复根目录中旧 `.uploads`、`.upload-locks`、逐执行 `state.json` 的发现须只读失败关闭，现存未交付事实由使用者确认处置。真实云账号、EIP、线路、拨号、生产部署和共享数据操作分别需要明确授权。
 
 ## SaaS、Dispatcher 与 Agent 的现行边界

deploys/test/saas-mock/README.md

@@ -1,6 +1,6 @@
 # SaaS 侧测试模拟服务（非生产）
 
-仅替代当前缺失的 SaaS，不改 Dispatcher 的正式 HTTP 配置接口或 MQ 归属。**当前只提供静态配置和预建队列；不会投递 `call.execute`、拨号或产生业务结果。** 虚构的合同示例不能充当真实 AI、线路或任务授权。
+仅替代当前缺失的 SaaS，不改 Dispatcher 的正式 HTTP 配置接口或 MQ 归属。默认只提供静态配置和预建队列；另有**显式单次** `call.execute` 投递命令，绝不自动拨号或产生业务结果。 虚构的合同示例不能充当真实 AI、线路或任务授权。
 
 ## 数据
 
@@ -12,7 +12,8 @@
 - 准备测试 HTTPS 证书与私钥；将 `SAAS_MOCK_DISPATCHER_SECRET` 和含凭据的 `SAAS_MOCK_RABBITMQ_URL` 放在受限环境文件，不在命令行、仓库或聊天中传输。
 - 启动：`go run ./deploys/test/saas-mock --data <私有目录> --dispatcher-id <UUID> --listen <地址:端口> --tls-cert <证书文件> --tls-key <私钥文件>`。
 - 服务以标准 `X-DISPATCHER-id` 和 `X-DISPATCHER-SECRET-KEY` 校验 Dispatcher，再提供五类只读配置。错误的归属、资源、租户、快照或消息队列会导致拒绝启动/读取。
+- 单次投递另起命令：`go run ./deploys/test/saas-mock --data <私有目录> --dispatcher-id <UUID> --publish-event-id <唯一事件号> --publish-task-id <单线路任务号> --publish-callee <原始白名单号码>`。必须在 `nonprod-call-evidence.sh --call-id <同一事件号> --trunk <任务唯一线路> --target <同一原始号码> -- <单次投递命令>` 启用并确认 SIP/RTP 抓包、PJSIP logger 和主机门禁之后运行；不得预投、批量投递、自动重试或换线。任务快照须只允许一条真实线路，投递仅含任务号和原始号码。RabbitMQ 必须用上述专用环境变量，不能借用默认或共享 vhost。发布确认只代表 MQ 接收，不代表 SaaS 已收到最终结果。
 
 测试：`go test ./deploys/test/saas-mock` 验证正式配置客户端；设置指向**单独隔离 vhost** 的 `SAAS_MOCK_TEST_BROKER_URL` 后，`TestSaaSMockProvisionsDispatcherTopology` 还将实际预建 MQ 并用 Dispatcher 被动读回。缺省测试不会连接共享 RabbitMQ。
 
-当前正式呼叫仍只允许 Mock，测试机 ARI/HTTP、真实 AI/TTS、录音、OSS 与拨号前抓包尚未完成验证。六组外呼均未执行。此服务通过测试也不代表真实 SaaS 或生产签收。
+正式入口已经具备独立 `nonprod-real` 代码路径，但测试机的正式链路、真实 AI/TTS、录音、OSS 与拨号前抓包仍未完成整体验证。六组外呼均未执行。此服务通过测试也不代表真实 SaaS 或生产签收。

deploys/test/saas-mock/main.go

@@ -1,8 +1,9 @@
 // saas-mock serves approved, static SaaS test snapshots through the formal
-// Dispatcher configuration API. It never creates calls or publishes MQ jobs.
+// Dispatcher configuration API. One-shot publishing is an explicit separate mode.
 package main
 
 import (
+	"context"
 	"errors"
 	"flag"
 	"log"
@@ -17,9 +18,28 @@ func main() {
 	listen := flag.String("listen", "", "explicit HTTPS listen address")
 	cert := flag.String("tls-cert", "", "HTTPS certificate file")
 	key := flag.String("tls-key", "", "HTTPS private key file")
+	publishID := flag.String("publish-event-id", "", "one-shot MQ event ID matching capture --call-id")
+	publishTask := flag.String("publish-task-id", "", "one approved single-trunk task")
+	publishCallee := flag.String("publish-callee", "", "one original allowlisted callee")
 	flag.Parse()
 	secret := os.Getenv("SAAS_MOCK_DISPATCHER_SECRET")
 	brokerURL := os.Getenv("SAAS_MOCK_RABBITMQ_URL")
+	if *publishID != "" || *publishTask != "" || *publishCallee != "" {
+		if *dataDir == "" || *dispatcherID == "" || brokerURL == "" || *publishID == "" || *publishTask == "" || *publishCallee == "" || flag.NArg() != 0 {
+			log.Fatal("one-shot publish requires private data, dispatcher ID, MQ environment and explicit event/task/callee")
+		}
+		data, err := loadDataset(*dataDir, *dispatcherID)
+		if err != nil {
+			log.Fatal(err)
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		if err := publishExecute(ctx, brokerURL, data, *publishID, *publishTask, *publishCallee); err != nil {
+			log.Fatal(err)
+		}
+		log.Printf("one-shot command broker-confirmed: event_id=%q task_id=%q (not SaaS result receipt)", *publishID, *publishTask)
+		return
+	}
 	if *dataDir == "" || *dispatcherID == "" || *listen == "" || *cert == "" || *key == "" || secret == "" || brokerURL == "" || flag.NArg() != 0 {
 		log.Fatal(errors.New("data, dispatcher-id, listen, tls-cert, tls-key, SAAS_MOCK_DISPATCHER_SECRET and SAAS_MOCK_RABBITMQ_URL are required"))
 	}

deploys/test/saas-mock/mq.go

@@ -12,12 +12,19 @@ import (
 
 // provisionMQ is SaaS-owned setup in a dedicated vhost. It never publishes,
 // consumes, purges, deletes, or resets an existing queue.
-func provisionMQ(brokerURL string, data dataset) error {
+func requireDedicatedVhost(brokerURL string) error {
 	address, err := url.Parse(brokerURL)
 	if err != nil || (address.Scheme != "amqp" && address.Scheme != "amqps") || address.Host == "" ||
 		!strings.HasPrefix(strings.TrimPrefix(address.Path, "/"), "saas-mock-") {
 		return errors.New("SaaS test MQ requires a dedicated saas-mock-* vhost")
 	}
+	return nil
+}
+
+func provisionMQ(brokerURL string, data dataset) error {
+	if err := requireDedicatedVhost(brokerURL); err != nil {
+		return err
+	}
 	conn, err := amqp.Dial(brokerURL)
 	if err != nil {
 		return fmt.Errorf("connect SaaS test MQ: %T", err)

deploys/test/saas-mock/publish.go

@@ -0,0 +1,106 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"regexp"
+	"time"
+
+	"git.ipao.vip/rogee/go-sip/internal/configread"
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+	"git.ipao.vip/rogee/go-sip/internal/mq"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+var mockCallID = regexp.MustCompile(`^[A-Za-z0-9._-]{1,100}$`)
+
+// buildExecute deliberately carries only the two approved call inputs. Trunk,
+// caller, AI and duration remain immutable properties of the SaaS task read.
+func buildExecute(data dataset, eventID, taskID, callee string, now time.Time) (string, []byte, error) {
+	shanghai, err := time.LoadLocation("Asia/Shanghai")
+	if err != nil {
+		return "", nil, err
+	}
+	hour := now.In(shanghai).Hour()
+	if !mockCallID.MatchString(eventID) || !mockCallID.MatchString(taskID) ||
+		(callee != "15003164745" && callee != "15830461047") || hour < 9 || hour >= 20 {
+		return "", nil, errors.New("one-shot command identity, whitelist or real call window rejected")
+	}
+	body, ok := data.tasks[taskID]
+	if !ok {
+		return "", nil, errors.New("one-shot command task is absent from the approved SaaS dataset")
+	}
+	var task configread.Task
+	if err := json.Unmarshal(body, &task); err != nil || task.DispatcherID != data.dispatcherID || task.TenantID != data.tenantID || task.TaskID != taskID || task.Status != "running" || len(task.AllowedTrunkIDs) != 1 {
+		return "", nil, errors.New("one-shot command requires a running task pinned to exactly one approved trunk")
+	}
+	command, err := json.Marshal(struct {
+		EventID      string `json:"event_id"`
+		EventType    string `json:"event_type"`
+		DispatcherID string `json:"dispatcher_id"`
+		TenantID     int64  `json:"tenant_id"`
+		IssuedAt     string `json:"issued_at"`
+		Payload      struct {
+			TaskID string `json:"task_id"`
+			Callee string `json:"callee"`
+		} `json:"payload"`
+	}{EventID: eventID, EventType: "call.execute", DispatcherID: data.dispatcherID, TenantID: data.tenantID, IssuedAt: now.UTC().Format(time.RFC3339Nano), Payload: struct {
+		TaskID string `json:"task_id"`
+		Callee string `json:"callee"`
+	}{TaskID: taskID, Callee: callee}})
+	if err != nil || contract.ValidateCurrent("mq", command) != nil {
+		return "", nil, errors.New("one-shot command violates current MQ contract")
+	}
+	return "d." + data.dispatcherID + ".task." + taskID + ".in", command, nil
+}
+
+// publishExecute publishes exactly once with mandatory routing and publisher
+// confirmation. An ambiguous response is NOT retried or called SaaS receipt.
+func publishExecute(ctx context.Context, brokerURL string, data dataset, eventID, taskID, callee string) error {
+	return publishExecuteAt(ctx, brokerURL, data, eventID, taskID, callee, time.Now())
+}
+
+func publishExecuteAt(ctx context.Context, brokerURL string, data dataset, eventID, taskID, callee string, now time.Time) error {
+	if err := requireDedicatedVhost(brokerURL); err != nil {
+		return err
+	}
+	routingKey, body, err := buildExecute(data, eventID, taskID, callee, now)
+	if err != nil {
+		return err
+	}
+	conn, err := amqp.Dial(brokerURL)
+	if err != nil {
+		return fmt.Errorf("connect one-shot test broker: %T", err)
+	}
+	defer conn.Close()
+	ch, err := conn.Channel()
+	if err != nil {
+		return fmt.Errorf("open one-shot publish channel: %T", err)
+	}
+	defer ch.Close()
+	if err := ch.Confirm(false); err != nil {
+		return fmt.Errorf("enable one-shot publisher confirmation: %T", err)
+	}
+	returned := ch.NotifyReturn(make(chan amqp.Return, 1))
+	confirmed := ch.NotifyPublish(make(chan amqp.Confirmation, 1))
+	if err := ch.PublishWithContext(ctx, mq.CommandsExchange, routingKey, true, false,
+		amqp.Publishing{ContentType: "application/json", DeliveryMode: amqp.Persistent, Body: body}); err != nil {
+		return fmt.Errorf("one-shot publish outcome unknown: %T", err)
+	}
+	select {
+	case <-ctx.Done():
+		return errors.New("one-shot publish confirmation unknown; do not republish")
+	case confirmation, ok := <-confirmed:
+		if !ok || !confirmation.Ack {
+			return errors.New("one-shot command was not publisher-confirmed; do not republish")
+		}
+		select {
+		case <-returned:
+			return errors.New("one-shot command was unroutable")
+		default:
+			return nil // broker accepted the command, NOT SaaS application receipt
+		}
+	}
+}

deploys/test/saas-mock/publish_test.go

@@ -0,0 +1,112 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+	"github.com/google/uuid"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+func TestSaaSMockPublishesOneCommandOnProvisionedRabbitMQ(t *testing.T) {
+	url := os.Getenv("SAAS_MOCK_TEST_BROKER_URL")
+	if url == "" {
+		t.Skip("requires an isolated ephemeral dedicated-vhost RabbitMQ")
+	}
+	data, err := loadDataset(testDataDir(t), testDispatcher)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := provisionMQ(url, data); err != nil {
+		t.Fatal(err)
+	}
+	found := false
+	for _, discovered := range data.discovery {
+		found = found || discovered.TaskID == "task-full"
+	}
+	if !found {
+		t.Fatal("fixture task-full was not preprovisioned")
+	}
+	conn, err := amqp.Dial(url)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	ch, err := conn.Channel()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ch.Close()
+	eventID := "test-" + uuid.NewString()
+	at := time.Date(2026, 10, 4, 2, 0, 0, 0, time.UTC)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	if err := publishExecuteAt(ctx, url, data, eventID, "task-full", "15003164745", at); err != nil {
+		t.Fatal(err)
+	}
+	queue := "agent-call.d." + testDispatcher + ".task.task-full.v1"
+	message, ok, err := ch.Get(queue, false)
+	if err != nil || !ok {
+		t.Fatalf("one command must reach exactly its preprovisioned task queue: present=%v err=%v", ok, err)
+	}
+	if err := contract.ValidateCurrent("mq", message.Body); err != nil || !strings.Contains(string(message.Body), eventID) || message.DeliveryMode != amqp.Persistent {
+		t.Fatalf("broker delivered wrong command identity or persistence: err=%v", err)
+	}
+	if err := message.Ack(false); err != nil {
+		t.Fatal(err)
+	}
+	if extra, ok, err := ch.Get(queue, false); err != nil || ok {
+		if ok {
+			_ = extra.Nack(false, true)
+		}
+		t.Fatalf("one publish created a second command: present=%v err=%v", ok, err)
+	}
+}
+
+func TestSaaSMockBuildsOneApprovedCommandForTheBoundTask(t *testing.T) {
+	data, err := loadDataset(testDataDir(t), testDispatcher)
+	if err != nil {
+		t.Fatal(err)
+	}
+	inside := time.Date(2026, 10, 4, 2, 0, 0, 0, time.UTC) // Shanghai 10:00
+	key, body, err := buildExecute(data, "event-once-1", "task-full", "15003164745", inside)
+	if err != nil || key != "d."+testDispatcher+".task.task-full.in" {
+		t.Fatalf("SaaS command must target only the assigned Dispatcher/task: key=%q err=%v", key, err)
+	}
+	if err := contract.ValidateCurrent("mq", body); err != nil {
+		t.Fatalf("call command violates the formal contract: %v", err)
+	}
+	var event struct {
+		EventID      string `json:"event_id"`
+		Type         string `json:"event_type"`
+		DispatcherID string `json:"dispatcher_id"`
+		TenantID     int64  `json:"tenant_id"`
+		Payload      struct {
+			TaskID string `json:"task_id"`
+			Callee string `json:"callee"`
+		} `json:"payload"`
+	}
+	if err := json.Unmarshal(body, &event); err != nil || event.EventID != "event-once-1" || event.Type != "call.execute" || event.DispatcherID != testDispatcher || event.TenantID != 1001 || event.Payload.TaskID != "task-full" || event.Payload.Callee != "15003164745" || strings.Contains(string(body), "trunk-mock") {
+		t.Fatalf("SaaS must not leak a trunk/caller/AI override into the execute command: %+v err=%v", event, err)
+	}
+	for _, test := range []struct {
+		id, task, callee string
+		at               time.Time
+	}{
+		{"../event", "task-full", "15003164745", inside},
+		{"event-2", "missing", "15003164745", inside},
+		{"event-3", "task-full", "708915003164745", inside},
+		{"event-4", "task-full", "13900000000", inside},
+		{"event-5", "task-full", "15003164745", inside.Add(-2 * time.Hour)},
+		{"event-6", "task-full", "15003164745", inside.Add(10 * time.Hour)},
+	} {
+		if _, _, err := buildExecute(data, test.id, test.task, test.callee, test.at); err == nil {
+			t.Fatalf("invalid or out-of-window command was allowed: event=%q task=%q", test.id, test.task)
+		}
+	}
+}

scripts/check-current-mq-mock.sh

@@ -45,12 +45,15 @@ d_pw=$(openssl rand -hex 12)
 if ! docker exec -u rabbitmq "$name" rabbitmqctl add_user saas_mock "$saas_pw" >/dev/null 2>&1 ||
    ! docker exec -u rabbitmq "$name" rabbitmqctl add_user dispatcher_mock "$d_pw" >/dev/null 2>&1 ||
    ! docker exec -u rabbitmq "$name" rabbitmqctl set_permissions -p / saas_mock '.*' '.*' '.*' >/dev/null 2>&1 ||
-   ! docker exec -u rabbitmq "$name" rabbitmqctl set_permissions -p / dispatcher_mock '^$' '^agent-call\.saas\.v1$' '^agent-call\.d\..*\.(control|task\..*)\.v1$' >/dev/null 2>&1; then
+   ! docker exec -u rabbitmq "$name" rabbitmqctl set_permissions -p / dispatcher_mock '^$' '^agent-call\.saas\.v1$' '^agent-call\.d\..*\.(control|task\..*)\.v1$' >/dev/null 2>&1 ||
+   ! docker exec -u rabbitmq "$name" rabbitmqctl add_vhost saas-mock-go-sip-test >/dev/null 2>&1 ||
+   ! docker exec -u rabbitmq "$name" rabbitmqctl set_permissions -p saas-mock-go-sip-test saas_mock '.*' '.*' '.*' >/dev/null 2>&1; then
   echo 'isolated RabbitMQ Mock user/permission setup failed' >&2
   exit 1
 fi
 export RABBITMQ_URL="amqp://dispatcher_mock:${d_pw}@127.0.0.1:${port}/"
 export RABBITMQ_PROVISIONER_URL="amqp://saas_mock:${saas_pw}@127.0.0.1:${port}/"
+export SAAS_MOCK_TEST_BROKER_URL="amqp://saas_mock:${saas_pw}@127.0.0.1:${port}/saas-mock-go-sip-test"
 run_required_test() {
   local package=$1 test_name=$2 output
   if ! output=$(go test -tags=integration "$package" -run "^${test_name}$" -count=1 -v 2>&1); then
@@ -66,3 +69,4 @@ run_required_test() {
 run_required_test ./internal/mq TestBrokerSharedResultQueueAndNoConfigure
 run_required_test ./internal/dispatcher TestRuntimeIsolatedControlBacklogExecuteAndSharedResult
 run_required_test ./cmd/sip-go-agent TestDispatcherCommandStartsWithIsolatedMQHTTPAndAgent
+run_required_test ./deploys/test/saas-mock TestSaaSMockPublishesOneCommandOnProvisionedRabbitMQ