修复套餐接续停机竞态与轮询兜底

cmd/audit-retention-simulate/main.go

@@ -141,9 +141,24 @@ func run(ctx context.Context) error {
 	if err := collectAfterCleanup(ctx, db, monthStart, monthEnd, &summary); err != nil {
 		return err
 	}
-	if summary.TargetRowsAfterCleanup != 0 || summary.BoundaryRowsAfterCleanup != 6 || summary.CleanupMarkersAfter != int64(summary.Days*2) {
+	if summary.TargetRowsAfterCleanup != 1 || summary.BoundaryRowsAfterCleanup != 6 || summary.CleanupMarkersAfter != int64(summary.Days*2) {
 		return fmt.Errorf("清理范围复核失败: target=%d boundary=%d markers=%d", summary.TargetRowsAfterCleanup, summary.BoundaryRowsAfterCleanup, summary.CleanupMarkersAfter)
 	}
+	if err := db.WithContext(ctx).Model(&model.IntegrationLog{}).
+		Where("integration_id = ?", "int_"+simulationPrefix+"-"+monthStart.Format("20060102")).
+		Updates(map[string]any{"result": constants.IntegrationResultSuccess, "updated_at": time.Now()}).Error; err != nil {
+		return err
+	}
+	if _, err := service.RetainDate(ctx, monthStart, true); err != nil {
+		return fmt.Errorf("pending 终结后的续跑清理演练失败: %w", err)
+	}
+	var remaining int64
+	if err := db.WithContext(ctx).Model(&model.IntegrationLog{}).Where("created_at >= ? AND created_at < ?", monthStart, monthEnd).Count(&remaining).Error; err != nil {
+		return err
+	}
+	if remaining != 0 {
+		return fmt.Errorf("pending 终结后的续跑清理未完成，剩余 %d 条记录", remaining)
+	}
 	if err := syncRetentionLogger(); err != nil {
 		return fmt.Errorf("刷新日留存仿真日志失败: %w", err)
 	}
@@ -270,16 +285,6 @@ func archiveMonth(ctx context.Context, db *gorm.DB, service *auditarchive.Servic
 	if err := service.ArchiveDate(ctx, start); err != nil {
 		return fmt.Errorf("Audit 故障重试演练失败: %w", err)
 	}
-	if err := db.WithContext(ctx).Model(&model.IntegrationLog{}).
-		Where("integration_id = ?", "int_"+simulationPrefix+"-"+start.Format("20060102")).
-		Updates(map[string]any{"result": constants.IntegrationResultSuccess, "updated_at": time.Now()}).Error; err != nil {
-		return err
-	}
-	for date := start; date.Before(end); date = date.AddDate(0, 0, 1) {
-		if err := service.FinalizeIntegrationDate(ctx, date); err != nil {
-			return err
-		}
-	}
 	return nil
 }
 

internal/application/auditarchive/integration.go

@@ -77,20 +77,20 @@ func (s *Service) archiveIntegrationDate(ctx context.Context, archiveDate time.T
 	if err != nil {
 		return err
 	}
+	if final && run.Status == constants.ArchiveStatusSuccess && run.IsFinal && run.CleanedAt != nil {
+		terminalCount, countErr := s.integrationTerminalCount(ctx, start, end)
+		if countErr != nil {
+			return countErr
+		}
+		if terminalCount == 0 {
+			return nil
+		}
+	}
 	file, err := s.buildIntegrationArchiveFile(ctx, start, end)
 	if err != nil {
 		return err
 	}
 	defer os.Remove(file.path)
-	if final {
-		pending, pendingErr := s.integrationPendingCount(ctx, start, end)
-		if pendingErr != nil {
-			return pendingErr
-		}
-		if pending > 0 {
-			return fmt.Errorf("仍有 %d 条 pending Integration Log，无法形成最终归档", pending)
-		}
-	}
 	if run.Status == constants.ArchiveStatusSuccess && run.RecordCount == file.recordCount && run.SHA256 == file.sha256 {
 		valid, validateErr := s.validateIntegrationRun(ctx, run)
 		if validateErr == nil && valid && (!final || run.IsFinal) {
@@ -146,7 +146,8 @@ func (s *Service) acquireIntegrationRun(ctx context.Context, run *model.LogArchi
 		Where("id = ? AND (status <> ? OR updated_at < ?)", run.ID, constants.ArchiveStatusRunning, now.Add(-3*time.Hour)).
 		Updates(map[string]any{
 			"status": constants.ArchiveStatusRunning, "revision": revision, "is_final": false,
-			"attempt_count": gorm.Expr("attempt_count + 1"), "error_summary": "", "completed_at": nil, "updated_at": now,
+			"attempt_count": gorm.Expr("attempt_count + 1"), "error_summary": "", "completed_at": nil,
+			"cleanup_started_at": nil, "cleaned_at": nil, "updated_at": now,
 		})
 	if result.Error != nil {
 		return false, fmt.Errorf("锁定 Integration Log 归档任务失败: %w", result.Error)
@@ -301,16 +302,6 @@ func (s *Service) integrationRecordCount(ctx context.Context, start, end time.Ti
 	return count, nil
 }
 
-func (s *Service) integrationPendingCount(ctx context.Context, start, end time.Time) (int64, error) {
-	var count int64
-	if err := s.db.WithContext(ctx).Model(&model.IntegrationLog{}).
-		Where("created_at >= ? AND created_at < ? AND result = ?", start, end, constants.IntegrationResultPending).
-		Count(&count).Error; err != nil {
-		return 0, fmt.Errorf("统计 pending Integration Log 失败: %w", err)
-	}
-	return count, nil
-}
-
 func integrationObjectKeys(date time.Time, revision int) (string, string) {
 	prefix := fmt.Sprintf("audit-archive/v1/%04d/%02d/%02d", date.Year(), date.Month(), date.Day())
 	name := fmt.Sprintf("integration-logs-%s-r%d", date.Format(time.DateOnly), revision)

internal/application/auditarchive/retention.go

@@ -7,7 +7,6 @@ import (
 	"io"
 	"os"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/bytedance/sonic"
@@ -50,11 +49,6 @@ func (e *RetentionBlockedError) Error() string {
 }
 func (e *RetentionBlockedError) Unwrap() error { return e.Err }
 
-type retentionRuns struct {
-	audit       *model.LogArchiveRun
-	integration *model.LogArchiveRun
-}
-
 // RetainPendingDays 从最早仍在线的日期连续处理至昨天。cleanup 为 false 时只读校验。
 func (s *Service) RetainPendingDays(ctx context.Context, cleanup bool) ([]RetentionResult, error) {
 	if s.db == nil || s.store == nil {
@@ -65,19 +59,39 @@ func (s *Service) RetainPendingDays(ctx context.Context, cleanup bool) ([]Retent
 		return nil, err
 	}
 	results := make([]RetentionResult, 0)
+	var auditBlocked, integrationBlocked *RetentionBlockedError
 	for date := start; date.Before(end); date = date.AddDate(0, 0, 1) {
-		result, err := s.retainDate(ctx, date, cleanup)
-		if err != nil {
-			source := "retention"
-			if strings.HasPrefix(err.Error(), "Audit") {
-				source = constants.AuditArchiveSource
+		startedAt := time.Now()
+		result := RetentionResult{ArchiveDate: date.Format(time.DateOnly)}
+		if auditBlocked == nil {
+			auditResult, auditErr := s.retainAuditDate(ctx, date, cleanup)
+			if auditErr != nil {
+				auditBlocked = &RetentionBlockedError{ArchiveDate: result.ArchiveDate, Source: constants.AuditArchiveSource, Err: auditErr}
+			} else {
+				result.EventCount, result.ResourceCount = auditResult.EventCount, auditResult.ResourceCount
+				result.ManifestKeys = append(result.ManifestKeys, auditResult.ManifestKeys...)
 			}
-			if strings.HasPrefix(err.Error(), "Integration") {
-				source = constants.IntegrationArchiveSource
-			}
-			return results, &RetentionBlockedError{ArchiveDate: date.Format(time.DateOnly), Source: source, Err: err}
 		}
-		results = append(results, result)
+		if integrationBlocked == nil {
+			integrationResult, integrationErr := s.retainIntegrationDate(ctx, date, cleanup)
+			if integrationErr != nil {
+				integrationBlocked = &RetentionBlockedError{ArchiveDate: result.ArchiveDate, Source: constants.IntegrationArchiveSource, Err: integrationErr}
+			} else {
+				result.IntegrationCount = integrationResult.IntegrationCount
+				result.ManifestKeys = append(result.ManifestKeys, integrationResult.ManifestKeys...)
+			}
+		}
+		result.EstimatedBatches = estimatedRetentionBatches(result)
+		result.Duration = time.Since(startedAt)
+		if auditBlocked == nil || integrationBlocked == nil {
+			results = append(results, result)
+		}
+	}
+	if auditBlocked != nil {
+		return results, auditBlocked
+	}
+	if integrationBlocked != nil {
+		return results, integrationBlocked
 	}
 	return results, nil
 }
@@ -91,9 +105,16 @@ func (s *Service) pendingRetentionRange(ctx context.Context) (time.Time, time.Ti
 	today := time.Now().In(s.location)
 	end := time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, s.location)
 	var earliest *time.Time
-	for _, item := range []struct{ table, column string }{{"tb_audit_event", "created_at"}, {"tb_integration_log", "created_at"}} {
+	for _, item := range []struct{ table, column, condition string }{
+		{"tb_audit_event", "created_at", ""},
+		{"tb_integration_log", "created_at", "result <> 'pending'"},
+	} {
+		query := s.db.WithContext(ctx).Table(item.table).Select("MIN(" + item.column + ")")
+		if item.condition != "" {
+			query = query.Where(item.condition)
+		}
 		var value *time.Time
-		if err := s.db.WithContext(ctx).Table(item.table).Select("MIN(" + item.column + ")").Scan(&value).Error; err != nil {
+		if err := query.Scan(&value).Error; err != nil {
 			return time.Time{}, time.Time{}, fmt.Errorf("查询日留存起点失败: %w", err)
 		}
 		if value != nil && (earliest == nil || value.Before(*earliest)) {
@@ -110,20 +131,52 @@ func (s *Service) pendingRetentionRange(ctx context.Context) (time.Time, time.Ti
 
 func (s *Service) retainDate(ctx context.Context, date time.Time, cleanup bool) (RetentionResult, error) {
 	startedAt := time.Now()
-	var existing []model.LogArchiveRun
-	if err := s.db.WithContext(ctx).Where("source IN ? AND archive_date = ? AND instance_id = ?", []string{constants.AuditArchiveSource, constants.IntegrationArchiveSource}, date.Format(time.DateOnly), s.instanceID).Find(&existing).Error; err != nil {
-		return RetentionResult{}, fmt.Errorf("读取日归档账本失败: %w", err)
+	auditResult, err := s.retainAuditDate(ctx, date, cleanup)
+	if err != nil {
+		return RetentionResult{}, err
 	}
-	statuses := map[string]string{}
-	for _, run := range existing {
-		statuses[run.Source] = run.Status
+	integrationResult, err := s.retainIntegrationDate(ctx, date, cleanup)
+	if err != nil {
+		return RetentionResult{}, err
 	}
-	if statuses[constants.AuditArchiveSource] != constants.ArchiveStatusSuccess {
+	result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), EventCount: auditResult.EventCount, ResourceCount: auditResult.ResourceCount, IntegrationCount: integrationResult.IntegrationCount, ManifestKeys: append(auditResult.ManifestKeys, integrationResult.ManifestKeys...)}
+	result.EstimatedBatches = estimatedRetentionBatches(result)
+	result.Duration = time.Since(startedAt)
+	return result, nil
+}
+
+func (s *Service) retainAuditDate(ctx context.Context, date time.Time, cleanup bool) (RetentionResult, error) {
+	run, err := s.retentionRun(ctx, date, constants.AuditArchiveSource)
+	if err != nil {
+		return RetentionResult{}, err
+	}
+	if run == nil || run.Status != constants.ArchiveStatusSuccess {
 		if err := s.ArchiveDate(ctx, date); err != nil {
 			return RetentionResult{}, fmt.Errorf("Audit 归档失败: %w", err)
 		}
+		run, err = s.retentionRun(ctx, date, constants.AuditArchiveSource)
+		if err != nil {
+			return RetentionResult{}, err
+		}
 	}
-	if statuses[constants.IntegrationArchiveSource] != constants.ArchiveStatusSuccess {
+	if err := s.validateAuditRetentionDay(ctx, date, run); err != nil {
+		return RetentionResult{}, fmt.Errorf("Audit 完整性校验失败: %w", err)
+	}
+	result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), EventCount: run.EventCount, ResourceCount: run.ResourceCount, ManifestKeys: []string{run.ManifestKey}}
+	if cleanup {
+		if err := s.cleanupAuditDate(ctx, date, run); err != nil {
+			return result, err
+		}
+	}
+	return result, nil
+}
+
+func (s *Service) retainIntegrationDate(ctx context.Context, date time.Time, cleanup bool) (RetentionResult, error) {
+	run, err := s.retentionRun(ctx, date, constants.IntegrationArchiveSource)
+	if err != nil {
+		return RetentionResult{}, err
+	}
+	if run == nil || run.Status != constants.ArchiveStatusSuccess {
 		if err := s.ArchiveIntegrationDate(ctx, date); err != nil {
 			return RetentionResult{}, fmt.Errorf("Integration Log 归档失败: %w", err)
 		}
@@ -131,48 +184,31 @@ func (s *Service) retainDate(ctx context.Context, date time.Time, cleanup bool)
 	if err := s.FinalizeIntegrationDate(ctx, date); err != nil {
 		return RetentionResult{}, fmt.Errorf("Integration Log 最终归档失败: %w", err)
 	}
-	runs, err := s.loadRetentionRuns(ctx, date)
+	run, err = s.retentionRun(ctx, date, constants.IntegrationArchiveSource)
 	if err != nil {
 		return RetentionResult{}, err
 	}
-	if err := s.validateAuditRetentionDay(ctx, date, runs.audit); err != nil {
-		return RetentionResult{}, fmt.Errorf("Audit 完整性校验失败: %w", err)
-	}
-	if err := s.validateIntegrationRetentionDay(ctx, date, runs.integration); err != nil {
+	if err := s.validateIntegrationRetentionDay(ctx, date, run); err != nil {
 		return RetentionResult{}, fmt.Errorf("Integration Log 完整性校验失败: %w", err)
 	}
-	result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), EventCount: runs.audit.EventCount, ResourceCount: runs.audit.ResourceCount, IntegrationCount: runs.integration.RecordCount, ManifestKeys: []string{runs.audit.ManifestKey, runs.integration.ManifestKey}}
-	result.EstimatedBatches = estimatedRetentionBatches(result)
+	result := RetentionResult{ArchiveDate: date.Format(time.DateOnly), IntegrationCount: run.RecordCount, ManifestKeys: []string{run.ManifestKey}}
 	if cleanup {
-		if err := s.cleanupAuditDate(ctx, date, runs.audit); err != nil {
-			return result, err
-		}
-		if err := s.cleanupIntegrationDate(ctx, date, runs.integration); err != nil {
+		if err := s.cleanupIntegrationDate(ctx, date, run); err != nil {
 			return result, err
 		}
 	}
-	result.Duration = time.Since(startedAt)
 	return result, nil
 }
 
-func (s *Service) loadRetentionRuns(ctx context.Context, date time.Time) (retentionRuns, error) {
-	var rows []model.LogArchiveRun
-	if err := s.db.WithContext(ctx).Where("source IN ? AND archive_date = ? AND instance_id = ?", []string{constants.AuditArchiveSource, constants.IntegrationArchiveSource}, date.Format(time.DateOnly), s.instanceID).Find(&rows).Error; err != nil {
-		return retentionRuns{}, fmt.Errorf("读取日归档账本失败: %w", err)
+func (s *Service) retentionRun(ctx context.Context, date time.Time, source string) (*model.LogArchiveRun, error) {
+	var runs []model.LogArchiveRun
+	if err := s.db.WithContext(ctx).Where("source = ? AND archive_date = ? AND instance_id = ?", source, date.Format(time.DateOnly), s.instanceID).Find(&runs).Error; err != nil {
+		return nil, fmt.Errorf("读取日归档账本失败: %w", err)
 	}
-	runs := retentionRuns{}
-	for index := range rows {
-		switch rows[index].Source {
-		case constants.AuditArchiveSource:
-			runs.audit = &rows[index]
-		case constants.IntegrationArchiveSource:
-			runs.integration = &rows[index]
-		}
+	if len(runs) == 0 {
+		return nil, nil
 	}
-	if runs.audit == nil || runs.integration == nil {
-		return retentionRuns{}, fmt.Errorf("Audit 或 Integration Log 归档账本缺失")
-	}
-	return runs, nil
+	return &runs[0], nil
 }
 
 func (s *Service) validateAuditRetentionDay(ctx context.Context, date time.Time, run *model.LogArchiveRun) error {
@@ -200,13 +236,17 @@ func (s *Service) validateIntegrationRetentionDay(ctx context.Context, date time
 	if err != nil {
 		return err
 	}
-	if run.CleanedAt != nil && count != 0 {
-		return fmt.Errorf("已标记清理完成但数据库仍有 %d 条记录", count)
+	terminalCount, err := s.integrationTerminalCount(ctx, run.RangeStart, run.RangeEnd)
+	if err != nil {
+		return err
+	}
+	if run.CleanedAt != nil && terminalCount != 0 {
+		return fmt.Errorf("已标记清理完成但数据库仍有 %d 条终态记录", terminalCount)
 	}
 	if run.CleanupStartedAt != nil && count > run.RecordCount {
 		return fmt.Errorf("续跑窗口记录数超过最终归档数量")
 	}
-	if run.CleanupStartedAt == nil {
+	if terminalCount > 0 && run.CleanupStartedAt == nil {
 		file, err := s.buildIntegrationArchiveFile(ctx, run.RangeStart, run.RangeEnd)
 		if err != nil {
 			return err
@@ -348,7 +388,11 @@ func (s *Service) cleanupAuditDate(ctx context.Context, date time.Time, run *mod
 	return s.markCleaned(ctx, constants.AuditArchiveSource, date)
 }
 func (s *Service) cleanupIntegrationDate(ctx context.Context, date time.Time, run *model.LogArchiveRun) error {
-	if run.CleanedAt != nil {
+	terminalCount, err := s.integrationTerminalCount(ctx, date, date.AddDate(0, 0, 1))
+	if err != nil {
+		return err
+	}
+	if run.CleanedAt != nil && terminalCount == 0 {
 		return nil
 	}
 	if err := s.markCleanupStarted(ctx, constants.IntegrationArchiveSource, date); err != nil {
@@ -356,7 +400,7 @@ func (s *Service) cleanupIntegrationDate(ctx context.Context, date time.Time, ru
 	}
 	end := date.AddDate(0, 0, 1)
 	for {
-		subquery := s.db.Model(&model.IntegrationLog{}).Select("id").Where("created_at >= ? AND created_at < ?", date, end).Order("id ASC").Limit(constants.AuditRetentionDeleteBatchSize)
+		subquery := s.db.Model(&model.IntegrationLog{}).Select("id").Where("created_at >= ? AND created_at < ? AND result <> ?", date, end, constants.IntegrationResultPending).Order("id ASC").Limit(constants.AuditRetentionDeleteBatchSize)
 		deleted := s.db.WithContext(ctx).Where("id IN (?)", subquery).Delete(&model.IntegrationLog{})
 		if deleted.Error != nil {
 			return fmt.Errorf("分批物理删除 Integration Log 失败: %w", deleted.Error)
@@ -367,6 +411,16 @@ func (s *Service) cleanupIntegrationDate(ctx context.Context, date time.Time, ru
 	}
 	return s.markCleaned(ctx, constants.IntegrationArchiveSource, date)
 }
+func (s *Service) integrationTerminalCount(ctx context.Context, start, end time.Time) (int64, error) {
+	var count int64
+	if err := s.db.WithContext(ctx).Model(&model.IntegrationLog{}).
+		Where("created_at >= ? AND created_at < ? AND result <> ?", start, end, constants.IntegrationResultPending).
+		Count(&count).Error; err != nil {
+		return 0, fmt.Errorf("统计终态 Integration Log 失败: %w", err)
+	}
+	return count, nil
+}
+
 func (s *Service) deleteAuditResources(ctx context.Context, start, end time.Time) error {
 	for {
 		subquery := s.db.Model(&model.AuditEventResource{}).Select("tb_audit_event_resource.id").Joins("JOIN tb_audit_event ON tb_audit_event.id = tb_audit_event_resource.audit_event_id").Where("tb_audit_event.created_at >= ? AND tb_audit_event.created_at < ?", start, end).Order("tb_audit_event_resource.id ASC").Limit(constants.AuditRetentionDeleteBatchSize)

internal/polling/package_activation_handler.go

@@ -253,7 +253,8 @@ func (h *PackageActivationHandler) findExpiredMainPackages(ctx context.Context)
 }
 
 // processExpiredPackage 处理单个过期套餐
-// 流程：先同步最新流量 → 事务内标记过期和失效加油包 → 提交后投递下一套餐并触发停机检查
+// 流程：先同步最新流量 → 事务内标记过期和失效加油包 → 提交后投递下一套餐；
+// 仅明确没有后续套餐时才触发停机检查，避免与异步激活任务竞态。
 func (h *PackageActivationHandler) processExpiredPackage(ctx context.Context, pkg *model.PackageUsage) error {
 	carrierType, carrierID := h.getCarrierInfo(pkg)
 
@@ -311,9 +312,15 @@ func (h *PackageActivationHandler) processExpiredPackage(ctx context.Context, pk
 
 	// 事务提交后再投递，确保消费者只能读取到旧套餐已经过期的状态。
 	if carrierType != "" && carrierID > 0 {
-		activationErr := h.activateNextPackage(ctx, h.db.WithContext(ctx), carrierType, carrierID)
+		activationEnqueued, err := h.activateNextPackage(ctx, h.db.WithContext(ctx), carrierType, carrierID)
+		if err != nil {
+			// 激活结果未知时不能将其当作无套餐；孤儿扫描和套餐轮询会继续兜底。
+			return err
+		}
+		if activationEnqueued {
+			return nil
+		}
 		h.triggerStopAfterExpiry(ctx, carrierType, carrierID)
-		return activationErr
 	}
 
 	return nil
@@ -368,7 +375,8 @@ func (h *PackageActivationHandler) getCarrierInfo(pkg *model.PackageUsage) (stri
 }
 
 // activateNextPackage 提交下一个待生效主套餐的异步激活任务。
-func (h *PackageActivationHandler) activateNextPackage(ctx context.Context, tx *gorm.DB, carrierType string, carrierID uint) error {
+// 返回 true 表示已找到并成功提交后续套餐；false 表示不存在后续套餐。
+func (h *PackageActivationHandler) activateNextPackage(ctx context.Context, tx *gorm.DB, carrierType string, carrierID uint) (bool, error) {
 	var nextPkg model.PackageUsage
 	query := tx.Where("status = ?", constants.PackageUsageStatusPending).
 		Where("master_usage_id IS NULL").
@@ -383,16 +391,19 @@ func (h *PackageActivationHandler) activateNextPackage(ctx context.Context, tx *
 
 	if err := query.First(&nextPkg).Error; err != nil {
 		if err == gorm.ErrRecordNotFound {
-			return nil
+			return false, nil
 		}
-		return err
+		return false, err
 	}
 
-	return h.enqueueActivationTask(ctx, nextPkg.ID, carrierType, carrierID, "queue")
+	if err := h.enqueueActivationTask(ctx, nextPkg.ID, carrierType, carrierID, "queue"); err != nil {
+		return false, err
+	}
+	return true, nil
 }
 
-// triggerStopAfterExpiry 套餐过期后异步触发停机检查
-// 仅在确认无后续生效套餐时有效；CheckAndStopCard 内部有幂等保护，重复调用安全
+// triggerStopAfterExpiry 在明确无后续套餐时异步触发停机检查。
+// 后续套餐存在时，停机重评估由激活任务在完成后顺序执行。
 func (h *PackageActivationHandler) triggerStopAfterExpiry(ctx context.Context, carrierType string, carrierID uint) {
 	if h.stopResumeCallback == nil {
 		return
@@ -431,6 +442,42 @@ func (h *PackageActivationHandler) triggerStopAfterExpiry(ctx context.Context, c
 	}
 }
 
+// reconcileCarrierAfterActivation 在套餐激活任务完成后按最新事实重评估停复机。
+// 它必须在激活调用返回后执行，不能与激活任务并发读取过期权益快照。
+func (h *PackageActivationHandler) reconcileCarrierAfterActivation(ctx context.Context, packageUsageID uint, carrierType string, carrierID uint) error {
+	if h.stopResumeCallback == nil {
+		h.logger.Warn("套餐激活后停复机回调未注入，跳过重评估",
+			zap.Uint("package_usage_id", packageUsageID),
+			zap.String("carrier_type", carrierType),
+			zap.Uint("carrier_id", carrierID))
+		return nil
+	}
+	if carrierID == 0 {
+		return errors.New(errors.CodeInvalidParam, "套餐使用记录缺少有效载体")
+	}
+
+	if carrierType == constants.AssetTypeIotCard {
+		if err := h.stopResumeCallback.CheckAndStopCard(ctx, carrierID); err != nil {
+			return err
+		}
+		return nil
+	}
+	if carrierType != constants.AssetTypeDevice {
+		return errors.New(errors.CodeInvalidParam, "套餐使用记录载体类型无效")
+	}
+
+	bindings, err := h.deviceSimBinding.ListByDeviceID(ctx, carrierID)
+	if err != nil {
+		return err
+	}
+	for _, binding := range bindings {
+		if err := h.stopResumeCallback.CheckAndStopCard(ctx, binding.IotCardID); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // enqueueActivationTask 提交套餐激活任务到 Asynq
 func (h *PackageActivationHandler) enqueueActivationTask(ctx context.Context, packageUsageID uint, carrierType string, carrierID uint, activationType string) error {
 	linkage := auditcontext.From(ctx)
@@ -499,14 +546,7 @@ func (h *PackageActivationHandler) HandlePackageQueueActivation(ctx context.Cont
 		return err
 	}
 
-	// 幂等性检查：如果已经是生效状态，跳过
-	if pkg.Status == constants.PackageUsageStatusActive {
-		h.logger.Info("套餐已激活，跳过",
-			zap.Uint("package_usage_id", payload.PackageUsageID))
-		return nil
-	}
-
-	// 调用 ActivationService 执行激活
+	// 调用 ActivationService 执行激活。即使套餐已由其他任务激活，仍须重新评估停复机。
 	if h.activationService != nil {
 		if err := h.activationService.ActivateSpecificPackage(ctx, payload.PackageUsageID); err != nil {
 			h.logger.Error("套餐激活失败",
@@ -521,8 +561,20 @@ func (h *PackageActivationHandler) HandlePackageQueueActivation(ctx context.Cont
 		return errors.New(errors.CodeInternalError, "激活服务未注入，无法执行套餐激活")
 	}
 
-	h.logger.Info("套餐激活成功",
-		zap.Uint("package_usage_id", payload.PackageUsageID))
+	carrierType, carrierID := h.getCarrierInfo(&pkg)
+	if err := h.reconcileCarrierAfterActivation(ctx, payload.PackageUsageID, carrierType, carrierID); err != nil {
+		h.logger.Error("套餐激活后停复机重评估失败",
+			zap.Uint("package_usage_id", payload.PackageUsageID),
+			zap.String("carrier_type", carrierType),
+			zap.Uint("carrier_id", carrierID),
+			zap.Error(err))
+		return err
+	}
+
+	h.logger.Info("套餐激活及停复机重评估完成",
+		zap.Uint("package_usage_id", payload.PackageUsageID),
+		zap.String("carrier_type", carrierType),
+		zap.Uint("carrier_id", carrierID))
 
 	return nil
 }

internal/polling/queue_manager.go

@@ -98,6 +98,20 @@ func (m *PollingQueueManager) Requeue(ctx context.Context, cardID uint, taskType
 	}).Err()
 }
 
+// EnsureQueued 仅在任务当前不在分片队列中时补入任务，不覆盖已有任务的执行时间。
+func (m *PollingQueueManager) EnsureQueued(ctx context.Context, cardID uint, taskType string, nextCheckAt time.Time) (bool, error) {
+	shardID := int(cardID) % m.shardCount
+	key := constants.RedisPollingShardQueueKey(shardID, taskType)
+	added, err := m.redis.ZAddArgs(ctx, key, redis.ZAddArgs{
+		NX: true,
+		Members: []redis.Z{{
+			Score:  float64(nextCheckAt.Unix()),
+			Member: fmt.Sprintf("%d", cardID),
+		}},
+	}).Result()
+	return added > 0, err
+}
+
 // RemoveFromAllQueues 从所有分片的所有5个队列（realname/carddata/package/protect/card_status）移除指定卡
 // 修复 Bug3：旧实现漏掉 protect 队列
 func (m *PollingQueueManager) RemoveFromAllQueues(ctx context.Context, cardID uint) error {

internal/service/package/activation_service.go

@@ -334,10 +334,10 @@ func (s *ActivationService) ActivateSpecificPackage(ctx context.Context, package
 		return errors.Wrap(errors.CodeRedisError, err, "获取分布式锁失败")
 	}
 	if !locked {
-		s.logger.Warn("套餐激活正在进行中，跳过",
+		s.logger.Warn("套餐激活正在进行中，等待任务重试",
 			zap.String("carrier_type", carrierType),
 			zap.Uint("carrier_id", carrierID))
-		return nil
+		return errors.New(errors.CodePackageActivationConflict)
 	}
 	defer s.redis.Del(ctx, lockKey)
 

internal/task/asset_package_batch_order.go

@@ -121,19 +121,13 @@ func (h *AssetPackageBatchOrderHandler) finishBatchOrderTask(ctx context.Context
 			return err
 		}
 		rootID := audit.TaskEventID(constants.AuditResourceAssetPackageBatchOrderTask, taskRecord.ID, "completed")
-		var childCount int64
-		if err := tx.WithContext(ctx).Model(&model.AuditEvent{}).
-			Where("parent_event_id = ? AND action_code = ?", rootID, constants.AuditActionOrderCreated).
-			Count(&childCount).Error; err != nil {
-			return err
-		}
-		result := batchAuditResult(int(childCount), failCount)
+		result := batchAuditResult(successCount, failCount)
 		return h.auditWriter.WriteTask(ctx, tx, audit.TaskInput{
 			EventID: rootID, ActionCode: constants.AuditActionAssetPackageBatchOrderTaskCompleted,
 			Summary: "完成资产套餐批量订购任务", TaskID: taskRecord.ID, TaskNo: taskRecord.TaskNo,
 			Result: result, CorrelationID: taskRecord.TaskNo,
 			ParentEventID: audit.TaskEventID(constants.AuditResourceAssetPackageBatchOrderTask, taskRecord.ID, "created"),
-			BatchTotal:    len(items), SuccessCount: int(childCount), FailCount: failCount,
+			BatchTotal:    len(items), SuccessCount: successCount, FailCount: failCount,
 			IdentitySnapshot: map[string]any{
 				"id": taskRecord.ID, "task_no": taskRecord.TaskNo, "file_name": taskRecord.FileName,
 				"package_id": taskRecord.PackageID, "package_code": taskRecord.PackageCode,

internal/task/polling_base.go

@@ -132,6 +132,29 @@ func (b *PollingBase) releaseCardTrafficSyncLock(_ context.Context, cardID uint,
 	}
 }
 
+// ensureMissingTask 根据数据库最新卡状态，仅在对应分片队列缺失时补入任务。
+func (b *PollingBase) ensureMissingTask(ctx context.Context, cardID uint, taskType string) error {
+	card, err := b.iotCardStore.GetByID(ctx, cardID)
+	if err != nil {
+		return err
+	}
+
+	info, ok := b.configMgr.MergedTaskIntervals(card)[taskType]
+	if !ok || info.Interval <= 0 {
+		return nil
+	}
+
+	added, err := b.queueMgr.EnsureQueued(ctx, cardID, taskType, time.Now())
+	if err != nil {
+		return err
+	}
+	if added {
+		b.logger.Info("卡状态轮询补齐缺失套餐任务",
+			zap.Uint("card_id", cardID), zap.String("task_type", taskType))
+	}
+	return nil
+}
+
 // requeueCardAt 使用独立短超时上下文执行真正的 ZADD 重入队。
 func (b *PollingBase) requeueCardAt(cardID uint, taskType string, nextCheckAt time.Time) error {
 	ctx, cancel := pollingFallbackContext()

internal/task/polling_carddata_handler.go

@@ -67,11 +67,15 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) e
 	}
 	result, err := h.gateway.QueryFlow(ctx, &gateway.FlowQueryReq{CardNo: card.ICCID})
 	if err != nil {
-		_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
+		if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
+			return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 流量 Integration Log 失败", logErr)
+		}
 		return h.failAndRequeue(ctx, cardID, startedAt, "查询流量失败", err)
 	}
 	if strings.TrimSpace(result.ICCID) == "" {
-		_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
+		if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
+			return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 流量 Integration Log 失败", logErr)
+		}
 		return h.failAndRequeue(ctx, cardID, startedAt, "流量查询响应缺少 ICCID", nil)
 	}
 	if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil {

internal/task/polling_cardstatus_handler.go

@@ -66,11 +66,15 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task)
 	}
 	result, err := h.gateway.QueryCardStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID})
 	if err != nil {
-		_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
+		if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
+			return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 网络 Integration Log 失败", logErr)
+		}
 		return h.failAndRequeue(ctx, cardID, startedAt, "查询卡状态失败", err)
 	}
 	if strings.TrimSpace(result.ICCID) == "" {
-		_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
+		if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
+			return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 网络 Integration Log 失败", logErr)
+		}
 		return h.failAndRequeue(ctx, cardID, startedAt, "卡状态查询响应缺少 ICCID", nil)
 	}
 	if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil {
@@ -102,6 +106,12 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task)
 		h.base.logger.Info("独立卡命中风险状态，已关闭轮询", zap.Uint("card_id", cardID), zap.String("gateway_extend", decision.GatewayExtend))
 		return nil
 	}
+
+	// 卡状态任务仍正常但套餐任务丢失时，仅补入缺失项；不改写已有套餐任务的执行时间。
+	h.base.invalidateCardCache(ctx, cardID)
+	if err := h.base.ensureMissingTask(ctx, cardID, constants.TaskTypePollingPackage); err != nil {
+		h.base.logger.Warn("卡状态轮询补齐套餐任务失败", zap.Uint("card_id", cardID), zap.Error(err))
+	}
 	return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus)
 }
 

internal/task/polling_integration_log.go

@@ -38,11 +38,13 @@ func completeGatewayAttempt(ctx context.Context, repository *integrationlog.Repo
 	if repository == nil || attempt == nil {
 		return errors.New(errors.CodeInternalError, "Gateway Integration Log 尝试不存在")
 	}
+	completionCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+	defer cancel()
 	result := constants.IntegrationResultFailed
 	if success {
 		result = constants.IntegrationResultSuccess
 	}
-	_, err := repository.Complete(ctx, attempt.IntegrationID, integrationlog.Completion{
+	_, err := repository.Complete(completionCtx, attempt.IntegrationID, integrationlog.Completion{
 		Result: result, DurationMS: time.Since(startedAt).Milliseconds(), StateChanged: false,
 		ResponseSummary: map[string]any{"success": success},
 	})

internal/task/polling_realname_handler.go

@@ -58,11 +58,15 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, task *asynq.Task) e
 	}
 	result, err := h.gateway.QueryRealnameStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID})
 	if err != nil {
-		_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
+		if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
+			return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 实名 Integration Log 失败", logErr)
+		}
 		return h.failAndRequeue(ctx, cardID, startedAt, "查询实名状态失败", err)
 	}
 	if strings.TrimSpace(result.ICCID) == "" {
-		_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
+		if logErr := completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt); logErr != nil {
+			return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 实名 Integration Log 失败", logErr)
+		}
 		return h.failAndRequeue(ctx, cardID, startedAt, "实名查询响应缺少 ICCID", nil)
 	}
 	if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil {

openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-08-21

openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/design.md

@@ -0,0 +1,53 @@
+## Context
+
+见 proposal.md。现有实现以 Audit 和 Integration 归档对作为原子清理单元，且在生成 Integration 最终 revision 前要求整日不存在 `pending`。2026-08-18 的 458 条 Gateway 同步轮询遗留 pending 因而阻断约 314 万条 Integration Log 和后续日期的清理。
+
+## Goals / Non-Goals
+
+**Goals:**
+
+- 只保留 pending Integration Log 在线，尽快释放已终态日志和审计数据。
+- 保持删除前可恢复归档和逐批续跑语义。
+- 防止旧 Gateway 轮询处理器静默制造新的永久 pending。
+
+**Non-Goals:**
+
+- 不直接修改历史 pending 的结果或删除它们。
+- 不改变 Integration Log 的写入频率、JSONL 格式、对象存储格式或数据库 Schema。
+- 不在 Worker 内执行 VACUUM 或表重写。
+
+## Decisions
+
+### 归档覆盖全日，删除只匹配终态
+
+仍按创建日生成包含全部记录的归档 revision；移除“存在 pending 即不得最终归档”的门禁。留存校验确认当前在线集合与该 revision 一致后，删除条件限定为 `result <> 'pending'`。这样每条已删除数据都在对象存储中存在恢复副本，而 pending 永不被删除。
+
+不按结果分别创建归档文件：会改变既有归档格式和恢复语义，且没有必要。
+
+### Audit 与 Integration 分来源推进
+
+Audit 归档校验成功后独立完成 Audit 资源和事件删除；Integration 的 pending 或残留不再使 Audit 退化为未清理。每个来源仍按自身失败日阻断该来源后续日期，避免跨越该来源的归档或校验故障。
+
+不继续要求两来源同日原子清理：该设计正是容量积压的根因，且两类数据已有独立的归档账本和清理断点。
+
+### 以剩余可删除记录决定 Integration 续跑
+
+已经删除过终态记录的日期不因只剩 pending 而反复阻断日期扫描。若旧 pending 后续进入终态，则该日期重新生成 revision、验证当前在线集合，并删除新增终态记录。查询仅检查可删除终态记录和对应清理断点，避免对只含 pending 的大历史范围反复做全表工作。
+
+### Gateway 失败分支不吞终结错误
+
+三个旧轮询 Handler 对已创建的尝试使用受控收口：终结写入失败返回任务错误并保留重试信号；不再使用 `_ = completeGatewayAttempt(...)` 静默丢弃错误。任务取消时仍尝试使用不受取消影响的短生命周期上下文终结该条日志。
+
+## Risks / Trade-offs
+
+- [终态日志在归档和删除间新增或变化] → 归档后重新统计和校验；不一致时不删除并在下次重建 revision。
+- [pending 永久不终结] → 仅保留这些记录，不能阻止终态数据和审计数据释放；通过调查接口和留存日志排查。
+- [PostgreSQL 文件空间不立即缩小] → DELETE 回收空间供后续写入复用；维护者按运行手册评估 VACUUM，不由 Worker 自动执行。
+- [旧 pending 终结后遗漏清理] → 续跑选择条件识别清理后进入终态的记录，并重建该日 revision。
+
+## Migration Plan
+
+1. 发布 Worker 二进制，保持既有物理清理开关状态。
+2. 若当前开关关闭，维护者在低峰期启用后重启调度 Worker。
+3. 观察 `logs/audit-retention.log`、`tb_log_archive_run` 清理断点和数据库磁盘指标；8 月 18 日的已终态记录应先被清理，458 条 pending 保留。
+4. 若归档或删除异常，关闭清理开关并重启 Worker；已删除记录通过对应归档 revision 恢复，未删除记录仍在线。

openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/proposal.md

@@ -0,0 +1,26 @@
+## Why
+
+日留存将某日存在的任意 `pending` Integration Log 视为整日删除的阻断条件。Gateway 同步轮询的遗留 pending 记录使 2026-08-18 及所有后续日期无法清理，在线日志持续积压并带来数据库容量风险。
+
+## What Changes
+
+- 日留存继续归档全天 Integration Log，但只将已进入公开终态的记录纳入可清理集合。
+- `pending` 记录保留在线，且不得阻断同日终态记录和后续日期的归档、验证与清理。
+- 已完成的 Gateway 轮询尝试在终结日志写入失败时不得静默忽略错误，避免新增永久 pending 记录。
+
+## Capabilities
+
+### New Capabilities
+
+无。
+
+### Modified Capabilities
+
+- `external-integration`: 外部交互日志日留存从“整日无 pending 才能清理”改为“仅清理已终态记录并保留 pending”。
+- `operations-audit`: 日留存不因 Integration Log 的 pending 记录阻断审计事件和资源快照的连续清理。
+
+## Impact
+
+- 代码：`internal/application/auditarchive/`、旧 Gateway 轮询任务处理器。
+- Worker：日留存会释放已终态的在线日志；pending 继续可调查和补偿。
+- 数据库：无迁移，不执行直接 SQL 数据修复。

openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/specs/external-integration/spec.md

@@ -0,0 +1,59 @@
+## MODIFIED Requirements
+
+### Requirement: 外部交互日志逐日物理留存
+
+系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位物理留存 `tb_integration_log`。删除一条已终态在线外部交互日志前，系统 MUST 已为其创建日生成可读取、可验证的归档对象和清单；该归档 MUST 包含该日当时全部外部交互日志，并验证日期范围、记录数量和校验摘要可作为恢复凭证。`pending` 记录 MUST 保留在线，不得被物理删除，且不得阻断同日已终态记录或更晚日期的归档、验证与物理清理。每次执行 SHALL 清理所有满足归档校验条件的已终态记录；某一来源的归档或校验失败时，系统 MUST 保留该来源当天及更晚日期的在线外部交互日志。
+
+#### Scenario: 最终归档通过后清理在线外部交互日志
+- **GIVEN** 某已结束自然日的外部交互归档对象和清单校验成功，且存在已终态在线外部交互日志
+- **WHEN** 日留存任务处理该日期
+- **THEN** 系统分批删除该日已终态的在线外部交互日志并保留归档对象
+
+#### Scenario: 存在待完成外部交互日志
+- **GIVEN** 某归档日仍存在待完成的外部交互日志
+- **WHEN** 日留存任务尝试处理该日期
+- **THEN** 系统保留该 pending 日志，但不得停止同日终态记录和更晚日期的清理
+
+#### Scenario: 含 pending 的日期清理终态日志
+- **GIVEN** 某已结束自然日同时存在已终态和 `pending` 的外部交互日志，且该日归档对象和清单校验成功
+- **WHEN** 日留存任务处理该日期
+- **THEN** 系统分批删除该日已终态的在线外部交互日志，保留 `pending` 记录，并保留归档对象
+
+#### Scenario: 历史积压按日期连续补清
+- **GIVEN** 存在多个昨天及更早日期尚未完成物理清理，且这些日期各自存在满足归档校验条件的已终态外部交互日志
+- **WHEN** 日留存任务执行
+- **THEN** 系统按日期从早到晚清理全部符合条件的终态日志
+
+#### Scenario: pending 不阻断后续日期
+- **GIVEN** 较早归档日只剩 `pending` 外部交互日志，且更晚归档日存在满足归档校验条件的已终态外部交互日志
+- **WHEN** 日留存任务执行
+- **THEN** 系统继续处理更晚归档日的已终态外部交互日志，不因较早日期的 pending 停止日期推进
+
+#### Scenario: pending 后续终结
+- **GIVEN** 某归档日的 pending 外部交互日志在该日首次留存后进入公开终态
+- **WHEN** 后续日留存任务执行
+- **THEN** 系统为包含该记录的当前在线集合重新验证归档，并删除该已终态记录
+
+#### Scenario: 归档或校验失败保留同来源后续数据
+- **GIVEN** 某归档日的外部交互归档缺失、归档校验失败或清单与在线数据不一致
+- **WHEN** 日留存任务处理该日期
+- **THEN** 系统不删除该日及更晚日期的在线外部交互日志，并将失败信息写入独立日留存日志
+
+### Requirement: 外部失败边界
+
+系统 SHALL 将第三方超时、渠道错误和无效响应转换为当前稳定的系统错误；已接入外部交互日志的渠道同时保留脱敏结果。Gateway 同步轮询在已建立 pending 外部交互日志后，无论查询成功、失败或任务上下文取消，MUST 尝试将该尝试终结为公开终态；终结失败 MUST 作为任务失败返回或记录为可重试错误，不得静默忽略。
+
+#### Scenario: 外部失败边界
+- **GIVEN** 外部系统超时或返回失败
+- **WHEN** 调用依赖该系统的操作
+- **THEN** 客户端收到当前稳定错误；已接入外部交互日志的调用记录脱敏渠道结果
+
+#### Scenario: Gateway 轮询查询失败
+- **GIVEN** Gateway 同步轮询已建立 pending 外部交互日志且查询返回错误
+- **WHEN** 轮询任务处理该错误
+- **THEN** 系统将该日志终结为失败或结果未知的公开终态，并按既有策略重新入队
+
+#### Scenario: Gateway 轮询终结日志失败
+- **GIVEN** Gateway 同步轮询已建立 pending 外部交互日志且终结写入失败
+- **WHEN** 轮询任务结束该次尝试
+- **THEN** 系统记录并返回该终结失败，不得把该错误静默丢弃

openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/specs/operations-audit/spec.md

@@ -0,0 +1,25 @@
+## MODIFIED Requirements
+
+### Requirement: 审计在线数据逐日物理留存
+
+系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位处理统一审计事件及其资源快照的在线留存。每次留存执行 SHALL 从最早尚未完成清理的归档日开始，连续处理至昨天；Integration Log 的 `pending` 记录不得阻断已通过完整性校验的审计事件及资源快照清理。任一审计归档日未通过完整性校验时，系统 MUST 保留该日及其后续日期的在线审计数据，且不得将它们标记为已清理。系统 MUST 先删除该日的审计资源快照，再删除该日的审计事件，并保留对象存储归档对象及清单作为恢复凭证。
+
+#### Scenario: 已验证日期完成物理清理
+- **GIVEN** 某已结束自然日的审计归档成功，归档对象和清单可读取且与在线记录范围、数量和校验摘要一致
+- **WHEN** 日留存任务处理该日期
+- **THEN** 系统分批删除该日在线审计资源快照和审计事件，并将该归档日标记为已清理
+
+#### Scenario: Integration Log pending 不阻断审计清理
+- **GIVEN** 某已结束自然日的审计归档已通过完整性校验，且同日存在 pending Integration Log
+- **WHEN** 日留存任务处理该日期
+- **THEN** 系统仍完成该日在线审计资源快照和审计事件的物理清理
+
+#### Scenario: 归档校验失败阻断连续清理
+- **GIVEN** 某尚未清理的审计归档日缺少归档、归档校验失败或清单与在线数据不一致
+- **WHEN** 日留存任务处理该日期
+- **THEN** 系统不删除该日或更晚日期的在线审计数据，并将失败信息写入独立日留存日志
+
+#### Scenario: 失败后从未完成日期续跑
+- **GIVEN** 日留存任务在某个审计归档日失败，较早日期已经标记为已清理
+- **WHEN** 后续日留存任务再次执行
+- **THEN** 系统从最早未完成清理的审计归档日继续处理，不重复删除已完成日期的数据

openspec/changes/archive/2026-08-21-fix-daily-retention-pending-cleanup/tasks.md

@@ -0,0 +1,15 @@
+## 1. 日留存按终态清理
+
+- [x] 1.1 调整 Integration Log 归档与留存校验，使归档保留全日快照、pending 不再阻断终态记录清理。
+- [x] 1.2 将 Integration Log 删除条件限制为公开终态，并支持同一日期遗留 pending 后续终结时重建归档 revision 和续跑删除。
+- [x] 1.3 让 Audit 留存独立于同日 Integration pending 完成归档校验和物理清理，并保持各来源的失败断点。
+
+## 2. Gateway 轮询收口
+
+- [x] 2.1 修改旧卡状态、流量和实名轮询 Handler，使已创建尝试的终结失败不被忽略，并在取消上下文下仍尝试终结。
+
+## 3. 验证与运行说明
+
+- [x] 3.1 格式化修改的 Go 文件并构建 `./cmd/worker`。
+- [ ] 3.2 以可控演练或最小验证确认：含 pending 的日期删除终态日志并继续后续日期，pending 本身保留在线。
+- [x] 3.3 更新生产运行说明，记录发布 Worker、启用既有清理开关、观察留存日志与 PostgreSQL 空间复用的步骤。

openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-08-26

openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/design.md

@@ -0,0 +1,32 @@
+## Context
+
+批量订购逐行调用后台订单创建，并在全部行处理结束后于同一事务保存任务结果和完成审计。当前完成审计以关联订单审计事件数量充当成功行数；该数量会包含失败记录和重试记录，可能大于输入总行数，违反审计表的批量统计约束并回滚任务完成状态。
+
+## Goals / Non-Goals
+
+**Goals:**
+
+- 使任务完成审计使用本次处理结果的业务成功数和失败数。
+- 保持任务结果、汇总计数和完成审计在同一事务提交。
+
+**Non-Goals:**
+
+- 不改变单笔订单、钱包余额不足或重复输入的业务规则。
+- 不恢复或重放历史任务；历史数据由维护者按生产运维流程处置。
+
+## Decisions
+
+- 以 `processRows` 返回的 `successCount` 和 `failCount` 作为任务完成审计的唯一批量统计来源。这些值与将写入任务记录的逐行结果来自同一次处理，满足审计表的计数不变量。
+- 移除对关联订单审计事件数量的查询。审计事件数量会因失败记录和重试而膨胀，不能表示批量业务成功数。
+
+## Risks / Trade-offs
+
+- [历史待处理任务不自动恢复] → 保持现有状态，避免对已扣款订单重复执行；由维护者核对后受控结案。
+- [完成审计仍依赖事务写入成功] → 使用与任务结果一致的统计值，消除本次约束失败原因，同时保留既有原子性。
+
+## Migration Plan
+
+1. 发布 API/Worker 中的修复后二进制，重点更新 Worker。
+2. 在隔离环境验证部分成功任务返回完成状态与一致计数。
+3. 生产历史任务由维护者在核对订单和源文件后手工结案；不重投原 Asynq 消息。
+4. 若发布异常，回滚 Worker 二进制；该修复不含迁移。

openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/proposal.md

@@ -0,0 +1,23 @@
+## Why
+
+资产套餐批量订购在部分行因钱包余额不足或输入重复而失败时，完成审计将审计事件条数误作成功行数，可能违反审计批量统计约束并回滚任务终态。已实际创建的订单因此在任务查询中显示为待处理且无结果，既不能反映业务结果，也有误重试风险。
+
+## What Changes
+
+- 批量订购任务完成时，使用本次逐行处理得到的成功数和失败数写入完成审计。
+- 确保部分成功、部分失败的批量订购任务能持久化为已完成，并可查询真实汇总和逐行结果。
+
+## Capabilities
+
+### New Capabilities
+
+无。
+
+### Modified Capabilities
+
+- `package-lifecycle`: 异步批量订购任务必须持久化并返回逐行处理后的终态统计，即使部分资产因余额不足或输入校验失败而未能购包。
+
+## Impact
+
+- `internal/task/asset_package_batch_order.go` 中的批量订购完成与审计统计。
+- 不新增 API、数据表或依赖；既有批量订购详情将正确展示部分完成结果。

openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/specs/package-lifecycle/spec.md

@@ -0,0 +1,17 @@
+## MODIFIED Requirements
+
+### Requirement: 批量操作可追踪
+
+系统 SHALL 为同步批量分配和调价直接返回处理结果；对异步批量订购返回任务标识并提供状态查询。异步批量订购完成后，系统 MUST 持久化每个输入行的成功或失败结果及与其一致的总数、成功数和失败数；部分资产因余额不足、资产校验或重复输入失败不得阻止任务进入完成终态。
+
+#### Scenario: 批量操作可追踪
+
+- **GIVEN** 操作者提交非空且有权处理的资源集合
+- **WHEN** 创建批量操作
+- **THEN** 同步操作直接返回结果；异步订购返回任务标识且可查询处理状态
+
+#### Scenario: 批量订购部分失败后查询结果
+
+- **GIVEN** 异步批量订购中的部分资产已成功创建订单，其他资产因钱包余额不足或输入重复失败
+- **WHEN** Worker 完成全部输入行的处理
+- **THEN** 任务状态为已完成，逐行结果保留成功订单与失败原因，且总数等于成功数与失败数之和

openspec/changes/archive/2026-08-26-fix-batch-order-task-completion-audit-count/tasks.md

@@ -0,0 +1,8 @@
+## 1. 批量订购完成统计修复
+
+- [x] 1.1 修改批量订购完成审计，使其使用逐行处理产生的成功数和失败数，不再统计关联审计事件数量。
+- [x] 1.2 格式化修改的 Go 文件并构建 Worker，验证部分成功任务的完成审计统计满足任务总行数约束。
+
+## 2. 规格验证
+
+- [x] 2.1 运行 OpenSpec 校验，确认变更工件一致。

openspec/changes/archive/2026-08-29-fix-package-expiry-stop-race/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-08-29

openspec/changes/archive/2026-08-29-fix-package-expiry-stop-race/design.md

@@ -0,0 +1,66 @@
+## Context
+
+见 proposal.md。当前到期处理在事务提交后投递异步套餐激活任务，同时以 goroutine 发起停机检查；两条路径没有共同顺序或共享状态结论。生产中已观察到新套餐激活后 1.9 至 15.7 秒仍写入 `no_package` 停机。
+
+现有 `ActivationService` 已负责待生效主套餐激活及激活后的复机回调；`StopResumeService` 已集中停复机条件。保留这两个职责边界，不增加新的状态表或外部依赖。
+
+## Goals / Non-Goals
+
+**Goals:**
+- 同一载体到期接续期间不以过时权益快照停机。
+- 没有后续可生效套餐时保留现有停机语义。
+- 激活投递或执行暂时失败时宁可交由既有恢复/轮询兜底，不误判为无套餐。
+- 给维护者提供基于当前事实的只读候选筛选与受控补偿流程。
+
+**Non-Goals:**
+- 不新增自动批量复机、管理接口或数据库迁移。
+- 不把 Redis/Asynq 运行排障改造成业务状态机。
+- 不改变手动停机、运营商风险停机或实名策略语义。
+
+## Decisions
+
+### 1. 将停机决策后移到套餐激活任务的最终重新评估
+
+到期处理找到后续待生效主套餐时，只投递激活任务，不并发启动停机检查。激活任务完成后以该载体的最新套餐状态调用现有停机评估：成功激活时不触发无套餐停机；队首套餐因实名等业务条件不能激活时，才按当前事实评估停机。
+
+这样复用现有激活锁、套餐条件和停复机服务，不让两个异步流程各自读取一次状态。
+
+备选方案：在到期处理内同步完成套餐激活。未采用，因为到期扫描单轮最多处理大量套餐，同步串行激活会放大调度 Worker 的数据库和 Redis 锁持有时间。
+
+### 2. 激活投递失败不是无套餐结论
+
+仅在明确不存在后续主套餐时，允许到期处理立即发起停机评估。存在后续套餐但投递失败时记录错误并保留当前网络状态，由既有孤儿套餐恢复扫描和套餐轮询重试；不得回退到并发停机。
+
+备选方案：投递失败立即停机。未采用，因为它重新引入“接续结果未知被当作无套餐”的错误路径。
+
+### 3. 存量恢复与代码修复分离
+
+代码发布后，维护者用只读查询筛选“有效主套餐、未耗尽、实名满足、`no_package` 停机”的候选卡；每张或受控小批次先查询运营商实时状态，再调用既有复机能力。补偿不直接更新数据库，也不自动对全部候选执行。
+
+备选方案：发布时自动批量复机全部候选。未采用，因为本地状态不足以代替运营商风险、销户和人工业务判断。
+
+### 4. 轮询只作兜底并单独核验运行覆盖
+
+停机卡配置的 `polling:package` 仍是后备恢复机制，不能作为避免竞态的主路径。维护者分别核验调度 Worker 心跳、Redis 分片队列深度、Asynq 队列积压与消费者吞吐；这些运行指标只用于定位未补偿原因。
+
+### 5. 卡状态轮询仅补齐缺失的套餐任务
+
+生产日志确认存在卡状态和流量任务仍持续执行、但同一卡的 `polling:package` 队列项缺失的情形。卡状态轮询成功且未命中风险停机后，按数据库最新卡状态重新匹配轮询配置；若该卡应参与套餐检查，则以 Redis `ZADD NX` 仅在对应分片 Sorted Set 不存在该卡时立即补入套餐任务。已存在任务不得覆盖其 score，也不新增重复任务。
+
+该做法不将卡状态轮询变为停复机决策路径：它只修复缺失的套餐调度项，最终仍由既有 `PollingPackageHandler → EvaluateAndAct` 判断权益、实名、停机原因和运营商结果。
+
+## Risks / Trade-offs
+
+- [激活任务投递失败时卡暂不因套餐到期自动停机] → 保留孤儿恢复扫描和套餐轮询；记录失败并在运行排障中核验任务恢复。
+- [激活任务内增加最终停机评估] → 保持幂等条件和现有 Redis 锁；同一载体状态变化必须重读数据库。
+- [存量候选包含运营商侧不可复机卡] → 补偿前逐卡或受控小批查询运营商状态，拒绝风险停机、销户或其他非本地套餐原因。
+- [轮询吞吐不足延迟兜底] → 发布前后分别记录队列深度、消费速率和最近检查时间分布，不以单次数据库字段判断 Worker 健康。
+- [卡状态轮询补齐套餐任务] → 仅使用 `ZADD NX` 补入缺失成员，不改写已有任务的执行时间；最终复机仍受既有权益、实名、停机原因和运营商校验约束。
+
+## Migration Plan
+
+1. 在隔离环境验证“旧套餐到期 + 后续套餐可激活”不会产生停机调用，及“无后续/不可激活”仍进入停机评估。
+2. 按生产运行说明发布 Worker；不需要数据库迁移。
+3. 维护者观察一轮套餐到期处理、激活任务与轮询队列，确认没有新的“有效套餐 + `no_package` 停机”记录。
+4. 使用只读筛选得到存量候选，按受控批次核验运营商状态后调用既有复机；失败项保留失败原因并人工处理。
+5. 若发布后出现意外停机语义，回滚 Worker 二进制；存量补偿不得直接写库。

openspec/changes/archive/2026-08-29-fix-package-expiry-stop-race/proposal.md

@@ -0,0 +1,28 @@
+## Why
+
+生产诊断发现套餐接续期间会并发投递下一套餐激活任务和无套餐停机检查。停机检查可在新套餐生效后才完成并回写停机，造成卡已具备有效套餐、实名和未耗尽流量却仍以 `no_package` 停机；当前已检出 117 张此类卡，且 13 张在新套餐生效后 120 秒内被停机。
+
+## What Changes
+
+- 将主套餐到期后的“接续下一套餐”和“无有效套餐停机”收敛为同一条按载体串行的决策流程，禁止待生效套餐接续期间以旧快照发起停机。
+- 只有在确认不存在可生效的后续主套餐，或后续套餐经业务校验确定不能生效后，才执行无套餐停机检查。
+- 套餐接续完成后重新依据当前卡与套餐事实判断停复机，确保有效套餐卡不遗留 `no_package` 停机状态。
+- 为既有异常卡提供只读筛选条件和受控人工补偿步骤；补偿前必须重新核验套餐、流量、实名和运营商实时状态，不引入自动批量复机。
+- 排查并记录轮询调度、入队与消费覆盖情况；当卡状态轮询仍正常但 `polling:package` 队列项缺失时，仅补入缺失的套餐任务，确保套餐轮询仍可作为即时接续失败后的兜底，不将 Redis/Asynq 运行状态误判为业务终态。
+
+## Capabilities
+
+### New Capabilities
+
+- 无。
+
+### Modified Capabilities
+
+- `package-lifecycle`: 主套餐到期、后续套餐接续及无套餐停机必须基于同一载体的最新权益事实，避免接续成功后错误停机。
+
+## Impact
+
+- 受影响代码：`internal/polling/package_activation_handler.go`、`internal/polling/queue_manager.go`、`internal/service/package/activation_service.go`、`internal/service/iot_card/stop_resume_service.go` 及相关轮询任务装配。
+- 受影响运行单元：调度 Worker、Asynq 套餐激活任务、套餐轮询任务。
+- 不新增 HTTP API、数据库 Schema 或第三方依赖。
+- 生产补偿仅由维护者执行受控外部复机；Agent 仅提供只读诊断与候选清单。

openspec/changes/archive/2026-08-29-fix-package-expiry-stop-race/specs/package-lifecycle/spec.md

@@ -0,0 +1,35 @@
+## MODIFIED Requirements
+
+### Requirement: 套餐状态流转
+
+系统 SHALL 按当前套餐和套餐使用状态控制上架、订购、激活、失效与到期处理。主套餐到期时，系统 MUST 先确定同一载体是否存在待生效的后续主套餐：存在时，后续套餐激活与停复机重新评估 MUST 由同一条顺序流程完成；系统 MUST NOT 依据后续套餐激活前的无套餐快照发起停机。后续套餐成功生效后，系统 MUST 依据最新套餐、流量和实名事实重新判断卡网络状态，且不得遗留 `no_package` 停机。不存在后续套餐或后续套餐经业务校验不能生效时，系统 SHALL 按现有停机规则评估卡状态。后续套餐激活结果未知或任务投递失败不得被当作无后续套餐处理并据此停机，系统 SHALL 保留既有激活恢复与轮询兜底路径。
+
+#### Scenario: 套餐状态流转
+
+- **GIVEN** 套餐或使用记录处于允许的前置状态
+- **WHEN** 执行状态操作
+- **THEN** 仅发生一次允许的状态变化；不满足前置状态时返回业务错误
+
+#### Scenario: 到期主套餐接续后续套餐
+
+- **GIVEN** 某载体的当前主套餐到期，且存在满足激活条件的待生效后续主套餐
+- **WHEN** 系统处理该主套餐到期
+- **THEN** 系统先完成后续套餐激活并按最新权益事实重新评估停复机，且不得因到期前的无套餐快照对该载体发起 `no_package` 停机
+
+#### Scenario: 到期主套餐无后续可生效套餐
+
+- **GIVEN** 某载体的当前主套餐到期，且不存在后续主套餐或队首后续主套餐不满足激活条件
+- **WHEN** 系统完成该套餐到期处理
+- **THEN** 系统按当前套餐、流量和实名事实执行既有停机评估
+
+#### Scenario: 后续套餐激活结果未知
+
+- **GIVEN** 某载体的当前主套餐到期，存在待生效后续主套餐，但激活任务投递或执行结果暂时未知
+- **WHEN** 系统处理该套餐到期
+- **THEN** 系统不得将该未知结果视为不存在后续套餐而依据旧快照发起停机，并保留既有激活恢复与套餐轮询兜底
+
+#### Scenario: 卡状态轮询发现缺失的套餐任务
+
+- **GIVEN** 启用轮询的卡匹配套餐检查配置，且其 `polling:package` 分片队列项因异常缺失
+- **WHEN** 卡状态轮询成功完成且未命中风险停机
+- **THEN** 系统基于最新卡状态仅补入缺失的套餐任务，不改写已存在套餐任务的执行时间；后续套餐任务仍按既有停复机条件评估该卡

openspec/changes/archive/2026-08-29-fix-package-expiry-stop-race/tasks.md

@@ -0,0 +1,21 @@
+## 1. 到期接续顺序
+
+- [x] 1.1 调整主套餐到期处理：明确区分“无后续主套餐”“已提交后续激活”和“后续激活投递失败”，仅无后续主套餐可立即进入停机评估。
+- [x] 1.2 调整套餐激活任务：在后续套餐激活完成或确认暂不能激活后，重读载体当前权益并调用既有停机评估；移除与激活任务并发的停机 goroutine。
+- [x] 1.3 保持激活后的既有复机回调、Redis 锁、审计与 Integration Log 语义，不新增状态表、接口或自动批量复机。
+
+## 2. 运行兜底与可观测性
+
+- [x] 2.1 核对停机卡匹配的 `polling:package` 配置及生命周期回调入队路径，修正本次改动涉及的入队或重排遗漏，确保其仅作为接续异常后的兜底。
+- [x] 2.2 为激活任务的最终停机评估保留可按载体、套餐使用记录和结果关联的中文结构化日志；不得记录敏感外部载荷。
+
+## 3. 验证
+
+- [ ] 3.1 在隔离环境人工演练“旧套餐到期且后续套餐可激活”“无后续套餐”“后续套餐暂不能激活”“激活投递失败”四种路径，核对停复机调用与套餐最终状态。
+- [x] 3.2 执行 `gofmt -w <changed-go-files>`、`go build ./cmd/api ./cmd/worker`、`openspec validate fix-package-expiry-stop-race --strict`。
+
+## 4. 生产受控恢复
+
+- [x] 4.1 维护者通过只读查询重新筛选“有效主套餐、流量未耗尽、实名满足且 `no_package` 停机”的候选卡，并记录筛选时点与数量。
+- [ ] 4.2 维护者按受控小批次查询运营商实时状态，仅对可复机卡调用既有复机能力；不得直接更新卡网络状态或停机原因。
+- [ ] 4.3 维护者核验调度 Worker 心跳、Redis 分片队列、Asynq 队列和消费者吞吐，并在发布后观察是否仍产生“有效套餐 + `no_package` 停机”记录。

openspec/specs/external-integration/spec.md

@@ -28,7 +28,7 @@
 
 ### Requirement: 外部失败边界
 
-系统 SHALL 将第三方超时、渠道错误和无效响应转换为当前稳定的系统错误；已接入外部交互日志的渠道同时保留脱敏结果。
+系统 SHALL 将第三方超时、渠道错误和无效响应转换为当前稳定的系统错误；已接入外部交互日志的渠道同时保留脱敏结果。Gateway 同步轮询在已建立 pending 外部交互日志后，无论查询成功、失败或任务上下文取消，MUST 尝试将该尝试终结为公开终态；终结失败 MUST 作为任务失败返回或记录为可重试错误，不得静默忽略。
 
 #### Scenario: 外部失败边界
 
@@ -36,6 +36,18 @@
 - **WHEN** 调用依赖该系统的操作
 - **THEN** 客户端收到当前稳定错误；已接入外部交互日志的调用记录脱敏渠道结果
 
+#### Scenario: Gateway 轮询查询失败
+
+- **GIVEN** Gateway 同步轮询已建立 pending 外部交互日志且查询返回错误
+- **WHEN** 轮询任务处理该错误
+- **THEN** 系统将该日志终结为失败或结果未知的公开终态，并按既有策略重新入队
+
+#### Scenario: Gateway 轮询终结日志失败
+
+- **GIVEN** Gateway 同步轮询已建立 pending 外部交互日志且终结写入失败
+- **WHEN** 轮询任务结束该次尝试
+- **THEN** 系统记录并返回该终结失败，不得把该错误静默丢弃
+
 ### Requirement: 外部调用重试边界
 
 系统 SHALL 仅对 Gateway 客户端超时、连接失败和 DNS 失败自动重试，默认最多重试两次且每次重新签名；HTTP 非 200、响应解析失败、Gateway 业务失败和调用方取消不重试。企业微信审批只有确认尚未调用提交接口的失败可释放后重试，提交结果未知时不得盲目重建审批。
@@ -103,22 +115,42 @@
 
 ### Requirement: 外部交互日志逐日物理留存
 
-系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位物理留存 `tb_integration_log`。删除某日在线外部交互日志前，系统 MUST 为该日生成最终归档版本，确认不存在待完成记录，并验证归档对象、清单、日期范围、记录数量和校验摘要可作为恢复凭证。每次执行 SHALL 从最早尚未完成清理的归档日连续处理至昨天；任一日期不能形成或验证最终归档时，系统 MUST 不删除该日及更晚日期的在线外部交互日志。
+系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位物理留存 `tb_integration_log`。删除一条已终态在线外部交互日志前，系统 MUST 已为其创建日生成可读取、可验证的归档对象和清单；该归档 MUST 包含该日当时全部外部交互日志，并验证日期范围、记录数量和校验摘要可作为恢复凭证。`pending` 记录 MUST 保留在线，不得被物理删除，且不得阻断同日已终态记录或更晚日期的归档、验证与物理清理。每次执行 SHALL 清理所有满足归档校验条件的已终态记录；某一来源的归档或校验失败时，系统 MUST 保留该来源当天及更晚日期的在线外部交互日志。
 
 #### Scenario: 最终归档通过后清理在线外部交互日志
-- **GIVEN** 某已结束自然日的外部交互日志均已进入终态，且该日最终归档对象和清单校验成功
+- **GIVEN** 某已结束自然日的外部交互归档对象和清单校验成功，且存在已终态在线外部交互日志
 - **WHEN** 日留存任务处理该日期
-- **THEN** 系统分批删除该日的在线外部交互日志并将该归档日标记为已清理，归档对象保持可用
+- **THEN** 系统分批删除该日已终态的在线外部交互日志并保留归档对象
 
 #### Scenario: 存在待完成外部交互日志
 - **GIVEN** 某归档日仍存在待完成的外部交互日志
 - **WHEN** 日留存任务尝试处理该日期
-- **THEN** 系统不删除该日及更晚日期的在线外部交互日志，并在独立日留存日志中记录该日期未形成最终归档的原因
+- **THEN** 系统保留该 pending 日志，但不得停止同日终态记录和更晚日期的清理
+
+#### Scenario: 含 pending 的日期清理终态日志
+- **GIVEN** 某已结束自然日同时存在已终态和 `pending` 的外部交互日志，且该日归档对象和清单校验成功
+- **WHEN** 日留存任务处理该日期
+- **THEN** 系统分批删除该日已终态的在线外部交互日志，保留 `pending` 记录，并保留归档对象
 
 #### Scenario: 历史积压按日期连续补清
-- **GIVEN** 存在多个昨天及更早日期尚未完成物理清理
-- **WHEN** 日留存任务执行且这些日期依次通过最终归档和完整性校验
-- **THEN** 系统按日期从早到晚清理全部连续合格日期
+- **GIVEN** 存在多个昨天及更早日期尚未完成物理清理，且这些日期各自存在满足归档校验条件的已终态外部交互日志
+- **WHEN** 日留存任务执行
+- **THEN** 系统按日期从早到晚清理全部符合条件的终态日志
+
+#### Scenario: pending 不阻断后续日期
+- **GIVEN** 较早归档日只剩 `pending` 外部交互日志，且更晚归档日存在满足归档校验条件的已终态外部交互日志
+- **WHEN** 日留存任务执行
+- **THEN** 系统继续处理更晚归档日的已终态外部交互日志，不因较早日期的 pending 停止日期推进
+
+#### Scenario: pending 后续终结
+- **GIVEN** 某归档日的 pending 外部交互日志在该日首次留存后进入公开终态
+- **WHEN** 后续日留存任务执行
+- **THEN** 系统为包含该记录的当前在线集合重新验证归档，并删除该已终态记录
+
+#### Scenario: 归档或校验失败保留同来源后续数据
+- **GIVEN** 某归档日的外部交互归档缺失、归档校验失败或清单与在线数据不一致
+- **WHEN** 日留存任务处理该日期
+- **THEN** 系统不删除该日及更晚日期的在线外部交互日志，并将失败信息写入独立日留存日志
 
 ## 可达操作索引
 

openspec/specs/operations-audit/spec.md

@@ -48,13 +48,18 @@
 
 ### Requirement: 审计在线数据逐日物理留存
 
-系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位处理统一审计事件及其资源快照的在线留存。每次留存执行 SHALL 从最早尚未完成清理的归档日开始，连续处理至昨天；任一归档日未通过完整性校验时，系统 MUST 保留该日及其后续日期的在线审计数据，且不得将它们标记为已清理。系统 MUST 先删除该日的审计资源快照，再删除该日的审计事件，并保留对象存储归档对象及清单作为恢复凭证。
+系统 SHALL 以 Asia/Shanghai 已结束的自然日为单位处理统一审计事件及其资源快照的在线留存。每次留存执行 SHALL 从最早尚未完成清理的归档日开始，连续处理至昨天；Integration Log 的 `pending` 记录不得阻断已通过完整性校验的审计事件及资源快照清理。任一审计归档日未通过完整性校验时，系统 MUST 保留该日及其后续日期的在线审计数据，且不得将它们标记为已清理。系统 MUST 先删除该日的审计资源快照，再删除该日的审计事件，并保留对象存储归档对象及清单作为恢复凭证。
 
 #### Scenario: 已验证日期完成物理清理
 - **GIVEN** 某已结束自然日的审计归档成功，归档对象和清单可读取且与在线记录范围、数量和校验摘要一致
 - **WHEN** 日留存任务处理该日期
 - **THEN** 系统分批删除该日在线审计资源快照和审计事件，并将该归档日标记为已清理
 
+#### Scenario: Integration Log pending 不阻断审计清理
+- **GIVEN** 某已结束自然日的审计归档已通过完整性校验，且同日存在 pending Integration Log
+- **WHEN** 日留存任务处理该日期
+- **THEN** 系统仍完成该日在线审计资源快照和审计事件的物理清理
+
 #### Scenario: 归档校验失败阻断连续清理
 - **GIVEN** 某尚未清理的归档日缺少归档、归档校验失败或清单与在线数据不一致
 - **WHEN** 日留存任务处理该日期

openspec/specs/package-lifecycle/spec.md

@@ -8,7 +8,7 @@
 
 ### Requirement: 套餐状态流转
 
-系统 SHALL 按当前套餐和套餐使用状态控制上架、订购、激活、失效与到期处理。
+系统 SHALL 按当前套餐和套餐使用状态控制上架、订购、激活、失效与到期处理。主套餐到期时，系统 MUST 先确定同一载体是否存在待生效的后续主套餐：存在时，后续套餐激活与停复机重新评估 MUST 由同一条顺序流程完成；系统 MUST NOT 依据后续套餐激活前的无套餐快照发起停机。后续套餐成功生效后，系统 MUST 依据最新套餐、流量和实名事实重新判断卡网络状态，且不得遗留 `no_package` 停机。不存在后续套餐或后续套餐经业务校验不能生效时，系统 SHALL 按现有停机规则评估卡状态。后续套餐激活结果未知或任务投递失败不得被当作无后续套餐处理并据此停机，系统 SHALL 保留既有激活恢复与轮询兜底路径。
 
 #### Scenario: 套餐状态流转
 
@@ -16,9 +16,33 @@
 - **WHEN** 执行状态操作
 - **THEN** 仅发生一次允许的状态变化；不满足前置状态时返回业务错误
 
+#### Scenario: 到期主套餐接续后续套餐
+
+- **GIVEN** 某载体的当前主套餐到期，且存在满足激活条件的待生效后续主套餐
+- **WHEN** 系统处理该主套餐到期
+- **THEN** 系统先完成后续套餐激活并按最新权益事实重新评估停复机，且不得因到期前的无套餐快照对该载体发起 `no_package` 停机
+
+#### Scenario: 到期主套餐无后续可生效套餐
+
+- **GIVEN** 某载体的当前主套餐到期，且不存在后续主套餐或队首后续主套餐不满足激活条件
+- **WHEN** 系统完成该套餐到期处理
+- **THEN** 系统按当前套餐、流量和实名事实执行既有停机评估
+
+#### Scenario: 后续套餐激活结果未知
+
+- **GIVEN** 某载体的当前主套餐到期，存在待生效后续主套餐，但激活任务投递或执行结果暂时未知
+- **WHEN** 系统处理该套餐到期
+- **THEN** 系统不得将该未知结果视为不存在后续套餐而依据旧快照发起停机，并保留既有激活恢复与套餐轮询兜底
+
+#### Scenario: 卡状态轮询发现缺失的套餐任务
+
+- **GIVEN** 启用轮询的卡匹配套餐检查配置，且其 `polling:package` 分片队列项因异常缺失
+- **WHEN** 卡状态轮询成功完成且未命中风险停机
+- **THEN** 系统基于最新卡状态仅补入缺失的套餐任务，不改写已存在套餐任务的执行时间；后续套餐任务仍按既有停复机条件评估该卡
+
 ### Requirement: 批量操作可追踪
 
-系统 SHALL 为同步批量分配和调价直接返回处理结果；对异步批量订购返回任务标识并提供状态查询。
+系统 SHALL 为同步批量分配和调价直接返回处理结果；对异步批量订购返回任务标识并提供状态查询。异步批量订购完成后，系统 MUST 持久化每个输入行的成功或失败结果及与其一致的总数、成功数和失败数；部分资产因余额不足、资产校验或重复输入失败不得阻止任务进入完成终态。
 
 #### Scenario: 批量操作可追踪
 
@@ -26,6 +50,12 @@
 - **WHEN** 创建批量操作
 - **THEN** 同步操作直接返回结果；异步订购返回任务标识且可查询处理状态
 
+#### Scenario: 批量订购部分失败后查询结果
+
+- **GIVEN** 异步批量订购中的部分资产已成功创建订单，其他资产因钱包余额不足或输入重复失败
+- **WHEN** Worker 完成全部输入行的处理
+- **THEN** 任务状态为已完成，逐行结果保留成功订单与失败原因，且总数等于成功数与失败数之和
+
 ### Requirement: 授权页面禁止重复选择套餐
 
 系统 SHALL 使代理系列授权页面能够区分目标店铺已授权和未授权套餐；已授权套餐 MUST 以不可新增的状态返回，首次创建系列授权和既有系列新增套餐均适用。