fix(processing): WP1 核心语义（ACM2-10 U06/U08/U10/U11）

- U06/N03：Dispatcher 逐条循环显式排除 KAFKA_SCHD（唯一出口 flushSchd 批量聚合），
  修复 schd 事件被无条件 markSent 吞掉、flush 永远 claim 不到的整链死路；flush 失败整批
  保持 PENDING、lastFlush 仅成功后推进（N24）
- U08 部分：Pump/Dispatcher loop 异常隔离（单次异常不静默死亡，U12 补告警日志）；
  processOne FAILED 入口 attempts 毒丸升级 DEAD（防退避到期后无限重试）；
  N28：backoffFor(attempt≤0) 下界保护不再抛异常
- U10/N21：未注册 handler → FAILED(UNSUPPORTED)+退避（绝不写终态）；占位安全化——
  HistorySweep.pickHistory 占位改空集（T07 修订）、SnapshotFlow staging 未实装改
  FAILED(UNSUPPORTED)+退避而非 DEAD（N05），CAS 冲突 FAILED(INFRA)+退避（N06 紧循环修复）
- U11/T06：decode 失败差异化——MALFORMED→DEAD 不重试；CODEC_ERROR→FAILED 可重放
- 新增 ErrorClass.UNSUPPORTED；单测：DispatcherTickTest(3)/MessageProcessorTest(6)/
  PipelinePropsTest(2)，19 个测试全绿

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/config/PipelineProps.kt

@@ -29,8 +29,11 @@ class PipelineProps {
         var backoffCapMs: Long = 60_000
         var headDeadline: Duration = Duration.ofMinutes(10)     // 最坏 HOL 上界（毒丸升级）
 
-        fun backoffFor(attempt: Int): Long =
-            backoffMs.drop(attempt - 1).firstOrNull()?.coerceAtMost(backoffCapMs) ?: backoffCapMs
+        /** N28：attempt ≤ 0（如 FAILED 未递增 attempts 的行）不得抛异常，取下界=首档退避。 */
+        fun backoffFor(attempt: Int): Long {
+            val index = (attempt - 1).coerceAtLeast(0)
+            return backoffMs.getOrNull(index)?.coerceAtMost(backoffCapMs) ?: backoffCapMs
+        }
     }
 
     @ConfigurationProperties("schd")

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

