feat(cleanup): sweep completed data at Shanghai midnight

cmd/sip-go-agent/agent_cleanup.go

@@ -0,0 +1,26 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+
+	"git.ipao.vip/rogee/go-sip/internal/agent"
+	"git.ipao.vip/rogee/go-sip/internal/daily"
+)
+
+// This owns only local delivered-file cleanup, never RPC/cloud recovery.
+func startAgentMidnightCleanup(parent context.Context, root string) func() {
+	ctx, cancel := context.WithCancel(parent)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		if err := daily.Run(ctx, func(ctx context.Context) error {
+			count, err := (&agent.RecordingRecovery{Root: root}).CleanupDelivered(ctx)
+			slog.InfoContext(ctx, "Agent midnight cleanup finished", "completed_recordings", count, "error", err)
+			return err
+		}); err != nil {
+			slog.Error("Agent midnight cleanup timer failed", "error", err)
+		}
+	}()
+	return func() { cancel(); <-done }
+}

cmd/sip-go-agent/agent_command.go

@@ -113,6 +113,10 @@ func newAgentCommand() *cobra.Command {
 			defer listener.Close()
 			server := grpc.NewServer(grpc.Creds(credentials.NewTLS(serverTLS)), grpc.ChainUnaryInterceptor(logging.UnaryServerInterceptor("agent", "dispatcher")))
 			agentpb.RegisterAgentControlServiceServer(server, handler)
+			if settings.RecoveryRoot != "" {
+				stopCleanup := startAgentMidnightCleanup(cmd.Context(), settings.RecoveryRoot)
+				defer stopCleanup()
+			}
 			if mode == "nonprod-real" {
 				startRealAgentRecovery(cmd.Context(), settings, recoveryClient, handler)
 			}

docs/evidence/sqlite-cleanup-v0.1-local.md

@@ -47,6 +47,15 @@
 - RejectData 写入和清理共享生命周期排除；清理结束后同源再来的新拒绝保留新原件和未确认回执，不误删为旧记录。移除了此前 admitted rejected 的“只要整个拒绝目录非空就保留”保护，改为精确本组资格。
 - MQ 确认后的真实控制器入口已验证三类拒绝，未产生执行或伪造最终结果；多原件、两阶段失败、损坏组与独立组、既有文件、未知临时文件和并发写入回归通过。`go test -race ./internal/store ./internal/dispatcher ./internal/rpc` 通过。
 
+### 每日午夜补扫
+
+- 采用固定 Asia/Shanghai 日历边界，启动只安排下一次午夜，不在启动时额外清理、不在 12:00 运行、不按消息年龄删除；每次重新计算下一日期。系统时钟回退不会提前执行，清理失败等待下一午夜，不形成高频重试。
+- Dispatcher 使用独立且可取消的后台任务，分页扫描 completed 候选并复用 MQ 确认后相同的逐条事务/原件资格；单条失败继续其他记录，不变更准入、任务状态、停止状态或发布消息。停服取消并等待清理任务结束。
+- 配置录音恢复根目录的 Agent 启动独立午夜任务，复用已有扫描和 delivered 删除；其他状态只校验保留，不申请授权、PUT 或上报结果。既有十秒真实恢复的用途和节奏不变。
+- 每条日志记录身份、保留原因或错误、实际文件数量与字节；Dispatcher 汇总扫描、保留、失败、删除行、逻辑数据字节、文件数量/字节及耗时。没有永久扫描进度或二次确认记录。
+- UTC/上海转换、12:00 不清理、午夜启动安排次日、时钟回退、失败等待次日、取消计时器，以及 205 条跨页候选、未确认/待拨邻居保留、取消后续处理、只清 delivered 不重试其他录音验证通过。
+- `go test -race ./internal/daily ./internal/store ./internal/dispatcher ./internal/agent ./cmd/sip-go-agent` 通过。
+
 ## 尚待完成
 
-午夜补扫，以及全量检查、隔离 MQ/HTTPS/mTLS、覆盖率和增长验收。上述局部验证不代表首期全部完成，不代表真实 SaaS、通话或生产验收。
+全量检查、隔离 MQ/HTTPS/mTLS、覆盖率和增长验收。上述局部验证不代表首期全部完成，不代表真实 SaaS、通话或生产验收。

internal/agent/recording_cleanup_test.go

@@ -238,6 +238,28 @@ func TestRecordingCleanupLeavesNoPermanentInMemoryLocks(t *testing.T) {
 	}
 }
 
