feat(saas-server): add explicit test publishing and feedback verification

cmd/saas-server/README.md

@@ -0,0 +1,132 @@
+# SaaS Server：最小外呼测试工具
+
+使用 Fiber/v3，代替浏览器操作：提供 Dispatcher 所需配置，手动向测试队列发送一条外呼，查看回执和最终结果。
+
+**不修改 Dispatcher/Agent，不绕过它们的业务检查。启动不自动发消息。仅支持专用 `saas-mock-*` vhost，不能和真实 SaaS 共用反馈队列。** 本工具不授权真实拨号或真实 AI/OSS 调用。
+
+## 启动
+
+```sh
+go build -o /tmp/saas-server ./cmd/saas-server
+/tmp/saas-server --config /absolute/path/server.json
+```
+
+`server.json` 示例（全部为占位，须换成已批准的测试配置）：
+
+```json
+{
+  "listen": "0.0.0.0:8089",
+  "dispatcher_id": "c046b893-8628-4589-ae50-619d049248a6",
+  "dispatcher_secret": "TEST_DISPATCHER_SECRET",
+  "broker_url": "amqp://TEST_USER:TEST_PASSWORD@127.0.0.1:5672/saas-mock-example",
+  "snapshots": "snapshots",
+  "feedback_dir": "feedback",
+  "tls_cert": "dispatcher-test.crt",
+  "tls_key": "dispatcher-test.key"
+}
+```
+
+- 相对路径以 `server.json` 所在目录为基准。
+- `listen` 省略时为 `0.0.0.0:8089`。
+- TLS 两项可同时省略，服务改为 HTTP；需要遵循现有 Dispatcher 客户端的连接要求，不能为测试放松客户端校验。
+- 启动文件和各快照必须是私有普通文件（如 `0600`）。反馈目录独立且私有（如 `0700`），不能与输入配置目录交叠。
+- MQ 用户必须能够在这个专用 vhost 声明、绑定和使用现行拓扑。已有拓扑不匹配或已有独占反馈消费者时明确失败；不会清队列或自动恢复连接。
+
+快照目录：
+
+```text
+snapshots/
+  sip.json
+  providers.json
+  quota.json
+  tasks/
+    <task_id>.json
+```
+
+JSON 内容直接采用 [`../../contracts/schema/`](../../contracts/schema/README.md) 的现行配置响应格式，不使用旧 ENV 替代。所有资源必须属于启动文件中的 Dispatcher；首期仅一个租户，任务文件名与 task_id 一致。任务列表由详情生成，不另建列表文件。
+
+AI 连接、模型、音色、参数、SIP 线路、额度和时段来自这些文件；合同示例只是结构参考，不保证线路具备执行参数，也不是拨号授权。
+
+另行启动使用这套测试配置及专用 MQ 的 Dispatcher/Agent；本工具不启动、停止、重启或修改现有服务。五类正式配置接口校验现有 `X-DISPATCHER-id` 和 `X-DISPATCHER-SECRET-KEY` 请求头。
+
+## 手动发送一条外呼
+
+以下命令以本地 HTTP 为例；启用 TLS 时改用 HTTPS 并使用测试 CA 的 `--cacert`，不要关闭证书校验。
+
+```sh
+BASE=http://127.0.0.1:8089
+# 任务、号码必须是本次获准的测试数据；123456 只是占位。
+curl --fail-with-body "$BASE/test/calls" \
+  -H 'Content-Type: application/json' \
+  -d '{"event_id":"test-numberID-001","task_id":"1","callee":"123456"}'
+```
+
+`event_id` 由调用者明确指定，原样保留。相同身份、相同请求在本次进程中不再发送；同身份不同内容报错。重启会清除发送记录，**不要通过重启、换 ID 或重复调用补拨**；Dispatcher 的既有重复消息规则不变。
+
+响应中的 `status`：
+
+| 值 | 含义 |
+| --- | --- |
+| `confirmed` | 队列已确认，不表示已拨号或已接通 |
+| `unroutable` | 消息被队列退回，没有对应路由 |
+| `rejected` | 队列明确拒绝 |
+| `not_sent` | 发送前失败 |
+| `unknown` | 是否发送成功不确定；禁止自动重发 |
+
+非 `confirmed` 返回 HTTP 503，并保留本次状态。输入错误返回 HTTP 400。工具没有自动重试、批量、定时或换线功能。
+
+## 查询和校验反馈
+
+```sh
+curl --fail-with-body --get "$BASE/test/events" \
+  --data-urlencode 'event_id=test-numberID-001'
+```
+
+返回本次进程内的 `requests` 与 `feedback`。派发回执和最终结果分别记录；`validation` 为 `valid` 或 `invalid`，错误原因会明确显示。等待不到反馈不等于通话失败，只有拒绝的未发起请求不一定有通话最终结果。
+
+查询只返回身份、状态、结果及校验摘要，不显示完整对话/音频等正文。收到的原消息保存在独立反馈目录的随机 `feedback-*.json` 文件中，文件保存并同步成功后才 ACK。保存失败停止消费并退出，不丢弃消息。重复或错误消息也保留证据，冲突不覆盖原记录。
+
+**重启后不恢复历史查询；旧反馈文件仍保留，供人工检查。本工具不会回写配置，也不会保存内存修改。**
+
+## 内存修改配置
+
+使用完整配置响应，不使用 PATCH 或字段合并：
+
+```sh
+curl --fail-with-body -X PUT "$BASE/test/config/sip" \
+  -H 'Content-Type: application/json' --data-binary @/path/candidate-sip.json
+curl --fail-with-body -X PUT "$BASE/test/config/providers" \
+  -H 'Content-Type: application/json' --data-binary @/path/candidate-providers.json
+curl --fail-with-body -X PUT "$BASE/test/config/quota" \
+  -H 'Content-Type: application/json' --data-binary @/path/candidate-quota.json
+curl --fail-with-body -X PUT "$BASE/test/config/tasks/1" \
+  -H 'Content-Type: application/json' --data-binary @/path/candidate-task.json
+```
+
+校验成功后整体替换内存，失败不改变当前值。只能修改已加载资源，不能增删任务或更换租户/Dispatcher 身份。重启重新加载原文件。
+
+修改配置不会自动通知 Dispatcher，需要显式操作：
+
+```sh
+# action 只支持 start、resume、edit；每条指令使用明确的源 event_id。
+curl --fail-with-body "$BASE/test/task-controls" \
+  -H 'Content-Type: application/json' \
+  -d '{"event_id":"test-edit-001","task_id":"1","action":"edit","reason":"test configuration change"}'
+
+# change_type 按现行合同：enabled、disabled、removed、created；revision 为正整数。
+curl --fail-with-body "$BASE/test/sip-refresh" \
+  -H 'Content-Type: application/json' \
+  -d '{"event_id":"test-sip-001","trunk_id":"trunk-mock","change_type":"enabled","revision":8}'
+```
+
+任务/额度按现行 start/resume/edit 流程读取，edit 不解除暂停、不恢复已停止任务。SIP 通知触发现行加载流程；已经发出的通话不改变快照。AI providers 只在 Dispatcher 启动读取，修改后须另行安排重启；本工具不重启服务。
+
+## 本地验证
+
+```sh
+go test -race ./cmd/saas-server
+# 需 Docker 和已存在的 rabbitmq:4.1-management-alpine 镜像，不自动拉取。
+bash cmd/saas-server/check-integration.sh
+```
+
+集成脚本只创建并清理自己的临时 RabbitMQ 容器，运行 Fiber HTTPS → RabbitMQ → 未修改的 Dispatcher → Mock Agent mTLS → 回执/最终结果测试。使用临时配置、数据库和合成音频，不拨号，不访问真实 AI/OSS。通过不代表真实服务或通话验收。

cmd/saas-server/check-integration.sh

