merge: integrate all business changes with safe debug logging

docs/evidence/structural-rejected-feedback-local-20261009.md

@@ -0,0 +1,31 @@
+# 数据结构拒绝反馈：本地开发验收
+
+## 范围
+
+使用者已确认：结构错误的数据若可核实原身份，保存原文并反馈拒绝及具体原因；身份无法核实的垃圾数据记错误后丢弃，不虚构身份。保持原 event_type，只使用 rejected 状态，不补缺失的源时间。本轮不部署、不调用真实 AI/OSS、不发布真实 SaaS 消息、不导入号码、不拨号。
+
+正式字段、Schema 与正常示例唯一入口为 [共享合同](../../contracts/schema/saas-dispatcher.md#数据结构拒绝及双方反馈)，固定提交 `8e6c332bb9d5b3d331cef3e5cbc35a1b8b4b7c86`，已提交并核对共享远端 main。
+
+## 已验证行为
+
+- 缺少 issued_at 的控制与执行消息不再单边丢弃。原身份核验通过后，先保存原字节与实际验证错误，再在现有 outbox 保存符合合同的 rejected 回执；不执行、不发 applied/dispatched、不伪造通话结果。
+- 私密原文保存在数据库同级 rejected-data/，目录 0700、文件 0600，文件与目录同步后才允许确认入站。普通日志不包含原文、配置值或凭据。原文或 outbox 保存失败时消息保留并重新交付，错误明确返回。
+- 无法解析 JSON、缺少原身份、Dispatcher/租户归属无法核实、把回执错误送回输入方向：记错误和摘要，不生成反馈身份。
+- 任务控制读取到结构不合格的 HTTP 配置，反馈关联原控制事件，并保存原控制与配置响应。SIP 通知等待旧通话结束时，原通知及内部加载代次在 sip-notification-sources/ 保持私密、可恢复；其 HTTP SIP 配置结构不合格时反馈原通知，继续关闭新呼叫。
+- 发布确认失败仍保留相同回执供重试；启动时先发送已有 outbox，不让坏配置阻断已经持久保存的反馈。MQ 确认不代表 SaaS 已处理。
+- rejected 身份不能因生产者补字段或改变正文而重新执行，只重发原拒绝回执。已经持久接纳的执行不能被后来坏消息改写成结构拒绝；未知执行及占用保留。已加载 SIP 的上下文清理不会删除更新且未加载通知的上下文。
+
+## 本地检查
+
+- `make check`：通过，包含格式、当前合同、Proto、go vet、完整 race 测试、构建与实际运行的隔离 RabbitMQ/HTTP(S)/双向 TLS 验收。
+- RabbitMQ 隔离验收实际投递缺少 issued_at 的控制消息，核实没有执行，原拒绝回执进入 SaaS 所有的共享结果队列，mandatory/publisher confirm 通过；垃圾消息不产生额外反馈。此队列是隔离夹具，不是真实 SaaS。
+- 新增原字节/权限、保存失败、发送确认失败、重启关联恢复、拒绝身份重放、已接纳执行保护、SIP 新旧通知清理及 HTTP 配置结构拒绝测试。
+- `make release-check-local`：通过；本地制品仍为 production_approval=false，未部署。
+- `scripts/coverage.sh`：本轮独立源码业务覆盖率 71.1%，高于 65%；包含并行日志工作的原工作区为 71.7%。
+- 从仅含本轮改动的 Git index 生成固定源码，在临时 detached 工作区、固定共享提交再次运行 `make check`、`make release-check-local` 与覆盖率检查，全部通过。没有新建分支；并行调试日志改动保留，未纳入本轮功能提交。
+
+## SaaS 对接通知与未验证事项
+
+共享合同已定义拒绝反馈。SaaS 必须按原事件关联消费 rejected、展示具体原因并结束对应处理中状态；按原身份去重，不重复执行。字段与示例请以共享合同为准，不从本证据维护第二份定义。
+
+真实 SaaS 是否已经采用新合同、是否消费拒绝并解除“开启中”，尚未验证。任务 4/5 的历史界面现象不能由本机 MQ 测试代签。没有部署授权或重新发送历史事件；浏览器、真实线路接通、AI/录音/OSS 全链路及生产签收均不属于本轮结果。

internal/configread/snapshots.go

@@ -210,16 +210,28 @@ func (c *Client) readResource(ctx context.Context, path, resource string, dst an
 		return err
 	}
 	if err := contract.ValidateCurrent("config-read", body); err != nil {
+		var failure *contract.ValidationFailure
+		if errors.As(err, &failure) {
+			failure.Resource = resource
+		}
 		return err
 	}
 	var identity struct {
 		Resource string `json:"resource"`
 	}
 	if err := json.Unmarshal(body, &identity); err != nil {
-		return err
+		return &contract.ValidationFailure{Surface: "config-read", Resource: resource, Raw: append([]byte(nil), body...), Reason: "invalid resource identity encoding"}
 	}
 	if identity.Resource != resource {
-		return fmt.Errorf("expected %s, got %s", resource, identity.Resource)
+		return &contract.ValidationFailure{Surface: "config-read", Resource: resource, Raw: append([]byte(nil), body...), Reason: "expected resource " + resource}
 	}
-	return json.Unmarshal(body, dst)
+	if err := json.Unmarshal(body, dst); err != nil {
+		reason := "JSON fields do not match " + resource
+		var typed *json.UnmarshalTypeError
+		if errors.As(err, &typed) {
+			reason += " at " + typed.Field
+		}
+		return &contract.ValidationFailure{Surface: "config-read", Resource: resource, Raw: append([]byte(nil), body...), Reason: reason}
+	}
+	return nil
 }

internal/contract/current.go

@@ -2,9 +2,7 @@ package contract
 
 import (
 	"bytes"
-	"errors"
 	"fmt"
-	"strings"
 	"sync"
 
 	"git.ipao.vip/rogee/go-sip/contracts"
@@ -25,7 +23,7 @@ func ValidateCurrent(surface string, raw []byte) error {
 	}
 	value, err := jsonschema.UnmarshalJSON(bytes.NewReader(raw))
 	if err != nil {
-		return fmt.Errorf("decode %s body: %w", surface, err)
+		return &ValidationFailure{Surface: surface, Raw: append([]byte(nil), raw...), Reason: "invalid JSON encoding"}
 	}
 	key := "current/" + name
 	cached, ok := compiledSchemas.Load(key)
@@ -41,11 +39,7 @@ func ValidateCurrent(surface string, raw []byte) error {
 		schema = actual.(*jsonschema.Schema)
 	}
 	if err := schema.Validate(value); err != nil {
-		var validation *jsonschema.ValidationError
-		if errors.As(err, &validation) {
-			return fmt.Errorf("%s validation failed at /%s", surface, strings.Join(validation.InstanceLocation, "/"))
-		}
-		return fmt.Errorf("%s validation failed", surface)
+		return newValidationFailure(surface, raw, value, err)
 	}
 	return nil
 }

internal/contract/failure.go

@@ -0,0 +1,68 @@
+package contract
+
+import (
+	"fmt"
+	"strings"
+
+	"git.ipao.vip/rogee/go-sip/contracts"
+	"github.com/santhosh-tekuri/jsonschema/v6"
+	"github.com/santhosh-tekuri/jsonschema/v6/kind"
+)
+
+// ValidationFailure retains original bytes for private evidence, never Error/log output.
+type ValidationFailure struct {
+	Surface  string
+	Resource string
+	Raw      []byte
+	Reason   string
+}
+
+func (e *ValidationFailure) Error() string { return e.Surface + " validation failed: " + e.Reason }
+
+func newValidationFailure(surface string, raw []byte, value any, cause error) error {
+	// Discriminate the MQ branch only to explain an already proven validation failure.
+	// This does not broaden acceptance or replace validation of the complete union.
+	if surface == "mq" {
+		if m, ok := value.(map[string]any); ok {
+			p, _ := m["payload"].(map[string]any)
+			if _, receipt := p["status"]; !receipt {
+				names := map[string]string{"task.control": "mq-task-control-request.schema.json", "call.execute": "mq-call-execute-request.schema.json", "sip.config": "mq-sip-config.schema.json"}
+				if name := names[fmt.Sprint(m["event_type"])]; name != "" {
+					if s, err := contracts.CompileCurrent(name); err == nil {
+						if detail := s.Validate(value); detail != nil {
+							cause = detail
+						}
+					}
+				}
+			}
+		}
+	}
+	var details []string
+	var walk func(*jsonschema.ValidationError)
+	walk = func(e *jsonschema.ValidationError) {
+		if len(details) >= 32 {
+			return
+		}
+		if len(e.Causes) > 0 {
+			for _, c := range e.Causes {
+				walk(c)
+			}
+			return
+		}
+		path := "/" + strings.Join(e.InstanceLocation, "/")
+		if r, ok := e.ErrorKind.(*kind.Required); ok {
+			for _, k := range r.Missing {
+				details = append(details, strings.TrimSuffix(path, "/")+"/"+k+": required field missing")
+			}
+			return
+		}
+		details = append(details, path+": "+strings.Join(e.ErrorKind.KeywordPath(), "/"))
+	}
+	if e, ok := cause.(*jsonschema.ValidationError); ok {
+		walk(e)
+	}
+	if len(details) == 0 {
+		details = append(details, "structure does not match current Schema")
+	}
+	return &ValidationFailure{Surface: surface, Raw: append([]byte(nil), raw...), Reason: strings.Join(details, "; ")}
+}

internal/contract/rejection_test.go

