fix(processing): U08 重试闭环 + FailureScheduler + U11 重放入口（ACM2-10）

- U08 闭环：失败状态迁移移到持有 head/事件的边界内——processOne/SnapshotFlow 统一经
  ProcFailure（FAILED+attempts+退避，达上限 DEAD/EXHAUSTED 并落库 attempts）；Dispatcher
  逐条与 SCHD 批量共用 retryOrDead；flushSchd 失败整批 attempts+1/指数退避，队首未到期不
  claim，达上限整批 DEAD(DLQ)；loop 只作最后防线——仅 catch Exception，InterruptedException
  恢复中断位并上抛，致命 Error 不吞（异常必可见）
- 统一策略：新增 FailureScheduler（可注入 java.time.Clock + TimeFactory，backoff/exhausted
  单一来源）与 ProcFailure，MessageProcessor/SnapshotFlow/Dispatcher 共用
- U11：ReplayService 显式重放入口 + ProcStateRepository.requeueByErrorClasses——仅白名单内
  可恢复类（CODEC_ERROR/UNSUPPORTED/INFRA/EXHAUSTED）从 FAILED/DEAD 回 PENDING(attempts 清零)，
  MALFORMED 永不重放
- 测试（30 个全绿）：①Pump 内部异常→FAILED(INFRA)→重试达上限 DEAD；②SCHD 连续失败退避递增→
  整批 DEAD 且 KAFKA:msg 不受影响；③AssertionError/InterruptedException 不被普通恢复吞掉；
  另有 ReplayService(3)、Dispatcher 批退避、InfraBindingStartupTest（DataSource 启动级绑定）
- 收尾：.gitattributes（行尾/二进制）；Gradle 10 弃用告警核查——仅来自 micronaut-application
  插件（BOM 注入 + IDE 文件生成），升级 Gradle 10 前需先升插件版本（已记录）

.gitattributes

@@ -0,0 +1,13 @@
+# 行尾/二进制规范（Gradle 10 前收尾项，ACM2-10 U02/U30）
+* text=auto
+
+# shell 启动脚本必须 LF（Windows 上保留可执行）
+*.sh text eol=lf
+gradlew text eol=lf
+*.bat text eol=crlf
+
+# 二进制一律不转行尾
+*.jar binary
+*.png binary
+*.jpg binary
+*.pdf binary

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/Dispatcher.kt

@@ -6,6 +6,7 @@ import com.gzzn.omms.msgexchange.nextgen.domain.EventStatus
 import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent
 import com.gzzn.omms.msgexchange.nextgen.domain.Targets
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.retry.FailureScheduler
 import jakarta.inject.Singleton
 import java.time.Duration
 import java.time.Instant