@@ -0,0 +1,30 @@
+#!/usr/bin/env bash
+set -euo pipefail
+cd "$(dirname "$0")/../.."
+image=rabbitmq:4.1-management-alpine
+command -v docker >/dev/null && docker info >/dev/null
+# Never pull images or connect to an existing broker implicitly.
+docker image inspect "$image" >/dev/null
+uid=$(docker run --rm --entrypoint id "$image" -u rabbitmq)
+gid=$(docker run --rm --entrypoint id "$image" -g rabbitmq)
+name="saas-server-test-$$"
+cleanup() { docker rm -f "$name" >/dev/null 2>&1; }
+trap cleanup EXIT
+docker run -d --name "$name" --tmpfs "/var/lib/rabbitmq:rw,uid=$uid,gid=$gid,mode=0700" \
+  -p 127.0.0.1::5672 "$image" >/dev/null
+ready=0
+for _ in $(seq 1 60); do
+  if docker exec -u rabbitmq "$name" rabbitmq-diagnostics -q check_running >/dev/null 2>&1; then ready=1; break; fi
+  sleep 1
+done
+if [ "$ready" -ne 1 ]; then echo 'isolated RabbitMQ not ready' >&2; exit 1; fi
+password=$(openssl rand -hex 12)
+docker exec -u rabbitmq "$name" rabbitmqctl add_user saas_test "$password" >/dev/null
+docker exec -u rabbitmq "$name" rabbitmqctl add_vhost saas-mock-server-test >/dev/null
+docker exec -u rabbitmq "$name" rabbitmqctl set_permissions -p saas-mock-server-test saas_test '.*' '.*' '.*' >/dev/null
+port=$(docker port "$name" 5672/tcp | awk -F: 'NR==1{print $NF}')
+export SAAS_MOCK_TEST_BROKER_URL="amqp://saas_test:${password}@127.0.0.1:${port}/saas-mock-server-test"
+output=$(go test -race -tags=integration ./cmd/saas-server -run '^Test(SaaSServerIsolatedHTTPSMQAgentMTLS|StartupShutdownPreservesConfiguration)$' -count=1 -v 2>&1) || { printf '%s\n' "$output"; exit 1; }
+printf '%s\n' "$output"
+grep -Fq -- '--- PASS: TestSaaSServerIsolatedHTTPSMQAgentMTLS (' <<<"$output"
+grep -Fq -- '--- PASS: TestStartupShutdownPreservesConfiguration (' <<<"$output"

cmd/saas-server/config.go