+func TestMidnightDeliveredOnlySweepPreservesUnfinishedFiles(t *testing.T) {
+	r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")}
+	completed := deliveredRecordingFixture(t, r)
+	pending := failedRecordingFixture()
+	pending.SourceEventID = "execute-pending"
+	pending.ObjectKey = "tenant/call/pending.wav"
+	if _, err := r.SaveFailure([]byte("pending audio"), pending, &UploadHTTPError{StatusCode: 503}); err != nil {
+		t.Fatal(err)
+	}
+	count, err := r.CleanupDelivered(context.Background())
+	if err != nil || count != 1 {
+		t.Fatalf("midnight result=%d err=%v", count, err)
+	}
+	_, _, completedInfo, _ := r.paths(completed.Bucket, completed.ObjectKey)
+	if _, err := os.Stat(completedInfo); !errors.Is(err, os.ErrNotExist) {
+		t.Fatal("delivered info remains")
+	}
+	if entry, err := r.Load(pending.Bucket, pending.ObjectKey); err != nil || entry.State != "retry_pending" {
+		t.Fatalf("unfinished entry changed: %+v %v", entry, err)
+	}
+}
+
 func TestMissingAudioDoesNotRelaxUnfinishedRecordingValidation(t *testing.T) {
 	r := &RecordingRecovery{Root: filepath.Join(t.TempDir(), "private")}
 	entry := failedRecordingFixture()

internal/agent/recording_scan.go

@@ -16,7 +16,16 @@ import (
 // RecoverDue discovers failure-only recording pairs after an Agent restart.
 // Blocked/corrupt entries remain untouched and are returned as errors while
 // other independent entries continue; none can produce a fabricated result.
-func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant, report RecordingResultReporter) (processedCount int, scanErr error) {
+func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant, report RecordingResultReporter) (int, error) {
+	return r.scanRecovery(ctx, grant, report, false)
+}
+
+// CleanupDelivered is a local midnight-only sweep, never an upload retry.
+func (r *RecordingRecovery) CleanupDelivered(ctx context.Context) (int, error) {
+	return r.scanRecovery(ctx, nil, nil, true)
+}
+
+func (r *RecordingRecovery) scanRecovery(ctx context.Context, grant RecoveryGrant, report RecordingResultReporter, deliveredOnly bool) (processedCount int, scanErr error) {
 	defer func() {
 		if scanErr != nil {
 			slog.ErrorContext(ctx, "Recording recovery scan failed", "stage", "recovery_scan_failed", "processed", processedCount, "error", scanErr)
@@ -106,6 +115,9 @@ func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant,
 			addBlocked(dir, "recovery_load", err)
 			return nil
 		}
+		if deliveredOnly && before.State != "delivered" {
+			return nil
+		}
 		after, err := r.Retry(ctx, bucket, objectKey, grant, report)
 		if errors.Is(err, ErrRecordingNotDue) {
 			return nil
@@ -116,6 +128,12 @@ func (r *RecordingRecovery) RecoverDue(ctx context.Context, grant RecoveryGrant,
 		}
 		if before.State != "delivered" && after.State == "delivered" {
 			processed++
+		} else if deliveredOnly && after.State == "delivered" {
+			if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
+				processed++
+			} else if err != nil {
+				addBlocked(path, "delivered_cleanup_stat", err)
+			}
 		}
 		return nil
 	})

internal/daily/midnight.go

@@ -0,0 +1,58 @@
+// Package daily provides the project's fixed Shanghai midnight calendar timer.
+// It is not configurable and does not track permanent sweep history.
+package daily
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"time"
+	_ "time/tzdata"
+)
+
+func NextMidnight(now time.Time) (time.Time, error) {
+	zone, err := time.LoadLocation("Asia/Shanghai")
+	if err != nil {
+		return time.Time{}, err
+	}
+	local := now.In(zone)
+	return time.Date(local.Year(), local.Month(), local.Day()+1, 0, 0, 0, 0, zone), nil
+}
+
+func Run(ctx context.Context, sweep func(context.Context) error) error {
+	return run(ctx, sweep, time.Now, func(delay time.Duration) (<-chan time.Time, func()) {
+		timer := time.NewTimer(delay)
+		return timer.C, func() { timer.Stop() }
+	})
+}
+
+func run(ctx context.Context, sweep func(context.Context) error, now func() time.Time, newTimer func(time.Duration) (<-chan time.Time, func())) error {
+	if ctx == nil || sweep == nil || now == nil || newTimer == nil {
+		return errors.New("midnight cleanup requires context and a sweep")
+	}
+	for {
+		if ctx.Err() != nil {
+			return nil
+		}
+		current := now()
+		next, err := NextMidnight(current)
+		if err != nil {
+			return err
+		}
+		slog.InfoContext(ctx, "Midnight cleanup scheduled", "timezone", "Asia/Shanghai", "next_at", next)
+		tick, stop := newTimer(next.Sub(current))
+		select {
+		case <-ctx.Done():
+			stop()
+			return nil
+		case <-tick:
+			stop()
+		}
+		if now().Before(next) {
+			continue
+		} // a wall-clock rollback must not run early
+		if err := sweep(ctx); err != nil && ctx.Err() == nil {
+			slog.ErrorContext(ctx, "Midnight cleanup incomplete; retry scheduled for next midnight", "error", err)
+		}
+	}
+}

