merge: include task identity and state in task control receipts

contracts/control_receipt_test.go

@@ -0,0 +1,52 @@
+package contracts
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestTaskControlReceiptFields(t *testing.T) {
+	schema, err := CompileCurrent("mq-task-control-receipt.schema.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	envelope := func(payload map[string]any) map[string]any {
+		return map[string]any{"event_id": "control-fields", "event_type": "task.control", "dispatcher_id": "c046b893-8628-4589-ae50-619d049248a6", "tenant_id": 1001, "payload": payload}
+	}
+	applied := map[string]any{"task_id": "task-a", "action": "pause", "status": "applied", "reason_code": "applied", "task_state": "paused"}
+	if err := schema.Validate(envelope(applied)); err != nil {
+		t.Fatalf("requested success structure rejected: %v", err)
+	}
+	for key := range applied {
+		raw, _ := json.Marshal(applied)
+		var missing map[string]any
+		_ = json.Unmarshal(raw, &missing)
+		delete(missing, key)
+		if err := schema.Validate(envelope(missing)); err == nil {
+			t.Errorf("success accepted without %s", key)
+		}
+	}
+	for _, payload := range []map[string]any{
+		{"status": "rejected", "reason_code": "invalid_data", "reason_message": "issued_at missing"},
+		{"task_id": "task-a", "action": "pause", "status": "rejected", "reason_code": "control_rejected", "task_state": "stopped", "reason_message": "task stopped"},
+		{"task_id": "task-a", "status": "rejected", "reason_code": "invalid_data", "task_state": "running", "reason_message": "invalid action"},
+	} {
+		if err := schema.Validate(envelope(payload)); err != nil {
+			t.Errorf("rejection with independently known fields rejected: %v", err)
+		}
+	}
+	for _, payload := range []map[string]any{
+		{"status": "applied"},
+		{"task_id": "task-a", "action": "pause", "status": "applied", "reason_code": "invalid_data", "task_state": "paused"},
+		{"status": "rejected", "reason_code": "invalid_data", "reason_message": ""},
+		{"status": "rejected", "reason_message": "invalid data"},
+		{"task_id": nil, "status": "rejected", "reason_code": "invalid_data", "reason_message": "invalid data"},
+		{"task_id": "bad task", "status": "rejected", "reason_code": "invalid_data", "reason_message": "invalid data"},
+		{"action": "unknown", "status": "rejected", "reason_code": "invalid_data", "reason_message": "invalid data"},
+		{"task_state": "invented", "status": "rejected", "reason_code": "invalid_data", "reason_message": "invalid data"},
+	} {
+		if err := schema.Validate(envelope(payload)); err == nil {
+			t.Errorf("invalid payload accepted: %+v", payload)
+		}
+	}
+}

docs/contracts.md

@@ -7,6 +7,7 @@ SaaS 开发人员与本项目共同维护业务规范、Schema、MQ 拓扑和正
 本项目的 `contracts/schema` 是 Git submodule，只记录消费的确定提交，不维护另一份本地定义。
 原 `contracts/local` 已移除；`docs/thirds/saas-dispatcher.md` 只提供导航。
 内部 `proto/agent/agent.proto` 仍属于本项目，不移入 SaaS 业务契约。
+任务控制回执及逐字段拒绝反馈规则统一见共享仓库的 `mq-task-control-receipt.schema.json`、`saas-dispatcher.md` 和 `examples/mq-control-{ack,rejected}.json`；SaaS 须同步采用这些入口，不以历史归档提案或旧的仅 status 回执作为现行结构。
 
 | 位置 | 用途 |
 | --- | --- |

docs/evidence/task-control-receipt-fields-20261010.md

