feat(dispatcher): clean confirmed completed call state transactionally

2026-10-11 18:48:22 +08:00
parent f07bf90759
commit 523ac84b2c
11 changed files with 1041 additions and 4 deletions
@@ -0,0 +1,33 @@
+# SQLite cleanup v0.1 本地迭代记录
+
+## 范围与状态
+
+- 开发分支：`feat/sqlite-cleanup-v0.1`；首期开发尚未全部完成。
+- 用户确认：仅本项目内部数据保留周期调整，不新增 `execution_id`，源 `event_id`（包括 numberID）原样回传；SaaS 如何处理再次返回的同 ID 结果不在范围内。
+- 共享合同固定 `ef3009d`，`make contracts-update` 的 43 个 JSON 检查与 8 项合同测试通过；接口字段、消息类型、路由、共享文件及 submodule 指针不变，没有共享发布。
+- 没有合入或推送本项目 main、部署、操作现存运行数据库/恢复目录、退役主机 `.approved`、拨号、抓包或调用真实 AI/OSS。
+
+## 已完成的分批验证
+
+### Agent 永久标记退役（`f07bf90`）
+
+- 先新增并运行反例，原实现因创建 `.approved`、依赖旧标记及完成后不允许重用而失败；删除相关生产读写代码后通过。
+- 验证同身份活动期间不能重叠执行，结束后可再次执行；旧 `.approved` 文件/目录不读取、不改写、不删除，新运行不重建该目录。
+- 仅在临时目录验证私密备份字节、长度和 SHA-256 一致后删除准确标记路径，相邻会话、`.results`、`.executions`、录音、上传恢复、锁及 Dispatcher 数据均保留。
+- `go test -race ./internal/rpc ./internal/agent ./internal/store ./internal/dispatcher` 通过。
+
+### SQLite 实时清理
+
+- 新增测试先因缺少清理入口失败，实现后全部通过。
+- 已派发呼叫需要已确认终结、通过现有结果 RPC 验证后持久提交的结果，以及派发回执、最终结果各自的 MQ publisher confirm。结果提交之后没有额外可能失败的业务验收步骤；不增加第二次 Agent 确认，不把 RPC 响应的网络到达或 MQ confirm 当作 SaaS 应用收讫。
+- 删除事务重新检查类型、源 ID、归属、号码、线路、冻结快照及上传结果；矛盾证据保留并报错。SQLite 结构仍为 4，没有新表或迁移。
+- 未发起拒绝仅清理已确认拒绝；原始拒绝文件仍须由后续文件收尾步骤保护其回执。活动、pending、dispatching、dispatched、unknown、任一确认缺失均保留。
+- stopped 的取消记录逐条事务、分批处理；单条失败不阻止其他记录，停止回执到剩余取消/待拨记录清完后才删，任务停止状态不删。205 条取消记录的批次、取消中断和 SQL 故障恢复测试通过。
+- 持有待拨指令或原派发调用期间排除清理；重新核对缓冲指令，停止/删除/替换后的旧批次不再拨号。极速终结与原派发返回并发时，先完成原调用再清理。
+- MQ publisher confirm 后触发清理；清理故障记录身份与原因，不作为业务准入故障返回。
+- 原 `event_id` 清理后可再次接纳、执行和回传；只删除本通数据，共享配置和任务状态保留。报告的是删除的逻辑行/正文字节，不声称 SQLite 文件立即变小。
+- `go test -race ./internal/store ./internal/dispatcher ./internal/rpc` 通过。
+
+## 尚待完成
+
+已交付录音、拒绝原始文件的准确关联与续处理、午夜补扫，以及全量检查、隔离 MQ/HTTPS/mTLS、覆盖率和增长验收。上述局部验证不代表首期全部完成，不代表真实 SaaS、通话或生产验收。
@@ -0,0 +1,33 @@
+package dispatcher
+
+import (
+	"context"
+	"log/slog"
+
+	"git.ipao.vip/rogee/go-sip/internal/store"
+)
+
+// A cleanup failure is local to completed data. Publication has already been
+// confirmed; it must not close admission or prevent independent deliveries.
+func (c *ExecuteController) cleanupConfirmedFeedback(ctx context.Context, event store.OutboxEvent) {
+	var result store.CleanupResult
+	var err error
+	switch event.EventType {
+	case "call.execute", "call.execute.result":
+		result, err = c.Store.CleanupCall(c.DispatcherID, event.EventID)
+	case "task.control":
+		result, err = c.Store.CleanupControlReply(ctx, c.DispatcherID, event.EventID)
+	default:
+		return
+	}
+	logger := slog.Default().With("dispatcher_id", c.DispatcherID, "event_type", event.EventType, "event_id", event.EventID, "stage", "confirmed_data_cleanup")
+	if err != nil {
+		logger.Error("Completed data cleanup failed; evidence retained", "deleted_rows", result.DeletedRows, "data_bytes", result.DataBytes, "error", err)
+		return
+	}
+	if result.DeletedRows != 0 {
+		logger.Info("Completed data cleaned", "deleted_rows", result.DeletedRows, "data_bytes", result.DataBytes)
+	} else {
+		logger.Debug("Completed data retained", "reason", result.RetainedReason)
+	}
+}
@@ -0,0 +1,160 @@
+package dispatcher
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"git.ipao.vip/rogee/go-sip/internal/store"
+)
+
+func cleanerResultPayload() []byte {
+	return []byte(`{"task_id":"task-asr","callee":"15003164745","trunk_id":"trunk-mock","started_at":"2026-09-21T01:30:01Z","ended_at":"2026-09-21T01:30:06Z","duration_ms":5000,"outcome":"no_answer","reason_code":486,"reason_message":"busy","status_line":null,"raw":null,"sip_capture_error":null,"transcript":[],"opt_out":false,"recording":{}}`)
+}
+
+func TestConfirmedResultCleansCallAndPreservesSourceIDForReuse(t *testing.T) {
+	controller, originator, publisher, s := newExecuteFixture(t)
+	id := "call:numberID:000042"
+	body := executeBody(t, id, "15003164745")
+	if err := controller.ProcessExecute(context.Background(), body); err != nil {
+		t.Fatal(err)
+	}
+	if err := controller.FlushOutbox(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.FinishExecute(controller.DispatcherID, id); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := s.RecordCallResult(controller.DispatcherID, id, cleanerResultPayload()); err != nil {
+		t.Fatal(err)
+	}
+	if err := controller.FlushOutbox(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.RequireReservedCall(controller.DispatcherID, id, 1001); err == nil {
+		t.Fatal("completed call was retained after both confirmations")
+	}
+	if err := controller.ProcessExecute(context.Background(), body); err != nil {
+		t.Fatal(err)
+	}
+	if len(originator.calls) != 2 {
+		t.Fatalf("same source event was not reusable: calls=%d", len(originator.calls))
+	}
+	if err := controller.FlushOutbox(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if len(publisher.bodies) != 3 {
+		t.Fatalf("feedback count=%d", len(publisher.bodies))
+	}
+	for _, event := range publisher.bodies {
+		var feedback struct {
+			EventID   string `json:"event_id"`
+			EventType string `json:"event_type"`
+		}
+		if err := json.Unmarshal(event, &feedback); err != nil || feedback.EventID != id {
+			t.Fatalf("source event ID changed: %+v err=%v", feedback, err)
+		}
+		var fields map[string]json.RawMessage
+		if err := json.Unmarshal(event, &fields); err != nil {
+			t.Fatal(err)
+		}
+		if _, added := fields["execution_id"]; added {
+			t.Fatal("cleanup added an external execution identity")
+		}
+	}
+}
+
+func TestConfirmedStopCleansCancelledBatchWithoutDispatchError(t *testing.T) {
+	controller, originator, _, s := newExecuteFixture(t)
+	cmd := store.ExecuteCommand{DispatcherID: controller.DispatcherID, EventID: "cancelled-buffer", TenantID: 1001, TaskID: "task-asr", Callee: "15003164745", IssuedAt: "2026-09-21T01:00:00Z"}
+	if _, _, err := s.RecordExecute(cmd); err != nil {
+		t.Fatal(err)
+	}
+	batch, err := s.ListPendingExecute(controller.DispatcherID)
+	if err != nil || len(batch) != 1 {
+		t.Fatalf("batch=%v err=%v", batch, err)
+	}
+	if err := s.PrepareControl(controller.DispatcherID, 1001, "task-asr", "stop"); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.CompleteControl(controller.DispatcherID, 1001, "task-asr", "stop", "stop-buffer"); err != nil {
+		t.Fatal(err)
+	}
+	if err := controller.FlushOutbox(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if err := controller.dispatchPending(context.Background(), batch[0]); err != nil {
+		t.Fatalf("cancelled buffered item closed normal processing: %v", err)
+	}
+	if len(originator.calls) != 0 {
+		t.Fatal("cancelled buffered instruction originated")
+	}
+	if admitted, err := s.CanAdmit(controller.DispatcherID, 1001, "task-asr"); err != nil || admitted {
+		t.Fatalf("durable stop state lost: %t %v", admitted, err)
+	}
+}
+
+type blockedCleanupOriginator struct{ started, release chan struct{} }
+
+func (o *blockedCleanupOriginator) LoadedTrunks(context.Context) (map[string]int64, error) {
+	return map[string]int64{"trunk-mock": 8}, nil
+}
+func (o *blockedCleanupOriginator) Originate(ctx context.Context, _ CallSpec) error {
+	close(o.started)
+	select {
+	case <-o.release:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func TestCleanupWaitsForOriginalDispatchHandlerToFinish(t *testing.T) {
+	controller, _, _, s := newExecuteFixture(t)
+	originator := &blockedCleanupOriginator{started: make(chan struct{}), release: make(chan struct{})}
+	controller.Originator = originator
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	body := executeBody(t, "rapid-end", "15003164745")
+	dispatched := make(chan error, 1)
+	go func() { dispatched <- controller.ProcessExecute(ctx, body) }()
+	select {
+	case <-originator.started:
+	case <-ctx.Done():
+		t.Fatal("originator did not start")
+	}
+	if err := s.FinishExecute(controller.DispatcherID, "rapid-end"); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := s.RecordCallResult(controller.DispatcherID, "rapid-end", cleanerResultPayload()); err != nil {
+		t.Fatal(err)
+	}
+	flushed := make(chan error, 1)
+	go func() { flushed <- controller.FlushOutbox(ctx) }()
+	select {
+	case err := <-flushed:
+		t.Fatalf("cleanup overtook active original dispatch: %v", err)
+	case <-time.After(30 * time.Millisecond):
+	}
+	close(originator.release)
+	select {
+	case err := <-dispatched:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-ctx.Done():
+		t.Fatal("dispatch did not finish")
+	}
+	select {
+	case err := <-flushed:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-ctx.Done():
+		t.Fatal("cleanup did not resume")
+	}
+	if err := s.RequireReservedCall(controller.DispatcherID, "rapid-end", 1001); err == nil {
+		t.Fatal("completed call remained after protected handler exit")
+	}
+}
@@ -65,6 +65,7 @@ func (c *ExecuteController) ProcessExecute(ctx context.Context, body []byte) err
 	if err := c.validate(); err != nil {
 		return err
 	}
+	defer c.Store.HoldCallLifecycle()()
 	if err := contract.ValidateCurrent("mq", body); err != nil {
 		return fmt.Errorf("invalid incoming call.execute: %w", err)
 	}
@@ -128,6 +129,7 @@ func (c *ExecuteController) ProcessPending(ctx context.Context) error {
 	if err := c.validate(); err != nil {
 		return err
 	}
+	defer c.Store.HoldCallLifecycle()()
 	commands, err := c.Store.ListPendingExecute(c.DispatcherID)
 	if err != nil {
 		return err
@@ -163,6 +165,14 @@ func (c *ExecuteController) dispatchPending(ctx context.Context, cmd store.Execu
 			logger.Error("Call admission or execution failed", "stage", phase, "error", dispatchErr, "duration_ms", time.Since(started).Milliseconds())
 		}
 	}()
+	current, err := c.Store.IsCurrentPending(cmd)
+	if err != nil {
+		return fmt.Errorf("recheck pending instruction %q: %w", cmd.EventID, err)
+	}
+	if !current {
+		logger.Info("Buffered call instruction no longer pending", "stage", "pending_batch_skip")
+		return nil
+	}
 	logger.Info("Call admission check", "stage", "select_line")
 	at := c.Now()
 	issued, err := time.Parse(time.RFC3339Nano, cmd.IssuedAt)
@@ -302,7 +312,8 @@ func (c *ExecuteController) dispatchPending(ctx context.Context, cmd store.Execu
 }

 // FlushOutbox uses the same event identity/body on every retry. Confirming a
-// bound queue is not a SaaS application receipt; confirmed rows remain stored.
+// bound queue is not a SaaS application receipt. Completed local data may be
+// cleaned only after its independent completion conditions are satisfied.
 func (c *ExecuteController) FlushOutbox(ctx context.Context) error {
 	if err := c.validate(); err != nil {
 		return err
@@ -319,7 +330,9 @@ func (c *ExecuteController) FlushOutbox(ctx context.Context) error {
 		}
 		if err := c.Store.MarkOutboxConfirmed(c.DispatcherID, event.EventType, event.EventID); err != nil {
 			failures = append(failures, fmt.Errorf("persist queue confirmation for %q/%q: %w", event.EventType, event.EventID, err))
+			continue
 		}
+		c.cleanupConfirmedFeedback(ctx, event)
 	}
 	return errors.Join(failures...)
 }
@@ -290,7 +290,10 @@ func (r *Runtime) processPending(ctx context.Context) error {
 	}
 	for _, command := range pending {
 		cmd := command
-		if err := r.withTask(cmd.TaskID, func() error { return r.Execute.dispatchPending(ctx, cmd) }); err != nil {
+		if err := r.withTask(cmd.TaskID, func() error {
+			defer r.Bootstrap.Store.HoldCallLifecycle()()
+			return r.Execute.dispatchPending(ctx, cmd)
+		}); err != nil {
 			return fmt.Errorf("retry eligible pending instruction %q: %w", cmd.EventID, err)
 		}
 	}
@@ -117,6 +117,20 @@ func (s *Store) RecordExecute(cmd ExecuteCommand) (Execution, bool, error) {
 	return found, affected == 1, nil
 }

+// IsCurrentPending rejects a detached batch item that has been cancelled,
+// completed, cleaned, or replaced since the batch was read.
+func (s *Store) IsCurrentPending(cmd ExecuteCommand) (bool, error) {
+	var current ExecuteCommand
+	err := s.db.QueryRow(`SELECT dispatcher_id,event_id,tenant_id,task_id,callee,issued_at FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=? AND status='pending'`, cmd.DispatcherID, cmd.EventID).Scan(&current.DispatcherID, &current.EventID, &current.TenantID, &current.TaskID, &current.Callee, &current.IssuedAt)
+	if errors.Is(err, sql.ErrNoRows) {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("inspect current pending call: %w", err)
+	}
+	return current == cmd, nil
+}
+
 // ReserveExecute is the final transactional quota/fencing boundary before
 // sending a single originate. A crash or timeout after this point leaves an
 // occupied dispatching/unknown execution, never an automatic redial.
@@ -0,0 +1,275 @@
+package store
+
+import (
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"git.ipao.vip/rogee/go-sip/internal/contract"
+)
+
+// CleanupResult reports deleted logical data, not a reduction in SQLite's
+// physical size. Completed pages may be reused without shrinking the file.
+type CleanupResult struct {
+	DeletedRows    int
+	DataBytes      int64
+	RetainedReason string
+}
+
+// HoldCallLifecycle protects in-flight handlers and detached pending batches
+// from cleanup. It does not create a lasting execution identity or journal.
+func (s *Store) HoldCallLifecycle() func() {
+	s.cleanupMu.RLock()
+	return s.cleanupMu.RUnlock
+}
+
+type cleanupFeedback struct {
+	EventID      string `json:"event_id"`
+	EventType    string `json:"event_type"`
+	DispatcherID string `json:"dispatcher_id"`
+	TenantID     int64  `json:"tenant_id"`
+	IssuedAt     string `json:"issued_at"`
+	Payload      struct {
+		Status    string          `json:"status"`
+		TaskID    string          `json:"task_id"`
+		Action    string          `json:"action"`
+		TaskState string          `json:"task_state"`
+		Callee    string          `json:"callee"`
+		TrunkID   string          `json:"trunk_id"`
+		Recording json.RawMessage `json:"recording"`
+	} `json:"payload"`
+}
+
+func confirmedCleanupFeedback(tx *sql.Tx, dispatcherID, eventType, eventID string) (cleanupFeedback, []byte, bool, error) {
+	var body []byte
+	var confirmed int
+	var confirmedAt sql.NullString
+	err := tx.QueryRow(`SELECT body,confirmed,confirmed_at FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type=? AND event_id=?`, dispatcherID, eventType, eventID).Scan(&body, &confirmed, &confirmedAt)
+	if errors.Is(err, sql.ErrNoRows) {
+		return cleanupFeedback{}, nil, false, nil
+	}
+	if err != nil {
+		return cleanupFeedback{}, nil, false, fmt.Errorf("inspect cleanup feedback: %w", err)
+	}
+	if confirmed != 1 {
+		return cleanupFeedback{}, nil, false, nil
+	}
+	if !confirmedAt.Valid || confirmedAt.String == "" {
+		return cleanupFeedback{}, nil, false, errors.New("confirmed feedback lacks its confirmation time")
+	}
+	if err := contract.ValidateCurrent("mq", body); err != nil {
+		return cleanupFeedback{}, nil, false, fmt.Errorf("invalid cleanup feedback: %w", err)
+	}
+	var event cleanupFeedback
+	if err := json.Unmarshal(body, &event); err != nil {
+		return event, nil, false, err
+	}
+	if event.DispatcherID != dispatcherID || event.EventID != eventID || event.EventType != eventType {
+		return event, nil, false, errors.New("cleanup feedback envelope differs from its durable identity")
+	}
+	return event, body, true, nil
+}
+
+// CleanupCall rechecks completion in the deletion transaction. The final-result
+// row is committed only after all fallible ReportCallResult validation succeeds;
+// no new Agent acknowledgment protocol or database layout is required.
+func (s *Store) CleanupCall(dispatcherID, eventID string) (result CleanupResult, err error) {
+	if dispatcherID == "" || eventID == "" {
+		return result, errors.New("cleanup requires Dispatcher and call identity")
+	}
+	s.cleanupMu.Lock()
+	defer s.cleanupMu.Unlock()
+	tx, err := s.db.Begin()
+	if err != nil {
+		return result, fmt.Errorf("begin call cleanup: %w", err)
+	}
+	defer tx.Rollback()
+	var tenantID int64
+	var state, taskID, callee, trunkID, issuedAt string
+	var snapshot []byte
+	err = tx.QueryRow(`SELECT tenant_id,task_id,callee,status,COALESCE(selected_trunk_id,''),issued_at,snapshot_json FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&tenantID, &taskID, &callee, &state, &trunkID, &issuedAt, &snapshot)
+	if errors.Is(err, sql.ErrNoRows) {
+		result.RetainedReason = "no_retained_execution"
+		return result, nil
+	}
+	if err != nil {
+		return result, fmt.Errorf("inspect call cleanup identity: %w", err)
+	}
+	if state != "finished" && state != "rejected" && state != "suppressed" {
+		result.RetainedReason = "execution_" + state
+		return result, nil
+	}
+	var ackBody, resultBody []byte
+	if state == "suppressed" {
+		if trunkID != "" || len(snapshot) != 0 {
+			return result, errors.New("suppressed call contains evidence of an issued reservation")
+		}
+		var references int
+		if err := tx.QueryRow(`SELECT (SELECT COUNT(*) FROM dispatcher_recordings WHERE dispatcher_id=? AND source_event_id=?)+(SELECT COUNT(*) FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=? AND event_type IN ('call.execute','call.execute.result'))`, dispatcherID, eventID, dispatcherID, eventID).Scan(&references); err != nil {
+			return result, err
+		}
+		if references != 0 {
+			return result, errors.New("suppressed call contains conflicting upload or call feedback")
+		}
+		ready, err := confirmedStoppedTask(tx, dispatcherID, tenantID, taskID)
+		if err != nil {
+			return result, err
+		}
+		if !ready {
+			result.RetainedReason = "stop_not_confirmed"
+			return result, nil
+		}
+	} else {
+		ack, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "call.execute", eventID)
+		if err != nil {
+			return result, err
+		}
+		if !ready {
+			result.RetainedReason = "call_receipt_not_confirmed"
+			return result, nil
+		}
+		if ack.TenantID != tenantID || ack.IssuedAt != issuedAt {
+			return result, errors.New("call cleanup receipt differs from its accepted command")
+		}
+		ackBody = body
+		if state == "rejected" {
+			if ack.Payload.Status != "rejected" {
+				return result, errors.New("rejected call has a non-rejection receipt")
+			}
+			var conflicting int
+			if err := tx.QueryRow(`SELECT (SELECT COUNT(*) FROM dispatcher_recordings WHERE dispatcher_id=? AND source_event_id=?)+(SELECT COUNT(*) FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=? AND event_type='call.execute.result')`, dispatcherID, eventID, dispatcherID, eventID).Scan(&conflicting); err != nil {
+				return result, err
+			}
+			if conflicting != 0 {
+				return result, errors.New("rejected call has recording or final-result evidence")
+			}
+			// File evidence cannot lose its confirmed receipt before its own
+			// coordinated cleanup has completed.
+			entries, err := os.ReadDir(filepath.Join(filepath.Dir(s.path), "rejected-data"))
+			if err != nil && !errors.Is(err, os.ErrNotExist) {
+				return result, fmt.Errorf("inspect rejection evidence before call cleanup: %w", err)
+			}
+			if len(entries) != 0 {
+				result.RetainedReason = "rejection_files_require_cleanup"
+				return result, nil
+			}
+		} else {
+			if ack.Payload.Status != "dispatched" || trunkID == "" || len(snapshot) == 0 {
+				return result, errors.New("finished call lacks its frozen dispatch evidence")
+			}
+			final, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "call.execute.result", eventID)
+			if err != nil {
+				return result, err
+			}
+			if !ready {
+				result.RetainedReason = "final_result_not_confirmed"
+				return result, nil
+			}
+			if final.TenantID != tenantID || final.Payload.TaskID != taskID || final.Payload.Callee != callee || final.Payload.TrunkID != trunkID {
+				return result, errors.New("final cleanup result differs from the finished call")
+			}
+			if err := validateCleanupRecording(tx, dispatcherID, eventID, final.Payload.Recording); err != nil {
+				return result, err
+			}
+			resultBody = body
+		}
+	}
+	queries := []struct {
+		query string
+		args  []any
+	}{
+		{`DELETE FROM dispatcher_recordings WHERE dispatcher_id=? AND source_event_id=?`, []any{dispatcherID, eventID}},
+		{`DELETE FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=? AND event_type IN ('call.execute','call.execute.result')`, []any{dispatcherID, eventID}},
+		{`DELETE FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=? AND status=?`, []any{dispatcherID, eventID, state}},
+	}
+	for _, deletion := range queries {
+		affected, err := tx.Exec(deletion.query, deletion.args...)
+		if err != nil {
+			return CleanupResult{}, fmt.Errorf("delete completed call data: %w", err)
+		}
+		n, err := affected.RowsAffected()
+		if err != nil {
+			return CleanupResult{}, err
+		}
+		result.DeletedRows += int(n)
+	}
+	if err := tx.Commit(); err != nil {
+		return CleanupResult{}, fmt.Errorf("commit completed call cleanup: %w", err)
+	}
+	result.DataBytes = int64(len(snapshot) + len(ackBody) + len(resultBody))
+	return result, nil
+}
+
+func confirmedStoppedTask(tx *sql.Tx, dispatcherID string, tenantID int64, taskID string) (bool, error) {
+	var stopped int
+	if err := tx.QueryRow(`SELECT COUNT(*) FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND control_state='stopped'`, dispatcherID, tenantID, taskID).Scan(&stopped); err != nil {
+		return false, err
+	}
+	if stopped == 0 {
+		return false, nil
+	}
+	rows, err := tx.Query(`SELECT event_id FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='task.control' AND confirmed=1
+        AND CASE WHEN json_valid(body) THEN json_extract(body,'$.tenant_id') END=?
+        AND CASE WHEN json_valid(body) THEN json_extract(body,'$.payload.task_id') END=?`, dispatcherID, tenantID, taskID)
+	if err != nil {
+		return false, err
+	}
+	var ids []string
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			rows.Close()
+			return false, err
+		}
+		ids = append(ids, id)
+	}
+	err = rows.Err()
+	rows.Close()
+	if err != nil {
+		return false, err
+	}
+	for _, id := range ids {
+		event, _, confirmed, err := confirmedCleanupFeedback(tx, dispatcherID, "task.control", id)
+		if err != nil {
+			return false, err
+		}
+		if confirmed && event.TenantID == tenantID && event.Payload.TaskID == taskID && event.Payload.Action == "stop" && event.Payload.Status == "applied" && event.Payload.TaskState == "stopped" {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func validateCleanupRecording(tx *sql.Tx, dispatcherID, eventID string, raw json.RawMessage) error {
+	grant, err := loadRecordingUpload(tx.QueryRow(`SELECT dispatcher_id,source_event_id,upload_id,recording_id,bucket,object_key,checksum_sha256,size_bytes,format,channels,sample_rate_hz,duration_ms,COALESCE(confirmed_at,'') FROM dispatcher_recordings WHERE dispatcher_id=? AND source_event_id=?`, dispatcherID, eventID))
+	var recording struct {
+		Status     string `json:"status"`
+		Bucket     string `json:"bucket"`
+		ObjectKey  string `json:"object_key"`
+		Checksum   string `json:"checksum_sha256"`
+		Size       int64  `json:"size_bytes"`
+		Format     string `json:"format"`
+		Channels   int    `json:"channels"`
+		SampleRate int    `json:"sample_rate_hz"`
+		Duration   int64  `json:"duration_ms"`
+	}
+	if decodeErr := json.Unmarshal(raw, &recording); decodeErr != nil {
+		return fmt.Errorf("decode final cleanup recording: %w", decodeErr)
+	}
+	if errors.Is(err, ErrUploadNotFound) {
+		if recording.Status != "" {
+			return errors.New("final cleanup recording has no original upload binding")
+		}
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	if grant.ConfirmedAt == "" || recording.Status != "uploaded" || recording.Bucket != grant.Bucket || recording.ObjectKey != grant.ObjectKey || recording.Checksum != grant.ChecksumSHA256 || recording.Size != grant.SizeBytes || recording.Format != grant.Format || recording.Channels != grant.Channels || recording.SampleRate != grant.SampleRateHz || recording.Duration != grant.DurationMS {
+		return errors.New("cleanup recording differs from its confirmed original upload")
+	}
+	return nil
+}
@@ -0,0 +1,172 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+)
+
+const cleanupBatchSize = 100
+
+// CleanupControlReply leaves durable task state untouched. Rejection evidence
+// has its own file/receipt cleanup; a successful stop first drains cancelled rows.
+func (s *Store) CleanupControlReply(ctx context.Context, dispatcherID, eventID string) (CleanupResult, error) {
+	event, ready, err := s.cleanupControlIdentity(dispatcherID, eventID)
+	if err != nil {
+		return CleanupResult{}, err
+	}
+	if !ready {
+		return CleanupResult{RetainedReason: "control_reply_not_confirmed"}, nil
+	}
+	if event.Payload.Status != "applied" {
+		return CleanupResult{RetainedReason: "rejection_files_require_cleanup"}, nil
+	}
+	if event.Payload.Action == "stop" {
+		return s.CleanupStoppedTask(ctx, dispatcherID, eventID)
+	}
+	s.cleanupMu.Lock()
+	defer s.cleanupMu.Unlock()
+	tx, err := s.db.Begin()
+	if err != nil {
+		return CleanupResult{}, err
+	}
+	defer tx.Rollback()
+	current, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "task.control", eventID)
+	if err != nil {
+		return CleanupResult{}, err
+	}
+	if !ready {
+		return CleanupResult{RetainedReason: "control_reply_not_confirmed"}, nil
+	}
+	if current.TenantID != event.TenantID || current.Payload.TaskID != event.Payload.TaskID || current.Payload.Action != event.Payload.Action || current.Payload.Status != event.Payload.Status || current.Payload.TaskState != event.Payload.TaskState {
+		return CleanupResult{}, errors.New("control reply changed during cleanup")
+	}
+	deleted, err := tx.Exec(`DELETE FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='task.control' AND event_id=? AND confirmed=1`, dispatcherID, eventID)
+	if err != nil {
+		return CleanupResult{}, err
+	}
+	n, err := deleted.RowsAffected()
+	if err != nil {
+		return CleanupResult{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return CleanupResult{}, err
+	}
+	return CleanupResult{DeletedRows: int(n), DataBytes: int64(len(body))}, nil
+}
+
+// CleanupStoppedTask deletes each cancelled call in its own transaction. The
+// stop receipt remains durable across interrupted batches and individual faults.
+func (s *Store) CleanupStoppedTask(ctx context.Context, dispatcherID, stopEventID string) (total CleanupResult, cleanupErr error) {
+	if err := ctx.Err(); err != nil {
+		return total, err
+	}
+	event, ready, err := s.cleanupControlIdentity(dispatcherID, stopEventID)
+	if err != nil {
+		return total, err
+	}
+	if !ready {
+		total.RetainedReason = "stop_not_confirmed"
+		return total, nil
+	}
+	if event.Payload.Action != "stop" || event.Payload.Status != "applied" || event.Payload.TaskState != "stopped" {
+		return total, errors.New("cancelled-call cleanup requires an applied stop receipt")
+	}
+	var failures []error
+	cursor := ""
+	for {
+		if err := ctx.Err(); err != nil {
+			return total, errors.Join(append(failures, err)...)
+		}
+		rows, err := s.db.Query(`SELECT event_id FROM dispatcher_inbox WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND status='suppressed' AND event_id>? ORDER BY event_id LIMIT ?`, dispatcherID, event.TenantID, event.Payload.TaskID, cursor, cleanupBatchSize)
+		if err != nil {
+			return total, errors.Join(append(failures, err)...)
+		}
+		var ids []string
+		for rows.Next() {
+			var id string
+			if err := rows.Scan(&id); err != nil {
+				rows.Close()
+				return total, errors.Join(append(failures, err)...)
+			}
+			ids = append(ids, id)
+		}
+		err = rows.Err()
+		rows.Close()
+		if err != nil {
+			return total, errors.Join(append(failures, err)...)
+		}
+		for _, id := range ids {
+			if err := ctx.Err(); err != nil {
+				return total, errors.Join(append(failures, err)...)
+			}
+			result, err := s.CleanupCall(dispatcherID, id)
+			total.DeletedRows += result.DeletedRows
+			total.DataBytes += result.DataBytes
+			if err != nil {
+				slog.Error("Cancelled call cleanup failed; stop proof retained", "dispatcher_id", dispatcherID, "event_id", id, "task_id", event.Payload.TaskID, "error", err)
+				failures = append(failures, fmt.Errorf("cancelled call %q: %w", id, err))
+			}
+		}
+		if len(ids) < cleanupBatchSize {
+			break
+		}
+		cursor = ids[len(ids)-1]
+	}
+	s.cleanupMu.Lock()
+	defer s.cleanupMu.Unlock()
+	tx, err := s.db.Begin()
+	if err != nil {
+		return total, errors.Join(append(failures, err)...)
+	}
+	defer tx.Rollback()
+	current, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "task.control", stopEventID)
+	if err != nil {
+		return total, errors.Join(append(failures, err)...)
+	}
+	if !ready {
+		total.RetainedReason = "stop_not_confirmed"
+		return total, errors.Join(failures...)
+	}
+	if current.TenantID != event.TenantID || current.Payload.TaskID != event.Payload.TaskID || current.Payload.Action != "stop" || current.Payload.Status != "applied" || current.Payload.TaskState != "stopped" {
+		return total, errors.Join(append(failures, errors.New("stop receipt changed during cancelled-call cleanup"))...)
+	}
+	var stopped, remaining int
+	if err := tx.QueryRow(`SELECT COUNT(*) FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND control_state='stopped'`, dispatcherID, event.TenantID, event.Payload.TaskID).Scan(&stopped); err != nil {
+		return total, errors.Join(append(failures, err)...)
+	}
+	if err := tx.QueryRow(`SELECT COUNT(*) FROM dispatcher_inbox WHERE dispatcher_id=? AND tenant_id=? AND task_id=? AND status IN ('suppressed','pending')`, dispatcherID, event.TenantID, event.Payload.TaskID).Scan(&remaining); err != nil {
+		return total, errors.Join(append(failures, err)...)
+	}
+	if stopped != 1 || remaining != 0 {
+		total.RetainedReason = "cancelled_calls_not_fully_cleaned"
+		return total, errors.Join(failures...)
+	}
+	deleted, err := tx.Exec(`DELETE FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='task.control' AND event_id=? AND confirmed=1`, dispatcherID, stopEventID)
+	if err != nil {
+		return total, errors.Join(append(failures, err)...)
+	}
+	n, err := deleted.RowsAffected()
+	if err != nil {
+		return total, errors.Join(append(failures, err)...)
+	}
+	if err := tx.Commit(); err != nil {
+		return total, errors.Join(append(failures, err)...)
+	}
+	total.DeletedRows += int(n)
+	total.DataBytes += int64(len(body))
+	return total, errors.Join(failures...)
+}
+
+func (s *Store) cleanupControlIdentity(dispatcherID, eventID string) (cleanupFeedback, bool, error) {
+	s.cleanupMu.Lock()
+	defer s.cleanupMu.Unlock()
+	tx, err := s.db.Begin()
+	if err != nil {
+		return cleanupFeedback{}, false, err
+	}
+	defer tx.Rollback()
+	event, _, ready, err := confirmedCleanupFeedback(tx, dispatcherID, "task.control", eventID)
+	return event, ready, err
+}
@@ -0,0 +1,121 @@
+package store
+
+import (
+	"context"
+	"fmt"
+	"testing"
+)
+
+func TestCleanupStoppedTaskBatchesAndKeepsDurableStopState(t *testing.T) {
+	s := preparedCurrentCallStore(t)
+	cmd := currentCall("cancelled-000")
+	for i := 0; i < 205; i++ {
+		cmd.EventID = fmt.Sprintf("cancelled-%03d", i)
+		if _, _, err := s.RecordExecute(cmd); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := s.PrepareControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop"); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.CompleteControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop", "stop-batch"); err != nil {
+		t.Fatal(err)
+	}
+	confirmCleanupFeedback(t, s, "stop-batch", "task.control")
+	result, err := s.CleanupStoppedTask(context.Background(), cmd.DispatcherID, "stop-batch")
+	if err != nil || result.DeletedRows != 206 {
+		t.Fatalf("batched cleanup=%+v err=%v", result, err)
+	}
+	requireCleanupCount(t, s, "dispatcher_inbox", 0)
+	requireCleanupCount(t, s, "dispatcher_outbox", 0)
+	requireCleanupCount(t, s, "dispatcher_tasks", 1)
+	if admitted, err := s.CanAdmit(cmd.DispatcherID, cmd.TenantID, cmd.TaskID); err != nil || admitted {
+		t.Fatalf("stopped task reopened: %t %v", admitted, err)
+	}
+}
+
+func TestCleanupStoppedTaskKeepsStopProofAcrossPartialFailure(t *testing.T) {
+	s := preparedCurrentCallStore(t)
+	cmd := currentCall("cancelled-bad")
+	if _, _, err := s.RecordExecute(cmd); err != nil {
+		t.Fatal(err)
+	}
+	cmd.EventID = "cancelled-good"
+	if _, _, err := s.RecordExecute(cmd); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.PrepareControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop"); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.CompleteControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop", "stop-partial"); err != nil {
+		t.Fatal(err)
+	}
+	confirmCleanupFeedback(t, s, "stop-partial", "task.control")
+	if _, err := s.db.Exec(`CREATE TRIGGER cleanup_one_fault BEFORE DELETE ON dispatcher_inbox WHEN OLD.event_id='cancelled-bad' BEGIN SELECT RAISE(ABORT,'one call failure'); END`); err != nil {
+		t.Fatal(err)
+	}
+	if result, err := s.CleanupStoppedTask(context.Background(), cmd.DispatcherID, "stop-partial"); err == nil || result.DeletedRows != 1 {
+		t.Fatalf("independent cancelled call was not cleaned: %+v %v", result, err)
+	}
+	requireCleanupCount(t, s, "dispatcher_inbox", 1)
+	requireCleanupCount(t, s, "dispatcher_outbox", 1)
+	if _, err := s.db.Exec(`DROP TRIGGER cleanup_one_fault`); err != nil {
+		t.Fatal(err)
+	}
+	if result, err := s.CleanupStoppedTask(context.Background(), cmd.DispatcherID, "stop-partial"); err != nil || result.DeletedRows != 2 {
+		t.Fatalf("partial stop cleanup did not resume: %+v %v", result, err)
+	}
+}
+
+func TestCleanupStoppedTaskPreservesAcknowledgmentWhenInterrupted(t *testing.T) {
+	s := preparedCurrentCallStore(t)
+	cmd := currentCall("cancelled-context")
+	if _, _, err := s.RecordExecute(cmd); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.PrepareControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop"); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.CompleteControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop", "stop-context"); err != nil {
+		t.Fatal(err)
+	}
+	confirmCleanupFeedback(t, s, "stop-context", "task.control")
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, err := s.CleanupStoppedTask(ctx, cmd.DispatcherID, "stop-context"); err == nil {
+		t.Fatal("interruption was ignored")
+	}
+	requireCleanupCount(t, s, "dispatcher_inbox", 1)
+	requireCleanupCount(t, s, "dispatcher_outbox", 1)
+	if result, err := s.CleanupStoppedTask(context.Background(), cmd.DispatcherID, "stop-context"); err != nil || result.DeletedRows != 2 {
+		t.Fatalf("restart-style retry=%+v err=%v", result, err)
+	}
+}
+
+func TestCleanupPendingBatchSeesCancellationAndChangedCommand(t *testing.T) {
+	s := preparedCurrentCallStore(t)
+	cmd := currentCall("buffered-stop")
+	if _, _, err := s.RecordExecute(cmd); err != nil {
+		t.Fatal(err)
+	}
+	pending, err := s.ListPendingExecute(cmd.DispatcherID)
+	if err != nil || len(pending) != 1 {
+		t.Fatalf("pending=%v err=%v", pending, err)
+	}
+	if err := s.PrepareControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop"); err != nil {
+		t.Fatal(err)
+	}
+	if current, err := s.IsCurrentPending(pending[0]); err != nil || current {
+		t.Fatalf("cancelled batch remained eligible: %t %v", current, err)
+	}
+	if err := s.CompleteControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop", "stop-buffered"); err != nil {
+		t.Fatal(err)
+	}
+	confirmCleanupFeedback(t, s, "stop-buffered", "task.control")
+	if _, err := s.CleanupStoppedTask(context.Background(), cmd.DispatcherID, "stop-buffered"); err != nil {
+		t.Fatal(err)
+	}
+	if current, err := s.IsCurrentPending(pending[0]); err != nil || current {
+		t.Fatalf("cleaned batch remained eligible: %t %v", current, err)
+	}
+}
@@ -0,0 +1,211 @@
+package store
+
+import (
+	"database/sql"
+	"errors"
+	"testing"
+	"time"
+)
+
+func confirmCleanupFeedback(t *testing.T, s *Store, id, eventType string) {
+	t.Helper()
+	if err := s.MarkOutboxConfirmed(currentCall(id).DispatcherID, eventType, id); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func requireCleanupCount(t *testing.T, s *Store, table string, expected int) {
+	t.Helper()
+	var count int
+	if err := s.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&count); err != nil || count != expected {
+		t.Fatalf("%s rows=%d expected=%d err=%v", table, count, expected, err)
+	}
+}
+
+func TestCleanupCallNeedsSuccessfulResultAndBothTypedConfirms(t *testing.T) {
+	for _, scenario := range []struct {
+		name                              string
+		result, ackConfirm, resultConfirm bool
+		deleted                           int
+	}{
+		{"no_result_rpc", false, true, false, 0},
+		{"no_dispatch_confirm", true, false, true, 0},
+		{"no_result_confirm", true, true, false, 0},
+		{"all_complete", true, true, true, 3},
+	} {
+		t.Run(scenario.name, func(t *testing.T) {
+			s := preparedCurrentCallStore(t)
+			cmd := currentResultCall(t, s, "numberID-cleanup", true)
+			if scenario.result {
+				if _, _, err := s.RecordCallResult(cmd.DispatcherID, cmd.EventID, currentResultPayload(t)); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if scenario.ackConfirm {
+				confirmCleanupFeedback(t, s, cmd.EventID, "call.execute")
+			}
+			if scenario.resultConfirm {
+				confirmCleanupFeedback(t, s, cmd.EventID, "call.execute.result")
+			}
+			result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID)
+			if err != nil || result.DeletedRows != scenario.deleted {
+				t.Fatalf("cleanup=%+v err=%v", result, err)
+			}
+			requireCleanupCount(t, s, "dispatcher_tasks", 1)
+			requireCleanupCount(t, s, "dispatcher_configs", 1)
+			if scenario.deleted == 0 {
+				requireCleanupCount(t, s, "dispatcher_inbox", 1)
+				if result.RetainedReason == "" {
+					t.Fatal("retained call has no reason")
+				}
+				return
+			}
+			requireCleanupCount(t, s, "dispatcher_inbox", 0)
+			requireCleanupCount(t, s, "dispatcher_outbox", 0)
+			if repeated, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || repeated.DeletedRows != 0 {
+				t.Fatalf("repeat cleanup=%+v err=%v", repeated, err)
+			}
+			if _, inserted, err := s.RecordExecute(cmd); err != nil || !inserted {
+				t.Fatalf("same event ID was not reusable: inserted=%t err=%v", inserted, err)
+			}
+		})
+	}
+}
+
+func TestCleanupRejectedNeedsOnlyItsConfirmedRefusal(t *testing.T) {
+	s := preparedCurrentCallStore(t)
+	cmd := currentCall("rejected-cleanup")
+	if _, _, err := s.RecordExecute(cmd); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.RejectExecute(cmd.DispatcherID, cmd.EventID, "not issued"); err != nil {
+		t.Fatal(err)
+	}
+	if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 0 {
+		t.Fatalf("unconfirmed refusal removed: %+v %v", result, err)
+	}
+	confirmCleanupFeedback(t, s, cmd.EventID, "call.execute")
+	if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 2 {
+		t.Fatalf("refusal cleanup=%+v err=%v", result, err)
+	}
+	requireCleanupCount(t, s, "dispatcher_inbox", 0)
+	requireCleanupCount(t, s, "dispatcher_outbox", 0)
+}
+
+func TestCleanupNeverRemovesUnfinishedOrUnknownCalls(t *testing.T) {
+	for _, state := range []string{"pending", "dispatching", "dispatched", "unknown"} {
+		t.Run(state, func(t *testing.T) {
+			s := preparedCurrentCallStore(t)
+			cmd := currentCall("retain-" + state)
+			if _, _, err := s.RecordExecute(cmd); err != nil {
+				t.Fatal(err)
+			}
+			if state != "pending" {
+				if err := s.ReserveExecute(cmd.DispatcherID, cmd.EventID, currentReservation(), time.Date(2026, 9, 21, 1, 30, 0, 0, time.UTC)); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if state == "dispatched" {
+				if err := s.MarkExecuteDispatched(cmd.DispatcherID, cmd.EventID); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if state == "unknown" {
+				if err := s.MarkExecuteUnknown(cmd.DispatcherID, cmd.EventID); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 0 || result.RetainedReason == "" {
+				t.Fatalf("unfinished cleanup=%+v err=%v", result, err)
+			}
+			requireCleanupCount(t, s, "dispatcher_inbox", 1)
+		})
+	}
+}
+
+func TestCleanupCallRollbackPreservesRecordingAndOriginalFacts(t *testing.T) {
+	s := preparedCurrentCallStore(t)
+	cmd := currentResultCall(t, s, "cleanup-rollback", true)
+	binding := currentUploadBinding(cmd)
+	if _, _, err := s.BindRecordingUpload(binding); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := s.RecordUploadedCallResult(cmd.DispatcherID, cmd.EventID, currentUploadedResultPayload(t, binding), currentUploadProof(binding)); err != nil {
+		t.Fatal(err)
+	}
+	confirmCleanupFeedback(t, s, cmd.EventID, "call.execute")
+	confirmCleanupFeedback(t, s, cmd.EventID, "call.execute.result")
+	if _, err := s.db.Exec(`CREATE TRIGGER cleanup_fault BEFORE DELETE ON dispatcher_inbox BEGIN SELECT RAISE(ABORT,'cleanup interrupted'); END`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err == nil {
+		t.Fatal("cleanup swallowed SQLite deletion failure")
+	}
+	requireCleanupCount(t, s, "dispatcher_recordings", 1)
+	requireCleanupCount(t, s, "dispatcher_outbox", 2)
+	requireCleanupCount(t, s, "dispatcher_inbox", 1)
+	if _, err := s.db.Exec(`DROP TRIGGER cleanup_fault`); err != nil {
+		t.Fatal(err)
+	}
+	if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 4 {
+		t.Fatalf("retry cleanup=%+v err=%v", result, err)
+	}
+	requireCleanupCount(t, s, "dispatcher_recordings", 0)
+}
+
+func TestCleanupSuppressedRequiresCompletedConfirmedStop(t *testing.T) {
+	s := preparedCurrentCallStore(t)
+	cmd := currentCall("cancelled-cleanup")
+	if _, _, err := s.RecordExecute(cmd); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.PrepareControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop"); err != nil {
+		t.Fatal(err)
+	}
+	if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 0 {
+		t.Fatalf("stopping task cleaned early: %+v %v", result, err)
+	}
+	if err := s.CompleteControl(cmd.DispatcherID, cmd.TenantID, cmd.TaskID, "stop", "stop-cleanup"); err != nil {
+		t.Fatal(err)
+	}
+	if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 0 {
+		t.Fatalf("unconfirmed stop cleaned early: %+v %v", result, err)
+	}
+	confirmCleanupFeedback(t, s, "stop-cleanup", "task.control")
+	if result, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err != nil || result.DeletedRows != 1 {
+		t.Fatalf("cancelled cleanup=%+v err=%v", result, err)
+	}
+	requireCleanupCount(t, s, "dispatcher_outbox", 1) // stop proof belongs to the task sweep
+	requireCleanupCount(t, s, "dispatcher_tasks", 1)
+	if admitted, err := s.CanAdmit(cmd.DispatcherID, cmd.TenantID, cmd.TaskID); err != nil || admitted {
+		t.Fatalf("stop barrier lost: %t %v", admitted, err)
+	}
+	if _, _, err := s.RecordExecute(cmd); !errors.Is(err, ErrStopped) {
+		t.Fatalf("cleaned stopped command was readmitted: %v", err)
+	}
+}
+
+func TestCleanupCallRefusesMismatchedFinalIdentity(t *testing.T) {
+	s := preparedCurrentCallStore(t)
+	cmd := currentResultCall(t, s, "cleanup-corrupt", true)
+	if _, _, err := s.RecordCallResult(cmd.DispatcherID, cmd.EventID, currentResultPayload(t)); err != nil {
+		t.Fatal(err)
+	}
+	confirmCleanupFeedback(t, s, cmd.EventID, "call.execute")
+	confirmCleanupFeedback(t, s, cmd.EventID, "call.execute.result")
+	if _, err := s.db.Exec(`UPDATE dispatcher_outbox SET body=json_set(body,'$.payload.task_id','other-task') WHERE event_type='call.execute.result'`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.CleanupCall(cmd.DispatcherID, cmd.EventID); err == nil {
+		t.Fatal("conflicting result identity was deleted")
+	}
+	requireCleanupCount(t, s, "dispatcher_outbox", 2)
+	requireCleanupCount(t, s, "dispatcher_inbox", 1)
+	if _, _, err := s.RecordExecute(currentCall("neighbor")); err != nil {
+		t.Fatal(err)
+	}
+	var state string
+	if err := s.db.QueryRow(`SELECT status FROM dispatcher_inbox WHERE event_id='neighbor'`).Scan(&state); errors.Is(err, sql.ErrNoRows) || err != nil || state != "pending" {
+		t.Fatalf("neighbor changed: state=%s err=%v", state, err)
+	}
+}
@@ -9,6 +9,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"

 	"git.ipao.vip/rogee/go-sip/internal/configread"
 	_ "modernc.org/sqlite"
@@ -17,8 +18,9 @@ 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
-	path string
+	db        *sql.DB
+	path      string
+	cleanupMu sync.RWMutex
 }

 var (