internal/daily/midnight_test.go

@@ -0,0 +1,72 @@
+package daily
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestNextMidnightUsesShanghaiCalendarNotNoonOrElapsed24Hours(t *testing.T) {
+	for _, test := range []struct{ now, want string }{
+		{"2026-10-10T03:59:00Z", "2026-10-11T00:00:00+08:00"},
+		{"2026-10-10T04:00:00Z", "2026-10-11T00:00:00+08:00"},
+		{"2026-10-10T15:59:59Z", "2026-10-11T00:00:00+08:00"},
+		{"2026-10-10T16:00:00Z", "2026-10-12T00:00:00+08:00"},
+	} {
+		now, _ := time.Parse(time.RFC3339, test.now)
+		got, err := NextMidnight(now)
+		if err != nil || got.Format(time.RFC3339) != test.want {
+			t.Fatalf("now=%s got=%s want=%s err=%v", test.now, got, test.want, err)
+		}
+	}
+}
+
+func TestWallClockRollbackDoesNotTriggerEarlyCleanup(t *testing.T) {
+	now, _ := time.Parse(time.RFC3339, "2026-10-10T15:59:00Z")
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	calls, waits := 0, []time.Duration{}
+	err := run(ctx, func(context.Context) error { calls++; return nil }, func() time.Time { return now }, func(delay time.Duration) (<-chan time.Time, func()) {
+		waits = append(waits, delay)
+		ch := make(chan time.Time, 1)
+		switch len(waits) {
+		case 1:
+			now = now.Add(delay - time.Hour)
+			ch <- now
+		case 2:
+			now = now.Add(delay)
+			ch <- now
+		default:
+			cancel()
+		}
+		return ch, func() {}
+	})
+	if err != nil || calls != 1 || len(waits) != 3 || waits[1] != time.Hour {
+		t.Fatalf("rollback ran early: calls=%d waits=%v err=%v", calls, waits, err)
+	}
+}
+
+func TestMidnightFailureRetriesOnlyTomorrowAndCancellationStopsTimer(t *testing.T) {
+	now, _ := time.Parse(time.RFC3339, "2026-10-10T15:59:00Z")
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	calls, stopped := 0, 0
+	durations := []time.Duration{}
+	clock := func() time.Time { return now }
+	timer := func(delay time.Duration) (<-chan time.Time, func()) {
+		durations = append(durations, delay)
+		ch := make(chan time.Time, 1)
+		if len(durations) <= 2 {
+			now = now.Add(delay)
+			ch <- now
+		} else {
+			cancel()
+		}
+		return ch, func() { stopped++ }
+	}
+	err := run(ctx, func(context.Context) error { calls++; return errors.New("one record failed") }, clock, timer)
+	if err != nil || calls != 2 || stopped != 3 || len(durations) != 3 || durations[0] != time.Minute || durations[1] != 24*time.Hour {
+		t.Fatalf("unexpected daily behavior: calls=%d stopped=%d delays=%v err=%v", calls, stopped, durations, err)
+	}
+}

internal/dispatcher/cleanup.go

@@ -9,9 +9,7 @@ import (
 
 // 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
+func (c *ExecuteController) cleanupConfirmedFeedback(ctx context.Context, event store.OutboxEvent) (result store.CleanupResult, err error) {
 	switch event.EventType {
 	case "call.execute", "call.execute.result":
 		result, err = c.Store.CleanupCall(c.DispatcherID, event.EventID)
@@ -26,16 +24,17 @@ func (c *ExecuteController) cleanupConfirmedFeedback(ctx context.Context, event
 	case "sip.config":
 		result, err = c.Store.CleanupRejectedFeedback(c.DispatcherID, event.EventType, event.EventID)
 	default:
-		return
+		return result, nil
 	}
 	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, "deleted_files", result.DeletedFiles, "file_bytes", result.FileBytes, "error", err)
-		return
+		return result, err
 	}
 	if result.DeletedRows != 0 || result.DeletedFiles != 0 {
 		logger.Info("Completed data cleaned", "deleted_rows", result.DeletedRows, "data_bytes", result.DataBytes, "deleted_files", result.DeletedFiles, "file_bytes", result.FileBytes)
 	} else {
-		logger.Debug("Completed data retained", "reason", result.RetainedReason)
+		logger.Info("Completed data retained", "reason", result.RetainedReason)
 	}
+	return result, nil
 }

