feat(wallet): safely purge verified outbox archives

2026-07-21 15:49:11 +08:00 · 2026-07-21 15:49:11 +08:00 · 48507a3861
commit 48507a3861
parent 7d092437d6
16 changed files with 1894 additions and 22 deletions
--- a/services/wallet-service/cmd/wallet-outbox-archive-restore/README.md
+++ b/services/wallet-service/cmd/wallet-outbox-archive-restore/README.md
@ -0,0 +1,53 @@
+# Wallet outbox archive restore rehearsal
+
+This command accepts the exact `data.ndjson.gz`, `manifest.json`, and `_SUCCESS`
+files from one private COS batch. It verifies `_SUCCESS` before opening MySQL,
+then checks manifest/data SHA-256, COS CRC64-ECMA, gzip integrity, row count,
+cursor bounds, UTC time bounds, unique source PKs, and both delta control totals.
+
+On a pay node, place the three files in a mode `0700` temporary directory using
+the attached `HyAppWalletOutboxArchiveRole`; do not copy the operator CSV key or
+long-lived credentials onto the node. Keep the files mode `0600` and remove the
+temporary directory after the report is captured.
+
+Verify files without database writes:
+
+```bash
+go run ./services/wallet-service/cmd/wallet-outbox-archive-restore \
+  --data /secure/batch/data.ndjson.gz \
+  --manifest /secure/batch/manifest.json \
+  --success /secure/batch/_SUCCESS
+```
+
+Restore to the standard isolation contract. The DSN is read from an environment
+variable so it does not appear in shell history or process arguments. The exact
+database check prevents a mistyped DSN from silently selecting another schema.
+
+```bash
+read -rsp 'Restore MySQL DSN: ' WALLET_OUTBOX_ARCHIVE_RESTORE_DSN && export WALLET_OUTBOX_ARCHIVE_RESTORE_DSN
+go run ./services/wallet-service/cmd/wallet-outbox-archive-restore \
+  --data /secure/batch/data.ndjson.gz \
+  --manifest /secure/batch/manifest.json \
+  --success /secure/batch/_SUCCESS \
+  --restore \
+  --expected-database wallet_archive_restore \
+  --rehearsal-id rehearsal-20260721-001 \
+  --confirm RESTORE:rehearsal-20260721-001:<batch-id>
+```
+
+The command creates only these versioned isolation tables:
+
+- `wallet_outbox_restore_rehearsals_v1`: one atomic state/evidence row per rehearsal.
+- `wallet_outbox_restore_rows_v1`: wallet_outbox-compatible fields, keyed by
+  `(rehearsal_id, app_code, event_id)` and tagged with `source_batch_id`.
+
+`RESTORE_VERIFIED` means the three files were cryptographically verified, every
+row was inserted into the isolation table, and SQL readback reproduced the
+manifest controls. It is not purge authorization. Any later purge workflow must
+also match the computed data/manifest/success CRC values to the owner receipt,
+prove all consumer watermarks are beyond the batch cursor, and delete only small
+exact `(app_code,event_id)` PK sets under its own concurrency gate.
+
+The batch states are intentionally monotonic: manifest `DATA_VERIFIED` -> marker
+`VERIFIED` -> CLI `FILES_VERIFIED` -> transactional `RESTORING` -> persisted
+`RESTORE_VERIFIED`. No state in this command maps to `PURGE_ELIGIBLE`.
--- a/services/wallet-service/cmd/wallet-outbox-archive-restore/main.go
+++ b/services/wallet-service/cmd/wallet-outbox-archive-restore/main.go
@ -0,0 +1,95 @@
+// Command wallet-outbox-archive-restore validates one downloaded COS batch and,
+// only after _SUCCESS and every checksum pass, writes it to isolation tables.
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"hyapp/services/wallet-service/internal/outboxarchiverestore"
+)
+
+const defaultDSNEnv = "WALLET_OUTBOX_ARCHIVE_RESTORE_DSN"
+
+func main() {
+	dataPath := flag.String("data", "", "local data.ndjson.gz downloaded from the private archive bucket")
+	manifestPath := flag.String("manifest", "", "local manifest.json from the same batch directory")
+	successPath := flag.String("success", "", "local _SUCCESS from the same batch directory")
+	restore := flag.Bool("restore", false, "persist the verified batch into fixed isolation tables")
+	dsnEnv := flag.String("dsn-env", defaultDSNEnv, "environment variable containing the restore-target MySQL DSN")
+	expectedDatabase := flag.String("expected-database", "", "exact restore-target database returned by SELECT DATABASE()")
+	rehearsalID := flag.String("rehearsal-id", "", "stable 1..64 character restore rehearsal key")
+	confirmation := flag.String("confirm", "", "restore confirmation RESTORE:<rehearsal-id>:<batch-id>")
+	timeout := flag.Duration("timeout", 10*time.Minute, "whole verification/restore timeout, at most 30m")
+	flag.Parse()
+
+	if strings.TrimSpace(*dataPath) == "" || strings.TrimSpace(*manifestPath) == "" || strings.TrimSpace(*successPath) == "" {
+		fatal(errors.New("--data, --manifest, and --success are required"))
+	}
+	if *timeout <= 0 || *timeout > 30*time.Minute {
+		fatal(errors.New("--timeout must be within (0,30m]"))
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), *timeout)
+	defer cancel()
+
+	verification, err := outboxarchiverestore.VerifyLocalFiles(*dataPath, *manifestPath, *successPath, outboxarchiverestore.DefaultLimits())
+	if err != nil {
+		fatal(err)
+	}
+	if !*restore {
+		printJSON(map[string]any{
+			"state": "FILES_VERIFIED", "batch_id": verification.Manifest.BatchID,
+			"app_code": verification.Manifest.AppCode, "row_count": verification.Manifest.RowCount,
+			"data_object_key": verification.Manifest.DataObjectKey,
+			"first_cursor":    verification.Manifest.FirstCursor, "last_cursor": verification.Manifest.LastCursor,
+			"ndjson_sha256": verification.Manifest.NDJSONSHA256, "gzip_sha256": verification.Manifest.GzipSHA256,
+			"data_crc64":      verification.Manifest.DataCRC64,
+			"manifest_sha256": verification.ManifestSHA256, "manifest_crc64": verification.ManifestCRC64,
+			"success_sha256": verification.SuccessSHA256, "success_crc64": verification.SuccessCRC64,
+			"available_delta_total": verification.Manifest.AvailableDeltaTotal,
+			"frozen_delta_total":    verification.Manifest.FrozenDeltaTotal, "no_database_writes": true,
+		})
+		return
+	}
+
+	rehearsal := strings.TrimSpace(*rehearsalID)
+	expectedConfirmation := "RESTORE:" + rehearsal + ":" + verification.Manifest.BatchID
+	if rehearsal == "" || strings.TrimSpace(*confirmation) != expectedConfirmation {
+		fatal(fmt.Errorf("--restore requires --rehearsal-id and --confirm=%s", expectedConfirmation))
+	}
+	envName := strings.TrimSpace(*dsnEnv)
+	if envName == "" {
+		fatal(errors.New("--dsn-env cannot be empty"))
+	}
+	dsn := strings.TrimSpace(os.Getenv(envName))
+	if dsn == "" {
+		fatal(fmt.Errorf("restore DSN environment variable %s is empty", envName))
+	}
+	result, err := outboxarchiverestore.Restore(ctx, dsn, *expectedDatabase, rehearsal, verification)
+	if err != nil {
+		fatal(err)
+	}
+	printJSON(struct {
+		outboxarchiverestore.RestoreResult
+		DSNEnv string `json:"dsn_env"`
+	}{RestoreResult: result, DSNEnv: envName})
+}
+
+func printJSON(value any) {
+	body, err := json.Marshal(value)
+	if err != nil {
+		fatal(err)
+	}
+	fmt.Println(string(body))
+}
+
+func fatal(err error) {
+	fmt.Fprintln(os.Stderr, err)
+	os.Exit(1)
+}
--- a/services/wallet-service/configs/config.docker.yaml
+++ b/services/wallet-service/configs/config.docker.yaml
@ -158,10 +158,17 @@ outbox_worker:
  initial_backoff: "5s"
  max_backoff: "5m"
 outbox_archive:
-  # 未配置私有 COS 的容器默认不扫描热表；显式启用后先用 dry_run 核对范围。
+  # 未配置私有 COS 的容器默认不扫描热表；purge 还需要显式租户、恢复凭证和生产授权记录。
  enabled: false
  mode: "dry_run"
  purge_enabled: false
+  purge_app_codes: []
+  purge_max_age: "720h"
+  purge_poll_interval: "5s"
+  purge_batch_size: 100
+  purge_receipt_grace: "1h"
+  purge_policy_version: "financial-v1"
+  purge_restore_proofs: []
  max_age: "720h"
  poll_interval: "5m"
  query_batch_size: 500
--- a/services/wallet-service/configs/config.tencent.example.yaml
+++ b/services/wallet-service/configs/config.tencent.example.yaml
@ -162,8 +162,15 @@ outbox_archive:
  # 模板默认关闭；生产私有配置补齐专用桶/角色后显式切 archive_only。
  enabled: false
  mode: "dry_run"
-  # 目录版本、notice repair、statistics replay 和 legacy projection 仍读热表，当前任何环境都禁止 purge。
+  # 自动清理必须再配置租户 allowlist、真实恢复凭证和生产库授权；模板永远不默认打开。
  purge_enabled: false
+  purge_app_codes: []
+  purge_max_age: "720h"
+  purge_poll_interval: "5s"
+  purge_batch_size: 100
+  purge_receipt_grace: "1h"
+  purge_policy_version: "financial-v1"
+  purge_restore_proofs: []
  max_age: "720h"
  poll_interval: "5m"
  query_batch_size: 500
--- a/services/wallet-service/configs/config.yaml
+++ b/services/wallet-service/configs/config.yaml
@ -160,10 +160,17 @@ outbox_worker:
  initial_backoff: "5s"
  max_backoff: "5m"
 outbox_archive:
-  # 未配置私有 COS 的环境默认不扫描热表；显式启用后先用 dry_run 核对范围。
+  # 未配置私有 COS 的环境默认不扫描热表；purge 还需要显式租户、恢复凭证和生产授权记录。
  enabled: false
  mode: "dry_run"
  purge_enabled: false
+  purge_app_codes: []
+  purge_max_age: "720h"
+  purge_poll_interval: "5s"
+  purge_batch_size: 100
+  purge_receipt_grace: "1h"
+  purge_policy_version: "financial-v1"
+  purge_restore_proofs: []
  max_age: "720h"
  poll_interval: "5m"
  query_batch_size: 500