@@ -0,0 +1,43 @@
+# 任务控制回执补齐 — 2026-10-10
+
+## 范围与来源
+
+- 使用者确认恢复任务控制成功回执的五个字段，覆盖 start/pause/resume/stop/edit。
+- 使用者进一步确认：整条消息未通过 Schema，不能据此丢弃其中有效的任务编号和动作；拒绝回执仅省略自身无效或确实无法取得的字段，不填 null，不推测任务状态。
+- 调研前已同步项目与共享合同，基线为项目 `e48ef85`、共享合同 `8e6c332`。
+- 共享合同变更 `e750fe4618e6472821ceaa74f6c6227d11f998e3` 已推送到共享仓库 main；消费方实现与合同指针一起提交为 `b1ae50a`，推送到本项目 `check-task-control-receipt-task-id` 分支。
+- 未合入本项目 main、未部署、未向真实 SaaS/MQ 发送消息、未拨号、未读取或重置现存业务数据。
+
+## SaaS 开发同步说明
+
+共享仓库 `git@gitee.com:zzmbac/sip-contracts.git` 的 main 已更新至 `e750fe4`。请 SaaS 开发同步读取以下权威入口；本文件不维护另一份字段定义：
+
+- `contracts/schema/mq-task-control-receipt.schema.json`
+- `contracts/schema/saas-dispatcher.md` 的任务控制与拒绝反馈规则
+- `contracts/schema/examples/mq-control-ack.json`
+- `contracts/schema/examples/mq-control-rejected.json`
+
+成功回执恢复任务关联和实际状态；拒绝回执仍保留具体原因，分别按共享字段定义读取有效身份与动作。已存的原回执不会因任务状态后来变化被改写。队列确认不代表 SaaS 应用已处理。
+
+本次共享合同已经发布，但消费方代码尚未合入 main 或部署，不能据此要求当前线上服务已经返回新结构。此说明供使用者直接转达 SaaS；没有独立的 SaaS 通知送达或应用签收证据。
+
+## 实现边界
+
+- 成功与业务拒绝的任务状态在持久回执所在事务内读取，限定 Dispatcher、租户、任务归属。
+- 结构拒绝逐字段使用共享 Schema 的字段定义验证；缺少 issued_at、reason 类型错误、无效动作或无效任务编号不会丢失其他有效字段。
+- 先按原始 JSON 字段拆分，再分别解码目标字段；其他字段含超出浮点范围的数值也不会抹掉有效任务编号。
+- 请求自带的状态不作为事实；查不到归属内任务时省略状态，数据库读取失败则明确返回错误并回滚，不能以省略状态掩盖故障。
+- 原回执重发保持原消息身份与原状态；不同动作不能复用同一回执身份。
+- 增加收发方向判定：补齐任务编号和动作后的回执仍不能作为入站控制命令执行。
+- 未新增业务表、旧库迁移、旧回执兼容或回退路径。
+
+## 验证结果
+
+- 新增 Schema、存储测试先失败后通过；入站方向与实际 MQ 收发断言通过。
+- `go test ./contracts ./internal/store ./internal/mq ./internal/dispatcher -count=1` 通过。
+- `make check` 通过：格式、Proto/来源、共享合同、Go vet、全部 race 测试、构建、隔离 RabbitMQ 与本机 HTTP/HTTPS/双向 TLS 测试。
+- 隔离 RabbitMQ 的 `TestBrokerSharedResultQueueAndNoConfigure` 直接读取队列里的成功和结构拒绝回执，核对任务关联、动作、实际状态、原因及发送前后正文相等；不是仅检查构造函数或本机数据库。
+- `make coverage` 通过，业务单元覆盖率 **72.6%**，高于要求的 65%。
+- 固定提交 `b1ae50a` 的 `make release-check-local` 通过，source dirty=false，合同固定为 `e750fe4`，本地制品 SHA-256 为 `a87168962e4e6bef828a02a7ce92eafb71c759707547cac8c2aba8978731684f`；production approved=false。
+
+本地隔离验证不代替真实 SaaS 接收或生产签收。本次未验证线上新回执、真实 SIP/AI/OSS、主机部署或呼叫。

internal/contract/rejection_test.go