internal/dispatcher/cleanup_scan.go

@@ -0,0 +1,63 @@
+package dispatcher
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"git.ipao.vip/rogee/go-sip/internal/store"
+)
+
+// SweepCompleted uses exactly the per-record qualifications used after MQ
+// confirmation. It never publishes, changes admission, or rewrites task state.
+func (c *ExecuteController) SweepCompleted(ctx context.Context) (total store.CleanupResult, sweepErr error) {
+	if c == nil || c.Store == nil || c.DispatcherID == "" {
+		return total, errors.New("cleanup sweep requires durable Dispatcher identity")
+	}
+	started := time.Now()
+	scanned, retained, failed := 0, 0, 0
+	defer func() {
+		slog.InfoContext(ctx, "Completed data sweep finished", "dispatcher_id", c.DispatcherID, "scanned", scanned, "retained", retained, "failed", failed, "deleted_rows", total.DeletedRows, "data_bytes", total.DataBytes, "deleted_files", total.DeletedFiles, "file_bytes", total.FileBytes, "elapsed_ms", time.Since(started).Milliseconds())
+	}()
+	cursorType, cursorID := "", ""
+	var firstFailure error
+	for {
+		if err := ctx.Err(); err != nil {
+			return total, errors.Join(firstFailure, err)
+		}
+		batch, err := c.Store.ListCleanupCandidates(c.DispatcherID, cursorType, cursorID)
+		if err != nil {
+			return total, errors.Join(firstFailure, err)
+		}
+		for _, event := range batch {
+			if err := ctx.Err(); err != nil {
+				return total, errors.Join(firstFailure, err)
+			}
+			scanned++
+			result, err := c.cleanupConfirmedFeedback(ctx, event)
+			total.DeletedRows += result.DeletedRows
+			total.DataBytes += result.DataBytes
+			total.DeletedFiles += result.DeletedFiles
+			total.FileBytes += result.FileBytes
+			if err != nil {
+				failed++
+				if firstFailure == nil {
+					firstFailure = err
+				}
+			} else if result.RetainedReason != "" {
+				retained++
+			}
+		}
+		if len(batch) == 0 {
+			break
+		}
+		last := batch[len(batch)-1]
+		cursorType, cursorID = last.EventType, last.EventID
+	}
+	if failed != 0 {
+		return total, fmt.Errorf("%d independent cleanup failures; first: %w", failed, firstFailure)
+	}
+	return total, nil
+}

internal/dispatcher/cleanup_scan_test.go