--- a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql
+++ b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql
@ -203,6 +203,60 @@ CREATE TABLE IF NOT EXISTS wallet_outbox_archive_receipts (
  KEY idx_wallet_outbox_archive_cursor (app_code, state, last_updated_at_ms, last_event_id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 外部归档读回验证回执';

+-- wallet_outbox 自动清理只新增独立窄表，不 ALTER 热 outbox，也不会触发表重写；member 随未清理归档行增减。
+-- 授权行必须由部署流程在隔离库完成 RESTORE_VERIFIED 并核对归档回执后手工写入；初始化脚本绝不生成清理授权。
+CREATE TABLE IF NOT EXISTS wallet_outbox_archive_members (
+  app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
+  event_id VARCHAR(128) NOT NULL COMMENT '已归档 wallet_outbox 事件 ID',
+  updated_at_ms BIGINT NOT NULL COMMENT '归档时 delivered 行的精确更新时间，UTC epoch ms',
+  batch_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '包含该精确事件的归档批次 ID',
+  manifest_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '包含该事件的 manifest SHA-256',
+  receipt_verified_at_ms BIGINT NOT NULL COMMENT '归档 receipt 完成 COS 读回验证的时间，UTC epoch ms',
+  PRIMARY KEY (app_code, event_id),
+  KEY idx_wallet_outbox_archive_member_purge (app_code, updated_at_ms, event_id),
+  KEY idx_wallet_outbox_archive_member_batch (app_code, batch_id, event_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 已验证归档精确成员';
+
+CREATE TABLE IF NOT EXISTS wallet_outbox_archive_purge_authorizations (
+  app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
+  archive_batch_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '完成恢复演练的归档批次内容指纹',
+  manifest_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '恢复演练核对通过的 manifest SHA-256',
+  schema_version INT NOT NULL COMMENT '恢复演练核对通过的归档 schema 版本',
+  row_count INT NOT NULL COMMENT '恢复演练核对通过的精确行数',
+  data_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 data COS CRC64',
+  manifest_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 manifest COS CRC64',
+  success_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 _SUCCESS COS CRC64',
+  rehearsal_receipt_id VARCHAR(64) NOT NULL COMMENT '隔离库 RESTORE_VERIFIED 演练回执 ID',
+  state VARCHAR(32) NOT NULL COMMENT '固定为 RESTORE_VERIFIED，其他状态不能授权清理',
+  verified_at_ms BIGINT NOT NULL COMMENT '恢复演练验证完成时间，UTC epoch ms',
+  created_at_ms BIGINT NOT NULL COMMENT '授权证据录入时间，UTC epoch ms',
+  updated_at_ms BIGINT NOT NULL COMMENT '授权证据更新时间，UTC epoch ms',
+  PRIMARY KEY (app_code, archive_batch_id, manifest_sha256),
+  KEY idx_wallet_outbox_purge_auth_verified (app_code, state, verified_at_ms, archive_batch_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 恢复演练清理授权证据';
+
+-- 一行只对应一个已提交的精确 PK 删除批次；删除与回执写入同事务，失败或回滚时不会留下虚假清理证据。
+CREATE TABLE IF NOT EXISTS wallet_outbox_archive_purge_receipts (
+  app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
+  purge_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '清理批次稳定指纹',
+  state VARCHAR(32) NOT NULL COMMENT '固定为 PURGED',
+  policy_version VARCHAR(32) NOT NULL COMMENT '正向安全事件白名单版本',
+  cutoff_ms BIGINT NOT NULL COMMENT '本批固定的 delivered 保留截止时间，UTC epoch ms，删除严格小于该值',
+  receipt_verified_before_ms BIGINT NOT NULL COMMENT '独立归档回执成熟截止时间，只有更早 VERIFIED receipt 可清理',
+  candidate_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '按游标排序的精确 (app_code,event_id) 候选集合 SHA-256',
+  candidate_count INT NOT NULL COMMENT '事务内锁定的候选行数',
+  deleted_count INT NOT NULL COMMENT '精确 PK 删除影响行数，必须与 candidate_count 一致',
+  first_updated_at_ms BIGINT NOT NULL COMMENT '首条候选 delivered 游标时间',
+  first_event_id VARCHAR(128) NOT NULL COMMENT '首条候选事件 ID',
+  last_updated_at_ms BIGINT NOT NULL COMMENT '末条候选 delivered 游标时间',
+  last_event_id VARCHAR(128) NOT NULL COMMENT '末条候选事件 ID',
+  purged_at_ms BIGINT NOT NULL COMMENT '清理提交时间，UTC epoch ms',
+  created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
+  updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
+  PRIMARY KEY (app_code, purge_id),
+  KEY idx_wallet_outbox_purge_receipt_time (app_code, purged_at_ms, purge_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 精确 PK 清理审计回执';
+
 -- 钱包运行侧收益政策实例：由 admin-server 发布写入，wallet 送礼、提现和结算只读本库运行快照。
 CREATE TABLE IF NOT EXISTS wallet_policy_instances (
  app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
--- a/services/wallet-service/deploy/mysql/migrations/013_wallet_outbox_archive_purge_receipts.sql
+++ b/services/wallet-service/deploy/mysql/migrations/013_wallet_outbox_archive_purge_receipts.sql
@ -0,0 +1,53 @@
+-- wallet_outbox 自动清理只新增独立窄表，不 ALTER 热 outbox，也不会触发表重写；member 随未清理归档行增减。
+-- 授权行必须由部署流程在隔离库完成 RESTORE_VERIFIED 并核对归档回执后手工写入；迁移本身绝不生成清理授权。
+CREATE TABLE IF NOT EXISTS wallet_outbox_archive_members (
+  app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
+  event_id VARCHAR(128) NOT NULL COMMENT '已归档 wallet_outbox 事件 ID',
+  updated_at_ms BIGINT NOT NULL COMMENT '归档时 delivered 行的精确更新时间，UTC epoch ms',
+  batch_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '包含该精确事件的归档批次 ID',
+  manifest_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '包含该事件的 manifest SHA-256',
+  receipt_verified_at_ms BIGINT NOT NULL COMMENT '归档 receipt 完成 COS 读回验证的时间，UTC epoch ms',
+  PRIMARY KEY (app_code, event_id),
+  KEY idx_wallet_outbox_archive_member_purge (app_code, updated_at_ms, event_id),
+  KEY idx_wallet_outbox_archive_member_batch (app_code, batch_id, event_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 已验证归档精确成员';
+
+CREATE TABLE IF NOT EXISTS wallet_outbox_archive_purge_authorizations (
+  app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
+  archive_batch_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '完成恢复演练的归档批次内容指纹',
+  manifest_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '恢复演练核对通过的 manifest SHA-256',
+  schema_version INT NOT NULL COMMENT '恢复演练核对通过的归档 schema 版本',
+  row_count INT NOT NULL COMMENT '恢复演练核对通过的精确行数',
+  data_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 data COS CRC64',
+  manifest_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 manifest COS CRC64',
+  success_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 _SUCCESS COS CRC64',
+  rehearsal_receipt_id VARCHAR(64) NOT NULL COMMENT '隔离库 RESTORE_VERIFIED 演练回执 ID',
+  state VARCHAR(32) NOT NULL COMMENT '固定为 RESTORE_VERIFIED，其他状态不能授权清理',
+  verified_at_ms BIGINT NOT NULL COMMENT '恢复演练验证完成时间，UTC epoch ms',
+  created_at_ms BIGINT NOT NULL COMMENT '授权证据录入时间，UTC epoch ms',
+  updated_at_ms BIGINT NOT NULL COMMENT '授权证据更新时间，UTC epoch ms',
+  PRIMARY KEY (app_code, archive_batch_id, manifest_sha256),
+  KEY idx_wallet_outbox_purge_auth_verified (app_code, state, verified_at_ms, archive_batch_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 恢复演练清理授权证据';
+
+-- 一行只对应一个已提交的精确 PK 删除批次；删除与回执写入同事务，失败或回滚时不会留下虚假清理证据。
+CREATE TABLE IF NOT EXISTS wallet_outbox_archive_purge_receipts (
+  app_code VARCHAR(32) NOT NULL COMMENT '应用编码',
+  purge_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '清理批次稳定指纹',
+  state VARCHAR(32) NOT NULL COMMENT '固定为 PURGED',
+  policy_version VARCHAR(32) NOT NULL COMMENT '正向安全事件白名单版本',
+  cutoff_ms BIGINT NOT NULL COMMENT '本批固定的 delivered 保留截止时间，UTC epoch ms，删除严格小于该值',
+  receipt_verified_before_ms BIGINT NOT NULL COMMENT '独立归档回执成熟截止时间，只有更早 VERIFIED receipt 可清理',
+  candidate_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '按游标排序的精确 (app_code,event_id) 候选集合 SHA-256',
+  candidate_count INT NOT NULL COMMENT '事务内锁定的候选行数',
+  deleted_count INT NOT NULL COMMENT '精确 PK 删除影响行数，必须与 candidate_count 一致',
+  first_updated_at_ms BIGINT NOT NULL COMMENT '首条候选 delivered 游标时间',
+  first_event_id VARCHAR(128) NOT NULL COMMENT '首条候选事件 ID',
+  last_updated_at_ms BIGINT NOT NULL COMMENT '末条候选 delivered 游标时间',
+  last_event_id VARCHAR(128) NOT NULL COMMENT '末条候选事件 ID',
+  purged_at_ms BIGINT NOT NULL COMMENT '清理提交时间，UTC epoch ms',
+  created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
+  updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms',
+  PRIMARY KEY (app_code, purge_id),
+  KEY idx_wallet_outbox_purge_receipt_time (app_code, purged_at_ms, purge_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 精确 PK 清理审计回执';
--- a/services/wallet-service/internal/app/app.go
+++ b/services/wallet-service/internal/app/app.go
@ -73,6 +73,10 @@ func New(cfg config.Config) (*App, error) {
 	if err != nil {
 		return nil, err
 	}
+	if err := requireWalletOutboxPurgeAuthorizations(startupCtx, repository, cfg.OutboxArchive); err != nil {
+		_ = repository.Close()
+		return nil, err
+	}
 	var outboxArchiveStore *walletOutboxArchiveCOSStore
 	if cfg.OutboxArchive.Enabled && cfg.OutboxArchive.Mode == config.OutboxArchiveModeArchiveOnly {
 		outboxArchiveStore, err = newWalletOutboxArchiveCOSStore(cfg.OutboxArchive.COS, nil)
--- a/services/wallet-service/internal/app/background_workers.go
+++ b/services/wallet-service/internal/app/background_workers.go
@ -20,9 +20,12 @@ func (a *App) runBackgroundWorkers() {
 		a.startWalletOutboxWorkers(ctx, "wallet-realtime-outbox", a.realtimeOutboxWorkerCfg, a.realtimeOutboxPartitions, a.realtimeOutboxProducer, a.realtimeWalletOutboxTopic, a.realtimeOutboxWorkerCfg.EventTypes, nil)
 	}
 	if a.outboxArchiveCfg.Enabled {
-		// archive-only 与 MQ fanout 独立；dry_run 也持续用有界索引查询暴露容量，但两种模式都不删除热表行。
+		// 归档与 MQ fanout 独立；purge 另有恢复凭证、exact membership 和 allowlist 三重门槛。
 		a.startWalletOutboxArchiveWorker(ctx)
 	}
+	if a.outboxArchiveCfg.PurgeEnabled {
+		a.startWalletOutboxPurgeWorker(ctx)
+	}
 	if a.externalRechargeReconcileWorkerCfg.Enabled {
 		a.startExternalRechargeReconcileWorker(ctx)
 	}
--- a/services/wallet-service/internal/app/outbox_archive.go
+++ b/services/wallet-service/internal/app/outbox_archive.go
@ -264,6 +264,13 @@ func (a *App) archiveWalletOutboxRecords(tenantCtx context.Context, targetAppCod
 		return err
 	}
 	nowMS := time.Now().UTC().UnixMilli()
+	members := make([]mysqlstorage.WalletOutboxArchiveMember, 0, len(records))
+	for _, record := range records {
+		// 精确成员证明与 VERIFIED receipt 同事务落库；purge 不再用游标范围猜测某一行是否真的存在于 COS 对象。
+		members = append(members, mysqlstorage.WalletOutboxArchiveMember{
+			AppCode: record.AppCode, EventID: record.EventID, UpdatedAtMS: record.UpdatedAtMS,
+		})
+	}
 	receiptCtx, receiptCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout)
 	err = a.mysqlRepo.RecordVerifiedWalletOutboxArchive(receiptCtx, mysqlstorage.WalletOutboxArchiveReceipt{
 		AppCode: targetAppCode, BatchID: artifact.manifest.BatchID, State: walletOutboxArchiveVerifiedState,
@ -275,6 +282,7 @@ func (a *App) archiveWalletOutboxRecords(tenantCtx context.Context, targetAppCod
 		NDJSONSHA256: artifact.manifest.NDJSONSHA256, GzipSHA256: artifact.manifest.GzipSHA256, ManifestSHA256: artifact.manifestSHA256,
 		DataCRC64: dataCRC64, ManifestCRC64: manifestCRC64, SuccessCRC64: successCRC64,
 		UncompressedBytes: artifact.manifest.UncompressedBytes, CompressedBytes: artifact.manifest.CompressedBytes, VerifiedAtMS: nowMS,
+		Members: members,
 	})
 	receiptCancel()
 	if err != nil {
--- a/services/wallet-service/internal/app/outbox_purge.go
+++ b/services/wallet-service/internal/app/outbox_purge.go
@ -0,0 +1,107 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"hyapp/pkg/appcode"
+	"hyapp/pkg/logx"
+	"hyapp/services/wallet-service/internal/config"
+	mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql"
+)
+
+func requireWalletOutboxPurgeAuthorizations(ctx context.Context, repository *mysqlstorage.Repository, cfg config.OutboxArchiveConfig) error {
+	if !cfg.PurgeEnabled {
+		return nil
+	}
+	if repository == nil {
+		return errors.New("wallet outbox purge repository is not configured")
+	}
+	for _, proof := range cfg.PurgeRestoreProofs {
+		proofCtx := appcode.WithContext(ctx, proof.AppCode)
+		if err := repository.RequireWalletOutboxPurgeAuthorization(proofCtx, proof.AppCode, proof.BatchID, proof.ManifestSHA256); err != nil {
+			// App 启动失败比“配置显示 purge=true、实际没有恢复凭证”更安全；滚动发布会停在第一台并保留另一台服务流量。
+			return fmt.Errorf("wallet outbox purge authorization rejected for app %s: %w", proof.AppCode, err)
+		}
+	}
+	return nil
+}
+
+func (a *App) startWalletOutboxPurgeWorker(ctx context.Context) {
+	a.workers.Add(1)
+	go func() {
+		defer a.workers.Done()
+		ticker := time.NewTicker(a.outboxArchiveCfg.PurgePollInterval)
+		defer ticker.Stop()
+		for {
+			if err := a.runWalletOutboxPurgeRound(ctx); err != nil && !errors.Is(err, context.Canceled) {
+				logx.Error(ctx, "wallet_outbox_purge_round_failed", err)
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
+}
+
+func (a *App) runWalletOutboxPurgeRound(ctx context.Context) error {
+	if a.mysqlRepo == nil {
+		return nil
+	}
+	now := time.Now().UTC()
+	cutoffMS := now.Add(-a.outboxArchiveCfg.PurgeMaxAge).UnixMilli()
+	receiptVerifiedBeforeMS := now.Add(-a.outboxArchiveCfg.PurgeReceiptGrace).UnixMilli()
+	for _, targetAppCode := range a.outboxArchiveCfg.PurgeAppCodes {
+		tenantCtx := appcode.WithContext(ctx, targetAppCode)
+		lockCtx, lockCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout)
+		release, acquired, err := a.mysqlRepo.AcquireWalletOutboxArchiveAppLock(lockCtx)
+		lockCancel()
+		if err != nil {
+			logx.Error(ctx, "wallet_outbox_purge_app_lock_failed", err, slog.String("app_code", targetAppCode))
+			continue
+		}
+		if !acquired {
+			continue
+		}
+
+		purgeCtx, purgeCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout)
+		receipt, purgeErr := a.mysqlRepo.PurgeDeliveredWalletOutboxBatch(
+			purgeCtx,
+			targetAppCode,
+			a.outboxArchiveCfg.PurgePolicyVersion,
+			cutoffMS,
+			receiptVerifiedBeforeMS,
+			a.outboxArchiveCfg.PurgeBatchSize,
+			now.UnixMilli(),
+		)
+		purgeCancel()
+		release()
+		if purgeErr != nil {
+			// 单 App 失败不能扩大到其他租户；事务会回滚，下一轮仍从同一 exact member 重试。
+			logx.Error(ctx, "wallet_outbox_purge_app_failed", purgeErr, slog.String("app_code", targetAppCode))
+			continue
+		}
+		if receipt.DeletedCount == 0 {
+			continue
+		}
+		logx.Info(ctx, "wallet_outbox_purged",
+			slog.String("app_code", receipt.AppCode),
+			slog.String("purge_id", receipt.PurgeID),
+			slog.String("policy_version", receipt.PolicyVersion),
+			slog.Int("deleted_count", receipt.DeletedCount),
+			slog.Int64("cutoff_ms", receipt.CutoffMS),
+			slog.Int64("receipt_verified_before_ms", receipt.ReceiptVerifiedBeforeMS),
+			slog.Int64("first_updated_at_ms", receipt.FirstCursor.UpdatedAtMS),
+			slog.String("first_event_id", receipt.FirstCursor.EventID),
+			slog.Int64("last_updated_at_ms", receipt.LastCursor.UpdatedAtMS),
+			slog.String("last_event_id", receipt.LastCursor.EventID),
+			slog.String("candidate_sha256", receipt.CandidateSHA256),
+		)
+	}
+	return nil
+}
--- a/services/wallet-service/internal/config/config.go
+++ b/services/wallet-service/internal/config/config.go
@ -1,6 +1,7 @@
 package config

 import (
+	"encoding/hex"
 	"errors"
 	"os"
 	"path/filepath"
@ -48,7 +49,7 @@ type Config struct {
 	ExternalRechargeReconcileWorker ExternalRechargeReconcileWorkerConfig `yaml:"external_recharge_reconcile_worker"`
 	// OutboxWorker 控制 wallet_outbox 到 MQ 的补偿投递。
 	OutboxWorker OutboxWorkerConfig `yaml:"outbox_worker"`
-	// OutboxArchive 只把已投递事实备份到外部存储；当前不提供 purge，避免破坏线上回放与修复读者。
+	// OutboxArchive 把已投递事实归档到外部存储；purge 仅对恢复验证且命中正向安全策略的精确成员开放。
 	OutboxArchive OutboxArchiveConfig `yaml:"outbox_archive"`
 	// RealtimeOutboxWorker 控制红包等实时 UI 事件到独立 MQ topic 的补偿投递。
 	RealtimeOutboxWorker OutboxWorkerConfig `yaml:"realtime_outbox_worker"`
@ -125,15 +126,28 @@ const (
 	OutboxArchiveModeArchiveOnly OutboxArchiveMode = "archive_only"
 )

-// OutboxArchiveConfig 控制 delivered wallet_outbox 的 archive-only 备份链路。
+// OutboxArchiveConfig 控制 delivered wallet_outbox 的归档与受控清理链路。
 type OutboxArchiveConfig struct {
-	Enabled        bool              `yaml:"enabled"`
-	Mode           OutboxArchiveMode `yaml:"mode"`
-	PurgeEnabled   bool              `yaml:"purge_enabled"`
-	MaxAge         time.Duration     `yaml:"max_age"`
-	PollInterval   time.Duration     `yaml:"poll_interval"`
-	QueryBatchSize int               `yaml:"query_batch_size"`
-	SegmentMaxRows int               `yaml:"segment_max_rows"`
+	Enabled bool              `yaml:"enabled"`
+	Mode    OutboxArchiveMode `yaml:"mode"`
+	// PurgeEnabled 只有在归档、真实恢复演练凭证和租户 allowlist 同时满足时才允许启动。
+	PurgeEnabled bool `yaml:"purge_enabled"`
+	// PurgeAppCodes 永远是显式 allowlist；清理不能像归档一样自动发现新租户。
+	PurgeAppCodes []string `yaml:"purge_app_codes"`
+	// PurgeMaxAge 与归档 cutoff 分离，防止调快归档任务时意外同步缩短热数据保留期。
+	PurgeMaxAge       time.Duration `yaml:"purge_max_age"`
+	PurgePollInterval time.Duration `yaml:"purge_poll_interval"`
+	PurgeBatchSize    int           `yaml:"purge_batch_size"`
+	// PurgeReceiptGrace 强制归档回执先稳定一段时间，避免刚完成上传就立即删除热表事实。
+	PurgeReceiptGrace time.Duration `yaml:"purge_receipt_grace"`
+	// PurgePolicyVersion 对应代码内正向事件类型清单；未知的新事件默认永不删除。
+	PurgePolicyVersion string `yaml:"purge_policy_version"`
+	// PurgeRestoreProofs 把一次真实 RESTORE_VERIFIED 结果绑定到 App 和归档 manifest；生产库还必须有相同授权记录。
+	PurgeRestoreProofs []OutboxArchivePurgeRestoreProofConfig `yaml:"purge_restore_proofs"`
+	MaxAge             time.Duration                          `yaml:"max_age"`
+	PollInterval       time.Duration                          `yaml:"poll_interval"`
+	QueryBatchSize     int                                    `yaml:"query_batch_size"`
+	SegmentMaxRows     int                                    `yaml:"segment_max_rows"`
 	// SegmentMaxUncompressedBytes caps one in-memory/archive artifact independently of row count because payload sizes vary.
 	SegmentMaxUncompressedBytes int64                  `yaml:"segment_max_uncompressed_bytes"`
 	RoundMaxRows                int                    `yaml:"round_max_rows"`
@ -145,6 +159,13 @@ type OutboxArchiveConfig struct {
 	COS                         OutboxArchiveCOSConfig `yaml:"cos"`
 }

+// OutboxArchivePurgeRestoreProofConfig 是自动删除的双重确认材料，不包含任何 COS 或数据库密钥。
+type OutboxArchivePurgeRestoreProofConfig struct {
+	AppCode        string `yaml:"app_code"`
+	BatchID        string `yaml:"batch_id"`
+	ManifestSHA256 string `yaml:"manifest_sha256"`
+}
+
 // OutboxArchiveCOSConfig 只允许服务端环境注入私有桶凭证；归档对象不生成公网 URL。
 type OutboxArchiveCOSConfig struct {
 	CredentialSource string        `yaml:"credential_source"`
@ -395,6 +416,11 @@ func Default() Config {
 			Enabled:                     false,
 			Mode:                        OutboxArchiveModeDryRun,
 			PurgeEnabled:                false,
+			PurgeMaxAge:                 30 * 24 * time.Hour,
+			PurgePollInterval:           5 * time.Second,
+			PurgeBatchSize:              100,
+			PurgeReceiptGrace:           time.Hour,
+			PurgePolicyVersion:          "financial-v1",
 			MaxAge:                      30 * 24 * time.Hour,
 			PollInterval:                5 * time.Minute,
 			QueryBatchSize:              500,
@ -918,15 +944,10 @@ func normalizeOutboxArchiveConfig(cfg OutboxArchiveConfig, defaults OutboxArchiv
 	if cfg.Mode == "" {
 		cfg.Mode = defaults.Mode
 	}
-	// wallet_outbox 仍是 GiftCatalogVersion、gift-wall/badge 回填、notice repair 和 statistics replay 的在线事实源。
-	// 在这些读者切到热表 + 归档恢复层之前，任何 purge 配置都必须启动失败，不允许静默降级为裸删。
-	if cfg.PurgeEnabled {
-		return OutboxArchiveConfig{}, errors.New("outbox_archive.purge_enabled is unavailable until wallet_outbox online readers use the archive restore path")
-	}
 	switch cfg.Mode {
 	case OutboxArchiveModeDryRun, OutboxArchiveModeArchiveOnly:
 	default:
-		return OutboxArchiveConfig{}, errors.New("outbox_archive.mode must be dry_run or archive_only; purge modes are intentionally unsupported")
+		return OutboxArchiveConfig{}, errors.New("outbox_archive.mode must be dry_run or archive_only")
 	}
 	if cfg.MaxAge <= 0 {
 		cfg.MaxAge = defaults.MaxAge
@ -934,6 +955,76 @@ func normalizeOutboxArchiveConfig(cfg OutboxArchiveConfig, defaults OutboxArchiv
 	if cfg.Enabled && cfg.MaxAge < 24*time.Hour {
 		return OutboxArchiveConfig{}, errors.New("outbox_archive.max_age must be at least 24h")
 	}
+	cfg.PurgeAppCodes = outboxpartition.Normalize(cfg.PurgeAppCodes, false)
+	if cfg.PurgeMaxAge <= 0 {
+		cfg.PurgeMaxAge = defaults.PurgeMaxAge
+	}
+	if cfg.PurgePollInterval <= 0 {
+		cfg.PurgePollInterval = defaults.PurgePollInterval
+	}
+	if cfg.PurgeBatchSize <= 0 {
+		cfg.PurgeBatchSize = defaults.PurgeBatchSize
+	}
+	if cfg.PurgeReceiptGrace <= 0 {
+		cfg.PurgeReceiptGrace = defaults.PurgeReceiptGrace
+	}
+	cfg.PurgePolicyVersion = strings.ToLower(strings.TrimSpace(cfg.PurgePolicyVersion))
+	if cfg.PurgePolicyVersion == "" {
+		cfg.PurgePolicyVersion = defaults.PurgePolicyVersion
+	}
+	proofsByApp := make(map[string]OutboxArchivePurgeRestoreProofConfig, len(cfg.PurgeRestoreProofs))
+	for _, proof := range cfg.PurgeRestoreProofs {
+		proof.AppCode = strings.ToLower(strings.TrimSpace(proof.AppCode))
+		proof.BatchID = strings.ToLower(strings.TrimSpace(proof.BatchID))
+		proof.ManifestSHA256 = strings.ToLower(strings.TrimSpace(proof.ManifestSHA256))
+		if proof.AppCode == "" || !isSHA256Hex(proof.BatchID) || !isSHA256Hex(proof.ManifestSHA256) {
+			return OutboxArchiveConfig{}, errors.New("outbox_archive purge restore proof requires app_code and 64-character hex batch/manifest hashes")
+		}
+		if _, duplicated := proofsByApp[proof.AppCode]; duplicated {
+			return OutboxArchiveConfig{}, errors.New("outbox_archive purge restore proofs must contain each app_code once")
+		}
+		proofsByApp[proof.AppCode] = proof
+	}
+	cfg.PurgeRestoreProofs = cfg.PurgeRestoreProofs[:0]
+	for _, appCode := range cfg.PurgeAppCodes {
+		proof, ok := proofsByApp[appCode]
+		if !ok && cfg.PurgeEnabled {
+			return OutboxArchiveConfig{}, errors.New("outbox_archive every purge app_code requires an exact restore proof")
+		}
+		if ok {
+			cfg.PurgeRestoreProofs = append(cfg.PurgeRestoreProofs, proof)
+			delete(proofsByApp, appCode)
+		}
+	}
+	if cfg.PurgeEnabled {
+		// 删除开关必须 fail closed：只能在持续 COS 归档模式下，对显式租户逐小批运行。
+		if !cfg.Enabled || cfg.Mode != OutboxArchiveModeArchiveOnly {
+			return OutboxArchiveConfig{}, errors.New("outbox_archive purge requires enabled archive_only mode")
+		}
+		if len(cfg.PurgeAppCodes) == 0 || len(proofsByApp) != 0 {
+			return OutboxArchiveConfig{}, errors.New("outbox_archive purge requires a non-empty app allowlist with no unmatched restore proofs")
+		}
+		for _, appCode := range cfg.PurgeAppCodes {
+			if appCode != "lalu" && appCode != "fami" {
+				return OutboxArchiveConfig{}, errors.New("outbox_archive financial-v1 purge only supports explicitly approved lalu and fami apps")
+			}
+		}
+		if cfg.PurgeMaxAge < 30*24*time.Hour || cfg.PurgeMaxAge < cfg.MaxAge {
+			return OutboxArchiveConfig{}, errors.New("outbox_archive purge_max_age must be at least 30d and not shorter than archive max_age")
+		}
+		if cfg.PurgePollInterval < 5*time.Second {
+			return OutboxArchiveConfig{}, errors.New("outbox_archive purge_poll_interval must be at least 5s")
+		}
+		if cfg.PurgeReceiptGrace < time.Hour {
+			return OutboxArchiveConfig{}, errors.New("outbox_archive purge_receipt_grace must be at least 1h")
+		}
+		if cfg.PurgePolicyVersion != "financial-v1" {
+			return OutboxArchiveConfig{}, errors.New("outbox_archive purge_policy_version must be financial-v1")
+		}
+		if cfg.PurgeBatchSize > 500 {
+			return OutboxArchiveConfig{}, errors.New("outbox_archive purge_batch_size must not exceed 500")
+		}
+	}
 	if cfg.PollInterval <= 0 {
 		cfg.PollInterval = defaults.PollInterval
 	}
@ -1011,6 +1102,14 @@ func normalizeOutboxArchiveConfig(cfg OutboxArchiveConfig, defaults OutboxArchiv
 	return cfg, nil
 }

+func isSHA256Hex(value string) bool {
+	if len(value) != 64 {
+		return false
+	}
+	decoded, err := hex.DecodeString(value)
+	return err == nil && len(decoded) == 32
+}
+
 func normalizeStringSlice(values []string) []string {
 	normalized := make([]string, 0, len(values))
 	seen := make(map[string]struct{}, len(values))
--- a/services/wallet-service/internal/outboxarchiverestore/mysql.go
+++ b/services/wallet-service/internal/outboxarchiverestore/mysql.go
@ -0,0 +1,440 @@
+package outboxarchiverestore
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+	"time"
+
+	_ "github.com/go-sql-driver/mysql"
+)
+
+const (
+	RehearsalTable = "wallet_outbox_restore_rehearsals_v1"
+	RowsTable      = "wallet_outbox_restore_rows_v1"
+	restoreState   = "RESTORING"
+)
+
+// RestoreResult is safe to print: it identifies the selected schema and fixed
+// table contract but never includes the DSN or credentials used to reach it.
+type RestoreResult struct {
+	State              string `json:"state"`
+	RehearsalID        string `json:"rehearsal_id"`
+	BatchID            string `json:"batch_id"`
+	AppCode            string `json:"app_code"`
+	RowCount           int    `json:"row_count"`
+	TargetDatabase     string `json:"target_database"`
+	RehearsalTable     string `json:"rehearsal_table"`
+	RowsTable          string `json:"rows_table"`
+	DataObjectKey      string `json:"data_object_key"`
+	FirstCursor        Cursor `json:"first_cursor"`
+	LastCursor         Cursor `json:"last_cursor"`
+	NDJSONSHA256       string `json:"ndjson_sha256"`
+	GzipSHA256         string `json:"gzip_sha256"`
+	DataCRC64          string `json:"data_crc64"`
+	ManifestSHA256     string `json:"manifest_sha256"`
+	ManifestCRC64      string `json:"manifest_crc64"`
+	SuccessSHA256      string `json:"success_sha256"`
+	SuccessCRC64       string `json:"success_crc64"`
+	AvailableTotal     string `json:"available_delta_total"`
+	FrozenTotal        string `json:"frozen_delta_total"`
+	VerifiedAtMS       int64  `json:"verified_at_ms"`
+	IdempotentReadback bool   `json:"idempotent_readback"`
+}
+
+// Restore persists a fully verified batch into versioned isolation tables. All
+// DDL targets new/fixed table names; all DML is scoped by rehearsal PK. It never
+// reads, locks, updates, or deletes wallet_outbox.
+func Restore(ctx context.Context, dsn, expectedDatabase, rehearsalID string, verification Verification) (RestoreResult, error) {
+	dsn = strings.TrimSpace(dsn)
+	expectedDatabase = strings.TrimSpace(expectedDatabase)
+	rehearsalID = strings.TrimSpace(rehearsalID)
+	if dsn == "" || expectedDatabase == "" {
+		return RestoreResult{}, errors.New("restore DSN and exact expected database are required")
+	}
+	if !validRehearsalID(rehearsalID) {
+		return RestoreResult{}, errors.New("rehearsal_id must be 1..64 ASCII letters, digits, dot, underscore, or dash")
+	}
+	if verification.State != filesVerifiedState || len(verification.Records) != verification.Manifest.RowCount {
+		return RestoreResult{}, errors.New("local archive verification is incomplete")
+	}
+
+	db, err := sql.Open("mysql", dsn)
+	if err != nil {
+		return RestoreResult{}, err
+	}
+	defer db.Close()
+	// A rehearsal is intentionally single-connection and serial; this prevents a
+	// restore tool from consuming the production-sized connection pool by mistake.
+	db.SetMaxOpenConns(1)
+	db.SetMaxIdleConns(1)
+	db.SetConnMaxLifetime(10 * time.Minute)
+	conn, err := db.Conn(ctx)
+	if err != nil {
+		return RestoreResult{}, err
+	}
+	defer conn.Close()
+
+	var databaseName string
+	if err := conn.QueryRowContext(ctx, `SELECT DATABASE()`).Scan(&databaseName); err != nil {
+		return RestoreResult{}, err
+	}
+	if databaseName != expectedDatabase {
+		return RestoreResult{}, fmt.Errorf("restore database mismatch: connected=%q expected=%q", databaseName, expectedDatabase)
+	}
+	if err := ensureIsolationSchema(ctx, conn); err != nil {
+		return RestoreResult{}, err
+	}
+	release, acquired, err := acquireRestoreLock(ctx, conn, rehearsalID)
+	if err != nil {
+		return RestoreResult{}, err
+	}
+	if !acquired {
+		return RestoreResult{}, errors.New("another process holds the same restore rehearsal lock")
+	}
+	defer release()
+
+	if existing, found, err := readExistingRestore(ctx, conn, rehearsalID, verification); err != nil {
+		return RestoreResult{}, err
+	} else if found {
+		return existing, nil
+	}
+
+	tx, err := conn.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
+	if err != nil {
+		return RestoreResult{}, err
+	}
+	defer tx.Rollback()
+	nowMS := time.Now().UTC().UnixMilli()
+	if err := insertRehearsal(ctx, tx, rehearsalID, databaseName, verification, nowMS); err != nil {
+		return RestoreResult{}, err
+	}
+	for offset := 0; offset < len(verification.Records); offset += 100 {
+		end := min(offset+100, len(verification.Records))
+		if err := insertRestoreRows(ctx, tx, rehearsalID, verification.Manifest.BatchID, nowMS, verification.Records[offset:end]); err != nil {
+			return RestoreResult{}, err
+		}
+	}
+	// These aggregates only scan one bounded rehearsal partition in the isolation
+	// table. They never aggregate the live outbox and are the database-level proof
+	// that JSON conversion and INSERT semantics preserved the archive controls.
+	if err := verifyRestoredControls(ctx, tx, rehearsalID, verification); err != nil {
+		return RestoreResult{}, err
+	}
+	if err := verifyRestoredRowsExact(ctx, tx, rehearsalID, verification.Records); err != nil {
+		return RestoreResult{}, err
+	}
+	result := newRestoreResult(rehearsalID, databaseName, verification, nowMS, false)
+	update, err := tx.ExecContext(ctx, `
+		UPDATE wallet_outbox_restore_rehearsals_v1
+		SET state = ?, verified_at_ms = ?, updated_at_ms = ?
+		WHERE rehearsal_id = ? AND state = ?`, restoreVerifiedState, nowMS, nowMS, rehearsalID, restoreState)
+	if err != nil {
+		return RestoreResult{}, err
+	}
+	if affected, err := update.RowsAffected(); err != nil || affected != 1 {
+		return RestoreResult{}, errors.New("restore rehearsal state transition did not affect exactly one row")
+	}
+	if err := tx.Commit(); err != nil {
+		return RestoreResult{}, err
+	}
+	return result, nil
+}
+
+func ensureIsolationSchema(ctx context.Context, conn *sql.Conn) error {
+	// CREATE IF NOT EXISTS takes metadata locks only on the dedicated table names;
+	// unlike ALTER wallet_outbox it cannot rebuild or block the hot owner table.
+	statements := []string{`
+		CREATE TABLE IF NOT EXISTS wallet_outbox_restore_rehearsals_v1 (
+		  rehearsal_id VARCHAR(64) NOT NULL,
+		  source_app_code VARCHAR(32) NOT NULL,
+		  source_batch_id CHAR(64) NOT NULL,
+		  state VARCHAR(32) NOT NULL,
+		  schema_version INT NOT NULL,
+		  data_object_key VARCHAR(512) NOT NULL,
+		  row_count INT NOT NULL,
+		  first_updated_at_ms BIGINT NOT NULL,
+		  first_event_id VARCHAR(128) NOT NULL,
+		  last_updated_at_ms BIGINT NOT NULL,
+		  last_event_id VARCHAR(128) NOT NULL,
+		  available_delta_total VARCHAR(80) NOT NULL,
+		  frozen_delta_total VARCHAR(80) NOT NULL,
+		  ndjson_sha256 CHAR(64) NOT NULL,
+		  gzip_sha256 CHAR(64) NOT NULL,
+		  data_crc64 VARCHAR(32) NOT NULL,
+		  manifest_sha256 CHAR(64) NOT NULL,
+		  manifest_crc64 VARCHAR(32) NOT NULL,
+		  success_sha256 CHAR(64) NOT NULL,
+		  success_crc64 VARCHAR(32) NOT NULL,
+		  target_database VARCHAR(128) NOT NULL,
+		  verified_at_ms BIGINT NOT NULL,
+		  created_at_ms BIGINT NOT NULL,
+		  updated_at_ms BIGINT NOT NULL,
+		  PRIMARY KEY (rehearsal_id),
+		  KEY idx_wallet_outbox_restore_batch (source_app_code, source_batch_id, state)
+		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox archive restore rehearsal evidence v1'`, `
+		CREATE TABLE IF NOT EXISTS wallet_outbox_restore_rows_v1 (
+		  rehearsal_id VARCHAR(64) NOT NULL,
+		  source_batch_id CHAR(64) NOT NULL,
+		  app_code VARCHAR(32) NOT NULL,
+		  event_id VARCHAR(128) NOT NULL,
+		  event_type VARCHAR(64) NOT NULL,
+		  transaction_id VARCHAR(96) NOT NULL,
+		  command_id VARCHAR(128) NOT NULL,
+		  user_id BIGINT NOT NULL,
+		  asset_type VARCHAR(32) NOT NULL,
+		  available_delta BIGINT NOT NULL,
+		  frozen_delta BIGINT NOT NULL,
+		  payload JSON NOT NULL,
+		  status VARCHAR(32) NOT NULL,
+		  worker_id VARCHAR(128) NOT NULL,
+		  lock_until_ms BIGINT NULL,
+		  retry_count INT NOT NULL,
+		  next_retry_at_ms BIGINT NULL,
+		  last_error TEXT NULL,
+		  created_at_ms BIGINT NOT NULL,
+		  updated_at_ms BIGINT NOT NULL,
+		  restored_at_ms BIGINT NOT NULL,
+		  PRIMARY KEY (rehearsal_id, app_code, event_id),
+		  KEY idx_wallet_outbox_restore_cursor (rehearsal_id, app_code, updated_at_ms, event_id),
+		  KEY idx_wallet_outbox_restore_batch (source_batch_id, app_code, event_id)
+		) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='isolated wallet_outbox-compatible restored rows v1'`}
+	for _, statement := range statements {
+		if _, err := conn.ExecContext(ctx, statement); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func insertRehearsal(ctx context.Context, tx *sql.Tx, rehearsalID, databaseName string, verification Verification, nowMS int64) error {
+	manifest := verification.Manifest
+	_, err := tx.ExecContext(ctx, `
+		INSERT INTO wallet_outbox_restore_rehearsals_v1 (
+		  rehearsal_id, source_app_code, source_batch_id, state, schema_version, data_object_key, row_count,
+		  first_updated_at_ms, first_event_id, last_updated_at_ms, last_event_id,
+		  available_delta_total, frozen_delta_total, ndjson_sha256, gzip_sha256, data_crc64,
+		  manifest_sha256, manifest_crc64, success_sha256, success_crc64, target_database,
+		  verified_at_ms, created_at_ms, updated_at_ms
+		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)`,
+		rehearsalID, manifest.AppCode, manifest.BatchID, restoreState, manifest.SchemaVersion, manifest.DataObjectKey, manifest.RowCount,
+		manifest.FirstCursor.UpdatedAtMS, manifest.FirstCursor.EventID, manifest.LastCursor.UpdatedAtMS, manifest.LastCursor.EventID,
+		manifest.AvailableDeltaTotal, manifest.FrozenDeltaTotal, manifest.NDJSONSHA256, manifest.GzipSHA256, manifest.DataCRC64,
+		verification.ManifestSHA256, verification.ManifestCRC64, verification.SuccessSHA256, verification.SuccessCRC64, databaseName,
+		nowMS, nowMS)
+	return err
+}
+
+func insertRestoreRows(ctx context.Context, tx *sql.Tx, rehearsalID, batchID string, restoredAtMS int64, records []Record) error {
+	if len(records) == 0 || len(records) > 100 {
+		return errors.New("restore insert batch must contain 1..100 rows")
+	}
+	const columns = `(rehearsal_id, source_batch_id, app_code, event_id, event_type, transaction_id, command_id,
+		user_id, asset_type, available_delta, frozen_delta, payload, status, worker_id, lock_until_ms,
+		retry_count, next_retry_at_ms, last_error, created_at_ms, updated_at_ms, restored_at_ms)`
+	placeholders := make([]string, 0, len(records))
+	args := make([]any, 0, len(records)*21)
+	for _, record := range records {
+		placeholders = append(placeholders, `(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
+		args = append(args,
+			rehearsalID, batchID, record.AppCode, record.EventID, record.EventType, record.TransactionID, record.CommandID,
+			record.UserID, record.AssetType, record.AvailableDelta, record.FrozenDelta, record.PayloadJSON, record.Status, record.WorkerID,
+			nullInt64(record.LockUntilMS), record.RetryCount, nullInt64(record.NextRetryAtMS), nullString(record.LastError),
+			record.CreatedAtMS, record.UpdatedAtMS, restoredAtMS,
+		)
+	}
+	_, err := tx.ExecContext(ctx, `INSERT INTO wallet_outbox_restore_rows_v1 `+columns+` VALUES `+strings.Join(placeholders, ","), args...)
+	return err
+}
+
+func verifyRestoredControls(ctx context.Context, query interface {
+	QueryRowContext(context.Context, string, ...any) *sql.Row
+}, rehearsalID string, verification Verification) error {
+	var count int
+	var availableTotal, frozenTotal string
+	var minCreated, maxCreated, minUpdated, maxUpdated int64
+	if err := query.QueryRowContext(ctx, `
+		SELECT COUNT(*),
+		       CAST(COALESCE(SUM(available_delta), 0) AS CHAR),
+		       CAST(COALESCE(SUM(frozen_delta), 0) AS CHAR),
+		       COALESCE(MIN(created_at_ms), 0), COALESCE(MAX(created_at_ms), 0),
+		       COALESCE(MIN(updated_at_ms), 0), COALESCE(MAX(updated_at_ms), 0)
+		FROM wallet_outbox_restore_rows_v1 FORCE INDEX (PRIMARY)
+		WHERE rehearsal_id = ?`, rehearsalID).Scan(&count, &availableTotal, &frozenTotal, &minCreated, &maxCreated, &minUpdated, &maxUpdated); err != nil {
+		return err
+	}
+	manifest := verification.Manifest
+	if count != manifest.RowCount || availableTotal != manifest.AvailableDeltaTotal || frozenTotal != manifest.FrozenDeltaTotal || minCreated != manifest.MinCreatedAtMS || maxCreated != manifest.MaxCreatedAtMS || minUpdated != manifest.MinUpdatedAtMS || maxUpdated != manifest.MaxUpdatedAtMS {
+		return errors.New("isolation-table row count, time bounds, or control totals do not match manifest")
+	}
+	var firstUpdated, lastUpdated int64
+	var firstEvent, lastEvent string
+	if err := query.QueryRowContext(ctx, `
+		SELECT updated_at_ms, event_id FROM wallet_outbox_restore_rows_v1 FORCE INDEX (idx_wallet_outbox_restore_cursor)
+		WHERE rehearsal_id = ? AND app_code = ? ORDER BY updated_at_ms ASC, event_id ASC LIMIT 1`, rehearsalID, manifest.AppCode).Scan(&firstUpdated, &firstEvent); err != nil {
+		return err
+	}
+	if err := query.QueryRowContext(ctx, `
+		SELECT updated_at_ms, event_id FROM wallet_outbox_restore_rows_v1 FORCE INDEX (idx_wallet_outbox_restore_cursor)
+		WHERE rehearsal_id = ? AND app_code = ? ORDER BY updated_at_ms DESC, event_id DESC LIMIT 1`, rehearsalID, manifest.AppCode).Scan(&lastUpdated, &lastEvent); err != nil {
+		return err
+	}
+	if (Cursor{UpdatedAtMS: firstUpdated, EventID: firstEvent}) != manifest.FirstCursor || (Cursor{UpdatedAtMS: lastUpdated, EventID: lastEvent}) != manifest.LastCursor {
+		return errors.New("isolation-table cursor bounds do not match manifest")
+	}
+	return nil
+}
+
+func readExistingRestore(ctx context.Context, conn *sql.Conn, rehearsalID string, verification Verification) (RestoreResult, bool, error) {
+	var state, appCode, batchID, manifestSHA, manifestCRC, successCRC, availableTotal, frozenTotal string
+	var rowCount int
+	var verifiedAtMS int64
+	err := conn.QueryRowContext(ctx, `
+		SELECT state, source_app_code, source_batch_id, row_count, manifest_sha256, manifest_crc64, success_crc64,
+		       available_delta_total, frozen_delta_total, verified_at_ms
+		FROM wallet_outbox_restore_rehearsals_v1 WHERE rehearsal_id = ?`, rehearsalID).
+		Scan(&state, &appCode, &batchID, &rowCount, &manifestSHA, &manifestCRC, &successCRC, &availableTotal, &frozenTotal, &verifiedAtMS)
+	if err == sql.ErrNoRows {
+		return RestoreResult{}, false, nil
+	}
+	if err != nil {
+		return RestoreResult{}, false, err
+	}
+	manifest := verification.Manifest
+	if state != restoreVerifiedState || appCode != manifest.AppCode || batchID != manifest.BatchID || rowCount != manifest.RowCount || manifestSHA != verification.ManifestSHA256 || manifestCRC != verification.ManifestCRC64 || successCRC != verification.SuccessCRC64 || availableTotal != manifest.AvailableDeltaTotal || frozenTotal != manifest.FrozenDeltaTotal {
+		return RestoreResult{}, false, errors.New("rehearsal_id already exists with non-matching or incomplete evidence")
+	}
+	if err := verifyRestoredControls(ctx, conn, rehearsalID, verification); err != nil {
+		return RestoreResult{}, false, fmt.Errorf("existing restore readback failed: %w", err)
+	}
+	if err := verifyRestoredRowsExact(ctx, conn, rehearsalID, verification.Records); err != nil {
+		return RestoreResult{}, false, fmt.Errorf("existing restore exact-row readback failed: %w", err)
+	}
+	result := newRestoreResult(rehearsalID, "", verification, verifiedAtMS, true)
+	if err := conn.QueryRowContext(ctx, `SELECT DATABASE()`).Scan(&result.TargetDatabase); err != nil {
+		return RestoreResult{}, false, err
+	}
+	return result, true, nil
+}
+
+func newRestoreResult(rehearsalID, databaseName string, verification Verification, verifiedAtMS int64, idempotent bool) RestoreResult {
+	manifest := verification.Manifest
+	return RestoreResult{
+		State: restoreVerifiedState, RehearsalID: rehearsalID, BatchID: manifest.BatchID, AppCode: manifest.AppCode,
+		RowCount: manifest.RowCount, TargetDatabase: databaseName, RehearsalTable: RehearsalTable, RowsTable: RowsTable,
+		DataObjectKey: manifest.DataObjectKey, FirstCursor: manifest.FirstCursor, LastCursor: manifest.LastCursor,
+		NDJSONSHA256: manifest.NDJSONSHA256, GzipSHA256: manifest.GzipSHA256, DataCRC64: manifest.DataCRC64,
+		ManifestSHA256: verification.ManifestSHA256, ManifestCRC64: verification.ManifestCRC64,
+		SuccessSHA256: verification.SuccessSHA256, SuccessCRC64: verification.SuccessCRC64,
+		AvailableTotal: manifest.AvailableDeltaTotal, FrozenTotal: manifest.FrozenDeltaTotal,
+		VerifiedAtMS: verifiedAtMS, IdempotentReadback: idempotent,
+	}
+}
+
+func verifyRestoredRowsExact(ctx context.Context, query interface {
+	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
+}, rehearsalID string, expected []Record) error {
+	rows, err := query.QueryContext(ctx, `
+		SELECT app_code, event_id, event_type, transaction_id, command_id, user_id, asset_type,
+		       available_delta, frozen_delta, CAST(payload AS CHAR), status, worker_id,
+		       lock_until_ms, retry_count, next_retry_at_ms, last_error, created_at_ms, updated_at_ms
+		FROM wallet_outbox_restore_rows_v1 FORCE INDEX (idx_wallet_outbox_restore_cursor)
+		WHERE rehearsal_id = ?
+		ORDER BY app_code ASC, updated_at_ms ASC, event_id ASC`, rehearsalID)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	index := 0
+	for rows.Next() {
+		if index >= len(expected) {
+			return errors.New("isolation table contains more rows than verified archive")
+		}
+		var actual Record
+		var lockUntilMS, nextRetryAtMS sql.NullInt64
+		var lastError sql.NullString
+		if err := rows.Scan(
+			&actual.AppCode, &actual.EventID, &actual.EventType, &actual.TransactionID, &actual.CommandID,
+			&actual.UserID, &actual.AssetType, &actual.AvailableDelta, &actual.FrozenDelta, &actual.PayloadJSON,
+			&actual.Status, &actual.WorkerID, &lockUntilMS, &actual.RetryCount, &nextRetryAtMS, &lastError,
+			&actual.CreatedAtMS, &actual.UpdatedAtMS,
+		); err != nil {
+			return err
+		}
+		if lockUntilMS.Valid {
+			value := lockUntilMS.Int64
+			actual.LockUntilMS = &value
+		}
+		if nextRetryAtMS.Valid {
+			value := nextRetryAtMS.Int64
+			actual.NextRetryAtMS = &value
+		}
+		if lastError.Valid {
+			value := lastError.String
+			actual.LastError = &value
+		}
+		// Source payload was archived from CAST(payload AS CHAR); casting the restored
+		// JSON back through MySQL proves JSON parsing did not alter the owner snapshot.
+		if !reflect.DeepEqual(actual, expected[index]) {
+			return fmt.Errorf("isolation table row %d does not exactly match archive", index+1)
+		}
+		index++
+	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	if index != len(expected) {
+		return errors.New("isolation table contains fewer rows than verified archive")
+	}
+	return nil
+}
+
+func acquireRestoreLock(ctx context.Context, conn *sql.Conn, rehearsalID string) (func(), bool, error) {
+	hash := sha256.Sum256([]byte(rehearsalID))
+	lockName := "wallet_outbox_restore:" + hex.EncodeToString(hash[:16])
+	var acquired sql.NullInt64
+	if err := conn.QueryRowContext(ctx, `SELECT GET_LOCK(?, 0)`, lockName).Scan(&acquired); err != nil {
+		return nil, false, err
+	}
+	if !acquired.Valid || acquired.Int64 != 1 {
+		return func() {}, false, nil
+	}
+	return func() {
+		releaseCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+		defer cancel()
+		var released sql.NullInt64
+		_ = conn.QueryRowContext(releaseCtx, `SELECT RELEASE_LOCK(?)`, lockName).Scan(&released)
+	}, true, nil
+}
+
+func validRehearsalID(value string) bool {
+	if len(value) == 0 || len(value) > 64 {
+		return false
+	}
+	for _, char := range value {
+		if (char >= 'a' && char <= 'z') || (char >= 'A' && char <= 'Z') || (char >= '0' && char <= '9') || char == '.' || char == '_' || char == '-' {
+			continue
+		}
+		return false
+	}
+	return true
+}
+
+func nullInt64(value *int64) any {
+	if value == nil {
+		return nil
+	}
+	return *value
+}
+
+func nullString(value *string) any {
+	if value == nil {
+		return nil
+	}
+	return *value
+}
--- a/services/wallet-service/internal/outboxarchiverestore/verify.go
+++ b/services/wallet-service/internal/outboxarchiverestore/verify.go
@ -0,0 +1,431 @@
+// Package outboxarchiverestore verifies one immutable wallet_outbox archive batch
+// and restores it into versioned isolation tables. It deliberately contains no
+// purge API: a successful rehearsal is evidence for a later retention gate, not
+// authorization to mutate the owner outbox.
+package outboxarchiverestore
+
+import (
+	"bufio"
+	"bytes"
+	"compress/gzip"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"hash/crc64"
+	"io"
+	"math/big"
+	"os"
+	"path"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const (
+	archiveSchema        = "hyapp.wallet_outbox.archive"
+	archiveSchemaVersion = 1
+	manifestState        = "DATA_VERIFIED"
+	successState         = "VERIFIED"
+	filesVerifiedState   = "FILES_VERIFIED"
+	restoreVerifiedState = "RESTORE_VERIFIED"
+
+	maxManifestBytes = 1 << 20
+	maxSuccessBytes  = 64 << 10
+	maxNDJSONLine    = 64 << 20
+)
+
+// Limits bounds local CPU, memory and disk exposure before the operator-selected
+// restore database is opened. Values are intentionally independent of worker
+// configuration so an accidentally supplied object cannot bypass the CLI guard.
+type Limits struct {
+	MaxRows              int
+	MaxCompressedBytes   int64
+	MaxUncompressedBytes int64
+}
+
+func DefaultLimits() Limits {
+	return Limits{MaxRows: 100000, MaxCompressedBytes: 512 << 20, MaxUncompressedBytes: 1 << 30}
+}
+
+type Cursor struct {
+	UpdatedAtMS int64  `json:"updated_at_ms"`
+	EventID     string `json:"event_id"`
+}
+
+type Manifest struct {
+	Schema              string `json:"schema"`
+	SchemaVersion       int    `json:"schema_version"`
+	State               string `json:"state"`
+	AppCode             string `json:"app_code"`
+	BatchID             string `json:"batch_id"`
+	RowCount            int    `json:"row_count"`
+	FirstCursor         Cursor `json:"first_cursor"`
+	LastCursor          Cursor `json:"last_cursor"`
+	MinCreatedAtMS      int64  `json:"min_created_at_ms"`
+	MaxCreatedAtMS      int64  `json:"max_created_at_ms"`
+	MinUpdatedAtMS      int64  `json:"min_updated_at_ms"`
+	MaxUpdatedAtMS      int64  `json:"max_updated_at_ms"`
+	AvailableDeltaTotal string `json:"available_delta_total"`
+	FrozenDeltaTotal    string `json:"frozen_delta_total"`
+	NDJSONSHA256        string `json:"ndjson_sha256"`
+	GzipSHA256          string `json:"gzip_sha256"`
+	DataCRC64           string `json:"data_crc64"`
+	UncompressedBytes   int64  `json:"uncompressed_bytes"`
+	CompressedBytes     int64  `json:"compressed_bytes"`
+	DataObjectKey       string `json:"data_object_key"`
+}
+
+type Success struct {
+	Schema         string `json:"schema"`
+	SchemaVersion  int    `json:"schema_version"`
+	State          string `json:"state"`
+	BatchID        string `json:"batch_id"`
+	ManifestSHA256 string `json:"manifest_sha256"`
+	ManifestCRC64  string `json:"manifest_crc64"`
+}
+
+// Record mirrors every wallet_outbox column. Restore tables add rehearsal and
+// source-batch keys around this payload rather than changing owner-table fields,
+// so later historical tools can compare one archived row without translation.
+type Record struct {
+	AppCode        string  `json:"app_code"`
+	EventID        string  `json:"event_id"`
+	EventType      string  `json:"event_type"`
+	TransactionID  string  `json:"transaction_id"`
+	CommandID      string  `json:"command_id"`
+	UserID         int64   `json:"user_id"`
+	AssetType      string  `json:"asset_type"`
+	AvailableDelta int64   `json:"available_delta"`
+	FrozenDelta    int64   `json:"frozen_delta"`
+	PayloadJSON    string  `json:"payload_json"`
+	Status         string  `json:"status"`
+	WorkerID       string  `json:"worker_id"`
+	LockUntilMS    *int64  `json:"lock_until_ms"`
+	RetryCount     int     `json:"retry_count"`
+	NextRetryAtMS  *int64  `json:"next_retry_at_ms"`
+	LastError      *string `json:"last_error"`
+	CreatedAtMS    int64   `json:"created_at_ms"`
+	UpdatedAtMS    int64   `json:"updated_at_ms"`
+}
+
+// Verification is the complete local evidence set. SuccessCRC64 is computed
+// locally because _SUCCESS cannot contain its own checksum; a future purge gate
+// must compare all three CRC values with wallet_outbox_archive_receipts.
+type Verification struct {
+	State          string   `json:"state"`
+	Manifest       Manifest `json:"manifest"`
+	Records        []Record `json:"-"`
+	ManifestSHA256 string   `json:"manifest_sha256"`
+	ManifestCRC64  string   `json:"manifest_crc64"`
+	SuccessSHA256  string   `json:"success_sha256"`
+	SuccessCRC64   string   `json:"success_crc64"`
+}
+
+// VerifyLocalFiles establishes the _SUCCESS gate before any restore DSN is
+// opened. The order is deliberate: an incomplete batch may have data/manifest
+// objects, but without a valid marker it can never reach a database write path.
+func VerifyLocalFiles(dataPath, manifestPath, successPath string, limits Limits) (Verification, error) {
+	limits = normalizeLimits(limits)
+	successBody, err := readBoundedFile(successPath, maxSuccessBytes)
+	if err != nil {
+		return Verification{}, fmt.Errorf("read _SUCCESS: %w", err)
+	}
+	var success Success
+	if err := decodeStrictJSON(successBody, &success); err != nil {
+		return Verification{}, fmt.Errorf("decode _SUCCESS: %w", err)
+	}
+	if success.Schema != archiveSchema || success.SchemaVersion != archiveSchemaVersion || success.State != successState || !validSHA256(success.ManifestSHA256) || !validCRC64(success.ManifestCRC64) || !validSHA256(success.BatchID) {
+		return Verification{}, errors.New("_SUCCESS schema, state, batch_id, or manifest checksum is invalid")
+	}
+
+	manifestBody, err := readBoundedFile(manifestPath, maxManifestBytes)
+	if err != nil {
+		return Verification{}, fmt.Errorf("read manifest: %w", err)
+	}
+	manifestSHA256, manifestCRC64 := checksumBytes(manifestBody)
+	if manifestSHA256 != success.ManifestSHA256 || manifestCRC64 != success.ManifestCRC64 {
+		return Verification{}, errors.New("manifest does not match _SUCCESS SHA256/CRC64")
+	}
+	var manifest Manifest
+	if err := decodeStrictJSON(manifestBody, &manifest); err != nil {
+		return Verification{}, fmt.Errorf("decode manifest: %w", err)
+	}
+	if err := validateManifest(manifest, success, limits); err != nil {
+		return Verification{}, err
+	}
+
+	records, err := verifyDataFile(dataPath, manifest, limits)
+	if err != nil {
+		return Verification{}, err
+	}
+	successSHA256, successCRC64 := checksumBytes(successBody)
+	return Verification{
+		State: filesVerifiedState, Manifest: manifest, Records: records,
+		ManifestSHA256: manifestSHA256, ManifestCRC64: manifestCRC64,
+		SuccessSHA256: successSHA256, SuccessCRC64: successCRC64,
+	}, nil
+}
+
+func normalizeLimits(limits Limits) Limits {
+	defaults := DefaultLimits()
+	if limits.MaxRows <= 0 {
+		limits.MaxRows = defaults.MaxRows
+	}
+	if limits.MaxCompressedBytes <= 0 {
+		limits.MaxCompressedBytes = defaults.MaxCompressedBytes
+	}
+	if limits.MaxUncompressedBytes <= 0 {
+		limits.MaxUncompressedBytes = defaults.MaxUncompressedBytes
+	}
+	return limits
+}
+
+func validateManifest(manifest Manifest, success Success, limits Limits) error {
+	if manifest.Schema != archiveSchema || manifest.SchemaVersion != archiveSchemaVersion || manifest.State != manifestState {
+		return errors.New("manifest schema, version, or state is invalid")
+	}
+	if manifest.BatchID != success.BatchID || !validSHA256(manifest.BatchID) || strings.TrimSpace(manifest.AppCode) == "" || strings.TrimSpace(manifest.DataObjectKey) == "" {
+		return errors.New("manifest identity does not match _SUCCESS")
+	}
+	if manifest.RowCount <= 0 || manifest.RowCount > limits.MaxRows {
+		return fmt.Errorf("manifest row_count must be within [1,%d]", limits.MaxRows)
+	}
+	if manifest.CompressedBytes <= 0 || manifest.CompressedBytes > limits.MaxCompressedBytes || manifest.UncompressedBytes <= 0 || manifest.UncompressedBytes > limits.MaxUncompressedBytes {
+		return errors.New("manifest byte counts exceed restore safety limits")
+	}
+	if !validSHA256(manifest.NDJSONSHA256) || !validSHA256(manifest.GzipSHA256) || !validCRC64(manifest.DataCRC64) {
+		return errors.New("manifest data checksums are invalid")
+	}
+	if _, ok := new(big.Int).SetString(manifest.AvailableDeltaTotal, 10); !ok {
+		return errors.New("manifest available_delta_total is invalid")
+	}
+	if _, ok := new(big.Int).SetString(manifest.FrozenDeltaTotal, 10); !ok {
+		return errors.New("manifest frozen_delta_total is invalid")
+	}
+	if manifest.FirstCursor.UpdatedAtMS <= 0 || manifest.LastCursor.UpdatedAtMS <= 0 || strings.TrimSpace(manifest.FirstCursor.EventID) == "" || strings.TrimSpace(manifest.LastCursor.EventID) == "" {
+		return errors.New("manifest cursor is incomplete")
+	}
+	deliveredDay := time.UnixMilli(manifest.FirstCursor.UpdatedAtMS).UTC().Format("2006-01-02")
+	if time.UnixMilli(manifest.LastCursor.UpdatedAtMS).UTC().Format("2006-01-02") != deliveredDay {
+		return errors.New("manifest crosses UTC delivered day")
+	}
+	expectedSuffix := path.Join("app_code="+manifest.AppCode, "delivered_day="+deliveredDay, "batch_id="+manifest.BatchID, "data.ndjson.gz")
+	cleanObjectKey := path.Clean(manifest.DataObjectKey)
+	if cleanObjectKey != expectedSuffix && !strings.HasSuffix(cleanObjectKey, "/"+expectedSuffix) {
+		return errors.New("manifest data_object_key does not match app/day/batch identity")
+	}
+	return nil
+}
+
+func verifyDataFile(path string, manifest Manifest, limits Limits) ([]Record, error) {
+	file, err := os.Open(strings.TrimSpace(path))
+	if err != nil {
+		return nil, fmt.Errorf("open data object: %w", err)
+	}
+	defer file.Close()
+	info, err := file.Stat()
+	if err != nil {
+		return nil, err
+	}
+	if info.Size() != manifest.CompressedBytes || info.Size() > limits.MaxCompressedBytes {
+		return nil, errors.New("compressed data size does not match manifest")
+	}
+
+	gzipHash := sha256.New()
+	gzipCRC := crc64.New(crc64.MakeTable(crc64.ECMA))
+	compressedReader := io.TeeReader(file, io.MultiWriter(gzipHash, gzipCRC))
+	gzipReader, err := gzip.NewReader(compressedReader)
+	if err != nil {
+		return nil, fmt.Errorf("open archived gzip: %w", err)
+	}
+	records, ndjsonSHA256, uncompressedBytes, totals, err := readAndValidateNDJSON(gzipReader, manifest, limits)
+	closeErr := gzipReader.Close()
+	_, drainErr := io.Copy(io.Discard, compressedReader)
+	if err != nil {
+		return nil, err
+	}
+	if closeErr != nil {
+		return nil, fmt.Errorf("close archived gzip: %w", closeErr)
+	}
+	if drainErr != nil {
+		return nil, fmt.Errorf("read archived gzip tail: %w", drainErr)
+	}
+	if hex.EncodeToString(gzipHash.Sum(nil)) != manifest.GzipSHA256 || strconv.FormatUint(gzipCRC.Sum64(), 10) != manifest.DataCRC64 {
+		return nil, errors.New("data object SHA256/CRC64 does not match manifest")
+	}
+	if ndjsonSHA256 != manifest.NDJSONSHA256 || uncompressedBytes != manifest.UncompressedBytes {
+		return nil, errors.New("NDJSON SHA256/byte count does not match manifest")
+	}
+	if totals.available.String() != manifest.AvailableDeltaTotal || totals.frozen.String() != manifest.FrozenDeltaTotal {
+		return nil, errors.New("NDJSON control totals do not match manifest")
+	}
+	return records, nil
+}
+
+type controlTotals struct {
+	available *big.Int
+	frozen    *big.Int
+}
+
+func readAndValidateNDJSON(reader io.Reader, manifest Manifest, limits Limits) ([]Record, string, int64, controlTotals, error) {
+	buffered := bufio.NewReaderSize(reader, 64<<10)
+	hash := sha256.New()
+	records := make([]Record, 0, manifest.RowCount)
+	seen := make(map[string]struct{}, manifest.RowCount)
+	totals := controlTotals{available: new(big.Int), frozen: new(big.Int)}
+	var byteCount int64
+	var minCreated, maxCreated int64
+	for {
+		line, err := readRawLine(buffered, maxNDJSONLine)
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return nil, "", 0, totals, err
+		}
+		byteCount += int64(len(line))
+		if byteCount > limits.MaxUncompressedBytes {
+			return nil, "", 0, totals, errors.New("NDJSON exceeds uncompressed byte limit")
+		}
+		_, _ = hash.Write(line)
+		var record Record
+		if err := decodeStrictJSON(bytes.TrimSuffix(line, []byte{'\n'}), &record); err != nil {
+			return nil, "", 0, totals, fmt.Errorf("decode NDJSON row %d: %w", len(records)+1, err)
+		}
+		if err := validateRecord(record, manifest, records, seen); err != nil {
+			return nil, "", 0, totals, fmt.Errorf("validate NDJSON row %d: %w", len(records)+1, err)
+		}
+		if len(records) == 0 {
+			minCreated, maxCreated = record.CreatedAtMS, record.CreatedAtMS
+		} else {
+			if record.CreatedAtMS < minCreated {
+				minCreated = record.CreatedAtMS
+			}
+			if record.CreatedAtMS > maxCreated {
+				maxCreated = record.CreatedAtMS
+			}
+		}
+		totals.available.Add(totals.available, big.NewInt(record.AvailableDelta))
+		totals.frozen.Add(totals.frozen, big.NewInt(record.FrozenDelta))
+		records = append(records, record)
+		if len(records) > limits.MaxRows || len(records) > manifest.RowCount {
+			return nil, "", 0, totals, errors.New("NDJSON row count exceeds manifest or safety limit")
+		}
+	}
+	if len(records) != manifest.RowCount {
+		return nil, "", 0, totals, errors.New("NDJSON row count does not match manifest")
+	}
+	first, last := records[0], records[len(records)-1]
+	if manifest.FirstCursor != (Cursor{UpdatedAtMS: first.UpdatedAtMS, EventID: first.EventID}) || manifest.LastCursor != (Cursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID}) || manifest.MinUpdatedAtMS != first.UpdatedAtMS || manifest.MaxUpdatedAtMS != last.UpdatedAtMS {
+		return nil, "", 0, totals, errors.New("NDJSON cursor bounds do not match manifest")
+	}
+	if manifest.MinCreatedAtMS != minCreated || manifest.MaxCreatedAtMS != maxCreated {
+		return nil, "", 0, totals, errors.New("NDJSON created_at bounds do not match manifest")
+	}
+	ndjsonSHA256 := hex.EncodeToString(hash.Sum(nil))
+	batchHash := sha256.Sum256([]byte(strings.Join([]string{
+		manifest.AppCode, strconv.Itoa(len(records)), strconv.FormatInt(first.UpdatedAtMS, 10), first.EventID,
+		strconv.FormatInt(last.UpdatedAtMS, 10), last.EventID, ndjsonSHA256,
+	}, "\n")))
+	if hex.EncodeToString(batchHash[:]) != manifest.BatchID {
+		return nil, "", 0, totals, errors.New("recomputed batch_id does not match manifest")
+	}
+	return records, ndjsonSHA256, byteCount, totals, nil
+}
+
+func validateRecord(record Record, manifest Manifest, prior []Record, seen map[string]struct{}) error {
+	if record.AppCode != manifest.AppCode || record.Status != "delivered" || strings.TrimSpace(record.EventID) == "" || strings.TrimSpace(record.EventType) == "" || record.UpdatedAtMS <= 0 || record.CreatedAtMS <= 0 {
+		return errors.New("row identity, state, or timestamp is invalid")
+	}
+	if !json.Valid([]byte(record.PayloadJSON)) {
+		return errors.New("payload_json is not valid JSON")
+	}
+	if _, exists := seen[record.EventID]; exists {
+		return errors.New("duplicate (app_code,event_id) archive PK")
+	}
+	seen[record.EventID] = struct{}{}
+	if len(prior) > 0 {
+		previous := prior[len(prior)-1]
+		if record.UpdatedAtMS < previous.UpdatedAtMS || (record.UpdatedAtMS == previous.UpdatedAtMS && record.EventID <= previous.EventID) {
+			return errors.New("rows are not strictly ordered by (updated_at_ms,event_id)")
+		}
+	}
+	return nil
+}
+
+func readRawLine(reader *bufio.Reader, maxBytes int) ([]byte, error) {
+	line := make([]byte, 0, 4096)
+	for {
+		fragment, err := reader.ReadSlice('\n')
+		if len(line)+len(fragment) > maxBytes {
+			return nil, fmt.Errorf("NDJSON line exceeds %d bytes", maxBytes)
+		}
+		line = append(line, fragment...)
+		switch err {
+		case nil:
+			return line, nil
+		case bufio.ErrBufferFull:
+			continue
+		case io.EOF:
+			if len(line) == 0 {
+				return nil, io.EOF
+			}
+			return nil, errors.New("NDJSON final row is missing newline terminator")
+		default:
+			return nil, err
+		}
+	}
+}
+
+func readBoundedFile(path string, maxBytes int64) ([]byte, error) {
+	file, err := os.Open(strings.TrimSpace(path))
+	if err != nil {
+		return nil, err
+	}
+	defer file.Close()
+	body, err := io.ReadAll(io.LimitReader(file, maxBytes+1))
+	if err != nil {
+		return nil, err
+	}
+	if int64(len(body)) > maxBytes || len(body) == 0 {
+		return nil, errors.New("file is empty or exceeds safety limit")
+	}
+	return body, nil
+}
+
+func decodeStrictJSON(body []byte, destination any) error {
+	decoder := json.NewDecoder(bytes.NewReader(body))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(destination); err != nil {
+		return err
+	}
+	var trailing any
+	if err := decoder.Decode(&trailing); err != io.EOF {
+		if err == nil {
+			return errors.New("JSON contains more than one value")
+		}
+		return err
+	}
+	return nil
+}
+
+func checksumBytes(body []byte) (string, string) {
+	sha := sha256.Sum256(body)
+	crc := crc64.Update(0, crc64.MakeTable(crc64.ECMA), body)
+	return hex.EncodeToString(sha[:]), strconv.FormatUint(crc, 10)
+}
+
+func validSHA256(value string) bool {
+	if len(value) != sha256.Size*2 {
+		return false
+	}
+	_, err := hex.DecodeString(value)
+	return err == nil
+}
+
+func validCRC64(value string) bool {
+	_, err := strconv.ParseUint(strings.TrimSpace(value), 10, 64)
+	return err == nil
+}
--- a/services/wallet-service/internal/storage/mysql/outbox_archive.go
+++ b/services/wallet-service/internal/storage/mysql/outbox_archive.go
@ -3,6 +3,7 @@ package mysql
 import (
 	"context"
 	"database/sql"
+	"fmt"
 	"strings"
 	"time"

@ -41,6 +42,14 @@ type WalletOutboxArchiveRecord struct {
 	UpdatedAtMS    int64   `json:"updated_at_ms"`
 }

+// WalletOutboxArchiveMember 是一个已经进入同批 COS data/manifest/_SUCCESS 并完成读回验证的精确事件键。
+// 它只保存后续清理所需的窄字段，不复制 payload；旧 receipt 没有 member 时天然不具备清理资格。
+type WalletOutboxArchiveMember struct {
+	AppCode     string
+	EventID     string
+	UpdatedAtMS int64
+}
+
 // WalletOutboxArchiveReceipt 只在 COS data/manifest/_SUCCESS 都读回校验后写入。
 // VERIFIED 只代表对象完整性，不代表已完成真实 MySQL 恢复演练。
 type WalletOutboxArchiveReceipt struct {
@ -67,6 +76,7 @@ type WalletOutboxArchiveReceipt struct {
 	UncompressedBytes   int64
 	CompressedBytes     int64
 	VerifiedAtMS        int64
+	Members             []WalletOutboxArchiveMember
 }

 // AcquireWalletOutboxArchiveAppLock 用 MySQL connection-scoped advisory lock 保证多副本下每个 App 只有一个归档者。
@ -186,13 +196,23 @@ func (r *Repository) RecordVerifiedWalletOutboxArchive(ctx context.Context, rece
 	if r == nil || r.db == nil {
 		return xerr.New(xerr.Unavailable, "mysql repository is not configured")
 	}
+	receipt.AppCode = appcode.Normalize(receipt.AppCode)
 	if receipt.State != walletOutboxArchiveStateVerified || receipt.BatchID == "" || receipt.RowCount <= 0 ||
 		receipt.DataObjectKey == "" || receipt.ManifestObjectKey == "" || receipt.SuccessObjectKey == "" ||
 		receipt.NDJSONSHA256 == "" || receipt.GzipSHA256 == "" || receipt.ManifestSHA256 == "" ||
-		receipt.DataCRC64 == "" || receipt.ManifestCRC64 == "" || receipt.SuccessCRC64 == "" {
+		receipt.DataCRC64 == "" || receipt.ManifestCRC64 == "" || receipt.SuccessCRC64 == "" ||
+		receipt.VerifiedAtMS <= 0 || len(receipt.Members) != receipt.RowCount {
 		return xerr.New(xerr.InvalidArgument, "verified archive receipt is incomplete")
 	}
-	_, err := r.db.ExecContext(ctx, `
+	if err := validateWalletOutboxArchiveMembers(receipt); err != nil {
+		return err
+	}
+	tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback() }()
+	_, err = tx.ExecContext(ctx, `
 		INSERT INTO wallet_outbox_archive_receipts (
 		  app_code, batch_id, state, data_object_key, manifest_object_key, success_object_key,
 		  schema_version, row_count, first_updated_at_ms, first_event_id, last_updated_at_ms, last_event_id,
@ -211,5 +231,111 @@ func (r *Repository) RecordVerifiedWalletOutboxArchive(ctx context.Context, rece
 		receipt.ManifestSHA256, receipt.UncompressedBytes, receipt.CompressedBytes,
 		receipt.VerifiedAtMS, receipt.VerifiedAtMS, receipt.VerifiedAtMS,
 	)
-	return err
+	if err != nil {
+		return err
+	}
+	if err := verifyWalletOutboxArchiveReceiptControls(ctx, tx, receipt); err != nil {
+		return err
+	}
+	// receipt 与 exact members 必须原子提交：只看到 receipt、看不到成员的中间态会让
+	// watermark 推断重新混入清理链路；只看到成员、receipt 未 VERIFIED 则无法证明 COS 对象完整。
+	for offset := 0; offset < len(receipt.Members); offset += 500 {
+		end := min(offset+500, len(receipt.Members))
+		if err := upsertAndVerifyWalletOutboxArchiveMembers(ctx, tx, receipt, receipt.Members[offset:end]); err != nil {
+			return err
+		}
+	}
+	return tx.Commit()
+}
+
+func verifyWalletOutboxArchiveReceiptControls(ctx context.Context, tx *sql.Tx, receipt WalletOutboxArchiveReceipt) error {
+	var state, manifestSHA, dataCRC64, manifestCRC64, successCRC64 string
+	var schemaVersion, rowCount int
+	var verifiedAtMS int64
+	if err := tx.QueryRowContext(ctx, `
+		SELECT state, schema_version, row_count, manifest_sha256,
+		       data_crc64, manifest_crc64, success_crc64, verified_at_ms
+		FROM wallet_outbox_archive_receipts FORCE INDEX (PRIMARY)
+		WHERE app_code = ? AND batch_id = ?`, receipt.AppCode, receipt.BatchID).
+		Scan(&state, &schemaVersion, &rowCount, &manifestSHA, &dataCRC64, &manifestCRC64, &successCRC64, &verifiedAtMS); err != nil {
+		return err
+	}
+	// batch_id 是内容指纹，但存储层仍逐字段核对恢复授权所依赖的控制值；同 PK 下任何
+	// 非幂等 receipt 都必须在写 members 前失败，不能留下永远无法与 receipt JOIN 的孤立证明。
+	if state != receipt.State || schemaVersion != receipt.SchemaVersion || rowCount != receipt.RowCount ||
+		manifestSHA != receipt.ManifestSHA256 || dataCRC64 != receipt.DataCRC64 ||
+		manifestCRC64 != receipt.ManifestCRC64 || successCRC64 != receipt.SuccessCRC64 || verifiedAtMS != receipt.VerifiedAtMS {
+		return xerr.New(xerr.Conflict, "verified archive receipt controls do not match existing batch")
+	}
+	return nil
+}
+
+func validateWalletOutboxArchiveMembers(receipt WalletOutboxArchiveReceipt) error {
+	seen := make(map[string]struct{}, len(receipt.Members))
+	for _, member := range receipt.Members {
+		if appcode.Normalize(member.AppCode) != receipt.AppCode || strings.TrimSpace(member.EventID) == "" || member.UpdatedAtMS <= 0 {
+			return xerr.New(xerr.InvalidArgument, "verified archive member is incomplete")
+		}
+		if _, exists := seen[member.EventID]; exists {
+			return xerr.New(xerr.InvalidArgument, "verified archive members contain duplicate event_id")
+		}
+		seen[member.EventID] = struct{}{}
+	}
+	return nil
+}
+
+func upsertAndVerifyWalletOutboxArchiveMembers(ctx context.Context, tx *sql.Tx, receipt WalletOutboxArchiveReceipt, members []WalletOutboxArchiveMember) error {
+	values := make([]string, 0, len(members))
+	args := make([]any, 0, len(members)*6)
+	for _, member := range members {
+		values = append(values, "(?, ?, ?, ?, ?, ?)")
+		args = append(args, receipt.AppCode, member.EventID, member.UpdatedAtMS, receipt.BatchID, receipt.ManifestSHA256, receipt.VerifiedAtMS)
+	}
+	// archive worker 的网络结果可能在 DB commit 后丢失；no-op duplicate 允许同一批重试，
+	// 随后的逐键读回会拒绝任何已经被另一 batch 占用或 metadata 不一致的成员。
+	if _, err := tx.ExecContext(ctx, `
+		INSERT INTO wallet_outbox_archive_members (
+		  app_code, event_id, updated_at_ms, batch_id, manifest_sha256, receipt_verified_at_ms
+		) VALUES `+strings.Join(values, ",")+`
+		ON DUPLICATE KEY UPDATE event_id = VALUES(event_id)`, args...); err != nil {
+		return err
+	}
+
+	placeholders := make([]string, 0, len(members))
+	lookupArgs := make([]any, 0, len(members)+1)
+	lookupArgs = append(lookupArgs, receipt.AppCode)
+	expected := make(map[string]int64, len(members))
+	for _, member := range members {
+		placeholders = append(placeholders, "?")
+		lookupArgs = append(lookupArgs, member.EventID)
+		expected[member.EventID] = member.UpdatedAtMS
+	}
+	rows, err := tx.QueryContext(ctx, `
+		SELECT event_id, updated_at_ms, batch_id, manifest_sha256, receipt_verified_at_ms
+		FROM wallet_outbox_archive_members FORCE INDEX (PRIMARY)
+		WHERE app_code = ? AND event_id IN (`+strings.Join(placeholders, ",")+`)`, lookupArgs...)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	matched := 0
+	for rows.Next() {
+		var eventID, batchID, manifestSHA string
+		var updatedAtMS, receiptVerifiedAtMS int64
+		if err := rows.Scan(&eventID, &updatedAtMS, &batchID, &manifestSHA, &receiptVerifiedAtMS); err != nil {
+			return err
+		}
+		expectedUpdatedAtMS, exists := expected[eventID]
+		if !exists || updatedAtMS != expectedUpdatedAtMS || batchID != receipt.BatchID || manifestSHA != receipt.ManifestSHA256 || receiptVerifiedAtMS != receipt.VerifiedAtMS {
+			return fmt.Errorf("wallet outbox archive member conflicts with verified receipt: event_id=%s", eventID)
+		}
+		matched++
+	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	if matched != len(members) {
+		return fmt.Errorf("wallet outbox archive member readback mismatch: expected=%d matched=%d", len(members), matched)
+	}
+	return nil
 }
--- a/services/wallet-service/internal/storage/mysql/outbox_purge.go
+++ b/services/wallet-service/internal/storage/mysql/outbox_purge.go
@ -0,0 +1,378 @@
+package mysql
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/binary"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"hash"
+	"strings"
+
+	"hyapp/pkg/appcode"
+	"hyapp/pkg/xerr"
+)
+
+const (
+	walletOutboxPurgeAuthorizationState = "RESTORE_VERIFIED"
+	walletOutboxPurgeReceiptState       = "PURGED"
+	walletOutboxPurgePolicyFinancialV1  = "financial-v1"
+)
+
+// financial-v1 是显式正向安全集合：只覆盖已经确认不被正常在线查询回读的账务事实。
+// 未知新事件、礼物/资源目录事件和低频资源/VIP 控制事件默认永久留热，新增清理类型必须发布新 policy version。
+var walletOutboxPurgeFinancialV1EventTypes = []string{
+	"AgencyPointCredited",
+	"HostPeriodDiamondCredited",
+	"HostPointCredited",
+	"WalletAgencyOpeningRewardCredited",
+	"WalletBalanceChanged",
+	"WalletCoinSellerCoinCompensated",
+	"WalletCoinSellerStockPurchased",
+	"WalletCoinSellerSubTransferred",
+	"WalletCoinSellerTransferred",
+	"WalletGiftDebited",
+	"WalletGiftIncomeCredited",
+	"WalletGooglePaymentCredited",
+	"WalletInviteActivityRewardCredited",
+	"WalletLuckyGiftRewardCredited",
+	"WalletRechargeRecorded",
+	"WalletRedPacketClaimed",
+	"WalletRedPacketCreated",
+	"WalletRedPacketRefunded",
+	"WalletRoomTurnoverRewardCredited",
+	"WalletTaskRewardCredited",
+	"WalletWheelRewardCredited",
+}
+
+// WalletOutboxPurgeReceipt 是一个已经提交的精确 PK 清理批次。
+// DeletedCount 为零时表示本轮没有成熟候选，此时 PurgeID 为空且数据库不会写空审计行。
+type WalletOutboxPurgeReceipt struct {
+	AppCode                 string
+	PurgeID                 string
+	State                   string
+	PolicyVersion           string
+	CutoffMS                int64
+	ReceiptVerifiedBeforeMS int64
+	CandidateSHA256         string
+	CandidateCount          int
+	DeletedCount            int
+	FirstCursor             WalletOutboxArchiveCursor
+	LastCursor              WalletOutboxArchiveCursor
+	PurgedAtMS              int64
+}
+
+type walletOutboxPurgeCandidate struct {
+	EventID     string
+	UpdatedAtMS int64
+}
+
+// RequireWalletOutboxPurgeAuthorization 要求生产 wallet DB 已手工固化一条与隔离库
+// RESTORE_VERIFIED 回执精确匹配的授权证据。迁移不会自动插入该表，因此仅部署人员
+// 明确核验过的 (App, archive batch, manifest hash) 能打开自动清理总闸。
+func (r *Repository) RequireWalletOutboxPurgeAuthorization(ctx context.Context, appCode, batchID, manifestSHA string) error {
+	if r == nil || r.db == nil {
+		return xerr.New(xerr.Unavailable, "mysql repository is not configured")
+	}
+	ctx, normalizedAppCode, err := walletOutboxPurgeContext(ctx, appCode)
+	if err != nil {
+		return err
+	}
+	batchID, err = normalizeWalletOutboxPurgeSHA256("archive_batch_id", batchID)
+	if err != nil {
+		return err
+	}
+	manifestSHA, err = normalizeWalletOutboxPurgeSHA256("manifest_sha256", manifestSHA)
+	if err != nil {
+		return err
+	}
+
+	var marker int
+	err = r.db.QueryRowContext(ctx, `
+		SELECT 1
+		FROM wallet_outbox_archive_purge_authorizations auth FORCE INDEX (PRIMARY)
+		JOIN wallet_outbox_archive_receipts receipt FORCE INDEX (PRIMARY)
+		  ON receipt.app_code = auth.app_code AND receipt.batch_id = auth.archive_batch_id
+		 AND receipt.manifest_sha256 = auth.manifest_sha256
+		 AND receipt.schema_version = auth.schema_version AND receipt.row_count = auth.row_count
+		 AND receipt.data_crc64 = auth.data_crc64 AND receipt.manifest_crc64 = auth.manifest_crc64
+		 AND receipt.success_crc64 = auth.success_crc64
+		WHERE auth.app_code = ? AND auth.archive_batch_id = ? AND auth.manifest_sha256 = ?
+		  AND auth.state = ? AND receipt.state = ?
+		LIMIT 1`,
+		normalizedAppCode, batchID, manifestSHA, walletOutboxPurgeAuthorizationState, walletOutboxArchiveStateVerified,
+	).Scan(&marker)
+	if errors.Is(err, sql.ErrNoRows) {
+		return xerr.New(xerr.Conflict, "wallet outbox purge restore rehearsal is not authorized")
+	}
+	return err
+}
+
+// PurgeDeliveredWalletOutboxBatch 在单个 READ COMMITTED 事务内锁定、删除并审计至多 limit 条旧事实。
+// 调用方必须已经持有 AcquireWalletOutboxArchiveAppLock 返回的 App 级 GET_LOCK。候选只能从
+// exact member 表产生；每一行都重新连接当前 outbox 和 VERIFIED receipt，杜绝用批次首尾游标
+// 推断中间行、迟到写入或旧 receipt 自动获得删除资格。
+func (r *Repository) PurgeDeliveredWalletOutboxBatch(ctx context.Context, appCode, policyVersion string, cutoffMS, receiptVerifiedBeforeMS int64, limit int, nowMS int64) (WalletOutboxPurgeReceipt, error) {
+	if r == nil || r.db == nil {
+		return WalletOutboxPurgeReceipt{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
+	}
+	ctx, normalizedAppCode, err := walletOutboxPurgeContext(ctx, appCode)
+	if err != nil {
+		return WalletOutboxPurgeReceipt{}, err
+	}
+	eventTypes, err := walletOutboxPurgeEventTypes(policyVersion)
+	if err != nil {
+		return WalletOutboxPurgeReceipt{}, err
+	}
+	if cutoffMS <= 0 || receiptVerifiedBeforeMS <= 0 || nowMS <= 0 || cutoffMS >= nowMS || receiptVerifiedBeforeMS >= nowMS {
+		return WalletOutboxPurgeReceipt{}, xerr.New(xerr.InvalidArgument, "wallet outbox purge cutoff, receipt grace, and now are invalid")
+	}
+	if limit <= 0 || limit > 500 {
+		return WalletOutboxPurgeReceipt{}, xerr.New(xerr.InvalidArgument, "wallet outbox purge limit must be within [1,500]")
+	}
+
+	tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
+	if err != nil {
+		return WalletOutboxPurgeReceipt{}, err
+	}
+	defer func() { _ = tx.Rollback() }()
+	receipt := WalletOutboxPurgeReceipt{
+		AppCode:                 normalizedAppCode,
+		State:                   walletOutboxPurgeReceiptState,
+		PolicyVersion:           policyVersion,
+		CutoffMS:                cutoffMS,
+		ReceiptVerifiedBeforeMS: receiptVerifiedBeforeMS,
+		PurgedAtMS:              nowMS,
+	}
+	candidates, err := selectWalletOutboxPurgeCandidates(ctx, tx, normalizedAppCode, cutoffMS, receiptVerifiedBeforeMS, eventTypes, limit)
+	if err != nil {
+		return WalletOutboxPurgeReceipt{}, err
+	}
+	if len(candidates) == 0 {
+		return receipt, nil
+	}
+
+	receipt.CandidateSHA256 = hashWalletOutboxPurgeCandidates(normalizedAppCode, candidates)
+	receipt.CandidateCount = len(candidates)
+	receipt.FirstCursor = WalletOutboxArchiveCursor{UpdatedAtMS: candidates[0].UpdatedAtMS, EventID: candidates[0].EventID}
+	last := candidates[len(candidates)-1]
+	receipt.LastCursor = WalletOutboxArchiveCursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID}
+	receipt.PurgeID = hashWalletOutboxPurgeReceipt(receipt)
+
+	deleted, err := deleteWalletOutboxPurgeCandidates(ctx, tx, normalizedAppCode, cutoffMS, eventTypes, candidates)
+	if err != nil {
+		return WalletOutboxPurgeReceipt{}, err
+	}
+	if deleted != int64(len(candidates)) {
+		return WalletOutboxPurgeReceipt{}, fmt.Errorf("wallet outbox purge candidate/delete mismatch: candidates=%d deleted=%d", len(candidates), deleted)
+	}
+	deletedMembers, err := deleteWalletOutboxArchiveMembers(ctx, tx, normalizedAppCode, candidates)
+	if err != nil {
+		return WalletOutboxPurgeReceipt{}, err
+	}
+	if deletedMembers != int64(len(candidates)) {
+		// outbox 与 member 删除必须共同提交；member 数不一致时回滚 outbox 删除，避免留下无法审计的半完成批次。
+		return WalletOutboxPurgeReceipt{}, fmt.Errorf("wallet outbox purge member/delete mismatch: candidates=%d deleted_members=%d", len(candidates), deletedMembers)
+	}
+	receipt.DeletedCount = int(deleted)
+	if err := insertWalletOutboxPurgeReceipt(ctx, tx, receipt); err != nil {
+		return WalletOutboxPurgeReceipt{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return WalletOutboxPurgeReceipt{}, err
+	}
+	return receipt, nil
+}
+
+func selectWalletOutboxPurgeCandidates(ctx context.Context, tx *sql.Tx, appCode string, cutoffMS, receiptVerifiedBeforeMS int64, eventTypes []string, limit int) ([]walletOutboxPurgeCandidate, error) {
+	placeholders := walletOutboxPurgePlaceholders(len(eventTypes))
+	args := make([]any, 0, len(eventTypes)+6)
+	args = append(args, appCode, cutoffMS, walletOutboxArchiveStateVerified, receiptVerifiedBeforeMS, outboxStatusDelivered)
+	for _, eventType := range eventTypes {
+		args = append(args, eventType)
+	}
+	args = append(args, limit)
+	rows, err := tx.QueryContext(ctx, `
+		SELECT STRAIGHT_JOIN m.event_id, m.updated_at_ms
+		FROM wallet_outbox_archive_members m FORCE INDEX (idx_wallet_outbox_archive_member_purge)
+		JOIN wallet_outbox_archive_receipts ar FORCE INDEX (PRIMARY)
+		  ON ar.app_code = m.app_code AND ar.batch_id = m.batch_id
+		 AND ar.manifest_sha256 = m.manifest_sha256 AND ar.verified_at_ms = m.receipt_verified_at_ms
+		JOIN wallet_outbox wo FORCE INDEX (PRIMARY)
+		  ON wo.app_code = m.app_code AND wo.event_id = m.event_id AND wo.updated_at_ms = m.updated_at_ms
+		WHERE m.app_code = ? AND m.updated_at_ms < ?
+		  AND ar.state = ? AND ar.verified_at_ms < ?
+		  AND wo.status = ? AND wo.event_type IN (`+placeholders+`)
+		ORDER BY m.updated_at_ms ASC, m.event_id ASC
+		LIMIT ?
+		FOR UPDATE SKIP LOCKED`, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	candidates := make([]walletOutboxPurgeCandidate, 0, limit)
+	for rows.Next() {
+		var candidate walletOutboxPurgeCandidate
+		if err := rows.Scan(&candidate.EventID, &candidate.UpdatedAtMS); err != nil {
+			return nil, err
+		}
+		candidates = append(candidates, candidate)
+	}
+	return candidates, rows.Err()
+}
+
+func deleteWalletOutboxPurgeCandidates(ctx context.Context, tx *sql.Tx, appCode string, cutoffMS int64, eventTypes []string, candidates []walletOutboxPurgeCandidate) (int64, error) {
+	primaryKeys, args, err := walletOutboxPurgeExactKeys(appCode, candidates)
+	if err != nil {
+		return 0, err
+	}
+	args = append(args, appCode, outboxStatusDelivered, cutoffMS)
+	for _, eventType := range eventTypes {
+		args = append(args, eventType)
+	}
+	// 候选 SELECT 已锁住这些行；删除仍逐键复核 app/event/updated、delivered、cutoff 和正向 policy，
+	// 即使未来有人修改候选 JOIN，也不能把未授权类型或时间发生变化的行带入删除。
+	result, err := tx.ExecContext(ctx, `
+		DELETE wo
+		FROM wallet_outbox wo FORCE INDEX (PRIMARY)
+		WHERE (wo.app_code, wo.event_id, wo.updated_at_ms) IN (`+primaryKeys+`)
+		  AND wo.app_code = ? AND wo.status = ? AND wo.updated_at_ms < ?
+		  AND wo.event_type IN (`+walletOutboxPurgePlaceholders(len(eventTypes))+`)`, args...)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
+func deleteWalletOutboxArchiveMembers(ctx context.Context, tx *sql.Tx, appCode string, candidates []walletOutboxPurgeCandidate) (int64, error) {
+	primaryKeys, args, err := walletOutboxPurgeExactKeys(appCode, candidates)
+	if err != nil {
+		return 0, err
+	}
+	result, err := tx.ExecContext(ctx, `
+		DELETE m
+		FROM wallet_outbox_archive_members m FORCE INDEX (PRIMARY)
+		WHERE (m.app_code, m.event_id, m.updated_at_ms) IN (`+primaryKeys+`)`, args...)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
+func walletOutboxPurgeExactKeys(appCode string, candidates []walletOutboxPurgeCandidate) (string, []any, error) {
+	if len(candidates) == 0 || len(candidates) > 500 {
+		return "", nil, xerr.New(xerr.InvalidArgument, "wallet outbox purge candidate count must be within [1,500]")
+	}
+	keys := make([]string, 0, len(candidates))
+	args := make([]any, 0, len(candidates)*3)
+	for _, candidate := range candidates {
+		keys = append(keys, "(?, ?, ?)")
+		args = append(args, appCode, candidate.EventID, candidate.UpdatedAtMS)
+	}
+	return strings.Join(keys, ","), args, nil
+}
+
+func insertWalletOutboxPurgeReceipt(ctx context.Context, tx *sql.Tx, receipt WalletOutboxPurgeReceipt) error {
+	result, err := tx.ExecContext(ctx, `
+		INSERT INTO wallet_outbox_archive_purge_receipts (
+		  app_code, purge_id, state, policy_version, cutoff_ms, receipt_verified_before_ms,
+		  candidate_sha256, candidate_count, deleted_count,
+		  first_updated_at_ms, first_event_id, last_updated_at_ms, last_event_id,
+		  purged_at_ms, created_at_ms, updated_at_ms
+		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		receipt.AppCode, receipt.PurgeID, receipt.State, receipt.PolicyVersion, receipt.CutoffMS, receipt.ReceiptVerifiedBeforeMS,
+		receipt.CandidateSHA256, receipt.CandidateCount, receipt.DeletedCount,
+		receipt.FirstCursor.UpdatedAtMS, receipt.FirstCursor.EventID, receipt.LastCursor.UpdatedAtMS, receipt.LastCursor.EventID,
+		receipt.PurgedAtMS, receipt.PurgedAtMS, receipt.PurgedAtMS,
+	)
+	if err != nil {
+		return err
+	}
+	affected, err := result.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if affected != 1 {
+		return fmt.Errorf("wallet outbox purge receipt insert affected %d rows", affected)
+	}
+	return nil
+}
+
+func walletOutboxPurgeContext(ctx context.Context, appCode string) (context.Context, string, error) {
+	if strings.TrimSpace(appCode) == "" {
+		return ctx, "", xerr.New(xerr.InvalidArgument, "wallet outbox purge app_code is required")
+	}
+	ctx = contextWithCommandApp(ctx, appCode)
+	normalized := appcode.FromContext(ctx)
+	// 本轮容量治理边界只有 Lalu/Fami；其他租户即使误配也必须在进入事务前失败。
+	if normalized != "lalu" && normalized != "fami" {
+		return ctx, "", xerr.New(xerr.InvalidArgument, "wallet outbox purge app_code is not allowed")
+	}
+	return ctx, normalized, nil
+}
+
+func walletOutboxPurgeEventTypes(policyVersion string) ([]string, error) {
+	policyVersion = strings.TrimSpace(policyVersion)
+	if policyVersion != walletOutboxPurgePolicyFinancialV1 {
+		return nil, xerr.New(xerr.InvalidArgument, "wallet outbox purge policy_version is unsupported")
+	}
+	return walletOutboxPurgeFinancialV1EventTypes, nil
+}
+
+func normalizeWalletOutboxPurgeSHA256(field string, value string) (string, error) {
+	value = strings.ToLower(strings.TrimSpace(value))
+	if len(value) != sha256.Size*2 {
+		return "", xerr.New(xerr.InvalidArgument, field+" must be a SHA-256 hex digest")
+	}
+	if _, err := hex.DecodeString(value); err != nil {
+		return "", xerr.New(xerr.InvalidArgument, field+" must be a SHA-256 hex digest")
+	}
+	return value, nil
+}
+
+func walletOutboxPurgePlaceholders(count int) string {
+	values := make([]string, count)
+	for index := range values {
+		values[index] = "?"
+	}
+	return strings.Join(values, ",")
+}
+
+func hashWalletOutboxPurgeCandidates(appCode string, candidates []walletOutboxPurgeCandidate) string {
+	digest := sha256.New()
+	writeWalletOutboxPurgeHashString(digest, appCode)
+	writeWalletOutboxPurgeHashUint64(digest, uint64(len(candidates)))
+	for _, candidate := range candidates {
+		// 候选已按 (updated_at_ms,event_id) 排序；哈希内容采用精确 PK 和归档时 updated_at 快照。
+		writeWalletOutboxPurgeHashString(digest, candidate.EventID)
+		writeWalletOutboxPurgeHashUint64(digest, uint64(candidate.UpdatedAtMS))
+	}
+	return hex.EncodeToString(digest.Sum(nil))
+}
+
+func hashWalletOutboxPurgeReceipt(receipt WalletOutboxPurgeReceipt) string {
+	digest := sha256.New()
+	writeWalletOutboxPurgeHashString(digest, receipt.AppCode)
+	writeWalletOutboxPurgeHashString(digest, receipt.PolicyVersion)
+	writeWalletOutboxPurgeHashString(digest, receipt.CandidateSHA256)
+	writeWalletOutboxPurgeHashUint64(digest, uint64(receipt.CutoffMS))
+	writeWalletOutboxPurgeHashUint64(digest, uint64(receipt.ReceiptVerifiedBeforeMS))
+	writeWalletOutboxPurgeHashUint64(digest, uint64(receipt.PurgedAtMS))
+	return hex.EncodeToString(digest.Sum(nil))
+}
+
+func writeWalletOutboxPurgeHashString(digest hash.Hash, value string) {
+	// 长度前缀消除字段拼接歧义，使相同候选在不同进程上得到完全一致的审计指纹。
+	var size [4]byte
+	binary.BigEndian.PutUint32(size[:], uint32(len(value)))
+	_, _ = digest.Write(size[:])
+	_, _ = digest.Write([]byte(value))
+}
+
+func writeWalletOutboxPurgeHashUint64(digest hash.Hash, value uint64) {
+	var encoded [8]byte
+	binary.BigEndian.PutUint64(encoded[:], value)
+	_, _ = digest.Write(encoded[:])
+}