@@ -0,0 +1,44 @@
+package contract
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+func TestRejectedFeedbackRequiresReasonAndPreservesOriginalType(t *testing.T) {
+	for _, kind := range []string{"task.control", "call.execute", "sip.config"} {
+		t.Run(kind, func(t *testing.T) {
+			body := map[string]any{"event_id": "original-event", "event_type": kind, "dispatcher_id": "c046b893-8628-4589-ae50-619d049248a6", "payload": map[string]any{"status": "rejected", "reason_message": "/issued_at: required field missing"}}
+			if kind != "sip.config" {
+				body["tenant_id"] = 1005
+			}
+			if kind == "call.execute" {
+				body["payload"].(map[string]any)["reason_code"] = nil
+			}
+			raw, err := json.Marshal(body)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := ValidateCurrent("mq", raw); err != nil {
+				t.Fatalf("rejection without source timestamp was rejected: %v", err)
+			}
+			delete(body["payload"].(map[string]any), "reason_message")
+			raw, err = json.Marshal(body)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if ValidateCurrent("mq", raw) == nil {
+				t.Fatal("reason-less rejection accepted")
+			}
+		})
+	}
+}
+
+func TestInvalidMessageReasonNamesMissingFieldWithoutLeakingValues(t *testing.T) {
+	raw := []byte(`{"event_id":"original-event","event_type":"task.control","dispatcher_id":"c046b893-8628-4589-ae50-619d049248a6","tenant_id":1005,"payload":{"task_id":"5","action":"resume","reason":"PRIVATE-CONTROL-TEXT"}}`)
+	err := ValidateCurrent("mq", raw)
+	if err == nil || !strings.Contains(err.Error(), "issued_at") || strings.Contains(err.Error(), "PRIVATE-CONTROL-TEXT") {
+		t.Fatalf("missing safe actionable rejection details: %v", err)
+	}
+}

internal/dispatcher/control.go

@@ -43,7 +43,7 @@ type ControlController struct {
 // command_id, expected revision, or control-message deduplication. The state
 // barrier is durable before Agent dispatch; applied state and MQ outbox are
 // committed atomically after the Agent accepts the instruction.
-func (c *ControlController) ProcessControl(ctx context.Context, body []byte) error {
+func (c *ControlController) ProcessControl(ctx context.Context, body []byte) (result error) {
 	if c == nil || c.DispatcherID == "" || c.Store == nil || c.Client == nil || c.Agent == nil || c.ApprovedSIP == nil || c.ApprovedProviders == nil || c.Now == nil {
 		return errors.New("control processing requires Dispatcher, durable store, HTTP client, Agent, approved SIP/provider snapshots, and clock")
 	}
@@ -71,6 +71,19 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) err
 	if event.EventType != "task.control" || event.DispatcherID != c.DispatcherID || event.Payload.TaskID == "" || event.Payload.Action == "" {
 		return errors.New("control event type, task, or Dispatcher owner mismatch")
 	}
+	defer func() {
+		var failure *contract.ValidationFailure
+		if errors.As(result, &failure) {
+			result = c.Store.RejectData(c.DispatcherID, body, result)
+		}
+	}()
+	rejected, err := c.Store.ReplayRejectedEvent(event.DispatcherID, event.EventID, event.EventType, event.TenantID)
+	if err != nil {
+		return err
+	}
+	if rejected {
+		return nil
+	}
 	issuedAt, err := time.Parse(time.RFC3339Nano, event.IssuedAt)
 	if err != nil {
 		return fmt.Errorf("invalid task.control issued_at: %w", err)
@@ -121,7 +134,7 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) err
 		if event.Payload.Action == "start" {
 			if err := c.Store.CompleteStart(snapshot, event.EventID); err != nil {
 				if errors.Is(err, store.ErrControlRejected) {
-					return c.Store.RejectControl(event.DispatcherID, event.TenantID, event.EventID)
+					return c.Store.RejectControl(event.DispatcherID, event.TenantID, event.EventID, err.Error())
 				}
 				return fmt.Errorf("start task %q: %w", event.Payload.TaskID, err)
 			}
@@ -140,7 +153,7 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) err
 	}
 	if err := c.Store.PrepareControl(event.DispatcherID, event.TenantID, event.Payload.TaskID, event.Payload.Action); err != nil {
 		if errors.Is(err, store.ErrControlRejected) {
-			if rejectErr := c.Store.RejectControl(event.DispatcherID, event.TenantID, event.EventID); rejectErr != nil {
+			if rejectErr := c.Store.RejectControl(event.DispatcherID, event.TenantID, event.EventID, err.Error()); rejectErr != nil {
 				return fmt.Errorf("persist rejected control %q: %w", event.EventID, rejectErr)
 			}
 			return nil

internal/dispatcher/execute.go

@@ -92,6 +92,10 @@ func (c *ExecuteController) ProcessExecute(ctx context.Context, body []byte) err
 	}
 	logger := slog.With("event_id", cmd.EventID, "task_id", cmd.TaskID, "tenant_id", cmd.TenantID)
 	stored, inserted, err := c.Store.RecordExecute(cmd)
+	if errors.Is(err, store.ErrEventRejected) {
+		logger.Info("Rejected call identity replayed without execution", "stage", "idempotent_rejection")
+		return nil
+	}
 	if errors.Is(err, store.ErrStopped) {
 		logger.Info("Call instruction ignored", "stage", "admission_closed", "reason", "task stopped")
 		// The stopped task never accepted this old queued command. RabbitMQ

internal/dispatcher/rejected_data.go

@@ -0,0 +1,30 @@
+package dispatcher
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"log/slog"
+
+	"git.ipao.vip/rogee/go-sip/internal/store"
+)
+
+func (r *Runtime) rejectInvalidData(ctx context.Context, queue, routing string, raw []byte, cause error) error {
+	originalCause := cause
+	if !r.Broker.OwnsRejectionRoute(queue, routing) {
+		cause = store.ErrUnidentifiableData
+	} else {
+		cause = r.Bootstrap.Store.RejectData(r.Bootstrap.DispatcherID, raw, cause)
+	}
+	if errors.Is(cause, store.ErrUnidentifiableData) {
+		sum := sha256.Sum256(raw)
+		slog.ErrorContext(ctx, "unidentifiable incoming garbage discarded", "queue", queue, "bytes", len(raw), "body_sha256", hex.EncodeToString(sum[:]), "reason", originalCause)
+		return nil
+	}
+	if cause != nil {
+		return cause
+	}
+	slog.WarnContext(ctx, "incoming data rejected; original and feedback persisted", "queue", queue, "bytes", len(raw))
+	return nil
+}

internal/dispatcher/rejected_data_test.go

@@ -0,0 +1,42 @@
+package dispatcher
+
+import (
+	"bytes"
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"git.ipao.vip/rogee/go-sip/internal/configread"
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+)
+
+func TestCorrelatedMalformedTaskHTTPReturnsRejectedControlWithoutApplying(t *testing.T) {
+	c, agent, s := newControlFixture(t)
+	invalid := []byte(`{"resource":"task_config","credentials":"PRIVATE-HTTP-CONTENT"}`)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write(invalid)
+	}))
+	defer server.Close()
+	client, err := configread.NewClient(server.URL, c.DispatcherID, "fixture", server.Client())
+	if err != nil {
+		t.Fatal(err)
+	}
+	c.Client = client
+	original := controlBody(t, "bad-http-original", "resume", "")
+	if err := c.ProcessControl(context.Background(), original); err != nil {
+		t.Fatal(err)
+	}
+	if len(agent.calls) != 0 {
+		t.Fatal("malformed HTTP configuration reached Agent")
+	}
+	outbox, err := s.ListPendingOutbox(c.DispatcherID)
+	if err != nil || len(outbox) != 1 {
+		t.Fatalf("no correlated feedback: %+v %v", outbox, err)
+	}
+	body := outbox[0].Body
+	if contract.ValidateCurrent("mq", body) != nil || !bytes.Contains(body, []byte(`"status":"rejected"`)) || bytes.Contains(body, []byte("PRIVATE-HTTP-CONTENT")) {
+		t.Fatalf("unsafe or untruthful feedback: %s", body)
+	}
+}

internal/dispatcher/rejection_identity_test.go

@@ -0,0 +1,34 @@
+package dispatcher
+
+import (
+	"context"
+	"errors"
+	"testing"
+)
+
+func TestCorrectedExecuteCannotReviveStructurallyRejectedIdentity(t *testing.T) {
+	c, agent, _, s := newExecuteFixture(t)
+	raw := []byte(`{"event_id":"call-1","event_type":"call.execute","dispatcher_id":"` + c.DispatcherID + `","tenant_id":1001,"payload":{"task_id":"task-asr","callee":"15003164745"}}`)
+	if err := s.RejectData(c.DispatcherID, raw, errors.New("issued_at required")); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.ProcessExecute(context.Background(), executeBody(t, "call-1", "15003164745")); err != nil {
+		t.Fatal(err)
+	}
+	if len(agent.calls) != 0 {
+		t.Fatal("producer correction revived a rejected execution identity")
+	}
+}
+func TestRepeatedControlCannotReviveRejectedIdentity(t *testing.T) {
+	c, agent, s := newControlFixture(t)
+	body := controlBody(t, "pause-original", "pause", "")
+	if err := s.RejectData(c.DispatcherID, body, errors.New("task HTTP configuration malformed")); err != nil {
+		t.Fatal(err)
+	}
+	if err := c.ProcessControl(context.Background(), body); err != nil {
+		t.Fatal(err)
+	}
+	if len(agent.calls) != 0 {
+		t.Fatalf("rejected identity reached Agent: %+v", agent.calls)
+	}
+}

internal/dispatcher/runtime.go

@@ -10,6 +10,7 @@ import (
 	"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"
 	"git.ipao.vip/rogee/go-sip/internal/store"
 	"git.ipao.vip/rogee/go-sip/internal/tenant"
@@ -74,6 +75,10 @@ func (r *Runtime) Serve(ctx context.Context) (result error) {
 		}
 	}()
 
+	r.Broker.SetInvalidMessageHandler(r.rejectInvalidData)
+	if err := r.Execute.FlushOutbox(ctx); err != nil {
+		return err
+	}
 	var sip configread.SIP
 	var providers map[string]configread.Provider
 	r.Bootstrap.SIP = &sip