@@ -0,0 +1,66 @@
+package dispatcher
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"git.ipao.vip/rogee/go-sip/internal/store"
+)
+
+func TestMidnightSweepUsesRealTimeConditionsAndKeepsUnconfirmedNeighbors(t *testing.T) {
+	c, originator, _, s := newExecuteFixture(t)
+	for i := 0; i < 205; i++ {
+		id := fmt.Sprintf("night-confirmed-%03d", i)
+		raw := []byte(`{"event_id":"` + id + `","event_type":"call.execute","dispatcher_id":"` + c.DispatcherID + `","tenant_id":1001,"payload":{}}`)
+		if err := s.RejectData(c.DispatcherID, raw, errors.New("invalid original")); err != nil {
+			t.Fatal(err)
+		}
+		if err := s.MarkOutboxConfirmed(c.DispatcherID, "call.execute", id); err != nil {
+			t.Fatal(err)
+		}
+	}
+	raw := []byte(`{"event_id":"night-unconfirmed","event_type":"call.execute","dispatcher_id":"` + c.DispatcherID + `","tenant_id":1001,"payload":{}}`)
+	if err := s.RejectData(c.DispatcherID, raw, errors.New("invalid original")); err != nil {
+		t.Fatal(err)
+	}
+	pending := store.ExecuteCommand{DispatcherID: c.DispatcherID, EventID: "night-pending", TenantID: 1001, TaskID: "task-asr", Callee: "15003164745", IssuedAt: "2026-09-21T01:00:00Z"}
+	if _, _, err := s.RecordExecute(pending); err != nil {
+		t.Fatal(err)
+	}
+	result, err := c.SweepCompleted(context.Background())
+	if err != nil || result.DeletedRows != 205 || result.DeletedFiles != 205 || len(originator.calls) != 0 {
+		t.Fatalf("sweep=%+v err=%v", result, err)
+	}
+	outbox, err := s.ListPendingOutbox(c.DispatcherID)
+	if err != nil || len(outbox) != 1 || outbox[0].EventID != "night-unconfirmed" {
+		t.Fatalf("unconfirmed proof lost: %v %v", outbox, err)
+	}
+	commands, err := s.ListPendingExecute(c.DispatcherID)
+	if err != nil || len(commands) != 1 || commands[0].EventID != pending.EventID {
+		t.Fatalf("pending neighbor changed: %v %v", commands, err)
+	}
+	if result, err := c.SweepCompleted(context.Background()); err != nil || result.DeletedRows != 0 {
+		t.Fatalf("repeat=%+v %v", result, err)
+	}
+}
+
+func TestMidnightSweepCancellationDoesNotInventCompletion(t *testing.T) {
+	c, _, _, s := newExecuteFixture(t)
+	body := []byte(`{"event_id":"night-cancelled","event_type":"call.execute","dispatcher_id":"` + c.DispatcherID + `","tenant_id":1001,"payload":{}}`)
+	if err := s.RejectData(c.DispatcherID, body, errors.New("invalid")); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.MarkOutboxConfirmed(c.DispatcherID, "call.execute", "night-cancelled"); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if result, err := c.SweepCompleted(ctx); !errors.Is(err, context.Canceled) || result.DeletedRows != 0 {
+		t.Fatalf("cancelled sweep=%+v %v", result, err)
+	}
+	if result, err := c.SweepCompleted(context.Background()); err != nil || result.DeletedRows != 1 {
+		t.Fatalf("resumed sweep=%+v %v", result, err)
+	}
+}

internal/dispatcher/runtime.go

@@ -11,6 +11,7 @@ import (
 
 	"git.ipao.vip/rogee/go-sip/internal/configread"
 	"git.ipao.vip/rogee/go-sip/internal/contract"
+	"git.ipao.vip/rogee/go-sip/internal/daily"
 	"git.ipao.vip/rogee/go-sip/internal/mq"
 	"git.ipao.vip/rogee/go-sip/internal/store"
 	"git.ipao.vip/rogee/go-sip/internal/tenant"
@@ -75,6 +76,19 @@ func (r *Runtime) Serve(ctx context.Context) (result error) {
 		}
 	}()
 
+	cleanupContext, stopCleanup := context.WithCancel(ctx)
+	cleanupDone := make(chan struct{})
+	go func() {
+		defer close(cleanupDone)
+		if err := daily.Run(cleanupContext, func(ctx context.Context) error {
+			_, err := r.Execute.SweepCompleted(ctx)
+			return err
+		}); err != nil {
+			slog.Error("Dispatcher midnight cleanup timer failed", "error", err)
+		}
+	}()
+	defer func() { stopCleanup(); <-cleanupDone }()
+
 	r.Broker.SetInvalidMessageHandler(r.rejectInvalidData)
 	if err := r.Execute.FlushOutbox(ctx); err != nil {
 		return err

internal/store/cleanup_scan.go

@@ -0,0 +1,27 @@
+package store
+
+// ListCleanupCandidates snapshots one bounded, stable-key page. Pending and
+// unknown executions are not candidates; missing confirmations stay protected
+// by the same transactional qualification used for real-time cleanup.
+func (s *Store) ListCleanupCandidates(dispatcherID, afterType, afterID string) ([]OutboxEvent, error) {
+	rows, err := s.db.Query(`SELECT event_type,event_id FROM (
+        SELECT 'call.execute' AS event_type,event_id FROM dispatcher_inbox
+          WHERE dispatcher_id=? AND status IN ('finished','rejected','suppressed')
+        UNION
+        SELECT event_type,event_id FROM dispatcher_outbox WHERE dispatcher_id=? AND confirmed=1
+          AND event_type IN ('call.execute','call.execute.result','task.control','sip.config')
+      ) WHERE event_type>? OR (event_type=? AND event_id>?) ORDER BY event_type,event_id LIMIT ?`, dispatcherID, dispatcherID, afterType, afterType, afterID, cleanupBatchSize)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var candidates []OutboxEvent
+	for rows.Next() {
+		event := OutboxEvent{}
+		if err := rows.Scan(&event.EventType, &event.EventID); err != nil {
+			return nil, err
+		}
+		candidates = append(candidates, event)
+	}
+	return candidates, rows.Err()
+}