@@ -16,6 +16,9 @@ func TestRejectedFeedbackRequiresReasonAndPreservesOriginalType(t *testing.T) {
 			if kind == "call.execute" {
 				body["payload"].(map[string]any)["reason_code"] = nil
 			}
+			if kind == "task.control" {
+				body["payload"].(map[string]any)["reason_code"] = "invalid_data"
+			}
 			raw, err := json.Marshal(body)
 			if err != nil {
 				t.Fatal(err)

internal/dispatcher/control.go

@@ -60,6 +60,7 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) (re
 			TaskID  string `json:"task_id"`
 			Action  string `json:"action"`
 			Reason  string `json:"reason"`
+			Status  string `json:"status"`
 			Options struct {
 				ActiveCallPolicy string `json:"active_call_policy"`
 			} `json:"options"`
@@ -68,7 +69,7 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) (re
 	if err := json.Unmarshal(body, &event); err != nil {
 		return fmt.Errorf("decode task.control: %w", err)
 	}
-	if event.EventType != "task.control" || event.DispatcherID != c.DispatcherID || event.Payload.TaskID == "" || event.Payload.Action == "" {
+	if event.EventType != "task.control" || event.DispatcherID != c.DispatcherID || event.Payload.TaskID == "" || event.Payload.Action == "" || event.Payload.Status != "" {
 		return errors.New("control event type, task, or Dispatcher owner mismatch")
 	}
 	defer func() {
@@ -134,7 +135,7 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) (re
 		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, err.Error())
+					return c.Store.RejectControl(event.DispatcherID, event.TenantID, event.EventID, event.Payload.TaskID, event.Payload.Action, err.Error())
 				}
 				return fmt.Errorf("start task %q: %w", event.Payload.TaskID, err)
 			}
@@ -153,7 +154,7 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) (re
 	}
 	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, err.Error()); rejectErr != nil {
+			if rejectErr := c.Store.RejectControl(event.DispatcherID, event.TenantID, event.EventID, event.Payload.TaskID, event.Payload.Action, err.Error()); rejectErr != nil {
 				return fmt.Errorf("persist rejected control %q: %w", event.EventID, rejectErr)
 			}
 			return nil

internal/mq/broker.go

@@ -415,6 +415,7 @@ func (b *Broker) validateInbound(queue, routingKey string, body []byte) error {
 			TaskID   string `json:"task_id"`
 			Action   string `json:"action"`
 			Callee   string `json:"callee"`
+			Status   string `json:"status"`
 			TrunkID  string `json:"trunk_id"`
 			Revision int64  `json:"revision"`
 		} `json:"payload"`
@@ -422,6 +423,9 @@ func (b *Broker) validateInbound(queue, routingKey string, body []byte) error {
 	if err := json.Unmarshal(body, &identity); err != nil {
 		return fmt.Errorf("decode MQ message identity: %w", err)
 	}
+	if identity.Payload.Status != "" {
+		return errors.New("outbound receipt cannot enter an inbound command queue")
+	}
 	if identity.DispatcherID != b.dispatcherID {
 		return errors.New("MQ message belongs to another Dispatcher")
 	}

internal/mq/broker_integration_test.go

@@ -3,6 +3,7 @@
 package mq
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -304,16 +305,69 @@ func TestBrokerSharedResultQueueAndNoConfigure(t *testing.T) {
 	var receipt struct {
 		EventID string `json:"event_id"`
 		Payload struct {
-			Status string `json:"status"`
-			Reason string `json:"reason_message"`
+			TaskID     string `json:"task_id"`
+			Action     string `json:"action"`
+			Status     string `json:"status"`
+			ReasonCode string `json:"reason_code"`
+			TaskState  string `json:"task_state"`
+			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") {
+	if err := json.Unmarshal(rejected.Body, &receipt); err != nil || receipt.EventID != "original-missing-issued-at" || receipt.Payload.TaskID != "task-asr" || receipt.Payload.Action != "resume" || receipt.Payload.TaskState != "running" || receipt.Payload.ReasonCode != "invalid_data" || 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)
 	}
+	// Inspect the actual queued success body as well as the correlated
+	// structural rejection above. This is isolated MQ, not external SaaS.
+	if err := s.PrepareControl(ids[0], 1001, "task-asr", "pause"); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.CompleteControl(ids[0], 1001, "task-asr", "pause", "control-applied-fields"); err != nil {
+		t.Fatal(err)
+	}
+	feedback, err = s.ListPendingOutbox(ids[0])
+	if err != nil {
+		t.Fatal(err)
+	}
+	var appliedBody []byte
+	for _, row := range feedback {
+		if row.EventID == "control-applied-fields" {
+			appliedBody = row.Body
+		}
+	}
+	if len(appliedBody) == 0 {
+		t.Fatal("applied receipt missing")
+	}
+	if err := brokers[0].Publish(context.Background(), resultDestination.Exchange, resultDestination.BindingKey, appliedBody); err != nil {
+		t.Fatal(err)
+	}
+	applied, ok, err := admin.Get(shared, false)
+	if err != nil || !ok || !bytes.Equal(applied.Body, appliedBody) {
+		t.Fatalf("success receipt changed or missing on MQ: %t %v", ok, err)
+	}
+	var success struct {
+		Payload map[string]any `json:"payload"`
+	}
+	if err := json.Unmarshal(applied.Body, &success); err != nil {
+		t.Fatal(err)
+	}
+	want := map[string]any{"task_id": "task-asr", "action": "pause", "status": "applied", "reason_code": "applied", "task_state": "paused"}
+	if len(success.Payload) != len(want) {
+		t.Fatalf("success receipt must contain exactly five fields: %+v", success.Payload)
+	}
+	for k, v := range want {
+		if success.Payload[k] != v {
+			t.Errorf("MQ success %s: got %v, want %v", k, success.Payload[k], v)
+		}
+	}
+	if err := applied.Ack(false); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.MarkOutboxConfirmed(ids[0], "control-applied-fields"); 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 {

internal/mq/inbound_test.go

@@ -22,6 +22,8 @@ func TestBrokerInboundRequiresExactQueueRouteOwnerAndSchema(t *testing.T) {
 		wantError                   bool
 	}{
 		{"control", control.Queue, control.BindingKey, "mq-control", false},
+		{"success receipt on control", control.Queue, control.BindingKey, "mq-control-ack", true},
+		{"rejected receipt on control", control.Queue, control.BindingKey, "mq-control-rejected", true},
 		{"sip change", control.Queue, control.BindingKey, "mq-sip-change", false},
 		{"execute", task.Queue, task.BindingKey, "mq-execute", false},
 		{"wrong queue", "agent-call.d.c046b893-8628-4589-ae50-619d049248a7.task.task-asr.v1", task.BindingKey, "mq-execute", true},

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", "task is stopped"); err != nil {
+	if err := s.RejectControl(currentDispatcherID, 1001, "resume-1", "task-asr", "resume", "task is stopped"); err != nil {
 		t.Fatal(err)
 	}
 	outbox, err := s.ListPendingOutbox(currentDispatcherID)

internal/store/control_flow.go

@@ -38,7 +38,7 @@ func (s *Store) CompleteStart(snapshot configread.Snapshot, eventID string) erro
 	if err := applyDiscoveredTasks(tx, task.DispatcherID, []configread.DiscoveredTask{{TenantID: task.TenantID, TaskID: task.TaskID, TaskRevision: task.TaskRevision, Status: task.Status}}); err != nil {
 		return err
 	}
-	if err := enqueueControlAck(tx, task.DispatcherID, task.TenantID, eventID, "applied"); err != nil {
+	if err := enqueueControlAck(tx, task.DispatcherID, task.TenantID, eventID, task.TaskID, "start", "applied", ""); err != nil {
 		return err
 	}
 	if err := tx.Commit(); err != nil {
@@ -64,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 := enqueueControlAckReason(tx, task.DispatcherID, task.TenantID, eventID, "rejected", "task is absent or stopped; edits are not allowed"); err != nil {
+		if err := enqueueControlAck(tx, task.DispatcherID, task.TenantID, eventID, task.TaskID, "edit", "rejected", "task is absent or stopped; edits are not allowed"); err != nil {
 			return "", err
 		}
 		if err := tx.Commit(); err != nil {
@@ -78,7 +78,7 @@ func (s *Store) CompleteEdit(snapshot configread.Snapshot, eventID string) (stri
 	if _, err := tx.Exec(`UPDATE dispatcher_tasks SET task_revision=? WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, task.TaskRevision, task.DispatcherID, task.TenantID, task.TaskID); err != nil {
 		return "", fmt.Errorf("persist edited task revision: %w", err)
 	}
-	if err := enqueueControlAck(tx, task.DispatcherID, task.TenantID, eventID, "applied"); err != nil {
+	if err := enqueueControlAck(tx, task.DispatcherID, task.TenantID, eventID, task.TaskID, "edit", "applied", ""); err != nil {
 		return "", err
 	}
 	if err := tx.Commit(); err != nil {
@@ -181,7 +181,7 @@ func (s *Store) CompleteControl(dispatcherID string, tenantID int64, taskID, act
 	if _, err := tx.Exec(`UPDATE dispatcher_tasks SET control_state=? WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, final, dispatcherID, tenantID, taskID); err != nil {
 		return fmt.Errorf("persist applied task control: %w", err)
 	}
-	if err := enqueueControlAck(tx, dispatcherID, tenantID, eventID, "applied"); err != nil {
+	if err := enqueueControlAck(tx, dispatcherID, tenantID, eventID, taskID, action, "applied", ""); err != nil {
 		return err
 	}
 	if err := tx.Commit(); err != nil {
@@ -190,13 +190,13 @@ func (s *Store) CompleteControl(dispatcherID string, tenantID int64, taskID, act
 	return nil
 }
 
-func (s *Store) RejectControl(dispatcherID string, tenantID int64, eventID, reason string) error {
+func (s *Store) RejectControl(dispatcherID string, tenantID int64, eventID, taskID, action, reason string) error {
 	tx, err := s.db.Begin()
 	if err != nil {
 		return err
 	}
 	defer tx.Rollback()
-	if err := enqueueControlAckReason(tx, dispatcherID, tenantID, eventID, "rejected", reason); err != nil {
+	if err := enqueueControlAck(tx, dispatcherID, tenantID, eventID, taskID, action, "rejected", reason); err != nil {
 		return err
 	}
 	if err := tx.Commit(); err != nil {
@@ -205,11 +205,7 @@ func (s *Store) RejectControl(dispatcherID string, tenantID int64, eventID, reas
 	return nil
 }
 
-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 {
+func enqueueControlAck(tx *sql.Tx, dispatcherID string, tenantID int64, eventID, taskID, action, status, reason string) error {
 	if status == "rejected" && strings.TrimSpace(reason) == "" {
 		return errors.New("rejected task control requires a concrete reason")
 	}
@@ -220,37 +216,53 @@ func enqueueControlAckReason(tx *sql.Tx, dispatcherID string, tenantID int64, ev
 	if err != nil {
 		return err
 	}
+	state, err := controlReceiptState(tx, dispatcherID, tenantID, taskID)
+	if err != nil {
+		return err
+	}
+	if status == "applied" && state == "" {
+		return errors.New("applied task control receipt requires a durable task state")
+	}
+	reasonCode := "applied"
+	if status == "rejected" {
+		reasonCode = "control_rejected"
+	}
+	payload := controlReceiptPayload{TaskID: taskID, Action: action, Status: status, ReasonCode: reasonCode, TaskState: state, ReasonMessage: reason}
+	// A retransmission reports the original state, not a later task state.
+	var oldType string
+	var oldBody []byte
+	err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&oldType, &oldBody)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return fmt.Errorf("inspect existing task control outbox: %w", err)
+	}
+	if err == nil {
+		var previous struct {
+			Payload controlReceiptPayload `json:"payload"`
+		}
+		if err := json.Unmarshal(oldBody, &previous); err != nil {
+			return fmt.Errorf("decode original control receipt: %w", err)
+		}
+		payload.TaskState = previous.Payload.TaskState
+	}
 	body, err := json.Marshal(struct {
-		EventID      string `json:"event_id"`
-		EventType    string `json:"event_type"`
-		DispatcherID string `json:"dispatcher_id"`
-		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"`
-		Reason string `json:"reason_message,omitempty"`
-	}{status, reason}})
+		EventID      string                `json:"event_id"`
+		EventType    string                `json:"event_type"`
+		DispatcherID string                `json:"dispatcher_id"`
+		TenantID     int64                 `json:"tenant_id"`
+		Payload      controlReceiptPayload `json:"payload"`
+	}{EventID: eventID, EventType: "task.control", DispatcherID: dispatcherID, TenantID: tenantID, Payload: payload})
 	if err != nil {
 		return fmt.Errorf("encode task control acknowledgment: %w", err)
 	}
 	if err := contract.ValidateCurrent("mq", body); err != nil {
 		return fmt.Errorf("task control acknowledgment violates MQ contract: %w", err)
 	}
-	var oldType string
-	var oldBody []byte
-	err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&oldType, &oldBody)
-	if errors.Is(err, sql.ErrNoRows) {
+	if oldBody == nil {
 		if _, err := tx.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body) VALUES(?,?,?,?,?)`, dispatcherID, eventID, "task.control", route.BindingKey, body); err != nil {
 			return fmt.Errorf("persist task control outbox: %w", err)
 		}
 		return nil
 	}
-	if err != nil {
-		return fmt.Errorf("inspect existing task control outbox: %w", err)
-	}
 	if oldType != "task.control" || string(oldBody) != string(body) {
 		return fmt.Errorf("control event identity %q conflicts with another outbound result", eventID)
 	}

internal/store/control_receipt.go

@@ -0,0 +1,79 @@
+package store
+
+import (
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"git.ipao.vip/rogee/go-sip/contracts"
+)
+
+type controlReceiptPayload struct {
+	TaskID        string `json:"task_id,omitempty"`
+	Action        string `json:"action,omitempty"`
+	Status        string `json:"status"`
+	ReasonCode    string `json:"reason_code"`
+	TaskState     string `json:"task_state,omitempty"`
+	ReasonMessage string `json:"reason_message,omitempty"`
+}
+
+// State is read in the same transaction as the receipt. A missing task is
+// distinguishable from a database failure; neither request data nor status
+// from another tenant/Dispatcher can supply this fact.
+func controlReceiptState(tx *sql.Tx, dispatcherID string, tenantID int64, taskID string) (string, error) {
+	var state string
+	err := tx.QueryRow(`SELECT CASE WHEN control_state<>'' THEN control_state ELSE status END FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, dispatcherID, tenantID, taskID).Scan(&state)
+	if errors.Is(err, sql.ErrNoRows) {
+		return "", nil
+	}
+	if err != nil {
+		return "", fmt.Errorf("read task control receipt state: %w", err)
+	}
+	return state, nil
+}
+
+// Validate each recoverable field against its own shared-schema definition,
+// not against the whole invalid request. An invalid sibling cannot erase a
+// valid task ID/action, and supplied task_state is never trusted.
+func rejectedControlPayload(tx *sql.Tx, dispatcherID string, tenantID int64, raw json.RawMessage, reasonCode, reason string) (controlReceiptPayload, error) {
+	result := controlReceiptPayload{Status: "rejected", ReasonCode: reasonCode, ReasonMessage: reason}
+	schema, err := contracts.CompileCurrent("mq-task-control-receipt.schema.json")
+	if err != nil {
+		return result, err
+	}
+	var fields map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &fields); err != nil {
+		// A non-object payload has no independently parseable fields.
+		return result, nil
+	}
+	for _, key := range []string{"task_id", "action"} {
+		rawValue, present := fields[key]
+		if !present {
+			continue
+		}
+		var value any
+		if err := json.Unmarshal(rawValue, &value); err != nil {
+			continue
+		}
+		if schema.Properties["payload"].Properties[key].Validate(value) != nil {
+			continue
+		}
+		text, ok := value.(string)
+		if !ok {
+			return result, fmt.Errorf("shared control receipt %s schema accepted a non-string", key)
+		}
+		if key == "task_id" {
+			result.TaskID = text
+		} else {
+			result.Action = text
+		}
+	}
+	if result.TaskID != "" {
+		result.TaskState, err = controlReceiptState(tx, dispatcherID, tenantID, result.TaskID)
+		if err != nil {
+			return result, err
+		}
+	}
+	return result, nil
+}

internal/store/control_receipt_test.go

@@ -0,0 +1,252 @@
+package store
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+)
+
+func controlPayload(t *testing.T, s *Store, eventID string) map[string]any {
+	t.Helper()
+	var body []byte
+	if err := s.db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, currentDispatcherID, eventID).Scan(&body); err != nil {
+		t.Fatal(err)
+	}
+	if err := contract.ValidateCurrent("mq", body); err != nil {
+		t.Fatalf("invalid receipt: %s: %v", body, err)
+	}
+	var event struct {
+		Payload map[string]any `json:"payload"`
+	}
+	if err := json.Unmarshal(body, &event); err != nil {
+		t.Fatal(err)
+	}
+	return event.Payload
+}
+
+func TestControlReceiptAllActionsReportDurableState(t *testing.T) {
+	for _, action := range []string{"start", "pause", "resume", "stop", "edit"} {
+		t.Run(action, func(t *testing.T) {
+			s := preparedCurrentCallStore(t)
+			snapshot := currentStoreSnapshot(t)
+			snapshot.SIP.Trunks = []byte(strings.Replace(string(snapshot.SIP.Trunks), `"max_concurrent_calls":null`, `"max_concurrent_calls":2`, 1))
+			state := "running"
+			var err error
+			switch action {
+			case "start":
+				err = s.CompleteStart(snapshot, "receipt")
+			case "edit":
+				if err = s.PrepareControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil {
+					t.Fatal(err)
+				}
+				if err = s.CompleteControl(currentDispatcherID, 1001, "task-asr", "pause", "prior"); err != nil {
+					t.Fatal(err)
+				}
+				_, err = s.CompleteEdit(snapshot, "receipt")
+				state = "paused"
+			default:
+				if action == "resume" {
+					if err = s.PrepareControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil {
+						t.Fatal(err)
+					}
+					if err = s.CompleteControl(currentDispatcherID, 1001, "task-asr", "pause", "prior"); err != nil {
+						t.Fatal(err)
+					}
+				}
+				if err = s.PrepareControl(currentDispatcherID, 1001, "task-asr", action); err != nil {
+					t.Fatal(err)
+				}
+				err = s.CompleteControl(currentDispatcherID, 1001, "task-asr", action, "receipt")
+				if action == "pause" {
+					state = "paused"
+				}
+				if action == "stop" {
+					state = "stopped"
+				}
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			payload := controlPayload(t, s, "receipt")
+			want := map[string]any{"task_id": "task-asr", "action": action, "status": "applied", "reason_code": "applied", "task_state": state}
+			if len(payload) != len(want) {
+				t.Fatalf("success must contain exactly five fields: %+v", payload)
+			}
+			for k, v := range want {
+				if payload[k] != v {
+					t.Errorf("%s: got %v, want %v", k, payload[k], v)
+				}
+			}
+		})
+	}
+}
+
+func TestRejectedControlExtractsValidFieldsIndependently(t *testing.T) {
+	for _, tc := range []struct{ name, payload, task, action, state string }{
+		{"missing timestamp", `{"task_id":"task-asr","action":"pause","reason":"private"}`, "task-asr", "pause", "running"},
+		{"invalid sibling", `{"task_id":"task-asr","action":"pause","reason":99}`, "task-asr", "pause", "running"},
+		{"overflowing sibling number", `{"task_id":"task-asr","action":"pause","reason":1e1000}`, "task-asr", "pause", "running"},
+		{"non-object payload", `["task-asr","pause"]`, "", "", ""},
+		{"null task", `{"task_id":null,"action":"pause"}`, "", "pause", ""},
+		{"invalid action type", `{"task_id":"task-asr","action":99}`, "task-asr", "", "running"},
+		{"unknown action", `{"task_id":"task-asr","action":"invalid"}`, "task-asr", "", "running"},
+		{"invalid task type", `{"task_id":99,"action":"pause"}`, "", "pause", ""},
+		{"invalid task syntax", `{"task_id":"bad task","action":"pause"}`, "", "pause", ""},
+		{"missing task", `{"action":"pause"}`, "", "pause", ""},
+		{"unknown task", `{"task_id":"absent","action":"pause"}`, "absent", "pause", ""},
+		{"other tenant task", `{"task_id":"foreign-task","action":"pause"}`, "foreign-task", "pause", ""},
+		{"fake requested state", `{"task_id":"task-asr","action":"pause","task_state":"stopped"}`, "task-asr", "pause", "running"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			s := preparedCurrentCallStore(t)
+			if _, err := s.db.Exec(`INSERT INTO dispatcher_tasks(dispatcher_id,tenant_id,task_id,task_revision,status,present) VALUES(?,2002,'foreign-task',1,'stopped',1)`, currentDispatcherID); err != nil {
+				t.Fatal(err)
+			}
+			raw := []byte(`{"event_id":"rejected-fields","event_type":"task.control","dispatcher_id":"` + currentDispatcherID + `","tenant_id":1001,"payload":` + tc.payload + `}`)
+			cause := contract.ValidateCurrent("mq", raw)
+			if cause == nil {
+				t.Fatal("fixture must be rejected")
+			}
+			if err := s.RejectData(currentDispatcherID, raw, cause); err != nil {
+				t.Fatal(err)
+			}
+			payload := controlPayload(t, s, "rejected-fields")
+			for k, want := range map[string]string{"task_id": tc.task, "action": tc.action, "task_state": tc.state} {
+				got, present := payload[k]
+				if want == "" {
+					if present {
+						t.Errorf("unverified %s must be omitted: %v", k, got)
+					}
+				} else if got != want {
+					t.Errorf("valid %s lost: %v, want %s", k, got, want)
+				}
+			}
+			if payload["reason_code"] != "invalid_data" || payload["status"] != "rejected" || payload["reason_message"] == "" {
+				t.Fatalf("missing rejection cause: %+v", payload)
+			}
+			var state string
+			if err := s.db.QueryRow(`SELECT control_state FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=1001 AND task_id='task-asr'`, currentDispatcherID).Scan(&state); err != nil || state != "" {
+				t.Fatalf("invalid request executed: %q %v", state, err)
+			}
+		})
+	}
+}
+
+func TestBusinessRejectedControlReportsTaskIdentityAndActualState(t *testing.T) {
+	for _, tc := range []struct{ name, task, action, state string }{
+		{"stopped task", "task-asr", "resume", "stopped"},
+		{"unknown task", "absent", "start", ""},
+		{"pending pause", "task-asr", "resume", "pausing"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			s := preparedCurrentCallStore(t)
+			if tc.state != "" {
+				if _, err := s.db.Exec(`UPDATE dispatcher_tasks SET control_state=? WHERE dispatcher_id=? AND tenant_id=1001 AND task_id='task-asr'`, tc.state, currentDispatcherID); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if err := s.RejectControl(currentDispatcherID, 1001, "business-rejection", tc.task, tc.action, "task cannot accept this control"); err != nil {
+				t.Fatal(err)
+			}
+			payload := controlPayload(t, s, "business-rejection")
+			for k, want := range map[string]string{"task_id": tc.task, "action": tc.action, "status": "rejected", "reason_code": "control_rejected", "reason_message": "task cannot accept this control"} {
+				if payload[k] != want {
+					t.Errorf("%s: got %v, want %v", k, payload[k], want)
+				}
+			}
+			if tc.state == "" {
+				if _, present := payload["task_state"]; present {
+					t.Fatal("invented unknown task state")
+				}
+			} else if payload["task_state"] != tc.state {
+				t.Fatalf("actual state missing: %+v", payload)
+			}
+		})
+	}
+}
+
+func TestRejectedControlReplayPreservesOriginalTaskState(t *testing.T) {
+	s := preparedCurrentCallStore(t)
+	raw := []byte(`{"event_id":"rejected-repeat","event_type":"task.control","dispatcher_id":"` + currentDispatcherID + `","tenant_id":1001,"payload":{"task_id":"task-asr","action":"pause"}}`)
+	cause := contract.ValidateCurrent("mq", raw)
+	if err := s.RejectData(currentDispatcherID, raw, cause); err != nil {
+		t.Fatal(err)
+	}
+	var original []byte
+	if err := s.db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE event_id='rejected-repeat'`).Scan(&original); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.MarkOutboxConfirmed(currentDispatcherID, "rejected-repeat"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.db.Exec(`UPDATE dispatcher_tasks SET control_state='stopped' WHERE dispatcher_id=? AND tenant_id=1001 AND task_id='task-asr'`, currentDispatcherID); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.RejectData(currentDispatcherID, raw, cause); err != nil {
+		t.Fatal(err)
+	}
+	rows, err := s.ListPendingOutbox(currentDispatcherID)
+	if err != nil || len(rows) != 1 || !bytes.Equal(original, rows[0].Body) {
+		t.Fatalf("rejected receipt changed on replay: %+v %v", rows, err)
+	}
+}
+
+func TestRejectedControlStateReadFailureDoesNotPersistFeedback(t *testing.T) {
+	s := preparedCurrentCallStore(t)
+	if _, err := s.db.Exec(`ALTER TABLE dispatcher_tasks RENAME COLUMN control_state TO unavailable_control_state`); err != nil {
+		t.Fatal(err)
+	}
+	raw := []byte(`{"event_id":"state-failure","event_type":"task.control","dispatcher_id":"` + currentDispatcherID + `","tenant_id":1001,"payload":{"task_id":"task-asr","action":"pause"}}`)
+	if err := s.RejectData(currentDispatcherID, raw, errors.New("invalid structure")); err == nil {
+		t.Fatal("state lookup failure silently omitted task_state")
+	}
+	var count int
+	if err := s.db.QueryRow(`SELECT count(*) FROM dispatcher_outbox`).Scan(&count); err != nil || count != 0 {
+		t.Fatalf("feedback persisted on state failure: %d %v", count, err)
+	}
+}
+
+func TestEditReceiptRetransmitsOriginalStateAfterTaskChanges(t *testing.T) {
+	s := preparedCurrentCallStore(t)
+	snapshot := currentStoreSnapshot(t)
+	snapshot.SIP.Trunks = []byte(strings.Replace(string(snapshot.SIP.Trunks), `"max_concurrent_calls":null`, `"max_concurrent_calls":2`, 1))
+	if _, err := s.CompleteEdit(snapshot, "edit-repeat"); err != nil {
+		t.Fatal(err)
+	}
+	var original []byte
+	if err := s.db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE event_id='edit-repeat'`).Scan(&original); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.MarkOutboxConfirmed(currentDispatcherID, "edit-repeat"); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.PrepareControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.CompleteControl(currentDispatcherID, 1001, "task-asr", "pause", "pause-later"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.CompleteEdit(snapshot, "edit-repeat"); err != nil {
+		t.Fatal(err)
+	}
+	var repeated []byte
+	if err := s.db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE event_id='edit-repeat'`).Scan(&repeated); err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(original, repeated) {
+		t.Fatalf("original receipt changed: %s -> %s", original, repeated)
+	}
+	if payload := controlPayload(t, s, "edit-repeat"); payload["task_state"] != "running" {
+		t.Fatalf("original state lost: %+v", payload)
+	}
+	if err := s.PrepareControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.CompleteControl(currentDispatcherID, 1001, "task-asr", "pause", "edit-repeat"); err == nil || !strings.Contains(err.Error(), "conflicts") {
+		t.Fatalf("event identity reused for another action: %v", err)
+	}
+}

internal/store/rejected_data.go

@@ -20,10 +20,11 @@ var ErrUnidentifiableData = errors.New("rejected data identity cannot be verifie
 // 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"`
+		EventID      string          `json:"event_id"`
+		EventType    string          `json:"event_type"`
+		DispatcherID string          `json:"dispatcher_id"`
+		TenantID     int64           `json:"tenant_id"`
+		Payload      json.RawMessage `json:"payload"`
 	}
 	if json.Unmarshal(original, &id) != nil || id.DispatcherID != dispatcherID || id.EventID == "" || len(id.EventID) > 255 {
 		return ErrUnidentifiableData
@@ -58,9 +59,27 @@ func (s *Store) RejectData(dispatcherID string, original []byte, cause error) er
 	if cause == nil {
 		return errors.New("rejection requires an actual failure reason")
 	}
-	payload := map[string]any{"status": "rejected", "reason_message": cause.Error()}
+	tx, err := s.db.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	var payload any = map[string]any{"status": "rejected", "reason_message": cause.Error()}
+	if id.EventType == "task.control" {
+		replayed, err := replayRejectedEvent(tx, dispatcherID, id.EventID, id.EventType, id.TenantID)
+		if err != nil {
+			return err
+		}
+		if replayed {
+			return tx.Commit()
+		}
+		payload, err = rejectedControlPayload(tx, dispatcherID, id.TenantID, id.Payload, "invalid_data", cause.Error())
+		if err != nil {
+			return err
+		}
+	}
 	if id.EventType == "call.execute" {
-		payload["reason_code"] = nil
+		payload = map[string]any{"status": "rejected", "reason_message": cause.Error(), "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" {
@@ -95,11 +114,6 @@ func (s *Store) RejectData(dispatcherID string, original []byte, cause error) er
 	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 {