@@ -21,7 +21,8 @@ type snapshots struct {
 
 func validateResource(raw []byte, resource, dispatcherID string) error {
 	if err := contract.ValidateCurrent("config-read", raw); err != nil {
-		return fmt.Errorf("%s violates configuration contract", resource)
+		// Existing contract diagnostics report field paths without private values.
+		return fmt.Errorf("%s: %w", resource, err)
 	}
 	var identity struct {
 		Resource     string `json:"resource"`

cmd/saas-server/diagnostics_test.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestSchemaErrorsIdentifyFieldWithoutPrivateValue(t *testing.T) {
+	raw, err := os.ReadFile("../../contracts/schema/examples/config-read-providers.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var body map[string]any
+	json.Unmarshal(raw, &body)
+	provider := body["providers"].([]any)[0].(map[string]any)
+	provider["provider_code"] = "PRIVATE_VALUE_MUST_NOT_APPEAR"
+	provider["api_key"] = map[string]string{"secret": "PRIVATE_VALUE_MUST_NOT_APPEAR"}
+	raw, _ = json.Marshal(body)
+	err = validateResource(raw, "ai_providers", body["dispatcher_id"].(string))
+	if err == nil || !strings.Contains(err.Error(), "api_key") || strings.Contains(err.Error(), "PRIVATE_VALUE_MUST_NOT_APPEAR") {
+		t.Fatalf("unsafe or opaque diagnostic: %v", err)
+	}
+}

cmd/saas-server/events.go

@@ -0,0 +1,231 @@
+package main
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"time"
+
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+)
+
+type envelope struct {
+	EventID      string         `json:"event_id"`
+	EventType    string         `json:"event_type"`
+	DispatcherID string         `json:"dispatcher_id"`
+	TenantID     int64          `json:"tenant_id,omitempty"`
+	IssuedAt     string         `json:"issued_at,omitempty"`
+	Payload      map[string]any `json:"payload"`
+}
+
+type commandInput struct {
+	EventID    string `json:"event_id"`
+	TaskID     string `json:"task_id,omitempty"`
+	Callee     string `json:"callee,omitempty"`
+	Action     string `json:"action,omitempty"`
+	Reason     string `json:"reason,omitempty"`
+	TrunkID    string `json:"trunk_id,omitempty"`
+	ChangeType string `json:"change_type,omitempty"`
+	Revision   *int64 `json:"revision,omitempty"`
+}
+
+type requestRecord struct {
+	Input     commandInput `json:"-"`
+	Digest    string       `json:"-"`
+	EventID   string       `json:"event_id"`
+	EventType string       `json:"event_type"`
+	Status    string       `json:"status"`
+	Error     string       `json:"error,omitempty"`
+}
+
+type feedbackRecord struct {
+	EventID      string `json:"event_id"`
+	EventType    string `json:"event_type"`
+	DispatcherID string `json:"dispatcher_id"`
+	Status       string `json:"status,omitempty"`
+	Outcome      string `json:"outcome,omitempty"`
+	Validation   string `json:"validation"`
+	Error        string `json:"error,omitempty"`
+	Duplicates   int    `json:"duplicates"`
+	Digest       string `json:"-"`
+}
+
+func digest(raw []byte) string { sum := sha256.Sum256(raw); return hex.EncodeToString(sum[:]) }
+func messageKey(dispatcherID, eventType, eventID string) string {
+	raw, _ := json.Marshal([]string{dispatcherID, eventType, eventID})
+	return string(raw)
+}
+
+func (s *service) send(ctx context.Context, kind string, input commandInput, now time.Time) (requestRecord, error) {
+	s.eventMu.Lock()
+	defer s.eventMu.Unlock()
+	s.mu.RLock()
+	data := s.data
+	s.mu.RUnlock()
+	msg := envelope{EventID: input.EventID, EventType: kind, DispatcherID: data.dispatcherID, Payload: make(map[string]any)}
+	route := "d." + data.dispatcherID + ".control.in"
+	switch kind {
+	case "call.execute", "task.control":
+		if _, ok := data.tasks[input.TaskID]; !ok {
+			return requestRecord{}, errors.New("unknown task")
+		}
+		msg.TenantID = data.tenantID
+		msg.IssuedAt = now.UTC().Format(time.RFC3339Nano)
+		msg.Payload["task_id"] = input.TaskID
+		if kind == "call.execute" {
+			msg.Payload["callee"] = input.Callee
+			route = "d." + data.dispatcherID + ".task." + input.TaskID + ".in"
+		} else {
+			if input.Action != "start" && input.Action != "resume" && input.Action != "edit" {
+				return requestRecord{}, errors.New("only start, resume and edit are supported")
+			}
+			msg.Payload["action"] = input.Action
+			msg.Payload["reason"] = input.Reason
+		}
+	case "sip.config":
+		msg.Payload["trunk_id"] = input.TrunkID
+		msg.Payload["change_type"] = input.ChangeType
+		if input.Revision != nil {
+			msg.Payload["revision"] = *input.Revision
+		}
+	default:
+		return requestRecord{}, errors.New("unsupported command")
+	}
+	raw, err := json.Marshal(msg)
+	if err != nil {
+		return requestRecord{}, err
+	}
+	if err = contract.ValidateCurrent("mq", raw); err != nil {
+		return requestRecord{}, fmt.Errorf("command: %w", err)
+	}
+	key := messageKey(data.dispatcherID, kind, input.EventID)
+	inputRaw, _ := json.Marshal(input)
+	hash := digest(inputRaw)
+	if previous, ok := s.requests[key]; ok {
+		if previous.Digest != hash {
+			return requestRecord{}, errors.New("event identity already used with different content")
+		}
+		return previous, nil
+	}
+	if s.publish == nil {
+		return requestRecord{}, errors.New("MQ publisher unavailable")
+	}
+	record := requestRecord{Input: input, Digest: hash, EventID: input.EventID, EventType: kind, Status: "unknown"}
+	s.requests[key] = record
+	slog.Info("Test command publishing", "event_id", input.EventID, "event_type", kind, "stage", "publish_requested")
+	status, publishErr := s.publish(ctx, route, raw)
+	record.Status = status
+	if publishErr != nil {
+		record.Error = publishErr.Error()
+	}
+	s.requests[key] = record
+	slog.Info("Test command publish outcome", "event_id", input.EventID, "event_type", kind, "status", record.Status, "error", record.Error)
+	return record, nil
+}
+
+func (s *service) receive(raw []byte, ack func() error) error {
+	s.eventMu.Lock()
+	defer s.eventMu.Unlock()
+	if err := saveFeedback(s.feedbackDir, raw); err != nil {
+		return err
+	}
+	s.mu.RLock()
+	dispatcherID, tenantID := s.data.dispatcherID, s.data.tenantID
+	s.mu.RUnlock()
+	var msg envelope
+	record := feedbackRecord{Validation: "valid", Digest: digest(raw)}
+	err := json.Unmarshal(raw, &msg)
+	record.EventID = msg.EventID
+	record.EventType = msg.EventType
+	record.DispatcherID = msg.DispatcherID
+	schemaErr := contract.ValidateCurrent("mq", raw)
+	// The shared schema validates these enum fields; do not maintain another value list.
+	if err == nil && schemaErr == nil {
+		record.Status, _ = msg.Payload["status"].(string)
+		record.Outcome, _ = msg.Payload["outcome"].(string)
+	}
+	invalid := func(reason string) { record.Validation = "invalid"; record.Error = reason }
+	if err != nil || schemaErr != nil {
+		invalid("feedback violates current MQ contract")
+	} else if msg.DispatcherID != dispatcherID {
+		invalid("feedback belongs to another dispatcher")
+	} else {
+		requestKind := msg.EventType
+		if msg.EventType == "call.execute.result" {
+			requestKind = "call.execute"
+		} else if (msg.EventType != "call.execute" && msg.EventType != "task.control" && msg.EventType != "sip.config") || record.Status == "" {
+			invalid("message is not an outbound receipt")
+		}
+		request, ok := s.requests[messageKey(msg.DispatcherID, requestKind, msg.EventID)]
+		if !ok {
+			invalid("feedback has no request in this process")
+		} else {
+			if msg.TenantID != 0 && msg.TenantID != tenantID {
+				invalid("feedback tenant mismatch")
+			}
+			for field, want := range map[string]string{"task_id": request.Input.TaskID, "callee": request.Input.Callee, "action": request.Input.Action} {
+				if value, exists := msg.Payload[field]; exists && value != want {
+					invalid("feedback request association mismatch")
+				}
+			}
+		}
+	}
+	for i, previous := range s.feedback {
+		if messageKey(previous.DispatcherID, previous.EventType, previous.EventID) != messageKey(record.DispatcherID, record.EventType, record.EventID) {
+			continue
+		}
+		if previous.Digest == record.Digest {
+			s.feedback[i].Duplicates++
+			return ack()
+		}
+		invalid("conflicting feedback for same identity")
+	}
+	s.feedback = append(s.feedback, record)
+	slog.Info("Test feedback saved and validated", "event_id", record.EventID, "event_type", record.EventType, "validation", record.Validation, "error", record.Error)
+	return ack()
+}
+
+func saveFeedback(dir string, raw []byte) error {
+	f, err := os.CreateTemp(dir, "feedback-*.json")
+	if err != nil {
+		return fmt.Errorf("create feedback evidence: %w", err)
+	}
+	if _, err = f.Write(raw); err == nil {
+		err = f.Sync()
+	}
+	closeErr := f.Close()
+	if err == nil {
+		err = closeErr
+	}
+	if err != nil {
+		return fmt.Errorf("save feedback evidence: %w", err)
+	}
+	parent, err := os.Open(filepath.Dir(f.Name()))
+	if err != nil {
+		return err
+	}
+	syncErr := parent.Sync()
+	closeErr = parent.Close()
+	if syncErr != nil {
+		return syncErr
+	}
+	return closeErr
+}
+
+func (s *service) events(id string) []feedbackRecord {
+	s.eventMu.Lock()
+	defer s.eventMu.Unlock()
+	records := []feedbackRecord{}
+	for _, record := range s.feedback {
+		if record.EventID == id {
+			records = append(records, record)
+		}
+	}
+	return records
+}

cmd/saas-server/events_test.go

@@ -0,0 +1,143 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestCommandIdentityDedupAndUnknown(t *testing.T) {
+	dir, id := fixture(t)
+	data, err := loadSnapshots(dir, id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	svc := newService(data, "secret")
+	sends := 0
+	svc.publish = func(_ context.Context, route string, raw []byte) (string, error) {
+		sends++
+		if !strings.HasSuffix(route, ".task.1.in") {
+			t.Fatal(route)
+		}
+		var msg envelope
+		if err := json.Unmarshal(raw, &msg); err != nil {
+			t.Fatal(err)
+		}
+		if msg.EventID != "source/numberID:42" {
+			t.Fatal(msg.EventID)
+		}
+		return "unknown", errors.New("confirmation missing")
+	}
+	req := commandInput{EventID: "source/numberID:42", TaskID: "1", Callee: "123456"}
+	first, err := svc.send(context.Background(), "call.execute", req, time.Now())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if first.Status != "unknown" {
+		t.Fatal(first)
+	}
+	second, err := svc.send(context.Background(), "call.execute", req, time.Now())
+	if err != nil || second.Status != "unknown" || sends != 1 {
+		t.Fatalf("duplicate sends=%d result=%+v err=%v", sends, second, err)
+	}
+	req.Callee = "999"
+	if _, err = svc.send(context.Background(), "call.execute", req, time.Now()); err == nil {
+		t.Fatal("accepted conflicting source identity")
+	}
+	req.EventID = ""
+	req.Callee = "123456"
+	if _, err = svc.send(context.Background(), "call.execute", req, time.Now()); err == nil {
+		t.Fatal("accepted empty source identity")
+	}
+}
+
+func TestFeedbackValidationAndSaveBeforeAck(t *testing.T) {
+	dir, id := fixture(t)
+	data, err := loadSnapshots(dir, id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	svc := newService(data, "secret")
+	svc.publish = func(context.Context, string, []byte) (string, error) { return "confirmed", nil }
+	if _, err = svc.send(context.Background(), "call.execute", commandInput{EventID: "call-example", TaskID: "1", Callee: "15003164745"}, time.Now()); err != nil {
+		t.Fatal(err)
+	}
+	svc.feedbackDir = t.TempDir()
+	raw, err := os.ReadFile("../../contracts/schema/examples/mq-execute-ack.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ack := false
+	if err = svc.receive(raw, func() error {
+		files, e := os.ReadDir(svc.feedbackDir)
+		if e != nil || len(files) != 1 {
+			t.Fatalf("ACK before save: %v %v", files, e)
+		}
+		ack = true
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if !ack {
+		t.Fatal("no ACK")
+	}
+	records := svc.events("call-example")
+	if len(records) != 1 || records[0].Validation != "valid" {
+		t.Fatalf("%+v", records)
+	}
+	if err = svc.receive(raw, func() error { return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if got := svc.events("call-example"); len(got) != 1 || got[0].Duplicates != 1 {
+		t.Fatalf("duplicates %+v", got)
+	}
+	result, err := os.ReadFile("../../contracts/schema/examples/mq-result-no-recording.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var msg envelope
+	json.Unmarshal(result, &msg)
+	msg.Payload["task_id"] = "1"
+	msg.Payload["outcome"] = "answered" // Regression: keep every schema-valid outcome visible.
+	result, _ = json.Marshal(msg)
+	if err = svc.receive(result, func() error { return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if records := svc.events("call-example"); len(records) != 2 || records[1].Outcome != "answered" {
+		t.Fatalf("receipt overwritten or valid outcome hidden: %+v", records)
+	}
+	msg.Payload["callee"] = "999"
+	result, _ = json.Marshal(msg)
+	if err = svc.receive(result, func() error { return nil }); err != nil {
+		t.Fatal(err)
+	}
+	got := svc.events("call-example")
+	if len(got) != 3 || got[2].Validation != "invalid" {
+		t.Fatalf("conflict=%+v", got)
+	}
+	bad := []byte(`{"event_id":"bad","token":"must-not-leak"}`)
+	if err = svc.receive(bad, func() error { return nil }); err != nil {
+		t.Fatal(err)
+	}
+	encoded, _ := json.Marshal(svc.events("bad"))
+	if strings.Contains(string(encoded), "must-not-leak") {
+		t.Fatal("raw body in query")
+	}
+	// A regular file instead of a directory forces saving to fail before ACK.
+	svc.feedbackDir = filepath.Join(svc.feedbackDir, "file")
+	os.WriteFile(svc.feedbackDir, []byte("x"), 0600)
+	ack = false
+	if err = svc.receive(raw, func() error { ack = true; return nil }); err == nil || ack {
+		t.Fatalf("save failure swallowed or ACKed: %v %v", err, ack)
+	}
+	fresh := newService(data, "secret")
+	fresh.feedbackDir = svc.feedbackDir
+	if len(fresh.events("call-example")) != 0 {
+		t.Fatal("restored history")
+	}
+}

cmd/saas-server/http.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"crypto/subtle"
 	"strconv"
 	"sync"
@@ -9,13 +10,18 @@ import (
 )
 
 type service struct {
-	app  *fiber.App
-	mu   sync.RWMutex
-	data snapshots
+	app         *fiber.App
+	mu          sync.RWMutex
+	data        snapshots
+	eventMu     sync.Mutex
+	requests    map[string]requestRecord
+	feedback    []feedbackRecord
+	feedbackDir string
+	publish     func(context.Context, string, []byte) (string, error)
 }
 
 func newService(data snapshots, secret string) *service {
-	s := &service{data: data, app: fiber.New(fiber.Config{Immutable: true})}
+	s := &service{data: data, app: fiber.New(fiber.Config{Immutable: true}), requests: make(map[string]requestRecord)}
 	const prefix = "/internal/v1/dispatcher"
 	s.app.Use(prefix, func(c fiber.Ctx) error {
 		if subtle.ConstantTimeCompare([]byte(c.Get("X-DISPATCHER-id")), []byte(data.dispatcherID)) != 1 || subtle.ConstantTimeCompare([]byte(c.Get("X-DISPATCHER-SECRET-KEY")), []byte(secret)) != 1 {
@@ -61,5 +67,6 @@ func newService(data snapshots, secret string) *service {
 	s.app.Get(prefix+"/tenant/:tenant_id/quota", serve("quota"))
 	s.app.Get(prefix+"/task/:task_id", serve("task"))
 	s.app.Get(prefix+"/tasks", serve("tasks"))
+	s.routes()
 	return s
 }

cmd/saas-server/integration_test.go

@@ -0,0 +1,307 @@
+//go:build integration
+
+package main
+
+import (
+	"bytes"
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/json"
+	"encoding/pem"
+	"errors"
+	"log/slog"
+	"math/big"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
+	"git.ipao.vip/rogee/go-sip/internal/agent"
+	"git.ipao.vip/rogee/go-sip/internal/ai"
+	"git.ipao.vip/rogee/go-sip/internal/configread"
+	"git.ipao.vip/rogee/go-sip/internal/dispatcher"
+	"git.ipao.vip/rogee/go-sip/internal/mq"
+	"git.ipao.vip/rogee/go-sip/internal/oss"
+	"git.ipao.vip/rogee/go-sip/internal/rpc"
+	"git.ipao.vip/rogee/go-sip/internal/store"
+	"github.com/gofiber/fiber/v3"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+)
+
+func localCertificates(t *testing.T) ([]byte, []byte, []byte, []byte, []byte, *x509.Certificate, *x509.Certificate) {
+	t.Helper()
+	makeKey := func() *ecdsa.PrivateKey {
+		k, e := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+		if e != nil {
+			t.Fatal(e)
+		}
+		return k
+	}
+	caKey := makeKey()
+	now := time.Now()
+	ca := &x509.Certificate{SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "isolated-saas-test"}, NotBefore: now.Add(-time.Minute), NotAfter: now.Add(time.Hour), IsCA: true, BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign}
+	caDER, e := x509.CreateCertificate(rand.Reader, ca, ca, &caKey.PublicKey, caKey)
+	if e != nil {
+		t.Fatal(e)
+	}
+	caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: caDER})
+	leaf := func(serial int64, name string) ([]byte, []byte, *x509.Certificate) {
+		key := makeKey()
+		cert := &x509.Certificate{SerialNumber: big.NewInt(serial), Subject: pkix.Name{CommonName: name}, DNSNames: []string{name}, IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, NotBefore: now.Add(-time.Minute), NotAfter: now.Add(time.Hour), KeyUsage: x509.KeyUsageDigitalSignature, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth}}
+		raw, err := x509.CreateCertificate(rand.Reader, cert, ca, &key.PublicKey, caKey)
+		if err != nil {
+			t.Fatal(err)
+		}
+		parsed, err := x509.ParseCertificate(raw)
+		if err != nil {
+			t.Fatal(err)
+		}
+		keyDER, err := x509.MarshalECPrivateKey(key)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: raw}), pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), parsed
+	}
+	dc, dk, dl := leaf(2, "dispatcher.local")
+	ac, ak, al := leaf(3, "agent.local")
+	return caPEM, dc, dk, ac, ak, dl, al
+}
+
+func TestSaaSServerIsolatedHTTPSMQAgentMTLS(t *testing.T) {
+	address := os.Getenv("SAAS_MOCK_TEST_BROKER_URL")
+	if address == "" {
+		t.Skip("requires dedicated isolated RabbitMQ vhost")
+	}
+	dir, id := fixture(t)
+	// Synthetic test snapshots permit this Mock run at any time; no real configuration is read.
+	task, err := os.ReadFile("../../contracts/schema/examples/config-read-task-asr.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var fields map[string]any
+	json.Unmarshal(task, &fields)
+	fields["task_id"] = "1"
+	weekly := map[string]any{}
+	for _, day := range []string{"monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"} {
+		weekly[day] = []map[string]string{{"start": "00:00", "end": "24:00"}}
+	}
+	fields["schedule"] = map[string]any{"time_zone": "Asia/Shanghai", "starts_at": "2000-01-01T00:00:00+08:00", "weekly_windows": weekly}
+	task, _ = json.Marshal(fields)
+	if err = os.WriteFile(filepath.Join(dir, "tasks/1.json"), task, 0600); err != nil {
+		t.Fatal(err)
+	}
+	sipRaw, err := os.ReadFile(filepath.Join(dir, "sip.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	fields = nil // JSON decoding into an existing map retains unrelated task fields.
+	json.Unmarshal(sipRaw, &fields)
+	for _, item := range fields["trunks"].([]any) {
+		trunk := item.(map[string]any)
+		trunk["transport"] = "udp"
+		trunk["auth_mode"] = "ip"
+		trunk["registration_required"] = false
+		trunk["max_concurrent_calls"] = float64(2)
+		trunk["schedule"] = map[string]any{"time_zone": "Asia/Shanghai", "weekly_windows": weekly}
+	}
+	sipRaw, _ = json.Marshal(fields)
+	if err = os.WriteFile(filepath.Join(dir, "sip.json"), sipRaw, 0600); err != nil {
+		t.Fatal(err)
+	}
+	before := map[string][]byte{}
+	if err = filepath.WalkDir(dir, func(path string, d os.DirEntry, e error) error {
+		if e != nil {
+			return e
+		}
+		if !d.IsDir() {
+			raw, e := os.ReadFile(path)
+			if e != nil {
+				return e
+			}
+			before[path] = raw
+		}
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	data, err := loadSnapshots(dir, id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, err := connectBroker(address, data)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer b.conn.Close()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s := newService(data, "isolated-secret")
+	s.feedbackDir = t.TempDir()
+	s.publish = b.publish
+	consumeDone := make(chan error, 1)
+	go func() { consumeDone <- b.consume(ctx, s) }()
+	defer func() {
+		cancel()
+		select {
+		case e := <-consumeDone:
+			if e != nil {
+				t.Error(e)
+			}
+		case <-time.After(5 * time.Second):
+			t.Error("consumer did not stop")
+		}
+	}()
+	ca, dc, dk, ac, ak, dl, al := localCertificates(t)
+	roots := x509.NewCertPool()
+	roots.AppendCertsFromPEM(ca)
+	pair, err := tls.X509KeyPair(dc, dk)
+	if err != nil {
+		t.Fatal(err)
+	}
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	go s.app.Listener(tls.NewListener(listener, &tls.Config{Certificates: []tls.Certificate{pair}, MinVersion: tls.VersionTLS12}), fiber.ListenConfig{DisableStartupMessage: true})
+	defer s.app.ShutdownWithTimeout(5 * time.Second)
+	httpClient := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS12}}, Timeout: 5 * time.Second}
+	client, err := configread.NewClient("https://"+listener.Addr().String(), id, "isolated-secret", httpClient)
+	if err != nil {
+		t.Fatal(err)
+	}
+	db, err := store.Open(filepath.Join(t.TempDir(), "dispatcher.sqlite"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	coordinator := dispatcher.NewAgentCoordinator(time.Now)
+	signer, err := oss.NewClient(oss.Config{Endpoint: "https://127.0.0.1:1", Region: "cn-test", Bucket: "mock-bucket", AccessKeyID: "test-id", AccessKeySecret: "test-secret", MaxAssetBytes: 4096})
+	if err != nil {
+		t.Fatal(err)
+	}
+	feedbackRPC := &rpc.RecordingServer{Store: db, OSS: signer, DispatcherID: id, TrustedFingerprints: map[string]struct{}{rpc.CertificateFingerprint(al): {}}, AuthorizeSession: coordinator.AuthorizeInboundMeta}
+	serveRPC := func(cert, key []byte, handler agentpb.AgentControlServiceServer) string {
+		cfg, e := rpc.NewServerTLSConfig(ca, cert, key)
+		if e != nil {
+			t.Fatal(e)
+		}
+		l, e := net.Listen("tcp", "127.0.0.1:0")
+		if e != nil {
+			t.Fatal(e)
+		}
+		server := grpc.NewServer(grpc.Creds(credentials.NewTLS(cfg)))
+		agentpb.RegisterAgentControlServiceServer(server, handler)
+		go server.Serve(l)
+		t.Cleanup(server.Stop)
+		return l.Addr().String()
+	}
+	connectRPC := func(target, name string, cert, key []byte) agentpb.AgentControlServiceClient {
+		cfg, e := rpc.NewClientTLSConfig(ca, cert, key, name)
+		if e != nil {
+			t.Fatal(e)
+		}
+		conn, e := grpc.NewClient(target, grpc.WithTransportCredentials(credentials.NewTLS(cfg)))
+		if e != nil {
+			t.Fatal(e)
+		}
+		t.Cleanup(func() { conn.Close() })
+		return agentpb.NewAgentControlServiceClient(conn)
+	}
+	agentToDispatcher := connectRPC(serveRPC(dc, dk, feedbackRPC), "dispatcher.local", ac, ak)
+	workerFailure := make(chan error, 1)
+	var handler *rpc.Server
+	worker := &rpc.ApprovedCallWorker{Lifecycle: ctx, Calls: &agent.TaskCalls{}, Prepare: func(execution rpc.ApprovedExecution) (func(context.Context) error, error) {
+		runner := &rpc.ApprovedRecordedMockCall{InboundPCM16: bytes.Repeat([]byte{1, 0}, 1600), Script: ai.ApprovedMockScript{Turns: []ai.ApprovedMockTurn{{Transcript: "synthetic ASR test"}}}, MaxWAVBytes: 4096, ExpectedRecording: false, Outcome: "answered", ReasonMessage: "isolated synthetic media", ReportTimeout: 5 * time.Second, Delivery: &agent.RecordingDelivery{Call: agent.RecordingClient{Client: agentToDispatcher, DispatcherID: execution.DispatcherID, TenantID: execution.TenantID, SourceEventID: execution.SourceEventID, Session: func(context.Context) (*agentpb.RequestMeta, error) { return handler.ActiveSessionMeta() }}}}
+		return runner.Prepare(execution)
+	}, OnFailure: func(_ rpc.ApprovedExecution, e error) error { workerFailure <- e; return nil }}
+	handler, err = rpc.NewApprovedAgentServer(rpc.ServerOptions{Mode: "mock", StatePath: filepath.Join(t.TempDir(), "session.json"), ApprovedDispatcherID: id, Status: &agentpb.AgentStatus{AgentId: "agent-mock", CellId: "cell-mock", BootId: "boot-test", ProtocolVersion: "agent.v1"}, LoadedSIP: func(context.Context) (map[string]int64, error) { return map[string]int64{"trunk-mock": 8}, nil }, RequirePeerCertificate: true, PeerCertificateFingerprints: map[string]struct{}{rpc.CertificateFingerprint(dl): {}}}, worker)
+	if err != nil {
+		t.Fatal(err)
+	}
+	agentClient := connectRPC(serveRPC(ac, ak, handler), "agent.local", dc, dk)
+	if err = coordinator.Register("agent-mock", agentClient); err != nil {
+		t.Fatal(err)
+	}
+	activateCtx, stop := context.WithTimeout(ctx, 5*time.Second)
+	defer stop()
+	status, err := coordinator.Probe(activateCtx, "agent-mock", "cell-mock")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err = coordinator.Activate(activateCtx, id, "agent-mock", "cell-mock", status.BootId, "test-epoch", 1); err != nil {
+		t.Fatal(err)
+	}
+	originator := &dispatcher.ApprovedOriginator{DispatcherID: id, Client: agentClient, Meta: func(ctx context.Context) (*agentpb.RequestMeta, error) {
+		return coordinator.ApprovedMeta(ctx, "agent-mock")
+	}}
+	dispatcherBroker, err := mq.Open(address, id, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer dispatcherBroker.Close()
+	runtime := &dispatcher.Runtime{Broker: dispatcherBroker, Bootstrap: dispatcher.Bootstrap{DispatcherID: id, Client: client, Store: db, VerifySIP: originator.VerifySIP}, Execute: dispatcher.ExecuteController{DispatcherID: id, Store: db, Originator: originator, Publisher: dispatcherBroker, Now: time.Now}, Control: dispatcher.ControlController{DispatcherID: id, Store: db, Client: client, Agent: originator, Now: time.Now}, PollInterval: 20 * time.Millisecond, Logger: slog.Default()}
+	runtimeDone := make(chan error, 1)
+	go func() { runtimeDone <- runtime.Serve(ctx) }()
+	defer func() {
+		cancel()
+		select {
+		case e := <-runtimeDone:
+			if e != nil && !errors.Is(e, context.Canceled) {
+				t.Error(e)
+			}
+		case <-time.After(5 * time.Second):
+			t.Error("Dispatcher did not stop")
+		}
+	}()
+	sourceID := "numberID-integration/Original:001"
+	// The first request may queue while bootstrap completes; no direct Agent call or forged feedback.
+	code, body := request(t, s, "POST", "/test/calls", []byte(`{"event_id":"`+sourceID+`","task_id":"1","callee":"123456"}`))
+	if code != 200 {
+		t.Fatalf("publish %d %s", code, body)
+	}
+	until := time.NewTimer(15 * time.Second)
+	defer until.Stop()
+	for {
+		records := s.events(sourceID)
+		if len(records) >= 2 {
+			seen := map[string]bool{}
+			for _, record := range records {
+				if record.Validation != "valid" {
+					t.Fatalf("feedback: %+v", record)
+				}
+				seen[record.EventType] = true
+			}
+			if !seen["call.execute"] || !seen["call.execute.result"] {
+				t.Fatalf("missing receipt or result: %+v", records)
+			}
+			break
+		}
+		select {
+		case e := <-workerFailure:
+			t.Fatal(e)
+		case <-until.C:
+			t.Fatalf("feedback timeout: %+v", records)
+		case <-time.After(20 * time.Millisecond):
+		}
+	}
+	for path, want := range before {
+		got, e := os.ReadFile(path)
+		if e != nil || !bytes.Equal(want, got) {
+			t.Fatalf("input changed: %s %v", path, e)
+		}
+	}
+	files, err := os.ReadDir(s.feedbackDir)
+	if err != nil || len(files) < 2 {
+		t.Fatalf("feedback evidence missing: %v %v", files, err)
+	}
+	t.Log("Fiber HTTPS → RabbitMQ → unchanged Dispatcher → Mock Agent mTLS → original-ID receipt/result verified; no SIP or real providers")
+}

cmd/saas-server/lifecycle_integration_test.go

@@ -0,0 +1,95 @@
+//go:build integration
+
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestStartupShutdownPreservesConfiguration(t *testing.T) {
+	address := os.Getenv("SAAS_MOCK_TEST_BROKER_URL")
+	if address == "" {
+		t.Skip("requires isolated test broker")
+	}
+	dir, id := fixture(t)
+	reserve, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	listen := reserve.Addr().String()
+	reserve.Close()
+	config := startupConfig{Listen: listen, DispatcherID: id, DispatcherSecret: "test-secret", BrokerURL: address, Snapshots: ".", Feedback: filepath.Join(t.TempDir(), "feedback")}
+	raw, _ := json.Marshal(config)
+	path := filepath.Join(dir, "server.json")
+	if err = os.WriteFile(path, raw, 0600); err != nil {
+		t.Fatal(err)
+	}
+	before := map[string][]byte{}
+	if err = filepath.WalkDir(dir, func(p string, d os.DirEntry, e error) error {
+		if e != nil {
+			return e
+		}
+		if !d.IsDir() {
+			before[p], e = os.ReadFile(p)
+		}
+		return e
+	}); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- run(ctx, path) }()
+	ready := false
+	deadline := time.Now().Add(5 * time.Second)
+	client := &http.Client{Timeout: time.Second}
+	for time.Now().Before(deadline) {
+		req, _ := http.NewRequest("GET", "http://"+listen+"/internal/v1/dispatcher/sip", nil)
+		req.Header.Set("X-DISPATCHER-id", id)
+		req.Header.Set("X-DISPATCHER-SECRET-KEY", "test-secret")
+		resp, e := client.Do(req)
+		if e == nil {
+			resp.Body.Close()
+			if resp.StatusCode == 200 {
+				ready = true
+				break
+			}
+		}
+		select {
+		case e = <-done:
+			t.Fatalf("startup failed: %v", e)
+		default:
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	if !ready {
+		t.Fatal("server did not become ready")
+	}
+	cancel()
+	select {
+	case err = <-done:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("shutdown timed out")
+	}
+	for p, want := range before {
+		got, e := os.ReadFile(p)
+		if e != nil || !bytes.Equal(got, want) {
+			t.Fatalf("configuration changed on exit: %s %v", p, e)
+		}
+	}
+	files, err := os.ReadDir(config.Feedback)
+	if err != nil || len(files) != 0 {
+		t.Fatalf("startup produced unsolicited feedback: %v %v", files, err)
+	}
+}

cmd/saas-server/main.go

@@ -0,0 +1,99 @@
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"flag"
+	"fmt"
+	"log/slog"
+	"net"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/gofiber/fiber/v3"
+)
+
+func run(ctx context.Context, path string) error {
+	config, err := loadStartup(path)
+	if err != nil {
+		return err
+	}
+	data, err := loadSnapshots(config.Snapshots, config.DispatcherID)
+	if err != nil {
+		return err
+	}
+	slog.Info("test configurations loaded", "dispatcher_id", data.dispatcherID, "tasks", len(data.tasks))
+	if err = os.MkdirAll(config.Feedback, 0700); err != nil {
+		return fmt.Errorf("create private feedback directory: %w", err)
+	}
+	info, err := os.Stat(config.Feedback)
+	if err != nil {
+		return err
+	}
+	if !info.IsDir() || info.Mode().Perm()&0077 != 0 {
+		return errors.New("feedback directory must be private")
+	}
+	b, err := connectBroker(config.BrokerURL, data)
+	if err != nil {
+		return err
+	}
+	defer b.conn.Close()
+	slog.Info("isolated MQ topology ready", "dispatcher_id", data.dispatcherID)
+	listener, err := net.Listen("tcp", config.Listen)
+	if err != nil {
+		return fmt.Errorf("listen: %w", err)
+	}
+	defer listener.Close()
+	if config.TLSCert != "" {
+		pair, e := tls.LoadX509KeyPair(config.TLSCert, config.TLSKey)
+		if e != nil {
+			return fmt.Errorf("load configured TLS pair: %w", e)
+		}
+		listener = tls.NewListener(listener, &tls.Config{Certificates: []tls.Certificate{pair}, MinVersion: tls.VersionTLS12})
+	}
+	s := newService(data, config.DispatcherSecret)
+	s.feedbackDir = config.Feedback
+	s.publish = b.publish
+	child, cancel := context.WithCancel(ctx)
+	defer cancel()
+	errs := make(chan error, 2)
+	go func() { errs <- b.consume(child, s) }()
+	go func() { errs <- s.app.Listener(listener, fiber.ListenConfig{DisableStartupMessage: true}) }()
+	slog.Info("SaaS test server listening", "address", listener.Addr().String(), "tls", config.TLSCert != "", "automatic_commands", false)
+	select {
+	case <-ctx.Done():
+		err = nil
+	case err = <-errs:
+		if err == nil && ctx.Err() == nil {
+			err = errors.New("SaaS test server stopped unexpectedly")
+		}
+	}
+	cancel()
+	if shutdownErr := s.app.ShutdownWithTimeout(5 * time.Second); shutdownErr != nil {
+		err = errors.Join(err, fmt.Errorf("HTTP shutdown: %w", shutdownErr))
+	}
+	// Closing the listener/connection unblocks both workers; do not leave consumers behind.
+	listener.Close()
+	b.conn.Close()
+	slog.Info("SaaS test server stopped")
+	return err
+}
+
+func main() {
+	var path string
+	flag.StringVar(&path, "config", "", "private startup JSON (required)")
+	flag.Parse()
+	if path == "" || flag.NArg() != 0 {
+		fmt.Fprintln(os.Stderr, "usage: saas-server --config /path/server.json")
+		os.Exit(2)
+	}
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer cancel()
+	if err := run(ctx, path); err != nil {
+		slog.Error("SaaS test server failed", "error", err)
+		os.Exit(1)
+	}
+}

cmd/saas-server/mq.go

@@ -0,0 +1,139 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"git.ipao.vip/rogee/go-sip/internal/mq"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+const feedbackQueue = "agent-call.saas.events.v1"
+
+type broker struct{ conn *amqp.Connection }
+
+func connectBroker(address string, data snapshots) (*broker, error) {
+	conn, err := amqp.DialConfig(address, amqp.Config{Dial: amqp.DefaultDial(5 * time.Second)})
+	if err != nil {
+		return nil, fmt.Errorf("connect test MQ: %w", err)
+	}
+	b := &broker{conn: conn}
+	if err = b.provision(data); err != nil {
+		conn.Close()
+		return nil, err
+	}
+	return b, nil
+}
+
+func (b *broker) provision(data snapshots) error {
+	ch, err := b.conn.Channel()
+	if err != nil {
+		return fmt.Errorf("open topology channel: %w", err)
+	}
+	defer ch.Close()
+	for _, exchange := range []string{mq.CommandsExchange, mq.ResultsExchange} {
+		if err = ch.ExchangeDeclare(exchange, "topic", true, false, false, false, nil); err != nil {
+			return fmt.Errorf("declare %s: %w", exchange, err)
+		}
+	}
+	bind := func(queue, route, exchange string) error {
+		if _, e := ch.QueueDeclare(queue, true, false, false, false, nil); e != nil {
+			return fmt.Errorf("declare %s: %w", queue, e)
+		}
+		if e := ch.QueueBind(queue, route, exchange, false, nil); e != nil {
+			return fmt.Errorf("bind %s: %w", queue, e)
+		}
+		return nil
+	}
+	if err = bind(mq.ControlQueueName(data.dispatcherID), "d."+data.dispatcherID+".control.in", mq.CommandsExchange); err != nil {
+		return err
+	}
+	for id := range data.tasks {
+		if err = bind("agent-call.d."+data.dispatcherID+".task."+id+".v1", "d."+data.dispatcherID+".task."+id+".in", mq.CommandsExchange); err != nil {
+			return err
+		}
+	}
+	return bind(feedbackQueue, "d."+data.dispatcherID+".out", mq.ResultsExchange)
+}
+
+func (b *broker) publish(ctx context.Context, route string, raw []byte) (string, error) {
+	ch, err := b.conn.Channel()
+	if err != nil {
+		return "not_sent", fmt.Errorf("open publisher: %w", err)
+	}
+	defer ch.Close()
+	if err = ch.Confirm(false); err != nil {
+		return "not_sent", fmt.Errorf("enable confirmation: %w", err)
+	}
+	confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))
+	returns := ch.NotifyReturn(make(chan amqp.Return, 1))
+	if err = ch.PublishWithContext(ctx, mq.CommandsExchange, route, true, false, amqp.Publishing{ContentType: "application/json", DeliveryMode: amqp.Persistent, Body: raw}); err != nil {
+		return "unknown", fmt.Errorf("MQ publish outcome unknown: %w", err)
+	}
+	return awaitConfirm(ctx, confirms, returns)
+}
+
+func awaitConfirm(ctx context.Context, confirms <-chan amqp.Confirmation, returns <-chan amqp.Return) (string, error) {
+	returned := false
+	for {
+		select {
+		case _, ok := <-returns:
+			if !ok {
+				returns = nil
+			} else {
+				returned = true
+			}
+		case confirm, ok := <-confirms:
+			if !ok {
+				return "unknown", errors.New("MQ confirmation stream closed")
+			}
+			// RabbitMQ emits return before confirmation; drain the buffered notification.
+			select {
+			case _, ok := <-returns:
+				if ok {
+					returned = true
+				}
+			default:
+			}
+			if returned {
+				return "unroutable", errors.New("MQ returned unroutable message")
+			}
+			if !confirm.Ack {
+				return "rejected", errors.New("MQ publisher NACK")
+			}
+			return "confirmed", nil
+		case <-ctx.Done():
+			return "unknown", errors.New("MQ confirmation deadline exceeded; do not resend")
+		}
+	}
+}
+
+func (b *broker) consume(ctx context.Context, s *service) error {
+	ch, err := b.conn.Channel()
+	if err != nil {
+		return fmt.Errorf("open feedback consumer: %w", err)
+	}
+	defer ch.Close()
+	if err = ch.Qos(1, 0, false); err != nil {
+		return fmt.Errorf("feedback prefetch: %w", err)
+	}
+	deliveries, err := ch.Consume(feedbackQueue, "", false, true, false, false, nil)
+	if err != nil {
+		return fmt.Errorf("start exclusive feedback consumer: %w", err)
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case delivery, ok := <-deliveries:
+			if !ok {
+				return errors.New("MQ feedback stream closed")
+			}
+			if err = s.receive(delivery.Body, func() error { return delivery.Ack(false) }); err != nil {
+				return fmt.Errorf("feedback save/ACK failed: %w", err)
+			}
+		}
+	}
+}

cmd/saas-server/mq_test.go

@@ -0,0 +1,41 @@
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+func TestPublishConfirmReturnAndUnknown(t *testing.T) {
+	for _, tc := range []struct {
+		name, want            string
+		ack, returned, closed bool
+	}{
+		{name: "ack", want: "confirmed", ack: true}, {name: "return", want: "unroutable", ack: true, returned: true}, {name: "nack", want: "rejected"}, {name: "closed", want: "unknown", closed: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			confirms := make(chan amqp.Confirmation, 1)
+			returns := make(chan amqp.Return, 1)
+			if tc.returned {
+				returns <- amqp.Return{ReplyCode: 312}
+			}
+			if tc.closed {
+				close(confirms)
+			} else {
+				confirms <- amqp.Confirmation{Ack: tc.ack}
+			}
+			got, err := awaitConfirm(context.Background(), confirms, returns)
+			if got != tc.want || (err == nil) != (got == "confirmed") {
+				t.Fatalf("%s: %s %v", tc.name, got, err)
+			}
+		})
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond)
+	defer cancel()
+	got, err := awaitConfirm(ctx, make(chan amqp.Confirmation), make(chan amqp.Return))
+	if got != "unknown" || err == nil {
+		t.Fatalf("timeout: %s %v", got, err)
+	}
+}

cmd/saas-server/operations.go

@@ -0,0 +1,140 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"time"
+
+	"git.ipao.vip/rogee/go-sip/internal/configread"
+	"github.com/gofiber/fiber/v3"
+)
+
+var errUnknownResource = errors.New("unknown resource")
+
+func (s *service) replace(resource, id string, raw []byte) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	candidate := s.data
+	candidate.tasks = make(map[string][]byte, len(s.data.tasks))
+	for key, value := range s.data.tasks {
+		candidate.tasks[key] = value
+	}
+	raw = bytes.Clone(raw)
+	switch resource {
+	case "sip":
+		if err := validateResource(raw, "sip_config", candidate.dispatcherID); err != nil {
+			return err
+		}
+		candidate.sip = raw
+	case "providers":
+		if err := validateResource(raw, "ai_providers", candidate.dispatcherID); err != nil {
+			return err
+		}
+		candidate.providers = raw
+	case "quota":
+		if err := validateResource(raw, "tenant_quota", candidate.dispatcherID); err != nil {
+			return err
+		}
+		var quota configread.Quota
+		if err := json.Unmarshal(raw, &quota); err != nil {
+			return errors.New("invalid quota")
+		}
+		if quota.TenantID != candidate.tenantID {
+			return errors.New("cannot change tenant identity")
+		}
+		candidate.quota = raw
+	case "task":
+		if _, ok := candidate.tasks[id]; !ok {
+			return errUnknownResource
+		}
+		if err := validateResource(raw, "task_config", candidate.dispatcherID); err != nil {
+			return err
+		}
+		var task configread.Task
+		if err := json.Unmarshal(raw, &task); err != nil {
+			return errors.New("invalid task")
+		}
+		if task.TaskID != id || task.TenantID != candidate.tenantID {
+			return errors.New("cannot change task identity")
+		}
+		candidate.tasks[id] = raw
+	default:
+		return errUnknownResource
+	}
+	if err := candidate.validateReferences(); err != nil {
+		return err
+	}
+	s.data = candidate
+	slog.Info("Test configuration changed in memory", "resource", resource, "task_id", id, "files_modified", false)
+	return nil
+}
+
+func (s *service) routes() {
+	update := func(resource string) fiber.Handler {
+		return func(c fiber.Ctx) error {
+			if err := s.replace(resource, c.Params("task_id"), c.Body()); err != nil {
+				status := 400
+				if errors.Is(err, errUnknownResource) {
+					status = 404
+				}
+				return c.Status(status).JSON(fiber.Map{"error": err.Error()})
+			}
+			note := "memory only; send an explicit refresh command to apply"
+			if resource == "providers" {
+				note = "memory only; Dispatcher reads providers on startup; no service restart performed"
+			}
+			return c.JSON(fiber.Map{"status": "updated", "note": note})
+		}
+	}
+	s.app.Put("/test/config/sip", update("sip"))
+	s.app.Put("/test/config/providers", update("providers"))
+	s.app.Put("/test/config/quota", update("quota"))
+	s.app.Put("/test/config/tasks/:task_id", update("task"))
+	command := func(kind string) fiber.Handler {
+		return func(c fiber.Ctx) error {
+			var input commandInput
+			dec := json.NewDecoder(bytes.NewReader(c.Body()))
+			dec.DisallowUnknownFields()
+			if !json.Valid(c.Body()) || dec.Decode(&input) != nil {
+				return c.Status(400).JSON(fiber.Map{"error": "invalid command JSON"})
+			}
+			// A single command has one purpose; do not silently accept fields from another command.
+			if (kind == "call.execute" && (input.Action != "" || input.Reason != "" || input.TrunkID != "" || input.ChangeType != "" || input.Revision != nil)) || (kind == "task.control" && (input.Callee != "" || input.TrunkID != "" || input.ChangeType != "" || input.Revision != nil)) || (kind == "sip.config" && (input.TaskID != "" || input.Callee != "" || input.Action != "" || input.Reason != "")) {
+				return c.Status(400).JSON(fiber.Map{"error": "fields do not belong to this command"})
+			}
+			ctx, cancel := context.WithTimeout(c.Context(), 5*time.Second)
+			defer cancel()
+			record, err := s.send(ctx, kind, input, time.Now())
+			if err != nil {
+				return c.Status(400).JSON(fiber.Map{"error": err.Error()})
+			}
+			code := 200
+			if record.Status != "confirmed" {
+				code = 503
+			}
+			return c.Status(code).JSON(record)
+		}
+	}
+	s.app.Post("/test/calls", command("call.execute"))
+	s.app.Post("/test/task-controls", command("task.control"))
+	s.app.Post("/test/sip-refresh", command("sip.config"))
+	s.app.Get("/test/events", func(c fiber.Ctx) error {
+		id := c.Query("event_id")
+		if id == "" {
+			return c.Status(400).JSON(fiber.Map{"error": "event_id is required"})
+		}
+		feedback := s.events(id)
+		requests := []requestRecord{}
+		s.eventMu.Lock()
+		for _, record := range s.requests {
+			if record.EventID == id {
+				requests = append(requests, record)
+			}
+		}
+		s.eventMu.Unlock()
+		return c.JSON(fiber.Map{"requests": requests, "feedback": feedback, "history": "current process only"})
+	})
+}

cmd/saas-server/operations_test.go

@@ -0,0 +1,145 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+)
+
+func request(t *testing.T, s *service, method, path string, body []byte) (int, []byte) {
+	t.Helper()
+	req, _ := http.NewRequest(method, "http://test"+path, bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := s.app.Test(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	raw, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return resp.StatusCode, raw
+}
+
+func TestMemoryUpdatesAndCommands(t *testing.T) {
+	dir, id := fixture(t)
+	data, err := loadSnapshots(dir, id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	s := newService(data, "secret")
+	var published []envelope
+	s.publish = func(_ context.Context, _ string, raw []byte) (string, error) {
+		var msg envelope
+		json.Unmarshal(raw, &msg)
+		published = append(published, msg)
+		return "confirmed", nil
+	}
+	before, err := os.ReadFile(filepath.Join(dir, "tasks/1.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var task map[string]any
+	json.Unmarshal(before, &task)
+	task["task_revision"] = float64(123)
+	updated, _ := json.Marshal(task)
+	code, raw := request(t, s, "PUT", "/test/config/tasks/1", updated)
+	if code != 200 {
+		t.Fatalf("update: %d %s", code, raw)
+	}
+	if len(published) != 0 {
+		t.Fatal("configuration update sent automatic command")
+	}
+	s.mu.RLock()
+	got := s.data.tasks["1"]
+	s.mu.RUnlock()
+	if !bytes.Equal(got, updated) {
+		t.Fatal("update not in memory")
+	}
+	code, _ = request(t, s, "PUT", "/test/config/tasks/1", []byte(`{"invalid":true}`))
+	if code != 400 {
+		t.Fatal(code)
+	}
+	task["task_id"] = "2"
+	bad, _ := json.Marshal(task)
+	code, _ = request(t, s, "PUT", "/test/config/tasks/1", bad)
+	if code != 400 {
+		t.Fatal(code)
+	}
+	code, _ = request(t, s, "PUT", "/test/config/tasks/2", updated)
+	if code != 404 {
+		t.Fatal(code)
+	}
+	for _, action := range []string{"start", "resume", "edit"} {
+		code, raw = request(t, s, "POST", "/test/task-controls", []byte(`{"event_id":"ctrl-`+action+`","task_id":"1","action":"`+action+`","reason":"manual test"}`))
+		if code != 200 {
+			t.Fatalf("%s %d %s", action, code, raw)
+		}
+	}
+	code, _ = request(t, s, "POST", "/test/task-controls", []byte(`{"event_id":"ctrl-stop","task_id":"1","action":"stop"}`))
+	if code != 400 {
+		t.Fatal(code)
+	}
+	code, raw = request(t, s, "POST", "/test/sip-refresh", []byte(`{"event_id":"sip-test","trunk_id":"trunk-mock","change_type":"enabled","revision":8}`))
+	if code != 200 {
+		t.Fatalf("sip %d %s", code, raw)
+	}
+	code, raw = request(t, s, "POST", "/test/calls", []byte(`{"event_id":"call-http","task_id":"1","callee":"123456"}`))
+	if code != 200 {
+		t.Fatalf("call %d %s", code, raw)
+	}
+	code, raw = request(t, s, "GET", "/test/events?event_id=call-http", nil)
+	if code != 200 || !bytes.Contains(raw, []byte(`"confirmed"`)) {
+		t.Fatalf("query: %d %s", code, raw)
+	}
+	code, _ = request(t, s, "POST", "/test/calls", []byte(`{"event_id":"bad","task_id":"1","callee":"123","unknown":true}`))
+	if code != 400 {
+		t.Fatal(code)
+	}
+	after, _ := os.ReadFile(filepath.Join(dir, "tasks/1.json"))
+	if !bytes.Equal(before, after) {
+		t.Fatal("file changed")
+	}
+	reloaded, err := loadSnapshots(dir, id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(reloaded.tasks["1"], before) {
+		t.Fatal("restart restored memory changes")
+	}
+}
+
+func TestConcurrentMemoryReadAndUpdate(t *testing.T) {
+	dir, id := fixture(t)
+	data, err := loadSnapshots(dir, id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	s := newService(data, "secret")
+	var wg sync.WaitGroup
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for j := 0; j < 10; j++ {
+				if err := s.replace("task", "1", data.tasks["1"]); err != nil {
+					t.Error(err)
+				}
+				s.mu.RLock()
+				_, err := s.data.page("")
+				s.mu.RUnlock()
+				if err != nil {
+					t.Error(err)
+				}
+			}
+		}()
+	}
+	wg.Wait()
+}