@@ -99,10 +104,17 @@ func (r *Runtime) Serve(ctx context.Context) (result error) {
 	}
 	if err := r.Bootstrap.Run(ctx); err != nil {
 		if !errors.Is(err, store.ErrSIPPending) {
-			return fmt.Errorf("bootstrap current Dispatcher: %w", err)
+			sources, sourceErr := r.Bootstrap.Store.PendingSIPSources()
+			if sourceErr != nil {
+				return errors.Join(err, sourceErr)
+			}
+			return fmt.Errorf("bootstrap current Dispatcher: %w", r.rejectSIPSources(ctx, sources, err))
 		}
 		r.Logger.Warn("control and outbox stay active while newer SIP revision waits; task admission remains closed", "dispatcher_id", r.Bootstrap.DispatcherID, "error", err)
 	}
+	if err := r.Bootstrap.Store.CompleteAppliedSIPSources(r.Bootstrap.DispatcherID); err != nil {
+		return err
+	}
 	if err := r.syncTaskConsumers(ctx, consumers); err != nil {
 		return err
 	}
@@ -175,10 +187,12 @@ func (r *Runtime) withTask(taskID string, process func() error) error {
 	return process()
 }
 
-func (r *Runtime) handleControl(ctx context.Context, _ string, body []byte) error {
+func (r *Runtime) handleControl(ctx context.Context, routing string, body []byte) error {
 	var event struct {
-		EventType string `json:"event_type"`
-		Payload   struct {
+		EventID      string `json:"event_id"`
+		DispatcherID string `json:"dispatcher_id"`
+		EventType    string `json:"event_type"`
+		Payload      struct {
 			TaskID   string `json:"task_id"`
 			Action   string `json:"action"`
 			Revision int64  `json:"revision"`
@@ -216,6 +230,10 @@ func (r *Runtime) handleControl(ctx context.Context, _ string, body []byte) erro
 		}
 		return r.withTask(event.Payload.TaskID, func() error {
 			if err := r.Control.ProcessControl(ctx, body); err != nil {
+				var failure *contract.ValidationFailure
+				if errors.As(err, &failure) || errors.Is(err, store.ErrUnidentifiableData) {
+					return r.rejectInvalidData(ctx, r.Broker.ControlQueue(), routing, body, err)
+				}
 				r.Logger.Error("task control failed", "dispatcher_id", r.Bootstrap.DispatcherID, "task_id", event.Payload.TaskID, "error", err)
 				r.signalFailure(err)
 				return err
@@ -224,7 +242,15 @@ func (r *Runtime) handleControl(ctx context.Context, _ string, body []byte) erro
 		})
 	case "sip.config":
 		r.gate.Lock()
+		rejected, replayErr := r.Bootstrap.Store.ReplayRejectedEvent(event.DispatcherID, event.EventID, event.EventType, 0)
+		if replayErr != nil || rejected {
+			r.gate.Unlock()
+			return replayErr
+		}
 		err := r.Bootstrap.Store.NoteSIPChange(r.Bootstrap.DispatcherID, event.Payload.Revision)
+		if err == nil {
+			err = r.Bootstrap.Store.RememberSIPSource(r.Bootstrap.DispatcherID, body)
+		}
 		r.gate.Unlock()
 		if err != nil {
 			failure := fmt.Errorf("persist SIP change before MQ ACK: %w", err)

internal/dispatcher/sip_reload.go

@@ -29,9 +29,13 @@ func (r *Runtime) refreshSIP(ctx context.Context, approvedSIP *configread.SIP) e
 		r.Logger.Debug("SIP reload waits for confirmed old-call drain", "dispatcher_id", r.Bootstrap.DispatcherID, "pending_revision", pending, "occupied_calls", occupied)
 		return nil
 	}
+	sources, err := r.Bootstrap.Store.PendingSIPSources()
+	if err != nil {
+		return err
+	}
 	current, err := r.Bootstrap.Client.ReadSIP(ctx)
 	if err != nil {
-		return r.closeSIPAdmission(fmt.Errorf("read approved SIP full snapshot: %w", err))
+		return r.closeSIPAdmission(r.rejectSIPSources(ctx, sources, fmt.Errorf("read approved SIP full snapshot: %w", err)))
 	}
 	current, err = r.Bootstrap.Store.PrepareSIP(current)
 	if err != nil {
@@ -65,6 +69,9 @@ func (r *Runtime) refreshSIP(ctx context.Context, approvedSIP *configread.SIP) e
 	if err := r.Bootstrap.Store.MarkReadyForSIP(r.Bootstrap.DispatcherID, current.Revision); err != nil {
 		return r.closeSIPAdmission(fmt.Errorf("verify and commit reloaded SIP revision: %w", err))
 	}
+	if err := r.Bootstrap.Store.CompleteAppliedSIPSources(r.Bootstrap.DispatcherID); err != nil {
+		return err
+	}
 	*approvedSIP = current
 	if r.Bootstrap.SIP != nil {
 		*r.Bootstrap.SIP = current

internal/dispatcher/sip_source.go

@@ -0,0 +1,31 @@
+package dispatcher
+
+import (
+	"context"
+	"errors"
+
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+	"git.ipao.vip/rogee/go-sip/internal/store"
+)
+
+func (r *Runtime) rejectSIPSources(ctx context.Context, sources []store.SIPSource, cause error) error {
+	var failure *contract.ValidationFailure
+	if !errors.As(cause, &failure) || failure.Resource != "sip_config" || len(sources) == 0 {
+		return cause
+	}
+	for _, source := range sources {
+		if err := r.Bootstrap.Store.RejectData(r.Bootstrap.DispatcherID, source.Original, cause); err != nil {
+			return errors.Join(cause, err)
+		}
+		if err := r.Bootstrap.Store.CompleteSIPSource(source); err != nil {
+			return errors.Join(cause, err)
+		}
+	}
+	if r.Execute.Publisher == nil {
+		return errors.Join(cause, errors.New("SIP rejection requires an outbox publisher"))
+	}
+	if err := r.Execute.FlushOutbox(ctx); err != nil {
+		return errors.Join(cause, err)
+	}
+	return cause // Invalid configuration still closes admission; feedback isn't a fix.
+}

internal/dispatcher/sip_source_test.go

@@ -0,0 +1,65 @@
+package dispatcher
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"git.ipao.vip/rogee/go-sip/internal/configread"
+)
+
+func TestMalformedSIPHTTPRejectsOriginalNotificationAndKeepsAdmissionClosed(t *testing.T) {
+	for _, publishFails := range []bool{false, true} {
+		t.Run(map[bool]string{false: "confirmed", true: "confirmation failure"}[publishFails], func(t *testing.T) {
+			execute, _, publisher, s := newExecuteFixture(t)
+			snapshot := policySnapshot(t)
+			id := snapshot.Task.DispatcherID
+			if publishFails {
+				publisher.err = errors.New("publisher confirmation lost")
+			}
+			if err := s.NoteSIPChange(id, 2); err != nil {
+				t.Fatal(err)
+			}
+			original := []byte(`{"event_id":"original-sip-event","event_type":"sip.config","dispatcher_id":"` + id + `","payload":{"trunk_id":"1","change_type":"enabled","revision":2}}`)
+			if err := s.RememberSIPSource(id, original); err != nil {
+				t.Fatal(err)
+			}
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				w.Header().Set("Content-Type", "application/json")
+				_, _ = w.Write([]byte(`{"resource":"sip_config","private":"PRIVATE-SIP-SOURCE"}`))
+			}))
+			defer server.Close()
+			client, err := configread.NewClient(server.URL, id, "fixture", server.Client())
+			if err != nil {
+				t.Fatal(err)
+			}
+			r := &Runtime{Bootstrap: Bootstrap{Store: s, DispatcherID: id, Client: client, VerifySIP: func(context.Context, configread.SIP) error {
+				t.Fatal("invalid configuration reached native verifier")
+				return nil
+			}}, Execute: *execute, Logger: slog.New(slog.NewTextHandler(io.Discard, nil))}
+			sip := snapshot.SIP
+			if err := r.refreshSIP(context.Background(), &sip); err == nil {
+				t.Fatal("malformed configuration hidden by feedback success")
+			}
+			if len(publisher.bodies) != 1 || !bytes.Contains(publisher.bodies[0], []byte(`"event_id":"original-sip-event"`)) || !bytes.Contains(publisher.bodies[0], []byte(`"status":"rejected"`)) || bytes.Contains(publisher.bodies[0], []byte("PRIVATE-SIP-SOURCE")) {
+				t.Fatalf("wrong or unsafe feedback: %s", publisher.bodies)
+			}
+			_, pending, err := s.SIPState(id)
+			if err != nil || pending == 0 {
+				t.Fatalf("malformed SIP removed the admission fence: %d %v", pending, err)
+			}
+			events, err := s.ListPendingOutbox(id)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if (publishFails && len(events) != 1) || (!publishFails && len(events) != 0) {
+				t.Fatalf("failed feedback not retained for retry: %+v", events)
+			}
+		})
+	}
+}

internal/mq/broker.go

@@ -23,13 +23,14 @@ const (
 )
 
 type Broker struct {
-	conn         *amqp.Connection
-	dispatcherID string
-	prefetch     int
-	controlQueue string
-	resultRoute  string
-	closed       <-chan *amqp.Error
-	mu           sync.Mutex
+	conn           *amqp.Connection
+	invalidHandler InvalidMessageHandler
+	dispatcherID   string
+	prefetch       int
+	controlQueue   string
+	resultRoute    string
+	closed         <-chan *amqp.Error
+	mu             sync.Mutex
 }
 
 // ControlQueueName returns this Dispatcher's SaaS-provisioned control queue.
@@ -131,7 +132,8 @@ func (b *Broker) Publish(ctx context.Context, exchange, routingKey string, body
 		return errors.New("outbound MQ dispatcher owner mismatch")
 	}
 	if identity.EventType != "call.execute.result" &&
-		!((identity.EventType == "task.control" || identity.EventType == "call.execute") && identity.Payload.Status != "") {
+		!((identity.EventType == "task.control" || identity.EventType == "call.execute") && identity.Payload.Status != "") &&
+		!(identity.EventType == "sip.config" && identity.Payload.Status == "rejected") {
 		return fmt.Errorf("unapproved outbound MQ event %q", identity.EventType)
 	}
 	if len(identity.EventID) > 255 {
@@ -363,21 +365,12 @@ func (b *Broker) DrainControlPredeclared(ctx context.Context, queue string, hand
 			return processed, nil
 		}
 		observeDelivery(ctx, queue, &delivery)
-		if len(delivery.Body) == 0 || len(delivery.Body) > MaxMessageBytes || !json.Valid(delivery.Body) {
-			if err := delivery.Reject(false); err != nil {
-				return processed, fmt.Errorf("reject invalid control backlog message: %w", err)
-			}
-			slog.Error("MQ control backlog invalid; admission stays closed", "dispatcher_id", b.dispatcherID, "delivery_tag", delivery.DeliveryTag, "reason", "invalid_size_or_json", "bytes", len(delivery.Body))
-			processed++
-			return processed, errors.New("invalid control backlog message; admission stays closed")
-		}
 		if err := b.validateInbound(queue, delivery.RoutingKey, delivery.Body); err != nil {
-			if rejectErr := delivery.Reject(false); rejectErr != nil {
-				return processed, fmt.Errorf("invalid control message %v; reject: %w", err, rejectErr)
+			if err := b.rejectInbound(ctx, queue, delivery, err); err != nil {
+				return processed, err
 			}
-			slog.Error("MQ control backlog invalid; admission stays closed", "dispatcher_id", b.dispatcherID, "delivery_tag", delivery.DeliveryTag, "reason", err)
 			processed++
-			return processed, fmt.Errorf("invalid control backlog message; admission stays closed: %w", err)
+			continue
 		}
 		slog.InfoContext(ctx, "MQ inbound validated", append(messageIdentity(delivery.Body), "direction", "mq->dispatcher", "queue", queue, "stage", "validated")...)
 		if err := handler(ctx, delivery.RoutingKey, delivery.Body); err != nil {
@@ -419,9 +412,11 @@ func (b *Broker) validateInbound(queue, routingKey string, body []byte) error {
 		DispatcherID string `json:"dispatcher_id"`
 		EventType    string `json:"event_type"`
 		Payload      struct {
-			TaskID string `json:"task_id"`
-			Action string `json:"action"`
-			Callee string `json:"callee"`
+			TaskID   string `json:"task_id"`
+			Action   string `json:"action"`
+			Callee   string `json:"callee"`
+			TrunkID  string `json:"trunk_id"`
+			Revision int64  `json:"revision"`
 		} `json:"payload"`
 	}
 	if err := json.Unmarshal(body, &identity); err != nil {
@@ -434,7 +429,7 @@ func (b *Broker) validateInbound(queue, routingKey string, body []byte) error {
 		if routingKey != "d."+b.dispatcherID+".control.in" {
 			return errors.New("control message used wrong routing key")
 		}
-		if identity.EventType == "sip.config" || (identity.EventType == "task.control" && identity.Payload.TaskID != "" && identity.Payload.Action != "") {
+		if (identity.EventType == "sip.config" && identity.Payload.TrunkID != "" && identity.Payload.Revision > 0) || (identity.EventType == "task.control" && identity.Payload.TaskID != "" && identity.Payload.Action != "") {
 			return nil
 		}
 		return errors.New("unapproved control-queue event")
@@ -474,33 +469,9 @@ func (b *Broker) consumeDeliveries(ctx context.Context, dispatcherID, queue stri
 				return errors.New("rabbitmq current delivery channel closed")
 			}
 			observeDelivery(ctx, queue, &delivery)
-			if len(delivery.Body) == 0 || len(delivery.Body) > MaxMessageBytes {
-				if err := delivery.Reject(false); err != nil {
-					return fmt.Errorf("reject invalid current message size: %w", err)
-				}
-				slog.Warn("MQ current message rejected", "dispatcher_id", dispatcherID, "delivery_tag", delivery.DeliveryTag, "reason", "invalid_message_size", "bytes", len(delivery.Body))
-				if queue == b.controlQueue {
-					return errors.New("invalid control message size; admission stays closed")
-				}
-				continue
-			}
-			if !json.Valid(delivery.Body) {
-				if err := delivery.Reject(false); err != nil {
-					return fmt.Errorf("reject invalid current JSON: %w", err)
-				}
-				slog.Warn("MQ current message rejected", "dispatcher_id", dispatcherID, "delivery_tag", delivery.DeliveryTag, "reason", "invalid_json")
-				if queue == b.controlQueue {
-					return errors.New("invalid control JSON; admission stays closed")
-				}
-				continue
-			}
 			if err := b.validateInbound(queue, delivery.RoutingKey, delivery.Body); err != nil {
-				if rejectErr := delivery.Reject(false); rejectErr != nil {
-					return fmt.Errorf("invalid inbound MQ message %v; reject: %w", err, rejectErr)
-				}
-				slog.Warn("MQ current message rejected", "dispatcher_id", dispatcherID, "delivery_tag", delivery.DeliveryTag, "reason", err)
-				if queue == b.controlQueue {
-					return fmt.Errorf("invalid control message; admission stays closed: %w", err)
+				if err := b.rejectInbound(ctx, queue, delivery, err); err != nil {
+					return err
 				}
 				continue
 			}

internal/mq/broker_integration_test.go

@@ -4,12 +4,17 @@ package mq
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
 
+	"git.ipao.vip/rogee/go-sip/internal/configread"
+	"git.ipao.vip/rogee/go-sip/internal/store"
 	"git.ipao.vip/rogee/go-sip/internal/tenant"
 	amqp "github.com/rabbitmq/amqp091-go"
 )
@@ -260,6 +265,55 @@ func TestBrokerSharedResultQueueAndNoConfigure(t *testing.T) {
 	if err := delivery.Ack(false); err != nil {
 		t.Fatal(err)
 	}
+	s, err := store.Open(filepath.Join(t.TempDir(), "rejections.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	if err := s.ApplyDiscoverySnapshot(ids[0], []configread.DiscoveredTask{{TaskID: "task-asr", TenantID: 1001, TaskRevision: 1, Status: "running"}}); err != nil {
+		t.Fatal(err)
+	}
+	brokers[0].SetInvalidMessageHandler(func(_ context.Context, _ string, _ string, raw []byte, cause error) error {
+		err := s.RejectData(ids[0], raw, cause)
+		if errors.Is(err, store.ErrUnidentifiableData) {
+			return nil
+		}
+		return err
+	})
+	badControl := []byte(`{"event_id":"original-missing-issued-at","event_type":"task.control","dispatcher_id":"` + ids[0] + `","tenant_id":1001,"payload":{"task_id":"task-asr","action":"resume","reason":"PRIVATE ORIGINAL"}}`)
+	controlSource, _ := tenant.ControlRoute(ids[0])
+	if err := admin.PublishWithContext(context.Background(), controlSource.Exchange, controlSource.BindingKey, true, false, amqp.Publishing{ContentType: "application/json", DeliveryMode: amqp.Persistent, Body: badControl}); err != nil {
+		t.Fatal(err)
+	}
+	count, err = brokers[0].DrainControlPredeclared(context.Background(), controlSource.Queue, func(context.Context, string, []byte) error { t.Fatal("malformed original executed"); return nil })
+	if err != nil || count != 1 {
+		t.Fatalf("malformed original not durably handled: %d %v", count, err)
+	}
+	feedback, err := s.ListPendingOutbox(ids[0])
+	if err != nil || len(feedback) != 1 {
+		t.Fatalf("correlated feedback missing: %+v %v", feedback, err)
+	}
+	resultDestination, _ := tenant.ResultRoute(ids[0])
+	if err := brokers[0].Publish(context.Background(), resultDestination.Exchange, resultDestination.BindingKey, feedback[0].Body); err != nil {
+		t.Fatal(err)
+	}
+	rejected, ok, err := admin.Get(shared, false)
+	if err != nil || !ok {
+		t.Fatalf("SaaS-owned queue did not receive rejection: %t %v", ok, err)
+	}
+	var receipt struct {
+		EventID string `json:"event_id"`
+		Payload struct {
+			Status string `json:"status"`
+			Reason string `json:"reason_message"`
+		} `json:"payload"`
+	}
+	if err := json.Unmarshal(rejected.Body, &receipt); err != nil || receipt.EventID != "original-missing-issued-at" || receipt.Payload.Status != "rejected" || !strings.Contains(receipt.Payload.Reason, "issued_at") || strings.Contains(string(rejected.Body), "PRIVATE ORIGINAL") {
+		t.Fatalf("untruthful/unsafe rejected feedback: %s %v", rejected.Body, err)
+	}
+	if err := rejected.Ack(false); err != nil {
+		t.Fatal(err)
+	}
 	for _, id := range ids {
 		route, _ := tenant.ResultRoute(id)
 		if err := admin.QueueUnbind(shared, route.BindingKey, route.Exchange, nil); err != nil {
@@ -281,7 +335,11 @@ func TestBrokerSharedResultQueueAndNoConfigure(t *testing.T) {
 		t.Fatal("invalid backlog control reached handler")
 		return nil
 	})
-	if err == nil || processed != 1 || !strings.Contains(err.Error(), "invalid control backlog") {
-		t.Fatalf("malformed control backlog was silently cleared: processed=%d err=%v", processed, err)
+	if err != nil || processed != 1 {
+		t.Fatalf("unidentifiable garbage was not handled without fabricated feedback: processed=%d err=%v", processed, err)
+	}
+	feedback, err = s.ListPendingOutbox(ids[0])
+	if err != nil || len(feedback) != 1 {
+		t.Fatalf("garbage invented another identity: %+v %v", feedback, err)
 	}
 }

internal/mq/pause_test.go

@@ -21,13 +21,20 @@ func TestConsumerRequestStopDoesNotWaitForItself(t *testing.T) {
 	consumer.RequestStop()
 }
 
-type currentRejectRecorder struct{ rejected int }
+type currentRejectRecorder struct {
+	rejected, acked, nacked int
+	requeued                bool
+}
 
-func (*currentRejectRecorder) Ack(uint64, bool) error        { return nil }
-func (*currentRejectRecorder) Nack(uint64, bool, bool) error { return nil }
-func (a *currentRejectRecorder) Reject(uint64, bool) error   { a.rejected++; return nil }
+func (a *currentRejectRecorder) Ack(uint64, bool) error { a.acked++; return nil }
+func (a *currentRejectRecorder) Nack(_ uint64, _ bool, requeue bool) error {
+	a.nacked++
+	a.requeued = requeue
+	return nil
+}
+func (a *currentRejectRecorder) Reject(uint64, bool) error { a.rejected++; return nil }
 
-func TestMalformedControlStopsConsumerAfterReject(t *testing.T) {
+func TestMalformedControlWithoutDurableHandlerRequeues(t *testing.T) {
 	id := "c046b893-8628-4589-ae50-619d049248a6"
 	route, err := tenant.ControlRoute(id)
 	if err != nil {
@@ -39,7 +46,7 @@ func TestMalformedControlStopsConsumerAfterReject(t *testing.T) {
 	deliveries <- amqp.Delivery{Body: []byte("{"), RoutingKey: route.BindingKey, Acknowledger: ack}
 	close(deliveries)
 	err = broker.consumeDeliveries(context.Background(), id, route.Queue, deliveries, func(context.Context, string, []byte) error { t.Fatal("invalid control reached handler"); return nil })
-	if err == nil || !strings.Contains(err.Error(), "invalid control") || ack.rejected != 1 {
+	if err == nil || !strings.Contains(err.Error(), "no durable rejection handler") || ack.rejected != 0 || ack.nacked != 1 || !ack.requeued {
 		t.Fatalf("malformed control was silently discarded before admission: rejected=%d err=%v", ack.rejected, err)
 	}
 }

internal/mq/rejected_receipt_test.go

@@ -0,0 +1,23 @@
+package mq
+
+import (
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+	"git.ipao.vip/rogee/go-sip/internal/tenant"
+	"testing"
+)
+
+func TestRejectedSIPReceiptCannotBecomeInboundNotification(t *testing.T) {
+	id := "c046b893-8628-4589-ae50-619d049248a6"
+	route, err := tenant.ControlRoute(id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw := []byte(`{"event_id":"original","event_type":"sip.config","dispatcher_id":"` + id + `","payload":{"status":"rejected","reason_message":"invalid configuration"}}`)
+	if err := contract.ValidateCurrent("mq", raw); err != nil {
+		t.Fatal(err)
+	}
+	b := &Broker{dispatcherID: id, controlQueue: route.Queue}
+	if b.validateInbound(route.Queue, route.BindingKey, raw) == nil {
+		t.Fatal("SIP feedback entered the notification handler")
+	}
+}

internal/mq/rejection.go

@@ -0,0 +1,49 @@
+package mq
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"git.ipao.vip/rogee/go-sip/internal/tenant"
+	"log/slog"
+	"strings"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+type InvalidMessageHandler func(context.Context, string, string, []byte, error) error
+
+// Configure before starting any consumers.
+func (b *Broker) OwnsRejectionRoute(queue, routing string) bool {
+	if b.validateQueue(queue, true) != nil {
+		return false
+	}
+	if queue == b.controlQueue {
+		return routing == "d."+b.dispatcherID+".control.in"
+	}
+	prefix := "agent-call.d." + b.dispatcherID + ".task."
+	id := strings.TrimSuffix(strings.TrimPrefix(queue, prefix), ".v1")
+	route, err := tenant.TaskRoute(b.dispatcherID, id)
+	return err == nil && route.BindingKey == routing
+}
+
+func (b *Broker) SetInvalidMessageHandler(handler InvalidMessageHandler) { b.invalidHandler = handler }
+func (b *Broker) rejectInbound(ctx context.Context, queue string, d amqp.Delivery, cause error) error {
+	if b.invalidHandler == nil {
+		if err := d.Nack(false, true); err != nil {
+			return errors.Join(cause, err)
+		}
+		return fmt.Errorf("invalid data has no durable rejection handler: %w", cause)
+	}
+	if err := b.invalidHandler(ctx, queue, d.RoutingKey, d.Body, cause); err != nil {
+		if nack := d.Nack(false, true); nack != nil {
+			return errors.Join(err, nack)
+		}
+		return fmt.Errorf("durable rejected-data handling failed: %w", err)
+	}
+	slog.WarnContext(ctx, "MQ invalid data handled", append(messageIdentity(d.Body), "transport", "mq", "direction", "mq->dispatcher", "queue", queue, "stage", "rejected_handled", "bytes", len(d.Body), "reason", cause)...)
+	if err := d.Ack(false); err != nil {
+		return fmt.Errorf("ack handled rejected data: %w", err)
+	}
+	return nil
+}

internal/mq/rejection_logging_test.go

@@ -0,0 +1,60 @@
+package mq
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"strings"
+	"testing"
+
+	"git.ipao.vip/rogee/go-sip/internal/logging"
+	"git.ipao.vip/rogee/go-sip/internal/tenant"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+func TestMergedRejectionKeepsReceiveLogsWireBytesAndDurableAck(t *testing.T) {
+	id := "c046b893-8628-4589-ae50-619d049248a6"
+	route, err := tenant.ControlRoute(id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, debug := range []bool{false, true} {
+		t.Run(map[bool]string{false: "normal", true: "debug"}[debug], func(t *testing.T) {
+			previous := slog.Default()
+			defer slog.SetDefault(previous)
+			var output bytes.Buffer
+			logging.Configure("dispatcher", debug, &output)
+			body := []byte(`{"event_id":"merge-rejection","event_type":"task.control","dispatcher_id":"` + id + `","tenant_id":1001,"payload":{"task_id":"4","action":"pause"},"api_key":"private-merge-key"}`)
+			original := bytes.Clone(body)
+			ack := &logAck{}
+			broker := &Broker{dispatcherID: id, controlQueue: route.Queue}
+			handled := false
+			broker.SetInvalidMessageHandler(func(_ context.Context, queue, key string, raw []byte, cause error) error {
+				if queue != route.Queue || key != route.BindingKey || !bytes.Equal(raw, original) || cause == nil || ack.acks != 0 {
+					t.Fatal("merged receive path lost original bytes/cause or acknowledged before handling")
+				}
+				handled = true
+				return nil
+			})
+			deliveries := make(chan amqp.Delivery, 1)
+			deliveries <- amqp.Delivery{Body: body, RoutingKey: route.BindingKey, DeliveryTag: 1, Acknowledger: ack}
+			close(deliveries)
+			err := broker.consumeDeliveries(context.Background(), id, route.Queue, deliveries, func(context.Context, string, []byte) error {
+				t.Fatal("invalid message reached the business handler")
+				return nil
+			})
+			if err == nil || !handled || ack.acks != 1 || ack.rejects != 0 || ack.nacks != 0 || !bytes.Equal(body, original) {
+				t.Fatalf("merged durable handling failed: %v %+v", err, ack)
+			}
+			text := output.String()
+			for _, marker := range []string{"received", "rejected_handled", "ack_sent", "merge-rejection", "issued_at"} {
+				if !strings.Contains(text, marker) {
+					t.Fatalf("missing %s in merged workflow: %s", marker, text)
+				}
+			}
+			if strings.Contains(text, "private-merge-key") || strings.Contains(text, `\"action\":\"pause\"`) != debug {
+				t.Fatal("merged logging changed privacy/mode behavior", text)
+			}
+		})
+	}
+}

internal/mq/rejection_test.go

@@ -0,0 +1,65 @@
+package mq
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"testing"
+
+	"git.ipao.vip/rogee/go-sip/internal/tenant"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+func TestInvalidDataAckWaitsForDurableHandling(t *testing.T) {
+	route, err := tenant.ControlRoute("c046b893-8628-4589-ae50-619d049248a6")
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw := []byte(`{"event_id":"original-event"}`)
+	for _, fail := range []bool{false, true} {
+		t.Run(map[bool]string{false: "durable", true: "storage failure"}[fail], func(t *testing.T) {
+			ack := &currentRejectRecorder{}
+			b := &Broker{controlQueue: route.Queue}
+			b.SetInvalidMessageHandler(func(ctx context.Context, queue, key string, body []byte, cause error) error {
+				if ack.acked != 0 || ack.nacked != 0 {
+					t.Fatal("input disposed before rejection persisted")
+				}
+				if queue != route.Queue || key != route.BindingKey || !bytes.Equal(body, raw) || cause == nil {
+					t.Fatal("original rejection context lost")
+				}
+				if fail {
+					return errors.New("evidence storage failed")
+				}
+				return nil
+			})
+			err := b.rejectInbound(context.Background(), route.Queue, amqp.Delivery{Body: raw, RoutingKey: route.BindingKey, Acknowledger: ack}, errors.New("issued_at required"))
+			if fail {
+				if err == nil || ack.acked != 0 || ack.nacked != 1 || !ack.requeued {
+					t.Fatalf("identified data discarded on persistence failure: %+v %v", ack, err)
+				}
+			} else {
+				if err != nil || ack.acked != 1 || ack.rejected != 0 {
+					t.Fatalf("durable rejection not acknowledged: %+v %v", ack, err)
+				}
+			}
+		})
+	}
+}
+func TestRejectionRouteMustBelongToDispatcher(t *testing.T) {
+	id := "c046b893-8628-4589-ae50-619d049248a6"
+	route, err := tenant.ControlRoute(id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	task, err := tenant.TaskRoute(id, "5")
+	if err != nil {
+		t.Fatal(err)
+	}
+	b := &Broker{dispatcherID: id, controlQueue: route.Queue}
+	if !b.OwnsRejectionRoute(route.Queue, route.BindingKey) || !b.OwnsRejectionRoute(task.Queue, task.BindingKey) {
+		t.Fatal("own source rejected")
+	}
+	if b.OwnsRejectionRoute(route.Queue, "d.other.control.in") || b.OwnsRejectionRoute("agent-call.d.other.control.v1", route.BindingKey) {
+		t.Fatal("foreign source accepted")
+	}
+}

internal/store/calls.go

@@ -65,6 +65,22 @@ func (s *Store) RecordExecute(cmd ExecuteCommand) (Execution, bool, error) {
 		return Execution{}, false, err
 	}
 	defer tx.Rollback()
+	var existing int
+	if err := tx.QueryRow(`SELECT count(*) FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=?`, cmd.DispatcherID, cmd.EventID).Scan(&existing); err != nil {
+		return Execution{}, false, err
+	}
+	if existing == 0 {
+		rejected, err := replayRejectedEvent(tx, cmd.DispatcherID, cmd.EventID, "call.execute", cmd.TenantID)
+		if err != nil {
+			return Execution{}, false, err
+		}
+		if rejected {
+			if err := tx.Commit(); err != nil {
+				return Execution{}, false, err
+			}
+			return Execution{}, false, ErrEventRejected
+		}
+	}
 	var controlState string
 	var present int
 	err = tx.QueryRow(`SELECT control_state,present FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, cmd.DispatcherID, cmd.TenantID, cmd.TaskID).Scan(&controlState, &present)

internal/store/control_ack_test.go

@@ -88,7 +88,7 @@ func TestStopRejectsResumeAndSuppressesOldCommands(t *testing.T) {
 	if err := s.PrepareControl(currentDispatcherID, 1001, "task-asr", "resume"); err == nil {
 		t.Fatal("stopped task accepted resume")
 	}
-	if err := s.RejectControl(currentDispatcherID, 1001, "resume-1"); err != nil {
+	if err := s.RejectControl(currentDispatcherID, 1001, "resume-1", "task is stopped"); err != nil {
 		t.Fatal(err)
 	}
 	outbox, err := s.ListPendingOutbox(currentDispatcherID)

internal/store/control_flow.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 
 	"git.ipao.vip/rogee/go-sip/internal/configread"
 	"git.ipao.vip/rogee/go-sip/internal/contract"
@@ -63,7 +64,7 @@ func (s *Store) CompleteEdit(snapshot configread.Snapshot, eventID string) (stri
 		return "", fmt.Errorf("read edit task state: %w", err)
 	}
 	if errors.Is(err, sql.ErrNoRows) || present != 1 || state == "stopped" || state == "stopping" || status == "stopped" || task.Status == "stopped" {
-		if err := enqueueControlAck(tx, task.DispatcherID, task.TenantID, eventID, "rejected"); err != nil {
+		if err := enqueueControlAckReason(tx, task.DispatcherID, task.TenantID, eventID, "rejected", "task is absent or stopped; edits are not allowed"); err != nil {
 			return "", err
 		}
 		if err := tx.Commit(); err != nil {
@@ -189,13 +190,13 @@ func (s *Store) CompleteControl(dispatcherID string, tenantID int64, taskID, act
 	return nil
 }
 
-func (s *Store) RejectControl(dispatcherID string, tenantID int64, eventID string) error {
+func (s *Store) RejectControl(dispatcherID string, tenantID int64, eventID, reason string) error {
 	tx, err := s.db.Begin()
 	if err != nil {
 		return err
 	}
 	defer tx.Rollback()
-	if err := enqueueControlAck(tx, dispatcherID, tenantID, eventID, "rejected"); err != nil {
+	if err := enqueueControlAckReason(tx, dispatcherID, tenantID, eventID, "rejected", reason); err != nil {
 		return err
 	}
 	if err := tx.Commit(); err != nil {
@@ -205,6 +206,13 @@ func (s *Store) RejectControl(dispatcherID string, tenantID int64, eventID strin
 }
 
 func enqueueControlAck(tx *sql.Tx, dispatcherID string, tenantID int64, eventID, status string) error {
+	return enqueueControlAckReason(tx, dispatcherID, tenantID, eventID, status, "")
+}
+
+func enqueueControlAckReason(tx *sql.Tx, dispatcherID string, tenantID int64, eventID, status, reason string) error {
+	if status == "rejected" && strings.TrimSpace(reason) == "" {
+		return errors.New("rejected task control requires a concrete reason")
+	}
 	if eventID == "" || len(eventID) > 255 || tenantID <= 0 || (status != "applied" && status != "rejected") {
 		return errors.New("invalid task control acknowledgment identity or status")
 	}
@@ -219,10 +227,12 @@ func enqueueControlAck(tx *sql.Tx, dispatcherID string, tenantID int64, eventID,
 		TenantID     int64  `json:"tenant_id"`
 		Payload      struct {
 			Status string `json:"status"`
+			Reason string `json:"reason_message,omitempty"`
 		} `json:"payload"`
 	}{EventID: eventID, EventType: "task.control", DispatcherID: dispatcherID, TenantID: tenantID, Payload: struct {
 		Status string `json:"status"`
-	}{status}})
+		Reason string `json:"reason_message,omitempty"`
+	}{status, reason}})
 	if err != nil {
 		return fmt.Errorf("encode task control acknowledgment: %w", err)
 	}

internal/store/private_data.go

@@ -0,0 +1,65 @@
+package store
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+)
+
+func persistPrivateData(dir, name string, data []byte) error {
+	if err := os.MkdirAll(dir, 0700); err != nil {
+		return err
+	}
+	info, err := os.Lstat(dir)
+	if err != nil || !info.IsDir() || info.Mode().Perm() != 0700 {
+		return errors.New("private data directory must have mode 0700")
+	}
+	target := filepath.Join(dir, name)
+	if info, err := os.Lstat(target); err == nil {
+		if !info.Mode().IsRegular() || info.Mode().Perm() != 0600 {
+			return errors.New("existing private data must be a regular 0600 file")
+		}
+		existing, err := os.ReadFile(target)
+		if err != nil {
+			return err
+		}
+		if !bytes.Equal(existing, data) {
+			return errors.New("private original data conflicts with existing file")
+		}
+		f, err := os.Open(target)
+		if err != nil {
+			return err
+		}
+		if err := errors.Join(f.Sync(), f.Close()); err != nil {
+			return err
+		}
+		return syncPrivateDirectory(dir)
+	} else if !errors.Is(err, os.ErrNotExist) {
+		return err
+	}
+	f, err := os.CreateTemp(dir, ".pending-")
+	if err != nil {
+		return err
+	}
+	temp := f.Name()
+	defer os.Remove(temp)
+	_, writeErr := f.Write(data)
+	syncErr := f.Sync()
+	closeErr := f.Close()
+	if err := errors.Join(writeErr, syncErr, closeErr); err != nil {
+		return fmt.Errorf("persist private original bytes: %w", err)
+	}
+	if err := os.Rename(temp, target); err != nil {
+		return err
+	}
+	return syncPrivateDirectory(dir)
+}
+func syncPrivateDirectory(dir string) error {
+	f, err := os.Open(dir)
+	if err != nil {
+		return err
+	}
+	return errors.Join(f.Sync(), f.Close())
+}

internal/store/rejected_data.go

@@ -0,0 +1,127 @@
+package store
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"path/filepath"
+
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+	"git.ipao.vip/rogee/go-sip/internal/tenant"
+)
+
+var ErrUnidentifiableData = errors.New("rejected data identity cannot be verified")
+
+// RejectData saves private original bytes before committing an immutable outbox
+// rejection. The caller may acknowledge the input only after this succeeds.
+func (s *Store) RejectData(dispatcherID string, original []byte, cause error) error {
+	var id struct {
+		EventID      string `json:"event_id"`
+		EventType    string `json:"event_type"`
+		DispatcherID string `json:"dispatcher_id"`
+		TenantID     int64  `json:"tenant_id"`
+	}
+	if json.Unmarshal(original, &id) != nil || id.DispatcherID != dispatcherID || id.EventID == "" || len(id.EventID) > 255 {
+		return ErrUnidentifiableData
+	}
+	if id.EventType != "task.control" && id.EventType != "call.execute" && id.EventType != "sip.config" {
+		return ErrUnidentifiableData
+	}
+	// A structurally valid receipt is not an original command identity. Never
+	// turn a bounced outbound receipt into another receipt.
+	if contract.ValidateCurrent("mq", original) == nil {
+		var envelope map[string]json.RawMessage
+		if err := json.Unmarshal(original, &envelope); err != nil {
+			return err
+		}
+		var payload map[string]json.RawMessage
+		if err := json.Unmarshal(envelope["payload"], &payload); err != nil {
+			return err
+		}
+		if _, receipt := payload["status"]; receipt {
+			return ErrUnidentifiableData
+		}
+	}
+	if id.EventType != "sip.config" {
+		var known int
+		if err := s.db.QueryRow(`SELECT count(*) FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=?`, dispatcherID, id.TenantID).Scan(&known); err != nil {
+			return err
+		}
+		if id.TenantID <= 0 || known == 0 {
+			return ErrUnidentifiableData
+		}
+	}
+	if cause == nil {
+		return errors.New("rejection requires an actual failure reason")
+	}
+	payload := map[string]any{"status": "rejected", "reason_message": cause.Error()}
+	if id.EventType == "call.execute" {
+		payload["reason_code"] = nil
+	}
+	receipt := map[string]any{"event_id": id.EventID, "event_type": id.EventType, "dispatcher_id": dispatcherID, "payload": payload}
+	if id.EventType != "sip.config" {
+		receipt["tenant_id"] = id.TenantID
+	}
+	body, err := json.Marshal(receipt)
+	if err != nil {
+		return err
+	}
+	if err := contract.ValidateCurrent("mq", body); err != nil {
+		return fmt.Errorf("rejection feedback violates current Schema: %w", err)
+	}
+	var failure *contract.ValidationFailure
+	var invalidSource []byte
+	if errors.As(cause, &failure) {
+		invalidSource = failure.Raw
+	}
+	evidence, err := json.Marshal(struct {
+		Original      []byte
+		InvalidSource []byte
+		Feedback      json.RawMessage
+	}{original, invalidSource, body})
+	if err != nil {
+		return err
+	}
+	sum := sha256.Sum256(evidence)
+	dir := filepath.Join(filepath.Dir(s.path), "rejected-data")
+	if err := persistPrivateData(dir, hex.EncodeToString(sum[:])+".json", evidence); err != nil {
+		return fmt.Errorf("persist rejected-data evidence: %w", err)
+	}
+	route, err := tenant.ResultRoute(dispatcherID)
+	if err != nil {
+		return err
+	}
+	tx, err := s.db.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	if id.EventType == "call.execute" {
+		var admitted int
+		if err := tx.QueryRow(`SELECT count(*) FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, id.EventID).Scan(&admitted); err != nil {
+			return err
+		}
+		if admitted != 0 {
+			return errors.New("structural rejection conflicts with an already durably admitted execution")
+		}
+	}
+	var oldType string
+	var old []byte
+	err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, id.EventID).Scan(&oldType, &old)
+	if errors.Is(err, sql.ErrNoRows) {
+		_, err = tx.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body) VALUES(?,?,?,?,?)`, dispatcherID, id.EventID, id.EventType, route.BindingKey, body)
+	} else if err == nil {
+		if oldType != id.EventType || !bytes.Equal(old, body) {
+			return errors.New("rejection conflicts with original event result")
+		}
+		_, err = tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_id=?`, dispatcherID, id.EventID)
+	}
+	if err != nil {
+		return fmt.Errorf("persist rejected-data feedback: %w", err)
+	}
+	return tx.Commit()
+}

internal/store/rejected_data_failure_test.go

@@ -0,0 +1,60 @@
+package store
+
+import (
+	"errors"
+	"git.ipao.vip/rogee/go-sip/internal/configread"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestRejectionEvidenceFailureCreatesNoFeedback(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state.db")
+	s, err := Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	if err := s.SaveSnapshot(currentStoreSnapshot(t)); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.DiscoveredTask{{TaskID: "task-asr", TenantID: 1001, TaskRevision: 1, Status: "running"}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Mkdir(filepath.Join(filepath.Dir(path), "rejected-data"), 0755); err != nil {
+		t.Fatal(err)
+	}
+	raw := []byte(`{"event_id":"failed-original","event_type":"task.control","dispatcher_id":"` + currentDispatcherID + `","tenant_id":1001,"payload":{"task_id":"task-asr","action":"resume"}}`)
+	if err := s.RejectData(currentDispatcherID, raw, errors.New("issued_at required")); err == nil {
+		t.Fatal("evidence failure swallowed")
+	}
+	events, err := s.ListPendingOutbox(currentDispatcherID)
+	if err != nil || len(events) != 0 {
+		t.Fatalf("feedback committed without original evidence: %v %v", events, err)
+	}
+}
+func TestRejectionOutboxFailurePreservesOriginalWithoutFalseFeedback(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state.db")
+	s, err := Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	if err := s.SaveSnapshot(currentStoreSnapshot(t)); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.DiscoveredTask{{TaskID: "task-asr", TenantID: 1001, TaskRevision: 1, Status: "running"}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.db.Exec(`CREATE TRIGGER reject_outbox_write BEFORE INSERT ON dispatcher_outbox BEGIN SELECT RAISE(FAIL,'injected outbox failure'); END`); err != nil {
+		t.Fatal(err)
+	}
+	raw := []byte(`{"event_id":"failed-original","event_type":"task.control","dispatcher_id":"` + currentDispatcherID + `","tenant_id":1001,"payload":{"task_id":"task-asr","action":"resume"}}`)
+	if err := s.RejectData(currentDispatcherID, raw, errors.New("issued_at required")); err == nil {
+		t.Fatal("outbox failure swallowed")
+	}
+	files, err := filepath.Glob(filepath.Join(filepath.Dir(path), "rejected-data", "*.json"))
+	if err != nil || len(files) != 1 {
+		t.Fatalf("original lost on outbox failure: %v %v", files, err)
+	}
+}

internal/store/rejected_data_test.go

@@ -0,0 +1,77 @@
+package store
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"git.ipao.vip/rogee/go-sip/internal/configread"
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+)
+
+func TestRejectedDataPersistsOriginalAndFeedbackWithoutExecuting(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "current.db")
+	s, err := Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	if err := s.SaveSnapshot(currentStoreSnapshot(t)); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.DiscoveredTask{{TaskID: "task-asr", TenantID: 1001, TaskRevision: 1, Status: "running"}}); err != nil {
+		t.Fatal(err)
+	}
+	raw := []byte(`{"event_id":"bad-original","event_type":"task.control","dispatcher_id":"` + currentDispatcherID + `","tenant_id":1001,"payload":{"task_id":"task-asr","action":"resume","reason":"PRIVATE ORIGINAL"}}`)
+	cause := contract.ValidateCurrent("mq", raw)
+	if cause == nil {
+		t.Fatal("fixture unexpectedly valid")
+	}
+	for i := 0; i < 2; i++ {
+		if err := s.RejectData(currentDispatcherID, raw, cause); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var body []byte
+	var count int
+	if err := s.db.QueryRow(`SELECT count(*),body FROM dispatcher_outbox WHERE event_id='bad-original'`).Scan(&count, &body); err != nil {
+		t.Fatal(err)
+	}
+	if count != 1 || contract.ValidateCurrent("mq", body) != nil || !bytes.Contains(body, []byte(`"rejected"`)) || !bytes.Contains(body, []byte("issued_at")) || bytes.Contains(body, []byte("PRIVATE ORIGINAL")) {
+		t.Fatalf("incorrect rejection: %s", body)
+	}
+	if err := s.db.QueryRow(`SELECT count(*) FROM dispatcher_inbox`).Scan(&count); err != nil || count != 0 {
+		t.Fatalf("rejection admitted a call: %d %v", count, err)
+	}
+	files, err := filepath.Glob(filepath.Join(filepath.Dir(path), "rejected-data", "*.json"))
+	if err != nil || len(files) != 1 {
+		t.Fatalf("original evidence absent: %v %v", files, err)
+	}
+	info, err := os.Stat(files[0])
+	if err != nil || info.Mode().Perm() != 0600 {
+		t.Fatalf("original not private: %v %v", info, err)
+	}
+	saved, err := os.ReadFile(files[0])
+	if err != nil {
+		t.Fatal(err)
+	}
+	var evidence struct{ Original []byte }
+	if json.Unmarshal(saved, &evidence) != nil || !bytes.Equal(evidence.Original, raw) {
+		t.Fatal("saved bytes differ from original")
+	}
+}
+func TestUnidentifiableOrForeignRejectedDataDoesNotFabricateFeedback(t *testing.T) {
+	s, err := Open(filepath.Join(t.TempDir(), "current.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	for _, raw := range []string{`{`, `{"event_type":"task.control"}`, `{"event_id":"x","event_type":"task.control","dispatcher_id":"00000000-0000-0000-0000-000000000001","tenant_id":1001}`} {
+		if err := s.RejectData(currentDispatcherID, []byte(raw), errors.New("invalid structure")); !errors.Is(err, ErrUnidentifiableData) {
+			t.Fatalf("unidentifiable/foreign data accepted: %v", err)
+		}
+	}
+}

internal/store/rejection_admitted_test.go

@@ -0,0 +1,44 @@
+package store
+
+import (
+	"errors"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"git.ipao.vip/rogee/go-sip/internal/configread"
+)
+
+func TestStructuralRejectionCannotMislabelAlreadyAdmittedExecution(t *testing.T) {
+	s, err := Open(filepath.Join(t.TempDir(), "state.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	snapshot := currentStoreSnapshot(t)
+	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.DiscoveredTask{{TaskID: snapshot.Task.TaskID, TenantID: snapshot.Task.TenantID, TaskRevision: 1, Status: "running"}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.SaveSnapshot(snapshot); err != nil {
+		t.Fatal(err)
+	}
+	cmd := ExecuteCommand{DispatcherID: currentDispatcherID, TenantID: 1001, EventID: "admitted-original", TaskID: snapshot.Task.TaskID, Callee: "15003164745", IssuedAt: time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC).Format(time.RFC3339Nano)}
+	if _, _, err := s.RecordExecute(cmd); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.db.Exec(`UPDATE dispatcher_inbox SET status='unknown' WHERE dispatcher_id=? AND event_id=?`, cmd.DispatcherID, cmd.EventID); err != nil {
+		t.Fatal(err)
+	}
+	raw := []byte(`{"event_id":"admitted-original","event_type":"call.execute","dispatcher_id":"` + currentDispatcherID + `","tenant_id":1001,"payload":{}}`)
+	if err := s.RejectData(currentDispatcherID, raw, errors.New("issued_at required")); err == nil {
+		t.Fatal("structural rejection contradicted an admitted execution")
+	}
+	var status string
+	if err := s.db.QueryRow(`SELECT status FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=?`, cmd.DispatcherID, cmd.EventID).Scan(&status); err != nil || status != "unknown" {
+		t.Fatalf("unknown execution changed: %s %v", status, err)
+	}
+	events, err := s.ListPendingOutbox(currentDispatcherID)
+	if err != nil || len(events) != 0 {
+		t.Fatalf("false rejection published: %+v %v", events, err)
+	}
+}

internal/store/rejection_identity.go

@@ -0,0 +1,58 @@
+package store
+
+import (
+	"database/sql"
+	"encoding/json"
+	"errors"
+)
+
+var ErrEventRejected = errors.New("original event identity was already rejected")
+
+// A rejected identity is terminal, even if its producer later changes the body.
+// Retransmit the original feedback, never adopt the changed command.
+func (s *Store) ReplayRejectedEvent(dispatcherID, eventID, eventType string, tenantID int64) (bool, error) {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return false, err
+	}
+	defer tx.Rollback()
+	rejected, err := replayRejectedEvent(tx, dispatcherID, eventID, eventType, tenantID)
+	if err != nil {
+		return false, err
+	}
+	if err := tx.Commit(); err != nil {
+		return false, err
+	}
+	return rejected, nil
+}
+func replayRejectedEvent(tx *sql.Tx, dispatcherID, eventID, eventType string, tenantID int64) (bool, error) {
+	var kind string
+	var body []byte
+	err := tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&kind, &body)
+	if errors.Is(err, sql.ErrNoRows) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	if kind != eventType {
+		return false, errors.New("original event identity belongs to another outbound result")
+	}
+	var receipt struct {
+		TenantID int64 `json:"tenant_id"`
+		Payload  struct {
+			Status string `json:"status"`
+		} `json:"payload"`
+	}
+	if err := json.Unmarshal(body, &receipt); err != nil {
+		return false, err
+	}
+	if receipt.TenantID != tenantID {
+		return false, errors.New("original rejection tenant identity conflicts with incoming data")
+	}
+	if receipt.Payload.Status != "rejected" {
+		return false, nil
+	}
+	_, err = tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID)
+	return err == nil, err
+}

internal/store/sip_source.go

@@ -0,0 +1,123 @@
+package store
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"os"
+	"path/filepath"
+
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+)
+
+type SIPSource struct {
+	Original   []byte
+	Generation int64
+	name       string
+}
+
+func (s *Store) sipSourceDir() string {
+	return filepath.Join(filepath.Dir(s.path), "sip-notification-sources")
+}
+
+// A valid notification is acknowledged before SIP can be loaded when old calls
+// must drain. Keep its exact source until the correlated read/apply completes.
+func (s *Store) RememberSIPSource(dispatcherID string, original []byte) error {
+	if err := contract.ValidateCurrent("mq", original); err != nil {
+		return err
+	}
+	var id struct {
+		EventType    string `json:"event_type"`
+		DispatcherID string `json:"dispatcher_id"`
+		Payload      struct {
+			Status string `json:"status"`
+		} `json:"payload"`
+	}
+	if json.Unmarshal(original, &id) != nil || id.EventType != "sip.config" || id.DispatcherID != dispatcherID || id.Payload.Status != "" {
+		return ErrUnidentifiableData
+	}
+	_, generation, err := s.SIPState(dispatcherID)
+	if err != nil {
+		return err
+	}
+	if generation <= 0 {
+		return errors.New("SIP source requires a pending internal generation")
+	}
+	data, err := json.Marshal(SIPSource{Original: original, Generation: generation})
+	if err != nil {
+		return err
+	}
+	sum := sha256.Sum256(data)
+	return persistPrivateData(s.sipSourceDir(), hex.EncodeToString(sum[:])+".json", data)
+}
+func (s *Store) PendingSIPSources() ([]SIPSource, error) {
+	entries, err := os.ReadDir(s.sipSourceDir())
+	if errors.Is(err, os.ErrNotExist) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	var sources []SIPSource
+	for _, entry := range entries {
+		name := filepath.Join(s.sipSourceDir(), entry.Name())
+		info, err := os.Lstat(name)
+		if err != nil {
+			return nil, err
+		}
+		if !info.Mode().IsRegular() || info.Mode().Perm() != 0600 {
+			return nil, errors.New("SIP notification source is not private")
+		}
+		raw, err := os.ReadFile(name)
+		if err != nil {
+			return nil, err
+		}
+		sum := sha256.Sum256(raw)
+		if filepath.Base(name) != hex.EncodeToString(sum[:])+".json" {
+			return nil, errors.New("SIP notification original bytes changed")
+		}
+		var source SIPSource
+		if err := json.Unmarshal(raw, &source); err != nil {
+			return nil, err
+		}
+		if source.Generation <= 0 {
+			return nil, errors.New("SIP source generation missing")
+		}
+		source.name = filepath.Base(name)
+		sources = append(sources, source)
+	}
+	return sources, nil
+}
+func (s *Store) CompleteAppliedSIPSources(dispatcherID string) error {
+	var applied int64
+	if err := s.db.QueryRow(`SELECT applied_sip_revision FROM dispatcher_state WHERE dispatcher_id=?`, dispatcherID).Scan(&applied); err != nil {
+		return err
+	}
+	sources, err := s.PendingSIPSources()
+	if err != nil {
+		return err
+	}
+	for _, source := range sources {
+		if source.Generation <= applied {
+			if err := s.CompleteSIPSource(source); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+func (s *Store) CompleteSIPSource(source SIPSource) error {
+	data, err := json.Marshal(source)
+	if err != nil {
+		return err
+	}
+	sum := sha256.Sum256(data)
+	if source.name != hex.EncodeToString(sum[:])+".json" {
+		return errors.New("invalid SIP notification source identity")
+	}
+	if err := os.Remove(filepath.Join(s.sipSourceDir(), source.name)); err != nil {
+		return err
+	}
+	return syncPrivateDirectory(s.sipSourceDir())
+}

internal/store/sip_source_generation_test.go

@@ -0,0 +1,39 @@
+package store
+
+import (
+	"bytes"
+	"path/filepath"
+	"testing"
+)
+
+func TestAppliedSIPCleanupPreservesNewerUnloadedNotification(t *testing.T) {
+	s, err := Open(filepath.Join(t.TempDir(), "state.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	first := []byte(`{"event_id":"first","event_type":"sip.config","dispatcher_id":"` + currentDispatcherID + `","payload":{"trunk_id":"1","change_type":"enabled","revision":1}}`)
+	if err := s.NoteSIPChange(currentDispatcherID, 1); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.RememberSIPSource(currentDispatcherID, first); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.db.Exec(`UPDATE dispatcher_state SET applied_sip_revision=1,pending_sip_revision=0 WHERE dispatcher_id=?`, currentDispatcherID); err != nil {
+		t.Fatal(err)
+	}
+	second := []byte(`{"event_id":"second","event_type":"sip.config","dispatcher_id":"` + currentDispatcherID + `","payload":{"trunk_id":"1","change_type":"enabled","revision":2}}`)
+	if err := s.NoteSIPChange(currentDispatcherID, 2); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.RememberSIPSource(currentDispatcherID, second); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.CompleteAppliedSIPSources(currentDispatcherID); err != nil {
+		t.Fatal(err)
+	}
+	pending, err := s.PendingSIPSources()
+	if err != nil || len(pending) != 1 || !bytes.Equal(pending[0].Original, second) {
+		t.Fatalf("new pending context lost or applied context retained: %+v %v", pending, err)
+	}
+}

internal/store/sip_source_test.go

@@ -0,0 +1,65 @@
+package store
+
+import (
+	"bytes"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+)
+
+func TestSIPSourceSurvivesRestartAndCorrelatesMalformedHTTP(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state.db")
+	s, err := Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.NoteSIPChange(currentDispatcherID, 1); err != nil {
+		t.Fatal(err)
+	}
+	raw := []byte(`{"event_id":"sip-original","event_type":"sip.config","dispatcher_id":"` + currentDispatcherID + `","payload":{"trunk_id":"1","change_type":"enabled","revision":1}}`)
+	if err := s.RememberSIPSource(currentDispatcherID, raw); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Close(); err != nil {
+		t.Fatal(err)
+	}
+	s, err = Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+	sources, err := s.PendingSIPSources()
+	if err != nil || len(sources) != 1 || !bytes.Equal(sources[0].Original, raw) {
+		t.Fatalf("notification context lost: %+v %v", sources, err)
+	}
+	invalid := []byte(`{"resource":"sip_config","private":"PRIVATE-SIP-HTTP"}`)
+	cause := contract.ValidateCurrent("config-read", invalid)
+	if cause == nil {
+		t.Fatal("invalid HTTP fixture accepted")
+	}
+	if err := s.RejectData(currentDispatcherID, raw, cause); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.CompleteSIPSource(sources[0]); err != nil {
+		t.Fatal(err)
+	}
+	events, err := s.ListPendingOutbox(currentDispatcherID)
+	if err != nil || len(events) != 1 || !bytes.Contains(events[0].Body, []byte(`"event_type":"sip.config"`)) {
+		t.Fatalf("notification rejection missing: %+v %v", events, err)
+	}
+	files, err := filepath.Glob(filepath.Join(filepath.Dir(path), "rejected-data", "*.json"))
+	if err != nil || len(files) != 1 {
+		t.Fatalf("private original missing: %v %v", files, err)
+	}
+	saved, err := os.ReadFile(files[0])
+	if err != nil {
+		t.Fatal(err)
+	}
+	var evidence struct{ Original, InvalidSource []byte }
+	if err := json.Unmarshal(saved, &evidence); err != nil || !bytes.Equal(evidence.Original, raw) || !bytes.Equal(evidence.InvalidSource, invalid) {
+		t.Fatal("source bytes changed or not saved")
+	}
+}

internal/store/store.go

@@ -16,7 +16,10 @@ import (
 
 // Store refuses an incompatible database instead of clearing or
 // replaying data whose owner and message identity cannot be proved safe.
-type Store struct{ db *sql.DB }
+type Store struct {
+	db   *sql.DB
+	path string
+}
 
 var (
 	ErrCommandConflict            = errors.New("command id reused with different body")
@@ -200,7 +203,7 @@ func Open(path string) (_ *Store, err error) {
 	} else if _, err := db.Exec(schema); err != nil {
 		return nil, fmt.Errorf("verify current SQLite schema: %w", err)
 	}
-	return &Store{db: db}, nil
+	return &Store{db: db, path: path}, nil
 }
 
 func (s *Store) Close() error { return s.db.Close() }