@@ -23,15 +24,19 @@ interface DeliveryPort {
 }
 
 /**
- * ACMA-8 流程 3：投递调度——每 target 严格 FIFO（I1 双层同策略）；
- * N03（U06）：KAFKA_SCHD 不走逐条循环（唯一出口是 flushSchd 批量聚合），
- * 逐条循环显式排除，避免 schd 事件被无条件 markSent 吞掉、flush 永远 claim 不到。
+ * ACMA-8 流程 3：投递调度——每 target 严格 FIFO（I1 双层同策略）。
+ * N03（U06）：KAFKA_SCHD 不走逐条循环（唯一出口是 flushSchd 批量聚合），逐条循环显式排除。
+ * U08 闭环：逐条与批量失败都在持有具体事件/批次的边界完成状态迁移——
+ *   · 逐条：scheduleRetry(attempts+1, backoff) / 达上限 markDead(EXHAUSTED)（DLQ 保留行）；
+ *   · 批量（flushSchd）：整批退避，队首 nextAttemptAt 未到不 claim；达上限整批 DEAD/DLQ；
+ *   · loop 只作最后防线，不吞 InterruptedException（致命/中断错误不被普通恢复吞掉）。
  */
 @Singleton
 class Dispatcher(
     private val msgEvents: MsgEventRepository,
     private val port: DeliveryPort,
     private val props: PipelineProps,
+    private val scheduler: FailureScheduler,
 ) {
     @Volatile
     private var running = true
@@ -42,8 +47,11 @@ class Dispatcher(
         while (running) {
             try {
                 tick()
+            } catch (e: InterruptedException) {
+                Thread.currentThread().interrupt()
+                return
             } catch (e: Exception) {
-                // U08/R02：投递循环不得因单次异常静默死亡；U12 补 ERROR 日志与告警出口。
+                // U08：投递循环只作最后防线；致命 Error 不吞。tick 内失败迁移已完成。
                 sleepQuietly(props.pipeline.pollInterval)
             }
             // N18：轮询间隔取参数表（下限 50ms，避免退避节律被吞）
@@ -56,7 +64,7 @@ class Dispatcher(
         for (t in targets) {
             if (t == Targets.KAFKA_SCHD) continue        // N03：schd 唯一出口 flushSchd
             val head = msgEvents.headUnsent(t) ?: continue
-            if (head.state == EventStatus.PENDING && head.nextAttemptAt != null && head.nextAttemptAt > Instant.now()) {
+            if (head.state == EventStatus.PENDING && head.nextAttemptAt != null && head.nextAttemptAt > scheduler.now()) {
                 // 队头退避未到期：等待，不跳过（保序）；超时升级归 U13（WP2）
                 continue
             }
@@ -68,30 +76,26 @@ class Dispatcher(
                     msgEvents.insertSync(listOf(MsgEvent(target = Targets.REDIS_FLIGHT_INFO, payloadJson = deleteOf(head))))
                 }
             } catch (e: Exception) {
-                val attempts = head.attempts + 1
-                if (attempts >= props.pipeline.maxAttempts) {
-                    msgEvents.markDead(head.eventId!!, ErrorClass.EXHAUSTED, e.message ?: "unknown")
-                } else {
-                    msgEvents.scheduleRetry(
-                        head.eventId!!,
-                        Instant.now().plusMillis(props.pipeline.backoffFor(attempts)),
-                        attempts,
-                    )
-                }
+                retryOrDead(head, e.message ?: "unknown")
             }
         }
         if (flushDue()) {
-            try {
-                flushSchd()
-            } catch (e: Exception) {
-                // U08/N24：批发送失败不得终止 loop——事件保持 PENDING，下个 flush 周期整批重试；
-                // 毒丸/告警随 U12/U13 补齐。lastFlush 仅在成功后推进（见 flushSchd）。
-            }
+            flushSchd()
         }
     }
 
     private fun flushDue(): Boolean =
-        Duration.between(lastFlush, Instant.now()) >= props.schd.flushPeriod
+        Duration.between(lastFlush, scheduler.now()) >= props.schd.flushPeriod
+
+    /** 单条事件失败迁移：attempts+1；达上限 DEAD(EXHAUSTED)（DLQ，attempts 落库审计），否则退避重试。 */
+    private fun retryOrDead(e: MsgEvent, lastError: String) {
+        val attempts = e.attempts + 1
+        if (scheduler.exhausted(attempts)) {
+            msgEvents.markDead(e.eventId!!, ErrorClass.EXHAUSTED, lastError, attempts)
+        } else {
+            msgEvents.scheduleRetry(e.eventId!!, scheduler.nextAttemptAt(attempts), attempts)
+        }
+    }
 
     private fun deliver(target: String, e: MsgEvent) = when (target) {
         Targets.KAFKA_MSG -> port.sendKafka("msg", e.payloadJson)
@@ -103,17 +107,30 @@ class Dispatcher(
 
     private fun deleteOf(e: MsgEvent) = """{"op":"delete","refs":${e.partitionKey ?: ""}}"""
 
-    /** 流程 3 flushSchd：批上限 + 每 FLID 最新一态聚合（topic "schd"，wire=FLTR JSON 数组）。 */
+    /**
+     * 流程 3 flushSchd：批上限 + 每 FLID 最新一态聚合（topic "schd"，wire=FLTR JSON 数组）。
+     * U08 批量闭环：队首退避未到期不 claim；发送失败 → 整批 attempts+1（退避）或达上限整批 DEAD/DLQ；
+     * lastFlush 仅在成功（含空批）后推进。
+     */
     internal fun flushSchd() {
         val pending = msgEvents.claimBatch(Targets.KAFKA_SCHD, limit = props.schd.flushLimit)
         if (pending.isEmpty()) {
-            lastFlush = Instant.now()
+            lastFlush = scheduler.now()
             return
         }
+        val due = pending.first()
+        if (due.nextAttemptAt != null && due.nextAttemptAt > scheduler.now()) {
+            return                                   // 队首仍在退避：整批等待，不推进 lastFlush（到期再试）
+        }
         val payload = SchdAggregation.latestPerFlight(pending).joinToString(",", "[", "]")
-        port.sendKafka("schd", payload)             // 失败整批退避（at-least-once）：异常上抛由 tick 隔离，事件保持 PENDING
+        try {
+            port.sendKafka("schd", payload)
+        } catch (e: Exception) {
+            pending.forEach { retryOrDead(it, "schd-send: ${e.message ?: "unknown"}") }
+            return                                   // 不推进 lastFlush：整批退避（含 DEAD 出队）后到期重试
+        }
         msgEvents.markAllSent(pending.mapNotNull { it.eventId })
-        lastFlush = Instant.now()                   // N24：仅在成功后推进
+        lastFlush = scheduler.now()
     }
 
     private fun sleepQuietly(d: Duration) {

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/persistence/Repositories.kt

@@ -29,6 +29,13 @@ interface ProcStateRepository {
         errorClass: ErrorClass? = null,
         lastError: String? = null,
     )
+
+    /**
+     * U11 显式重放入口（ReplayService）：仅把给定 errorClass 集合中的行从 FAILED/DEAD 置回 PENDING，
+     * 以便主泵重新领取。实现约定：ATTEMPTS=0、NEXT_ATTEMPT_AT=NULL（立即重试），
+     * ERROR_CLASS/LAST_ERROR 保留作审计。返回受影响行数。
+     */
+    fun requeueByErrorClasses(errorClasses: List<ErrorClass>): Int
 }
 
 interface MsgEventRepository {
@@ -45,7 +52,7 @@ interface MsgEventRepository {
 
     fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int)
 
-    fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String)
+    fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int? = null)
 
     /** 阶段 B（定案 2）：Delivery 同线程在 ES 投递成功后同步 enqueue 删除事件。 */
     fun insertSync(events: List<MsgEvent>)

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/FailureScheduler.kt

@@ -0,0 +1,36 @@
+package com.gzzn.omms.msgexchange.nextgen.infra.retry
+
+import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps
+import io.micronaut.context.annotation.Factory
+import jakarta.inject.Singleton
+import java.time.Clock
+import java.time.Instant
+
+/**
+ * 统一重试策略（ACM2-10 U08）：供 MessageProcessor / SnapshotFlow（ProcState 侧）与
+ * Dispatcher（MsgEvent 侧）共用——attempts 递增后按 backoff 表给 nextAttemptAt；
+ * exhausted 判定与两侧同源（maxAttempts）。时间一律经可注入 Clock（测试用固定钟，避免脆弱睡眠）。
+ */
+@Singleton
+class FailureScheduler(
+    private val props: PipelineProps,
+    private val clock: Clock,
+) {
+    /** 便捷构造：默认系统时钟（生产路径）。 */
+    constructor(props: PipelineProps) : this(props, Clock.systemUTC())
+
+    fun now(): Instant = clock.instant()
+
+    fun exhausted(attempts: Int): Boolean = attempts >= props.pipeline.maxAttempts
+
+    /** attempts 指递增后的值；N28：attempt ≤ 0 由 backoffFor 兜底为首档。 */
+    fun nextAttemptAt(attemptsAfterIncrement: Int): Instant =
+        now().plusMillis(props.pipeline.backoffFor(attemptsAfterIncrement))
+}
+
+/** 提供可注入 Clock（java.time.Clock）；测试可用 Clock.fixed(...) 或自定义可变钟覆盖。 */
+@Factory
+class TimeFactory {
+    @Singleton
+    fun systemClock(): Clock = Clock.systemUTC()
+}

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ProcFailure.kt

@@ -0,0 +1,38 @@
+package com.gzzn.omms.msgexchange.nextgen.infra.retry
+
+import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass
+import com.gzzn.omms.msgexchange.nextgen.domain.ProcState
+import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository
+import jakarta.inject.Singleton
+
+/**
+ * ProcState 侧统一失败迁移（U08/U10）：处理/快照路径共用——
+ * attempts+1 后若 exhausted → DEAD(EXHAUSTED)（终态，errorClass 规范化，原因保留在 lastError）；
+ * 否则 FAILED + attempts + nextAttemptAt(退避)（可重放）。任何“失败”都不得在无退避下直接终态化。
+ */
+@Singleton
+class ProcFailure(
+    private val procState: ProcStateRepository,
+    val scheduler: FailureScheduler,
+) {
+    fun fail(head: ProcState, ec: ErrorClass, reason: String) {
+        val attempts = head.attempts + 1
+        if (scheduler.exhausted(attempts)) {
+            procState.update(
+                head.cminmsgsId, ProcStatus.DEAD,
+                attempts = attempts,
+                errorClass = ErrorClass.EXHAUSTED,
+                lastError = "$reason; attempts=$attempts",
+            )
+        } else {
+            procState.update(
+                head.cminmsgsId, ProcStatus.FAILED,
+                attempts = attempts,
+                nextAttemptAt = scheduler.nextAttemptAt(attempts),
+                errorClass = ec,
+                lastError = reason,
+            )
+        }
+    }
+}

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayService.kt

@@ -0,0 +1,28 @@
+package com.gzzn.omms.msgexchange.nextgen.infra.retry
+
+import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository
+import jakarta.inject.Singleton
+
+/**
+ * U11 显式重放入口：只允许“可恢复”的错误类从 FAILED/DEAD 回到 PENDING（主泵重领）。
+ * 不可恢复类（MALFORMED——报文非法，重放必再失败）与未知类一律不在白名单内。
+ */
+@Singleton
+class ReplayService(
+    private val procState: ProcStateRepository,
+) {
+    /** 可恢复错误类：codec 修复可重放 / 未实装补齐可重放 / 基础设施抖动可重放 / 重试耗尽后人工复核可重放。 */
+    val replayableErrorClasses: Set<ErrorClass> =
+        setOf(ErrorClass.CODEC_ERROR, ErrorClass.UNSUPPORTED, ErrorClass.INFRA, ErrorClass.EXHAUSTED)
+
+    /** 只重放白名单内的类；请求含 MALFORMED 等非法类时静默忽略该类。 */
+    fun replay(requested: Collection<ErrorClass>): Int {
+        val allowed = requested.filter { it in replayableErrorClasses }
+        if (allowed.isEmpty()) return 0
+        return procState.requeueByErrorClasses(allowed)
+    }
+
+    /** 默认入口：重放全部可恢复类。 */
+    fun replayAll(): Int = replay(replayableErrorClasses)
+}

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/Pump.kt

@@ -12,6 +12,7 @@ import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.PumpJobRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.redis.FlightRedisClient
+import com.gzzn.omms.msgexchange.nextgen.infra.retry.ProcFailure
 import com.gzzn.omms.msgexchange.nextgen.jobs.JobExecutor
 import jakarta.inject.Singleton
 import java.time.Duration
@@ -37,9 +38,12 @@ class Pump(
         while (running) {
             try {
                 tick()
-            } catch (t: Throwable) {
-                // U08/R02：主泵不得因单次异常静默死亡——任一 codec/DB/Redis 抖动不得终止队列；
-                // 具体失败落 FAILED/DEAD 与告警见 MessageProcessor/U12。
+            } catch (e: InterruptedException) {
+                Thread.currentThread().interrupt()
+                return
+            } catch (e: Exception) {
+                // U08：loop 只作最后防线——失败状态迁移已在 processOne/execute 的边界内完成；
+                // 致命 Error 不在此捕获（任其终止进程，保证“异常必可见”）。
                 sleepQuietly(props.pipeline.pollInterval)
             }
         }
@@ -56,11 +60,12 @@ class Pump(
             head == null -> sleepQuietly(props.pipeline.pollInterval)
             head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > Instant.now() ->
                 if (poisoned(head)) {
-                    upgradeDead(head, head.lastError ?: "head-deadline-exceeded")
+                    procState.update(head.cminmsgsId, ProcStatus.DEAD,
+                        errorClass = ErrorClass.EXHAUSTED, lastError = head.lastError ?: "head-deadline-exceeded")
                 } else {
                     sleepQuietly(Duration.between(Instant.now(), head.nextAttemptAt))
                 }
-            // PENDING、或 FAILED 退避已到期：交处理入口（入口再做 attempts 毒丸守卫）
+            // PENDING、或 FAILED 退避已到期：交处理入口（内部有边界化失败迁移与 attempts 守卫）
             else -> processor.processOne(head)
         }
     }
@@ -69,10 +74,6 @@ class Pump(
         head.attempts >= props.pipeline.maxAttempts ||
             Duration.between(head.updatedAt, Instant.now()) > props.pipeline.headDeadline
 
-    private fun upgradeDead(head: ProcState, reason: String) {
-        procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED, lastError = reason)
-    }
-
     /** JOB 与队头消息的先后由入队时间近似；实装以统一序号列保证（阶段 1 后续）。 */
     private fun jobBefore(job: PumpJobRepository.Job, head: ProcState?) =
         head == null || head.state == ProcStatus.FAILED
@@ -94,9 +95,9 @@ class Pump(
 
 /**
  * ACMA-8 流程 2 processOne：解码→绑定→纯函数决策→提交。
- * U08/U10/U11（N06/N21/T06）：统一失败治理——「未实装/未注册/CODEC_ERROR/staging 未实装」
- * 一律 FAILED（可重放，带 attempts+nextAttemptAt，绝不直接写终态）；仅 MALFORMED（报文非法）
- * 与「重试耗尽」写 DEAD 终态。attempts ≥ maxAttempts 的 FAILED 在入口即升级 DEAD（防退避到期后无限重试）。
+ * U08/U10/U11：失败状态迁移全部在“持有具体 head”的边界内完成——
+ * 任何意外异常 → FAILED(INFRA)+退避（ProcFailure）；attempts 达上限 → DEAD(EXHAUSTED)；
+ * MALFORMED 直接 DEAD；InterruptedException 恢复中断位并上抛（不被普通恢复逻辑吞掉）。
  */
 @Singleton
 class MessageProcessor(
@@ -107,11 +108,24 @@ class MessageProcessor(
     private val handlers: HandlerHolder,
     private val redis: FlightRedisClient,
     private val snapshotFlow: SnapshotFlow,
+    private val procFailure: ProcFailure,
     private val props: PipelineProps,
 ) {
     fun processOne(head: ProcState) {
-        // U08 入口守卫：FAILED 行退避到期后再处理前，先判 attempts 毒丸（与 tick poisoned 同值）
-        if (head.state == ProcStatus.FAILED && head.attempts >= props.pipeline.maxAttempts) {
+        try {
+            processInternal(head)
+        } catch (e: InterruptedException) {
+            Thread.currentThread().interrupt()
+            throw e
+        } catch (e: Exception) {
+            // U08：边界化——异常归于本条 head，写 FAILED/DEAD，而不是穿出杀 pump
+            procFailure.fail(head, ErrorClass.INFRA, e.message ?: e.javaClass.simpleName)
+        }
+    }
+
+    private fun processInternal(head: ProcState) {
+        // 守卫：手工/遗留 FAILED 行若 attempts 已达上限，直接终态（防止退避到期后无限重试）
+        if (head.state == ProcStatus.FAILED && procFailure.scheduler.exhausted(head.attempts)) {
             procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED,
                 lastError = head.lastError ?: "max-attempts")
             return
@@ -129,7 +143,7 @@ class MessageProcessor(
                     procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED,
                         lastError = r.failure.detail)
                 } else {
-                    failWithBackoff(head, r.failure.errorClass, r.failure.detail)
+                    procFailure.fail(head, r.failure.errorClass, r.failure.detail)
                 }
                 return
             }
@@ -149,7 +163,7 @@ class MessageProcessor(
         val handler = handlers.registry.dispatcherFor(decoded)
         if (handler == null) {
             // U10（N21）：未注册 ≠ 报文非法——写 FAILED（可重放），绝不写终态
-            failWithBackoff(head, ErrorClass.UNSUPPORTED, "no-handler:${decoded.typeTag}")
+            procFailure.fail(head, ErrorClass.UNSUPPORTED, "no-handler:${decoded.typeTag}")
             return
         }
         val schdKind = decoded.kind as? com.gzzn.omms.msgexchange.nextgen.domain.MsgKind.Schd
@@ -175,18 +189,6 @@ class MessageProcessor(
         procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
         // 阶段 B：flightState.apply(decision.flightChanges) 进入同一事务；投影事件（ES/REDIS）追加。
     }
-
-    /** 统一失败路径：FAILED + attempts+1 + nextAttemptAt=backoff（N28 下界由 backoffFor 保证）。 */
-    private fun failWithBackoff(head: ProcState, ec: ErrorClass, reason: String) {
-        val attempts = head.attempts + 1
-        procState.update(
-            head.cminmsgsId, ProcStatus.FAILED,
-            attempts = attempts,
-            nextAttemptAt = Instant.now().plusMillis(props.pipeline.backoffFor(attempts)),
-            errorClass = ec,
-            lastError = reason,
-        )
-    }
 }
 
 /** 延迟装配占位（阶段 1 后续以 Micronaut Bean 替换直连构造）。 */

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/SnapshotFlow.kt

@@ -1,6 +1,5 @@
 package com.gzzn.omms.msgexchange.nextgen.processing
 
-import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps
 import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage
 import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass
 import com.gzzn.omms.msgexchange.nextgen.domain.ProcState
@@ -9,28 +8,28 @@ import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.redis.FlightRedisClient
 import com.gzzn.omms.msgexchange.nextgen.infra.redis.RedisScript
+import com.gzzn.omms.msgexchange.nextgen.infra.retry.ProcFailure
 import jakarta.inject.Singleton
-import java.time.Instant
 
 /**
  * ACMA-8 流程 4：日计划快照（generation，主泵内执行）。
  * staging（内存瞬态，崩溃从 raw 整包重放）→ Lua 原子“覆盖+按代差删”（I4/I5）
  * → putGenIfVersion CAS + SUCCEEDED（重放幂等，版本不二次自增——恢复协议细节见 U09）。
- * U10/T07 占位安全化：staging 未实装 → FAILED(UNSUPPORTED)+退避（可重放），绝不写 DEAD 终态；
- * CAS 冲突 → FAILED(INFRA)+退避（不再无 nextAttemptAt 紧循环，N06/N28）。
+ * U10/T07 占位安全化 + U08 统一失败迁移（ProcFailure）：staging 未实装 → FAILED(UNSUPPORTED)+退避
+ * （可重放，绝不写终态）；CAS 冲突 → FAILED(INFRA)+退避；达上限统一 DEAD(EXHAUSTED)。
  */
 @Singleton
 class SnapshotFlow(
     private val procState: ProcStateRepository,
     private val refData: RefDataRepository,
     private val redis: FlightRedisClient,
-    private val props: PipelineProps,
+    private val procFailure: ProcFailure,
 ) {
     fun publishSnapshot(head: ProcState, msg: DecodedMessage) {
         // 1) staging：流式解析 + 整包校验（TODO(阶段2): 流式 codec；千级 FLTR 为 MB 级，内存瞬态）
         val staged = StageResult.stagingOf(msg) // 骨架：TODO 解析 FLTR 集与重组（KEEP 现役 MAFL/登机桥规则）
         if (staged is StageResult.Invalid) {
-            failWithBackoff(head, ErrorClass.UNSUPPORTED, staged.reason)   // U10：未实装 → 可重放，非终态
+            procFailure.fail(head, ErrorClass.UNSUPPORTED, staged.reason)   // U10：未实装 → 可重放，非终态
             return
         }
         val normalized = (staged as StageResult.Ok).flights            // (flid, payloadJson)
@@ -48,24 +47,13 @@ class SnapshotFlow(
             // 重放路径：version 已是目标值 → no-op 视为成功
             val again = refData.getGen(day)
             if (again == null || again.version != expected + 1) {
-                failWithBackoff(head, ErrorClass.INFRA, "gen-cas-conflict")   // N06/N28：带退避，禁止紧循环
+                procFailure.fail(head, ErrorClass.INFRA, "gen-cas-conflict")   // N06/N28：带退避，禁止紧循环
                 return
             }
         }
         procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED)
     }
 
-    private fun failWithBackoff(head: ProcState, ec: ErrorClass, reason: String) {
-        val attempts = head.attempts + 1
-        procState.update(
-            head.cminmsgsId, ProcStatus.FAILED,
-            attempts = attempts,
-            nextAttemptAt = Instant.now().plusMillis(props.pipeline.backoffFor(attempts)),
-            errorClass = ec,
-            lastError = reason,
-        )
-    }
-
     /** staging 结果（骨架）。 */
     sealed interface StageResult {
         data class Ok(val day: String, val flights: List<Pair<String, String>>) : StageResult

src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/TestClocks.kt

@@ -0,0 +1,27 @@
+package com.gzzn.omms.msgexchange.nextgen
+
+import java.time.Clock
+import java.time.Instant
+import java.time.ZoneId
+import java.time.ZoneOffset
+
+/** 可手动推进的可注入时钟（测试用），替代真实时间避免脆弱睡眠测试。 */
+internal class MutableClock(
+    initial: Instant,
+    private val zone: ZoneId = ZoneOffset.UTC,
+) : Clock() {
+
+    var instant: Instant = initial
+
+    override fun instant(): Instant = instant
+
+    override fun getZone(): ZoneId = zone
+
+    override fun withZone(zoneId: ZoneId): Clock = MutableClock(instant, zoneId)
+
+    fun advance(millis: Long) { instant = instant.plusMillis(millis) }
+
+    companion object {
+        val BASE: Instant = Instant.parse("2026-09-06T02:00:00Z")
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/config/InfraBindingStartupTest.kt

@@ -0,0 +1,39 @@
+package com.gzzn.omms.msgexchange.nextgen.config
+
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest
+import jakarta.inject.Inject
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Test
+import java.sql.Connection
+import javax.sql.DataSource
+
+/**
+ * U03（N02/R09/R01a 验收的“启动级”版本）：基础设施配置在真实上下文启动时绑定生效——
+ * 注入 DataSource 并建立真实连接（H2 内存，application-test.yml），证明 datasources.default.*
+ * 键位与驱动解析正确（而非“看似配置实则未生效”）。Pump 等业务 bean 懒加载，不依赖实仓储。
+ */
+@MicronautTest
+class InfraBindingStartupTest {
+
+    @Inject
+    lateinit var dataSource: DataSource
+
+    @Inject
+    lateinit var props: PipelineProps
+
+    @Test
+    fun `datasource binds and yields a live connection`() {
+        assertNotNull(dataSource)
+        dataSource.connection.use { c: Connection ->
+            assertNotNull(c.metaData)
+            c.createStatement().use { it.execute("SELECT 1") }
+        }
+    }
+
+    @Test
+    fun `registration disabled by test profile`() {
+        assertNotNull(props)
+        // eureka 注册在 application-test.yml 经 msgx.register-eureka=false 关闭——启动级验证绑定可达
+        assertNotNull(dataSource)
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/delivery/DispatcherTickTest.kt

@@ -1,21 +1,24 @@
 package com.gzzn.omms.msgexchange.nextgen.delivery
 
+import com.gzzn.omms.msgexchange.nextgen.MutableClock
 import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps
 import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass
 import com.gzzn.omms.msgexchange.nextgen.domain.EventStatus
 import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent
 import com.gzzn.omms.msgexchange.nextgen.domain.Targets
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.retry.FailureScheduler
 import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Assertions.assertNull
 import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Test
-import java.time.Duration
 import java.time.Instant
-import kotlin.test.assertFails
 
 /**
- * U06（N03）：KAFKA_SCHD 不被逐条循环吞掉——唯一出口 flushSchd 批量聚合；
- * 聚合只发每 FLID 最新一态；发送失败整批保持 PENDING 且不推进 lastFlush（N24）。
+ * U06（N03）+ U08 闭环：schd 出口只走 flushSchd；
+ * 批发送失败 → 整批 attempts 递增 + 指数退避（队首未到期不 claim），达上限整批 DEAD/DLQ；
+ * 时间经可注入 Clock（MutableClock），不依赖真实睡眠。
  */
 class DispatcherTickTest {
 
@@ -47,9 +50,10 @@ class DispatcherTickTest {
             if (i >= 0) events[i] = events[i].copy(state = EventStatus.PENDING, attempts = attempts, nextAttemptAt = nextAttemptAt)
         }
 
-        override fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String) {
+        override fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int?) {
             val i = events.indexOfFirst { it.eventId == eventId }
-            if (i >= 0) events[i] = events[i].copy(state = EventStatus.DEAD, errorClass = errorClass, lastError = lastError)
+            if (i >= 0) events[i] = events[i].copy(state = EventStatus.DEAD, errorClass = errorClass, lastError = lastError,
+                attempts = attempts ?: events[i].attempts)
         }
 
         override fun insertSync(events: List<MsgEvent>) = Unit
@@ -68,7 +72,10 @@ class DispatcherTickTest {
         override fun projectRedis(payloadJson: String) = Unit
     }
 
-    private fun props(): PipelineProps = PipelineProps()   // schd.flush-period 默认 3s
+    private val clock = MutableClock(MutableClock.BASE)
+
+    private fun dispatcher(repo: FakeRepo, port: FakePort, p: PipelineProps = PipelineProps()): Dispatcher =
+        Dispatcher(repo, port, p, FailureScheduler(p, clock))
 
     private fun ev(id: Long, target: String, key: String?, payload: String) =
         MsgEvent(eventId = id, target = target, partitionKey = key, payloadJson = payload, state = EventStatus.PENDING)
@@ -77,57 +84,101 @@ class DispatcherTickTest {
     fun `schd flushed once via batch path, per-target tick never consumes it`() {
         val repo = FakeRepo()
         val port = FakePort()
-        val d = Dispatcher(repo, port, props())
+        val d = dispatcher(repo, port)
         repo.enqueue(ev(1, Targets.KAFKA_MSG, null, """{"msg":1}"""))
         repo.enqueue(ev(2, Targets.KAFKA_SCHD, "F1", """{"FLID":"F1","v":1}"""))
 
         d.tick()   // lastFlush=EPOCH → 首 tick 即到 flush 周期
 
-        // KAFKA:msg 逐条投递；schd 经 flushSchd 聚合发出一次（而非逐条 markSent 吞掉）
         assertEquals(1, port.sent.count { it.first == "msg" })
         assertEquals(1, port.sent.count { it.first == "schd" })
         assertEquals(EventStatus.SENT, repo.events.first { it.eventId == 2L }.state)
 
-        // 新的 schd 事件：flush 周期未到 → 逐条循环不得消费它（仍 PENDING、未发）
         repo.enqueue(ev(4, Targets.KAFKA_SCHD, "F1", """{"FLID":"F1","v":4}"""))
-        d.tick()
+        d.tick()   // 时钟未推进 → flush 周期未到；逐条循环不得消费 schd
         assertEquals(EventStatus.PENDING, repo.events.first { it.eventId == 4L }.state)
-        assertEquals(1, port.sent.count { it.first == "schd" })   // 第二次 tick 未额外发送
+        assertEquals(1, port.sent.count { it.first == "schd" })
     }
 
     @Test
     fun `flushSchd aggregates latest per flight and marks all sent`() {
         val repo = FakeRepo()
         val port = FakePort()
-        val d = Dispatcher(repo, port, props())
+        val d = dispatcher(repo, port)
         repo.enqueue(ev(1, Targets.KAFKA_SCHD, "F1", "old"))
         repo.enqueue(ev(2, Targets.KAFKA_SCHD, "F2", "only"))
         repo.enqueue(ev(3, Targets.KAFKA_SCHD, "F1", "new"))
 
         d.flushSchd()
 
-        assertEquals(1, port.sent.count { it.first == "schd" })
         val payload = port.sent.first { it.first == "schd" }.second
-        assertTrue(payload.startsWith("[") && payload.endsWith("]"))
         assertTrue(payload.contains("new") && payload.contains("only") && !payload.contains("old"))
         assertEquals(2, payload.removeSurrounding("[", "]").split(",").size)   // 同 FLID 只发最新（矩阵 #7）
         assertTrue(repo.events.all { it.state == EventStatus.SENT })
     }
 
     @Test
-    fun `flush failure keeps batch PENDING and retries after next due`() {
+    fun `flush failure schedules per-event backoff and retries after due`() {
         val repo = FakeRepo()
         val port = FakePort().apply { failTopic = "schd" }
-        val d = Dispatcher(repo, port, props())
+        val d = dispatcher(repo, port)
         repo.enqueue(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"1"}"""))
 
-        assertFails { d.flushSchd() }              // 发送失败 → 异常上抛（tick 侧隔离），事件保持 PENDING
-        assertEquals(EventStatus.PENDING, repo.events.single().state)
+        d.flushSchd()
+        val e1 = repo.events.single()
+        assertEquals(EventStatus.PENDING, e1.state)
+        assertEquals(1, e1.attempts)
+        assertNotNull(e1.nextAttemptAt)                 // 指数退避落库（attempts=1 → 首档）
         assertEquals(0, port.sent.size)
 
+        clock.advance(1000)                             // 退避到期
         port.failTopic = null
-        d.flushSchd()                              // lastFlush 未推进 → 下个周期整批重试成功
+        d.flushSchd()
         assertEquals(EventStatus.SENT, repo.events.single().state)
         assertEquals(1, port.sent.count { it.first == "schd" })
     }
+
+    @Test
+    fun `repeated batch failures backoff grows and whole batch goes DEAD DLQ at limit`() {
+        val props = PipelineProps().apply { pipeline.maxAttempts = 2 }
+        val repo = FakeRepo()
+        val port = FakePort().apply { failTopic = "schd" }
+        val d = dispatcher(repo, port, props)
+        repo.enqueue(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"1"}"""))
+        repo.enqueue(ev(2, Targets.KAFKA_SCHD, "F2", """{"v":"2"}"""))
+
+        d.flushSchd()                                   // 失败#1：attempts=1，退避 backoff[1]=1000ms
+        val after1 = repo.events.map { it.attempts to it.state }
+        assertEquals(listOf(1 to EventStatus.PENDING, 1 to EventStatus.PENDING), after1)
+        assertTrue(repo.events.all { it.nextAttemptAt == MutableClock.BASE.plusMillis(1000) })
+
+        d.flushSchd()                                   // 时钟未推进 → 队首未到期，不 claim（不推进 lastFlush）
+        assertEquals(1, repo.events.first { it.eventId == 1L }.attempts)
+
+        clock.advance(1000)
+        d.flushSchd()                                   // 失败#2：attempts=2 == maxAttempts → 整批 DEAD
+        assertTrue(repo.events.all { it.state == EventStatus.DEAD && it.errorClass == ErrorClass.EXHAUSTED })
+        assertTrue(repo.events.all { it.attempts == 2 })
+        assertEquals(0, port.sent.size)
+
+        // DEAD 行不再参与 claim：空批 → 无新发送、无异常
+        d.flushSchd()
+        assertEquals(0, port.sent.size)
+    }
+
+    @Test
+    fun `KAFKA msg delivery unaffected while schd batch is retrying`() {
+        val props = PipelineProps().apply { pipeline.maxAttempts = 2 }
+        val repo = FakeRepo()
+        val port = FakePort().apply { failTopic = "schd" }
+        val d = dispatcher(repo, port, props)
+        repo.enqueue(ev(1, Targets.KAFKA_SCHD, "F1", """{"v":"1"}"""))
+        repo.enqueue(ev(2, Targets.KAFKA_MSG, null, """{"m":1}"""))
+
+        d.tick()
+
+        assertEquals(EventStatus.SENT, repo.events.first { it.eventId == 2L }.state)   // msg 正常
+        assertEquals(1, repo.events.first { it.eventId == 1L }.attempts)               // schd 进退避
+        assertEquals(EventStatus.PENDING, repo.events.first { it.eventId == 1L }.state)
+    }
 }

src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/infra/retry/ReplayServiceTest.kt

@@ -0,0 +1,94 @@
+package com.gzzn.omms.msgexchange.nextgen.infra.retry
+
+import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass
+import com.gzzn.omms.msgexchange.nextgen.domain.ProcState
+import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertNull
+import org.junit.jupiter.api.Test
+import java.time.Instant
+
+/**
+ * U11：显式重放入口只允许可恢复错误从 FAILED/DEAD 返回 PENDING（attempts 清零、立即重试）；
+ * MALFORMED（报文非法）永不被重放。
+ */
+class ReplayServiceTest {
+
+    private class FakeRepo : ProcStateRepository {
+        val rows = linkedMapOf<Long, ProcState>()
+        val requeueCalls = mutableListOf<List<ErrorClass>>()
+
+        fun seed(id: Long, status: ProcStatus, ec: ErrorClass?) {
+            rows[id] = ProcState(id, status, identityKey = "k$id", attempts = 3, errorClass = ec, lastError = "x")
+        }
+
+        override fun insert(cminmsgsId: Long, state: ProcStatus) = Unit
+        override fun headUnfinished(): ProcState? = null
+        override fun tryBindIdentity(cminmsgsId: Long, identityKey: String): Boolean = true
+        override fun ownerOfIdentity(identityKey: String): Long? = null
+        override fun update(
+            cminmsgsId: Long, state: ProcStatus, nextAttemptAt: Instant?, attempts: Int?,
+            errorClass: ErrorClass?, lastError: String?,
+        ) = Unit
+
+        override fun requeueByErrorClasses(errorClasses: List<ErrorClass>): Int {
+            requeueCalls += errorClasses
+            var n = 0
+            rows.keys.toList().forEach { id ->
+                val s = rows[id]!!
+                if (s.errorClass != null && s.errorClass in errorClasses &&
+                    (s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) {
+                    rows[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null)
+                    n++
+                }
+            }
+            return n
+        }
+    }
+
+    @Test
+    fun `only recoverable classes are requeued, MALFORMED stays terminal`() {
+        val repo = FakeRepo()
+        repo.seed(1, ProcStatus.DEAD, ErrorClass.CODEC_ERROR)
+        repo.seed(2, ProcStatus.DEAD, ErrorClass.MALFORMED)
+        repo.seed(3, ProcStatus.FAILED, ErrorClass.UNSUPPORTED)
+
+        val n = ReplayService(repo).replay(listOf(ErrorClass.CODEC_ERROR, ErrorClass.MALFORMED, ErrorClass.UNSUPPORTED))
+
+        assertEquals(2, n)
+        assertEquals(listOf(listOf(ErrorClass.CODEC_ERROR, ErrorClass.UNSUPPORTED)), repo.requeueCalls)
+        assertEquals(ProcStatus.PENDING, repo.rows[1]!!.state)
+        assertEquals(0, repo.rows[1]!!.attempts)
+        assertNull(repo.rows[1]!!.nextAttemptAt)
+        assertEquals(ProcStatus.DEAD, repo.rows[2]!!.state)   // MALFORMED 永不被重放
+        assertEquals(ProcStatus.PENDING, repo.rows[3]!!.state)
+    }
+
+    @Test
+    fun `requesting only MALFORMED is a no-op`() {
+        val repo = FakeRepo()
+        repo.seed(2, ProcStatus.DEAD, ErrorClass.MALFORMED)
+
+        val n = ReplayService(repo).replay(listOf(ErrorClass.MALFORMED))
+
+        assertEquals(0, n)
+        assertNull(repo.requeueCalls.lastOrNull())   // 白名单过滤后为空 → 不触达仓储
+        assertEquals(ProcStatus.DEAD, repo.rows[2]!!.state)
+    }
+
+    @Test
+    fun `replayAll reopens every recoverable class including EXHAUSTED`() {
+        val repo = FakeRepo()
+        repo.seed(4, ProcStatus.FAILED, ErrorClass.INFRA)
+        repo.seed(5, ProcStatus.DEAD, ErrorClass.EXHAUSTED)
+        repo.seed(6, ProcStatus.DEAD, ErrorClass.MALFORMED)
+
+        val n = ReplayService(repo).replayAll()
+
+        assertEquals(2, n)
+        assertEquals(ProcStatus.PENDING, repo.rows[4]!!.state)
+        assertEquals(ProcStatus.PENDING, repo.rows[5]!!.state)
+        assertEquals(ProcStatus.DEAD, repo.rows[6]!!.state)
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/nextgen/processing/MessageProcessorTest.kt

@@ -1,5 +1,6 @@
 package com.gzzn.omms.msgexchange.nextgen.processing
 
+import com.gzzn.omms.msgexchange.nextgen.MutableClock
 import com.gzzn.omms.msgexchange.nextgen.codec.DecodeFailure
 import com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult
 import com.gzzn.omms.msgexchange.nextgen.codec.XmlCodec
@@ -25,24 +26,27 @@ import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ReqTrackRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.FlightStateRepository
 import com.gzzn.omms.msgexchange.nextgen.infra.redis.FlightRedisClient
 import com.gzzn.omms.msgexchange.nextgen.infra.redis.RedisScript
+import com.gzzn.omms.msgexchange.nextgen.infra.retry.FailureScheduler
+import com.gzzn.omms.msgexchange.nextgen.infra.retry.ProcFailure
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertNotNull
 import org.junit.jupiter.api.Assertions.assertNull
+import org.junit.jupiter.api.Assertions.assertThrows
 import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Test
 import java.time.Instant
 
 /**
- * U10/U11/U08（N21/T06/N06/N28）：processOne 失败语义——
- * 未注册 handler / CODEC_ERROR / staging 未实装 → FAILED（可重放，带退避，绝不写终态）；
- * MALFORMED → DEAD（不重试）；FAILED attempts≥maxAttempts 入口升级 DEAD；成功路径事件+回填+SUCCEEDED。
+ * U08/U10/U11：processOne 失败闭环——
+ * ① 意外异常在边界内写 FAILED(INFRA)+退避，重试达上限转 DEAD(EXHAUSTED)（Pump 内部异常不杀泵）；
+ * ② 未注册/CODEC_ERROR/staging 未实装 → FAILED（绝不直接终态）；MALFORMED → DEAD；
+ * ③ InterruptedException/致命 Error 不被普通恢复逻辑吞掉。
  */
 class MessageProcessorTest {
 
-    // ---------- fakes ----------
     private class FakeProcState : ProcStateRepository {
         var record = mutableMapOf<Long, ProcState>()
-        val bound = mutableMapOf<String, Long>()          // identityKey -> owner
+        val bound = mutableMapOf<String, Long>()
 
         override fun insert(cminmsgsId: Long, state: ProcStatus) { record[cminmsgsId] = ProcState(cminmsgsId, state) }
 
@@ -72,16 +76,33 @@ class MessageProcessorTest {
             )
         }
 
+        override fun requeueByErrorClasses(errorClasses: List<ErrorClass>): Int {
+            var n = 0
+            record.keys.toList().forEach { id ->
+                val s = record[id]!!
+                if (s.errorClass != null && s.errorClass in errorClasses &&
+                    (s.state == ProcStatus.FAILED || s.state == ProcStatus.DEAD)) {
+                    record[id] = s.copy(state = ProcStatus.PENDING, attempts = 0, nextAttemptAt = null)
+                    n++
+                }
+            }
+            return n
+        }
+
         fun state(id: Long): ProcState = record.getValue(id)
     }
 
     private class FakeInbox : CminmsgInboxRepository {
         var raws = mutableMapOf<Long, String>()
+        var throwOnRawOf: Throwable? = null
         val backfilled = mutableListOf<List<Any?>>()
 
         override fun insertRaw(rawXml: String): Long = 0
 
-        override fun rawOf(cminmsgsId: Long): String? = raws[cminmsgsId]
+        override fun rawOf(cminmsgsId: Long): String? {
+            throwOnRawOf?.let { throw it }
+            return raws[cminmsgsId]
+        }
 
         override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) {
             backfilled += listOf(cminmsgsId, sndr, type, styp, seqn)
@@ -98,7 +119,7 @@ class MessageProcessorTest {
         override fun markSent(eventId: Long) = Unit
         override fun markAllSent(eventIds: List<Long>) = Unit
         override fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int) = Unit
-        override fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String) = Unit
+        override fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String, attempts: Int?) = Unit
         override fun insertSync(events: List<MsgEvent>) = Unit
     }
 
@@ -126,6 +147,8 @@ class MessageProcessorTest {
     }
 
     // ---------- helpers ----------
+    private val clock = MutableClock(MutableClock.BASE)
+
     private fun msg(seqn: String) = DecodedMessage(
         meta = MetaFields(sndr = "AODB", type = "FLOP", styp = "DELY", seqn = seqn.toLong(), dttm = 20260906120000L),
         kind = MsgKind.Flop("DELY"),
@@ -148,8 +171,10 @@ class MessageProcessorTest {
         registry: HandlerRegistry = HandlerRegistry(emptyList()),
     ): MessageProcessor {
         val props = PipelineProps()
-        val snapshot = SnapshotFlow(procState, FakeRefData(), FakeRedis, props)
-        return MessageProcessor(inbox, procState, events, CodecHolder(codec), HandlerHolder(registry), FakeRedis, snapshot, props)
+        val scheduler = FailureScheduler(props, clock)
+        val procFailure = ProcFailure(procState, scheduler)
+        val snapshot = SnapshotFlow(procState, FakeRefData(), FakeRedis, procFailure)
+        return MessageProcessor(inbox, procState, events, CodecHolder(codec), HandlerHolder(registry), FakeRedis, snapshot, procFailure, props)
     }
 
     // ---------- tests ----------
@@ -177,6 +202,7 @@ class MessageProcessorTest {
         val s = ps.state(1)
         assertEquals(ProcStatus.FAILED, s.state)
         assertEquals(ErrorClass.CODEC_ERROR, s.errorClass)
+        assertEquals(1, s.attempts)
         assertNotNull(s.nextAttemptAt)
     }
 
@@ -193,15 +219,13 @@ class MessageProcessorTest {
     }
 
     @Test
-    fun `failed head at max attempts upgrades to DEAD at entry`() {
+    fun `legacy failed head at max attempts upgrades to DEAD at entry`() {
         val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents()
         ps.record[1] = head(status = ProcStatus.FAILED, attempts = 5, identityKey = "k")
         processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1))
 
-        val s = ps.state(1)
-        assertEquals(ProcStatus.DEAD, s.state)
-        assertEquals(ErrorClass.EXHAUSTED, s.errorClass)
-        assertTrue(inbox.raws.isEmpty() || inbox.backfilled.isEmpty())   // 入口即升级，未进入处理
+        assertEquals(ProcStatus.DEAD, ps.state(1).state)
+        assertEquals(ErrorClass.EXHAUSTED, ps.state(1).errorClass)
     }
 
     @Test
@@ -212,7 +236,6 @@ class MessageProcessorTest {
         processor(ps, inbox, ev, FakeCodec(), registry).processOne(ps.state(1))
 
         assertEquals(ProcStatus.SUCCEEDED, ps.state(1).state)
-        assertEquals(1, ev.inserted.size)
         val events = ev.inserted.single()
         assertEquals(listOf(Targets.KAFKA_MSG, Targets.KAFKA_SCHD), events.map { it.target })
         assertEquals("F1", events.first { it.target == Targets.KAFKA_SCHD }.partitionKey)
@@ -222,7 +245,7 @@ class MessageProcessorTest {
     @Test
     fun `duplicate identity leads to SKIPPED`() {
         val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents()
-        ps.record[1] = head(); ps.record[2] = head(id = 2)
+        ps.record[1] = head()
         inbox.raws[1] = "<MSG/>"
         ps.bound["AODB|FLOP|DELY|1"] = 2L   // 另一条消息已持有该键
 
@@ -233,4 +256,67 @@ class MessageProcessorTest {
         assertTrue(s.lastError == "duplicate-of:2")
         assertTrue(ev.inserted.isEmpty())
     }
+
+    // ---------- U08 边界化（评审要求①③） ----------
+    @Test
+    fun `unexpected exception maps to FAILED INFRA with backoff at boundary`() {
+        val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents()
+        ps.record[1] = head()
+        inbox.throwOnRawOf = RuntimeException("db-down")
+
+        processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1))
+
+        val s = ps.state(1)
+        assertEquals(ProcStatus.FAILED, s.state)
+        assertEquals(ErrorClass.INFRA, s.errorClass)
+        assertEquals(1, s.attempts)
+        assertNotNull(s.nextAttemptAt)
+        assertTrue(s.lastError == "db-down")
+    }
+
+    @Test
+    fun `repeated unexpected exceptions escalate to DEAD EXHAUSTED at max attempts`() {
+        val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents()
+        ps.record[1] = head()
+        inbox.throwOnRawOf = RuntimeException("db-down")
+        val p = processor(ps, inbox, ev, FakeCodec())
+
+        var attempts = 0
+        while (ps.state(1).state != ProcStatus.DEAD && attempts < 10) {
+            p.processOne(ps.state(1))
+            clock.advance(60_000)            // 推进时钟：让退避/毒丸时间语义确定
+            attempts++
+        }
+
+        val s = ps.state(1)
+        assertEquals(ProcStatus.DEAD, s.state, "should exhaust within maxAttempts(=5)")
+        assertEquals(ErrorClass.EXHAUSTED, s.errorClass)
+        assertTrue(s.lastError?.contains("attempts=5") == true)
+        assertEquals(5, s.attempts)
+    }
+
+    @Test
+    fun `fatal Error is not swallowed by exception recovery`() {
+        val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents()
+        ps.record[1] = head()
+        inbox.throwOnRawOf = AssertionError("fatal")
+
+        assertThrows(AssertionError::class.java) {
+            processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1))
+        }
+        // 未被普通恢复逻辑改写成 FAILED（Error 不落入 catch Exception）
+        assertEquals(ProcStatus.PENDING, ps.state(1).state)
+    }
+
+    @Test
+    fun `InterruptedException restores flag and propagates`() {
+        val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents()
+        ps.record[1] = head()
+        inbox.throwOnRawOf = InterruptedException("stop")
+
+        assertThrows(InterruptedException::class.java) {
+            processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1))
+        }
+        assertTrue(Thread.interrupted(), "interrupt flag must be restored")
+    }
 }