@@ -24,7 +24,8 @@ interface DeliveryPort {
 
 /**
  * ACMA-8 流程 3：投递调度——每 target 严格 FIFO（I1 双层同策略）；
- * 阶段 B（定案 2）：ES 投递成功后同线程同步 enqueue REDIS:flightInfo 删除事件。
+ * N03（U06）：KAFKA_SCHD 不走逐条循环（唯一出口是 flushSchd 批量聚合），
+ * 逐条循环显式排除，避免 schd 事件被无条件 markSent 吞掉、flush 永远 claim 不到。
  */
 @Singleton
 class Dispatcher(
@@ -39,17 +40,24 @@ class Dispatcher(
 
     fun loop() {
         while (running) {
-            tick()
-            Thread.sleep(200)
+            try {
+                tick()
+            } catch (e: Exception) {
+                // U08/R02：投递循环不得因单次异常静默死亡；U12 补 ERROR 日志与告警出口。
+                sleepQuietly(props.pipeline.pollInterval)
+            }
+            // N18：轮询间隔取参数表（下限 50ms，避免退避节律被吞）
+            sleepQuietly(props.pipeline.pollInterval.coerceAtLeast(Duration.ofMillis(50)))
         }
     }
 
     internal fun tick() {
         val targets = if (props.phase == PipelineProps.Phase.A) Targets.phaseA else Targets.phaseB
         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()) {
-                // 队头退避未到期：等待，不跳过（保序）；毒丸升级由实装补 attempts/deadline 判定
+                // 队头退避未到期：等待，不跳过（保序）；超时升级归 U13（WP2）
                 continue
             }
             try {
@@ -72,15 +80,22 @@ class Dispatcher(
                 }
             }
         }
-        if (Duration.between(lastFlush, Instant.now()) >= props.schd.flushPeriod) {
-            lastFlush = Instant.now()
-            flushSchd()
+        if (flushDue()) {
+            try {
+                flushSchd()
+            } catch (e: Exception) {
+                // U08/N24：批发送失败不得终止 loop——事件保持 PENDING，下个 flush 周期整批重试；
+                // 毒丸/告警随 U12/U13 补齐。lastFlush 仅在成功后推进（见 flushSchd）。
+            }
         }
     }
 
+    private fun flushDue(): Boolean =
+        Duration.between(lastFlush, Instant.now()) >= props.schd.flushPeriod
+
     private fun deliver(target: String, e: MsgEvent) = when (target) {
         Targets.KAFKA_MSG -> port.sendKafka("msg", e.payloadJson)
-        Targets.KAFKA_SCHD -> Unit          // schd 走 flushSchd 批量路径
+        Targets.KAFKA_SCHD -> error("KAFKA_SCHD must go through flushSchd (N03)")
         Targets.ES_FLIGHT_HTS -> port.indexFlightHts(e.payloadJson)
         Targets.REDIS_FLIGHT_INFO -> port.projectRedis(e.payloadJson)
         else -> error("unknown target $target")
@@ -91,9 +106,17 @@ class Dispatcher(
     /** 流程 3 flushSchd：批上限 + 每 FLID 最新一态聚合（topic "schd"，wire=FLTR JSON 数组）。 */
     internal fun flushSchd() {
         val pending = msgEvents.claimBatch(Targets.KAFKA_SCHD, limit = props.schd.flushLimit)
-        if (pending.isEmpty()) return
+        if (pending.isEmpty()) {
+            lastFlush = Instant.now()
+            return
+        }
         val payload = SchdAggregation.latestPerFlight(pending).joinToString(",", "[", "]")
-        port.sendKafka("schd", payload)             // 失败整批退避（at-least-once）
+        port.sendKafka("schd", payload)             // 失败整批退避（at-least-once）：异常上抛由 tick 隔离，事件保持 PENDING
         msgEvents.markAllSent(pending.mapNotNull { it.eventId })
+        lastFlush = Instant.now()                   // N24：仅在成功后推进
+    }
+
+    private fun sleepQuietly(d: Duration) {
+        if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1))
     }
 }

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/domain/ProcState.kt

@@ -5,7 +5,7 @@ package com.gzzn.omms.msgexchange.nextgen.domain
  */
 enum class ProcStatus { PENDING, FAILED, SUCCEEDED, SKIPPED, DEAD }
 
-enum class ErrorClass { MALFORMED, CODEC_ERROR, EXHAUSTED, INFRA }
+enum class ErrorClass { MALFORMED, CODEC_ERROR, EXHAUSTED, INFRA, UNSUPPORTED }
 
 data class ProcState(
     val cminmsgsId: Long,

src/main/kotlin/com/gzzn/omms/msgexchange/nextgen/jobs/JobExecutor.kt

@@ -38,14 +38,16 @@ class HistorySweepJob(
 ) {
     fun run() {
         val all = redis.hgetAllFlightInfo()
-        val history = pickHistory(all)                         // TODO(阶段2): 沿用现役判史规则（KEEP）
-        val success = emptyList<String>()                      // TODO(阶段2): esFlightHts.saveSync(history)
+        val history = pickHistory(all)          // U10/T07：saveSync 接线前判史为空集——禁止“全量可删”默认
+        val success = emptyList<String>()       // TODO(阶段2): esFlightHts.saveSync(history)（返回成功集后再接线删除）
         if (success.isNotEmpty()) {
             redis.eval(RedisScript.BATCH_DELETE, delFields = success)
         }
     }
 
-    private fun pickHistory(all: Map<String, String>): Map<String, String> = all   // TODO(阶段2)
+    /** U10/T07（修订）：接入 ES success 集之前的门禁——占位默认返回空集；
+     *  现役五条判史规则（SODT 3 天 / CNCL 1 小时 / 备降 / 离港 / 到港）golden 通过后才允许接线。 */
+    private fun pickHistory(all: Map<String, String>): Map<String, String> = emptyMap()   // TODO(阶段2)
 }
 
 /** 流程 5 runJob(ARCHIVE)：3:00 归档——1 天前且仅终态可迁（矩阵 #12）。 */

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

@@ -1,8 +1,10 @@
 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.MsgEvent
+import com.gzzn.omms.msgexchange.nextgen.domain.ProcState
 import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus
 import com.gzzn.omms.msgexchange.nextgen.domain.Targets
 import com.gzzn.omms.msgexchange.nextgen.infra.persistence.CminmsgInboxRepository
@@ -33,7 +35,13 @@ class Pump(
 
     fun loop() {
         while (running) {
-            tick()
+            try {
+                tick()
+            } catch (t: Throwable) {
+                // U08/R02：主泵不得因单次异常静默死亡——任一 codec/DB/Redis 抖动不得终止队列；
+                // 具体失败落 FAILED/DEAD 与告警见 MessageProcessor/U12。
+                sleepQuietly(props.pipeline.pollInterval)
+            }
         }
     }
 
@@ -48,25 +56,25 @@ class Pump(
             head == null -> sleepQuietly(props.pipeline.pollInterval)
             head.state == ProcStatus.FAILED && (head.nextAttemptAt ?: Instant.EPOCH) > Instant.now() ->
                 if (poisoned(head)) {
-                    // 毒丸出队：DLQ + 告警，队列继续（I1）
-                    procState.update(
-                        head.cminmsgsId, ProcStatus.DEAD,
-                        errorClass = ErrorClass.EXHAUSTED,
-                        lastError = head.lastError ?: "head-deadline-exceeded",
-                    )
+                    upgradeDead(head, head.lastError ?: "head-deadline-exceeded")
                 } else {
                     sleepQuietly(Duration.between(Instant.now(), head.nextAttemptAt))
                 }
-            else -> processor.processOne(head.cminmsgsId)
+            // PENDING、或 FAILED 退避已到期：交处理入口（入口再做 attempts 毒丸守卫）
+            else -> processor.processOne(head)
         }
     }
 
-    private fun poisoned(head: com.gzzn.omms.msgexchange.nextgen.domain.ProcState): Boolean =
+    private fun poisoned(head: ProcState): Boolean =
         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: com.gzzn.omms.msgexchange.nextgen.domain.ProcState?) =
+    private fun jobBefore(job: PumpJobRepository.Job, head: ProcState?) =
         head == null || head.state == ProcStatus.FAILED
 
     private fun execute(job: PumpJobRepository.Job) {
@@ -84,7 +92,12 @@ class Pump(
     }
 }
 
-/** ACMA-8 流程 2 processOne：解码→绑定→纯函数决策→提交。 */
+/**
+ * ACMA-8 流程 2 processOne：解码→绑定→纯函数决策→提交。
+ * U08/U10/U11（N06/N21/T06）：统一失败治理——「未实装/未注册/CODEC_ERROR/staging 未实装」
+ * 一律 FAILED（可重放，带 attempts+nextAttemptAt，绝不直接写终态）；仅 MALFORMED（报文非法）
+ * 与「重试耗尽」写 DEAD 终态。attempts ≥ maxAttempts 的 FAILED 在入口即升级 DEAD（防退避到期后无限重试）。
+ */
 @Singleton
 class MessageProcessor(
     private val inbox: CminmsgInboxRepository,
@@ -96,28 +109,38 @@ class MessageProcessor(
     private val snapshotFlow: SnapshotFlow,
     private val props: PipelineProps,
 ) {
-    fun processOne(cminmsgsId: Long) {
-        val raw = inbox.rawOf(cminmsgsId) ?: run {
-            procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "raw-missing")
+    fun processOne(head: ProcState) {
+        // U08 入口守卫：FAILED 行退避到期后再处理前，先判 attempts 毒丸（与 tick poisoned 同值）
+        if (head.state == ProcStatus.FAILED && head.attempts >= props.pipeline.maxAttempts) {
+            procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.EXHAUSTED,
+                lastError = head.lastError ?: "max-attempts")
+            return
+        }
+
+        val raw = inbox.rawOf(head.cminmsgsId) ?: run {
+            procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "raw-missing")
             return
         }
         val decoded = when (val r = codecHolder.codec.decode(raw)) {
             is com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult.Ok -> r.message
             is com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult.Err -> {
-                // MALFORMED 不重试；CODEC_ERROR 可一键重放（errorClass 入库）
-                procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = r.failure.errorClass, lastError = r.failure.detail)
+                // T06（U11）：MALFORMED（报文非法）→ DEAD 不重试；CODEC_ERROR（可随 codec 修复重放）→ FAILED 退避
+                if (r.failure.errorClass == ErrorClass.MALFORMED) {
+                    procState.update(head.cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED,
+                        lastError = r.failure.detail)
+                } else {
+                    failWithBackoff(head, r.failure.errorClass, r.failure.detail)
+                }
                 return
             }
         }
 
-        val state = procState.headUnfinished()
-        // I3：identity 仅首次绑定；FAILED 重试不重绑（绑定失败 = 另一条同键消息 → SKIPPED）
-        // TODO(阶段1后续): 以行内 IDENTITY_KEY 判定而非 head 复查（含 FAILED 持久化字段读取）
-        if (state != null && state.cminmsgsId == cminmsgsId && state.identityKey == null) {
+        // I3：identity 仅首次绑定（head.identityKey == null）；FAILED 重试不重绑（N25：入参传递 head，不再二次查询）
+        if (head.identityKey == null) {
             val identity = Identity.of(decoded, props.identity)
-            if (!procState.tryBindIdentity(cminmsgsId, identity)) {
+            if (!procState.tryBindIdentity(head.cminmsgsId, identity)) {
                 val owner = procState.ownerOfIdentity(identity) ?: -1L
-                procState.update(cminmsgsId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner")
+                procState.update(head.cminmsgsId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner")
                 return
             }
         }
@@ -125,14 +148,15 @@ class MessageProcessor(
         // 快照消息走专属流程（流程 4：staging→Lua→putGenIfVersion CAS 同事务）
         val handler = handlers.registry.dispatcherFor(decoded)
         if (handler == null) {
-            procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = "no-handler:${decoded.typeTag}")
+            // U10（N21）：未注册 ≠ 报文非法——写 FAILED（可重放），绝不写终态
+            failWithBackoff(head, ErrorClass.UNSUPPORTED, "no-handler:${decoded.typeTag}")
             return
         }
         val schdKind = decoded.kind as? com.gzzn.omms.msgexchange.nextgen.domain.MsgKind.Schd
         if (schdKind != null &&
             schdKind.subtype == com.gzzn.omms.msgexchange.nextgen.domain.MsgKind.SchdSubtype.DNLD
         ) {
-            snapshotFlow.publishSnapshot(cminmsgsId, decoded)
+            snapshotFlow.publishSnapshot(head, decoded)
             return
         }
 
@@ -147,10 +171,22 @@ class MessageProcessor(
             decision.schdPush.forEach { add(MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = it.flid, payloadJson = it.fltrJson)) }
         }
         msgEvents.insertAll(events)
-        inbox.backfillOnSuccess(cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn)
-        procState.update(cminmsgsId, ProcStatus.SUCCEEDED)
+        inbox.backfillOnSuccess(head.cminmsgsId, decoded.meta.sndr, decoded.meta.type, decoded.meta.styp, decoded.meta.seqn)
+        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,31 +1,36 @@
 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
 import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus
 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 jakarta.inject.Singleton
+import java.time.Instant
 
 /**
  * ACMA-8 流程 4：日计划快照（generation，主泵内执行）。
  * staging（内存瞬态，崩溃从 raw 整包重放）→ Lua 原子“覆盖+按代差删”（I4/I5）
- * → putGenIfVersion CAS + SUCCEEDED 同一 MySQL 事务（重放幂等，版本不二次自增）。
- * 整包失败 = 现役等价（KEEP）；跳坏行进 quarantine = CONFIRM（矩阵 #9）。
+ * → putGenIfVersion CAS + SUCCEEDED（重放幂等，版本不二次自增——恢复协议细节见 U09）。
+ * U10/T07 占位安全化：staging 未实装 → FAILED(UNSUPPORTED)+退避（可重放），绝不写 DEAD 终态；
+ * CAS 冲突 → FAILED(INFRA)+退避（不再无 nextAttemptAt 紧循环，N06/N28）。
  */
 @Singleton
 class SnapshotFlow(
     private val procState: ProcStateRepository,
     private val refData: RefDataRepository,
     private val redis: FlightRedisClient,
+    private val props: PipelineProps,
 ) {
-    fun publishSnapshot(cminmsgsId: Long, msg: DecodedMessage) {
+    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) {
-            procState.update(cminmsgsId, ProcStatus.DEAD, errorClass = ErrorClass.MALFORMED, lastError = staged.reason)
+            failWithBackoff(head, ErrorClass.UNSUPPORTED, staged.reason)   // U10：未实装 → 可重放，非终态
             return
         }
         val normalized = (staged as StageResult.Ok).flights            // (flid, payloadJson)
@@ -43,11 +48,22 @@ class SnapshotFlow(
             // 重放路径：version 已是目标值 → no-op 视为成功
             val again = refData.getGen(day)
             if (again == null || again.version != expected + 1) {
-                procState.update(cminmsgsId, ProcStatus.FAILED, lastError = "gen-cas-conflict")
+                failWithBackoff(head, ErrorClass.INFRA, "gen-cas-conflict")   // N06/N28：带退避，禁止紧循环
                 return
             }
         }
-        procState.update(cminmsgsId, ProcStatus.SUCCEEDED)
+        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 结果（骨架）。 */

src/main/resources/db/migration/V2.0.0__aux_tables.sql

@@ -15,7 +15,7 @@ CREATE TABLE PROC_STATE (
   IDENTITY_KEY    VARCHAR(200) NULL,       -- SNDR|TYPE|STYP|SEQN，decode 后首次绑定（I3）
   ATTEMPTS        INT          NOT NULL DEFAULT 0,
   NEXT_ATTEMPT_AT TIMESTAMP    NULL,
-  ERROR_CLASS     VARCHAR(20)  NULL,       -- MALFORMED/CODEC_ERROR/EXHAUSTED/INFRA
+  ERROR_CLASS     VARCHAR(20)  NULL,       -- MALFORMED/CODEC_ERROR/EXHAUSTED/INFRA/UNSUPPORTED
   LAST_ERROR      VARCHAR(1000) NULL,
   UPDATED_AT      TIMESTAMP    NOT NULL,
   UNIQUE KEY UK_PROC_IDENTITY (IDENTITY_KEY),

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

@@ -0,0 +1,27 @@
+package com.gzzn.omms.msgexchange.nextgen.config
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+/**
+ * U08/N28：backoffFor 必须对 attempt ≤ 0（FAILED 行未递增 attempts）给出首档退避而非抛异常。
+ */
+class PipelinePropsTest {
+
+    private val pipeline = PipelineProps().pipeline
+
+    @Test
+    fun `backoff follows table then caps`() {
+        assertEquals(1000, pipeline.backoffFor(1))
+        assertEquals(2000, pipeline.backoffFor(2))
+        assertEquals(16000, pipeline.backoffFor(5))
+        assertEquals(60_000, pipeline.backoffFor(6))   // 表外 → 封顶
+        assertEquals(60_000, pipeline.backoffFor(99))
+    }
+
+    @Test
+    fun `non-positive attempt never throws and falls back to first slot`() {
+        assertEquals(1000, pipeline.backoffFor(0))
+        assertEquals(1000, pipeline.backoffFor(-1))
+    }
+}

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

@@ -0,0 +1,133 @@
+package com.gzzn.omms.msgexchange.nextgen.delivery
+
+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 org.junit.jupiter.api.Assertions.assertEquals
+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）。
+ */
+class DispatcherTickTest {
+
+    private class FakeRepo : MsgEventRepository {
+        val events = mutableListOf<MsgEvent>()
+
+        fun enqueue(e: MsgEvent) { events += e }
+
+        override fun insertAll(events: List<MsgEvent>) = events.map { it.eventId ?: 0L }
+
+        override fun headUnsent(target: String): MsgEvent? =
+            events.filter { it.target == target && it.state != EventStatus.SENT && it.state != EventStatus.DEAD }
+                .minByOrNull { it.eventId ?: Long.MAX_VALUE }
+
+        override fun claimBatch(target: String, limit: Int): List<MsgEvent> =
+            events.filter { it.target == target && it.state == EventStatus.PENDING }
+                .sortedBy { it.eventId ?: Long.MAX_VALUE }
+                .take(limit)
+
+        override fun markSent(eventId: Long) {
+            val i = events.indexOfFirst { it.eventId == eventId }
+            if (i >= 0) events[i] = events[i].copy(state = EventStatus.SENT)
+        }
+
+        override fun markAllSent(eventIds: List<Long>) = eventIds.forEach(::markSent)
+
+        override fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int) {
+            val i = events.indexOfFirst { it.eventId == eventId }
+            if (i >= 0) events[i] = events[i].copy(state = EventStatus.PENDING, attempts = attempts, nextAttemptAt = nextAttemptAt)
+        }
+
+        override fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String) {
+            val i = events.indexOfFirst { it.eventId == eventId }
+            if (i >= 0) events[i] = events[i].copy(state = EventStatus.DEAD, errorClass = errorClass, lastError = lastError)
+        }
+
+        override fun insertSync(events: List<MsgEvent>) = Unit
+    }
+
+    private class FakePort : DeliveryPort {
+        val sent = mutableListOf<Pair<String, String>>()
+        var failTopic: String? = null
+
+        override fun sendKafka(topic: String, payloadJson: String) {
+            if (failTopic == topic) throw RuntimeException("send-fail:$topic")
+            sent += topic to payloadJson
+        }
+
+        override fun indexFlightHts(payloadJson: String) = Unit
+        override fun projectRedis(payloadJson: String) = Unit
+    }
+
+    private fun props(): PipelineProps = PipelineProps()   // schd.flush-period 默认 3s
+
+    private fun ev(id: Long, target: String, key: String?, payload: String) =
+        MsgEvent(eventId = id, target = target, partitionKey = key, payloadJson = payload, state = EventStatus.PENDING)
+
+    @Test
+    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())
+        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()
+        assertEquals(EventStatus.PENDING, repo.events.first { it.eventId == 4L }.state)
+        assertEquals(1, port.sent.count { it.first == "schd" })   // 第二次 tick 未额外发送
+    }
+
+    @Test
+    fun `flushSchd aggregates latest per flight and marks all sent`() {
+        val repo = FakeRepo()
+        val port = FakePort()
+        val d = Dispatcher(repo, port, props())
+        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`() {
+        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"}"""))
+
+        assertFails { d.flushSchd() }              // 发送失败 → 异常上抛（tick 侧隔离），事件保持 PENDING
+        assertEquals(EventStatus.PENDING, repo.events.single().state)
+        assertEquals(0, port.sent.size)
+
+        port.failTopic = null
+        d.flushSchd()                              // lastFlush 未推进 → 下个周期整批重试成功
+        assertEquals(EventStatus.SENT, repo.events.single().state)
+        assertEquals(1, port.sent.count { it.first == "schd" })
+    }
+}

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

@@ -0,0 +1,236 @@
+package com.gzzn.omms.msgexchange.nextgen.processing
+
+import com.gzzn.omms.msgexchange.nextgen.codec.DecodeFailure
+import com.gzzn.omms.msgexchange.nextgen.codec.DecodeResult
+import com.gzzn.omms.msgexchange.nextgen.codec.XmlCodec
+import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps
+import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage
+import com.gzzn.omms.msgexchange.nextgen.domain.Decision
+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.MsgKind
+import com.gzzn.omms.msgexchange.nextgen.domain.MetaFields
+import com.gzzn.omms.msgexchange.nextgen.domain.NotifyPayload
+import com.gzzn.omms.msgexchange.nextgen.domain.ProcState
+import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus
+import com.gzzn.omms.msgexchange.nextgen.domain.SchdPush
+import com.gzzn.omms.msgexchange.nextgen.domain.Targets
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.CminmsgInboxRepository
+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.persistence.RefDataRepository
+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 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.Instant
+
+/**
+ * U10/U11/U08（N21/T06/N06/N28）：processOne 失败语义——
+ * 未注册 handler / CODEC_ERROR / staging 未实装 → FAILED（可重放，带退避，绝不写终态）；
+ * MALFORMED → DEAD（不重试）；FAILED attempts≥maxAttempts 入口升级 DEAD；成功路径事件+回填+SUCCEEDED。
+ */
+class MessageProcessorTest {
+
+    // ---------- fakes ----------
+    private class FakeProcState : ProcStateRepository {
+        var record = mutableMapOf<Long, ProcState>()
+        val bound = mutableMapOf<String, Long>()          // identityKey -> owner
+
+        override fun insert(cminmsgsId: Long, state: ProcStatus) { record[cminmsgsId] = ProcState(cminmsgsId, state) }
+
+        override fun headUnfinished(): ProcState? = null
+
+        override fun tryBindIdentity(cminmsgsId: Long, identityKey: String): Boolean {
+            val owner = bound[identityKey]
+            if (owner != null && owner != cminmsgsId) return false
+            bound[identityKey] = cminmsgsId
+            record[cminmsgsId] = record[cminmsgsId]?.copy(identityKey = identityKey) ?: ProcState(cminmsgsId, ProcStatus.PENDING, identityKey = identityKey)
+            return true
+        }
+
+        override fun ownerOfIdentity(identityKey: String): Long? = bound[identityKey]
+
+        override fun update(
+            cminmsgsId: Long, state: ProcStatus, nextAttemptAt: Instant?, attempts: Int?,
+            errorClass: ErrorClass?, lastError: String?,
+        ) {
+            val old = record[cminmsgsId] ?: ProcState(cminmsgsId, state)
+            record[cminmsgsId] = old.copy(
+                state = state,
+                nextAttemptAt = nextAttemptAt ?: old.nextAttemptAt,
+                attempts = attempts ?: old.attempts,
+                errorClass = errorClass ?: old.errorClass,
+                lastError = lastError ?: old.lastError,
+            )
+        }
+
+        fun state(id: Long): ProcState = record.getValue(id)
+    }
+
+    private class FakeInbox : CminmsgInboxRepository {
+        var raws = mutableMapOf<Long, String>()
+        val backfilled = mutableListOf<List<Any?>>()
+
+        override fun insertRaw(rawXml: String): Long = 0
+
+        override fun rawOf(cminmsgsId: Long): String? = raws[cminmsgsId]
+
+        override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) {
+            backfilled += listOf(cminmsgsId, sndr, type, styp, seqn)
+        }
+    }
+
+    private class FakeEvents : MsgEventRepository {
+        val inserted = mutableListOf<List<MsgEvent>>()
+
+        override fun insertAll(events: List<MsgEvent>): List<Long> { inserted += events; return events.indices.map { it.toLong() + 1 } }
+
+        override fun headUnsent(target: String): MsgEvent? = null
+        override fun claimBatch(target: String, limit: Int): List<MsgEvent> = emptyList()
+        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 insertSync(events: List<MsgEvent>) = Unit
+    }
+
+    private class FakeCodec : XmlCodec {
+        var result: DecodeResult = DecodeResult.Ok(
+            DecodedMessage(
+                meta = MetaFields(sndr = "AODB", type = "FLOP", styp = "DELY", seqn = 1L, dttm = 20260906120000L),
+                kind = MsgKind.Flop("DELY"),
+                rawXml = "<MSG/>",
+            ),
+        )
+        override fun decode(rawXml: String): DecodeResult = result
+        override fun encodeRqrd(kind: String, rangeJson: String): String = ""
+    }
+
+    private object FakeRedis : FlightRedisClient {
+        override fun eval(script: RedisScript, setPairs: List<Pair<String, String>>, delFields: List<String>) = Unit
+        override fun hgetAllFlightInfo(): Map<String, String> = emptyMap()
+    }
+
+    private class FakeRefData : RefDataRepository {
+        override fun getGen(day: String): RefDataRepository.GenMeta? = null
+        override fun putGenIfVersion(day: String, expected: Long, new: RefDataRepository.GenMeta): Boolean = true
+        override fun upsertAll(rows: List<Triple<String, String, String>>) = Unit
+    }
+
+    // ---------- helpers ----------
+    private fun msg(seqn: String) = DecodedMessage(
+        meta = MetaFields(sndr = "AODB", type = "FLOP", styp = "DELY", seqn = seqn.toLong(), dttm = 20260906120000L),
+        kind = MsgKind.Flop("DELY"),
+        rawXml = "<MSG/>",
+    )
+
+    private fun head(id: Long = 1, status: ProcStatus = ProcStatus.PENDING, attempts: Int = 0, identityKey: String? = null) =
+        ProcState(id, status, identityKey = identityKey, attempts = attempts)
+
+    private fun deliveHandler() = object : Handler {
+        override val kind: MsgKind = MsgKind.Flop("DELY")
+        override fun decide(flightView: Map<String, String>, msg: DecodedMessage): Decision = Decision(
+            msgNotifies = listOf(NotifyPayload("""{"n":1}""")),
+            schdPush = listOf(SchdPush(flid = "F1", fltrJson = """{"FLID":"F1","v":2}""")),
+        )
+    }
+
+    private fun processor(
+        procState: FakeProcState, inbox: FakeInbox, events: FakeEvents, codec: FakeCodec,
+        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)
+    }
+
+    // ---------- tests ----------
+    @Test
+    fun `no handler is FAILED UNSUPPORTED with backoff - never terminal`() {
+        val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents()
+        ps.record[1] = head(); inbox.raws[1] = "<MSG/>"
+        processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1))
+
+        val s = ps.state(1)
+        assertEquals(ProcStatus.FAILED, s.state)
+        assertEquals(ErrorClass.UNSUPPORTED, s.errorClass)
+        assertEquals(1, s.attempts)
+        assertNotNull(s.nextAttemptAt)
+        assertTrue(s.lastError?.startsWith("no-handler:FLOP-DELY") == true)
+    }
+
+    @Test
+    fun `CODEC_ERROR decode failure is FAILED retryable`() {
+        val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents()
+        ps.record[1] = head(); inbox.raws[1] = "<MSG/>"
+        val codec = FakeCodec().apply { result = DecodeResult.Err(DecodeFailure(ErrorClass.CODEC_ERROR, "boom")) }
+        processor(ps, inbox, ev, codec).processOne(ps.state(1))
+
+        val s = ps.state(1)
+        assertEquals(ProcStatus.FAILED, s.state)
+        assertEquals(ErrorClass.CODEC_ERROR, s.errorClass)
+        assertNotNull(s.nextAttemptAt)
+    }
+
+    @Test
+    fun `MALFORMED decode failure stays DEAD terminal`() {
+        val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents()
+        ps.record[1] = head(); inbox.raws[1] = "<MSG/>"
+        val codec = FakeCodec().apply { result = DecodeResult.Err(DecodeFailure(ErrorClass.MALFORMED, "bad-xml")) }
+        processor(ps, inbox, ev, codec).processOne(ps.state(1))
+
+        val s = ps.state(1)
+        assertEquals(ProcStatus.DEAD, s.state)
+        assertEquals(ErrorClass.MALFORMED, s.errorClass)
+    }
+
+    @Test
+    fun `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())   // 入口即升级，未进入处理
+    }
+
+    @Test
+    fun `success path inserts events backfills and marks SUCCEEDED`() {
+        val ps = FakeProcState(); val inbox = FakeInbox(); val ev = FakeEvents()
+        ps.record[1] = head(); inbox.raws[1] = "<MSG/>"
+        val registry = HandlerRegistry(listOf(deliveHandler()))
+        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)
+        assertEquals(listOf(1L, "AODB", "FLOP", "DELY", 1L), inbox.backfilled.single())
+    }
+
+    @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)
+        inbox.raws[1] = "<MSG/>"
+        ps.bound["AODB|FLOP|DELY|1"] = 2L   // 另一条消息已持有该键
+
+        processor(ps, inbox, ev, FakeCodec()).processOne(ps.state(1))
+
+        val s = ps.state(1)
+        assertEquals(ProcStatus.SKIPPED, s.state)
+        assertTrue(s.lastError == "duplicate-of:2")
+        assertTrue(ev.inserted.isEmpty())
+    }
+}