feat: scaffold Micronaut+Kotlin service per ACMA-8 v4 architecture

- 4 modules (ingress/processing/delivery/reference) + codec + infra + jobs - Flyway V2.0.0: six aux tables (PROC_STATE/MSG_EVENT/REF_DATA/REQ_TRACK/PUMP_JOB/FLIGHT_STATE) - Lua: snapshot_replace (atomic cover+gen-diff-delete), batch_delete (3:30 sweep) - config: env-externalized secrets, phase A/B switch, ACMA-8 param-table defaults - logback: logstash TCP JSON + traceId MDC - pure-logic tests: identity (I3), schd aggregation max-by-EVENT_ID (FIX #7) - build verified: gradle compile+test green on JDK 25 (Micronaut 5.1 requires JVM 25+) Migrated from legacy repo subdirectory per repo-strategy decision (separate repo per service, org convention). Refs: ACMA-9, ACMA-8 v4, ACMA-6
2026-09-06 16:13:50 +08:00
commit b2f3b896cc
33 changed files with 1414 additions and 0 deletions
@@ -0,0 +1,5 @@
+.gradle/
+build/
+out/
+*.iml
+.idea/
@@ -0,0 +1,54 @@
+# msgexchange-api-v2（新一代消息交换服务）
+
+依据 **ACMA-8 v4 综合架构**（单写者严格 FIFO 管道 + 两阶段权威）与 **ACMA-6 技术选型**
+（Micronaut 5.1 + Kotlin 2.3）搭建的新一代消息交换服务工程。本仓库独立于 legacy
+`msgexchange-api`（Java 8 / Spring Boot 1.5 / Maven）——过渡期两套系统并存（影子对拍→
+切流→旧仓库冻结），legacy 维护不受本仓库影响。
+
+> **JDK 口径实测修正**：Micronaut 5.1 系构件（如 micronaut-http-server-netty:5.1.10）
+> 要求 JVM 25+，计划原定 JDK 21 不可行；工程已按 **JDK 25** 配置（ACMA-9 记录）。
+
+## 包结构 → ACMA-8 架构映射
+
+| 包 | 职责 | 对应 ACMA-8 |
+|---|---|---|
+| `ingress/` | Ingress & Inbox：接收事务（事务1），不解析报文 | 流程 1，I3 |
+| `processing/` | Processing 主泵：严格 FIFO 领取、identity 绑定、纯函数决策、事务2 | 流程 2/4，I1/I2/I5 |
+| `delivery/` | Delivery & Projection：每 target 严格 FIFO 投递、schd 聚合 | 流程 3 |
+| `reference/` | Reference & Query：21 类同步 + 15 类请求状态机 | 流程 6 |
+| `jobs/` | 泵作业：ARCHIVE / HISTORY_SWEEP / PROJECTION_REBUILD | 流程 4/5/7，I4 |
+| `codec/` | XML codec（阶段 1 先 vendor 复用 legacy POJO，见 ACMA-6 选型） | 决策 4 前置 |
+| `domain/` | 领域模型：Decision、事件、状态机枚举、Phase 开关 | I1–I5 |
+| `infra/` | 仓储接口、Redis Lua 装载、配置 | 数据模型节 |
+
+## 资源
+
+- `db/migration/V2.0.0__aux_tables.sql`：六表 DDL（PROC_STATE / MSG_EVENT / REF_DATA /
+  REQ_TRACK / PUMP_JOB / FLIGHT_STATE），与 ACMA-8 v4 数据模型节逐字一致；影子实例在
+  独立 schema 执行同一脚本。
+- `lua/snapshot_replace.lua`：同一 hash 原子“覆盖新代 + 按代差删”（流程 4，I4/I5）。
+- `lua/batch_delete.lua`：3:30 清场批量删除（仅 ES 写成功集，I4）。
+- `application.yml`：口令全部环境变量外置（零入库）；`msgx.phase` 为阶段 A/B 总开关；
+  pipeline 参数 = ACMA-8 参数表初值。
+
+## 未完成（按计划属于后续阶段，不是本脚手架遗漏）
+
+1. **Handler 业务（3+29）**：`processing/HandlerRegistry` 仅注册骨架，翻译属阶段 2/3。
+2. **codec 实装**：vendor 复用 legacy `entity/msg` POJO + Jackson XML（ACMA-6 选型），
+   阶段 1 后续项。
+3. **依赖版本锁定**：`gradle/libs.versions.toml` 中版本为计划口径，需阶段 0
+   「Micronaut×现网 Eureka 互操作冒烟 + logstash + ES REST」通过后固化。
+4. **仓储实装**：`infra/persistence/Repositories.kt` 目前是接口（Micronaut Data JDBC
+   实装属阶段 1 后续），主泵/调度循环以接口驱动，纯逻辑已抽离可单测。
+
+## 构建
+
+```bash
+gradle build          # 需网络拉取依赖；内网环境见 gradle.properties 注释
+gradle test           # 纯逻辑单测（identity / schd 聚合）
+```
+
+## 关联
+
+Plane `airport_chengdu_msgexchange_api`：ACMA-9（本阶段跟踪）、ACMA-8 v4（架构）、
+ACMA-6（技术选型）、ACMA-3（总计划）、ACMA-4（行为对拍基线）。
@@ -0,0 +1,55 @@
+plugins {
+    alias(libs.plugins.kotlin.jvm)
+    alias(libs.plugins.micronaut.application)
+}
+
+group = "com.gzzn.omms"
+version = "2.0.0-SNAPSHOT"
+
+repositories {
+    mavenCentral()
+}
+
+java {
+    toolchain {
+        // Micronaut 5.1 系构件要求 JVM 25+（实测依赖解析结论，修正计划的 JDK 21 口径）
+        languageVersion = JavaLanguageVersion.of(25)
+    }
+}
+
+kotlin {
+    jvmToolchain(25)
+}
+
+micronaut {
+    runtime("netty")
+    testRuntime("junit5")
+    processing {
+        incremental(true)
+        annotations("com.gzzn.omms.msgexchange.nextgen.*")
+    }
+}
+
+dependencies {
+    implementation(libs.bundles.micronaut.runtime)
+    implementation(libs.micronaut.data.jdbc)
+    implementation(libs.micronaut.jdbc.hikari)
+    implementation(libs.micronaut.flyway)
+    implementation(libs.micronaut.redis.lettuce)
+    implementation(libs.micronaut.kafka)
+    implementation(libs.micronaut.discovery.eureka)
+    implementation(libs.jackson.dataformat.xml)
+    implementation(libs.jackson.module.kotlin)
+    implementation(libs.logstash.logback.encoder)
+
+    runtimeOnly(libs.mysql.connector.j)
+    runtimeOnly("org.yaml:snakeyaml")
+
+    testImplementation(libs.junit.jupiter)
+    testImplementation("io.micronaut.test:micronaut-test-junit5")
+    testImplementation("org.jetbrains.kotlin:kotlin-test")
+}
+
+tasks.test {
+    useJUnitPlatform()
+}
@@ -0,0 +1,3 @@
+org.gradle.jvmargs=-Xmx1g
+kotlin.code.style=official
+# 阶段 0 锁定版本冒烟通过后，可将依赖镜像指向内网 Maven 仓库（沿用现网环境前提）
@@ -0,0 +1,34 @@
+# 版本为 ACMA-6 定稿口径（Micronaut 5.1 / Kotlin 2.3 / JDK 21）。
+# 注意：具体版本号在阶段 0「锁定版本互操作冒烟」（现网 Eureka/logstash/ES）通过后固化，
+# 见 ACMA-8 遗留开放点与 ACMA-9 Checks。
+[versions]
+micronaut = "5.1.0"
+kotlin = "2.3.0"
+junit = "5.11.4"
+
+[plugins]
+kotlin-jvm = { id = "org.jetbrains.kotlin.jvm", version.ref = "kotlin" }
+micronaut-application = { id = "io.micronaut.application", version = "4.6.0" }
+
+[libraries]
+micronaut-http-server-netty = { module = "io.micronaut:micronaut-http-server-netty" }
+micronaut-jackson-databind = { module = "io.micronaut:micronaut-jackson-databind" }
+micronaut-data-jdbc = { module = "io.micronaut.data:micronaut-data-jdbc" }
+micronaut-jdbc-hikari = { module = "io.micronaut.sql:micronaut-jdbc-hikari" }
+micronaut-flyway = { module = "io.micronaut.flyway:micronaut-flyway" }
+micronaut-redis-lettuce = { module = "io.micronaut.redis:micronaut-redis-lettuce" }
+micronaut-kafka = { module = "io.micronaut.kafka:micronaut-kafka" }
+micronaut-discovery-eureka = { module = "io.micronaut.discovery:micronaut-discovery-client" }
+micronaut-management = { module = "io.micronaut:micronaut-management" }
+jackson-dataformat-xml = { module = "com.fasterxml.jackson.dataformat:jackson-dataformat-xml", version = "2.18.2" }
+jackson-module-kotlin = { module = "com.fasterxml.jackson.module:jackson-module-kotlin", version = "2.18.2" }
+logstash-logback-encoder = { module = "net.logstash.logback:logstash-logback-encoder", version = "8.0" }
+mysql-connector-j = { module = "com.mysql:mysql-connector-j", version = "9.1.0" }
+junit-jupiter = { module = "org.junit.jupiter:junit-jupiter", version.ref = "junit" }
+
+[bundles]
+micronaut-runtime = [
+    "micronaut-http-server-netty",
+    "micronaut-jackson-databind",
+    "micronaut-management",
+]
@@ -0,0 +1,11 @@
+rootProject.name = "msgexchange-api-v2"
+
+pluginManagement {
+    repositories {
+        gradlePluginPortal()
+        mavenCentral()
+    }
+}
+
+// 新一代消息交换服务：ACMA-8 v4 架构（单写者严格 FIFO 管道 + 两阶段权威）。
+// 独立仓库，与 legacy msgexchange-api 并存至切流完成。
@@ -0,0 +1,7 @@
+package com.gzzn.omms.msgexchange.nextgen
+
+import io.micronaut.runtime.Micronaut
+
+fun main(args: Array<String>) {
+    Micronaut.run(*args)
+}
@@ -0,0 +1,24 @@
+package com.gzzn.omms.msgexchange.nextgen.codec
+
+import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage
+import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass
+
+/**
+ * ACMA-8 流程 2：解码失败分类（矩阵/状态机：MALFORMED 不重试；CODEC_ERROR 可一键重放）。
+ */
+data class DecodeFailure(val errorClass: ErrorClass, val detail: String)
+
+sealed interface DecodeResult {
+    data class Ok(val message: DecodedMessage) : DecodeResult
+    data class Err(val failure: DecodeFailure) : DecodeResult
+}
+
+/**
+ * XML codec：XXE 防护（禁 DTD/外部实体）；阶段 1 先 vendor 复用 legacy entity/msg POJO +
+ * jackson-dataformat-xml + JaxbAnnotationIntrospector（现役已验证组合，ACMA-6 选型）。
+ * TODO(阶段1后续): vendor POJO 引入与 BDPB workaround 配置化（golden 固化）。
+ */
+interface XmlCodec {
+    fun decode(rawXml: String): DecodeResult
+    fun encodeRqrd(kind: String, rangeJson: String): String
+}
@@ -0,0 +1,47 @@
+package com.gzzn.omms.msgexchange.nextgen.config
+
+import io.micronaut.context.annotation.ConfigurationProperties
+import java.time.Duration
+
+/**
+ * ACMA-8 参数表（v4）初值；阶段 0 现网基线校准。
+ */
+@ConfigurationProperties("msgx")
+class PipelineProps {
+    var phase: Phase = Phase.A
+    var serviceName: String = "msgexchangeapi"
+    var registerEureka: Boolean = true
+    var pipeline: Pipeline = Pipeline()
+    var schd: Schd = Schd()
+    var identity: Identity = Identity()
+    var consistencyCheck: ConsistencyCheck = ConsistencyCheck()
+
+    enum class Phase { A, B }
+
+    class Pipeline {
+        var pollInterval: Duration = Duration.ofSeconds(1)      // KEEP 现役节奏
+        var claimBatch: Int = 50
+        var maxAttempts: Int = 5                                // 处理/投递同值
+        var backoffMs: List<Long> = listOf(1000, 2000, 4000, 8000, 16000)
+        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
+    }
+
+    class Schd {
+        var flushPeriod: Duration = Duration.ofSeconds(3)       // KEEP 现役节律
+        var flushLimit: Int = 500
+    }
+
+    class Identity {
+        /** CONFIRM（矩阵 #11）：SEQN 重置作用域确认前保持 false，计算集中此处（I3）。 */
+        var includeDayBoundary: Boolean = false
+    }
+
+    class ConsistencyCheck {
+        var onStartup: Boolean = true                           // 阶段 A 必选哨兵
+        var dailySampleRatio: Double = 0.01
+    }
+}
@@ -0,0 +1,99 @@
+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 jakarta.inject.Singleton
+import java.time.Duration
+import java.time.Instant
+
+/** 对外投递端口（Kafka 同步确认；阶段 B 追加 ES/Redis 投影写入）。 */
+interface DeliveryPort {
+    /** Kafka 发送（同步确认，at-least-once）。topic 由 target 映射：KAFKA:msg→"msg"，KAFKA:schd→"schd"。 */
+    fun sendKafka(topic: String, payloadJson: String)
+
+    /** 阶段 B：ES flight_hts 写入。 */
+    fun indexFlightHts(payloadJson: String)
+
+    /** 阶段 B：Redis 投影写（仅阶段 B；I5：Delivery 阶段 A 不写 Redis）。 */
+    fun projectRedis(payloadJson: String)
+}
+
+/**
+ * ACMA-8 流程 3：投递调度——每 target 严格 FIFO（I1 双层同策略）；
+ * 阶段 B（定案 2）：ES 投递成功后同线程同步 enqueue REDIS:flightInfo 删除事件。
+ */
+@Singleton
+class Dispatcher(
+    private val msgEvents: MsgEventRepository,
+    private val port: DeliveryPort,
+    private val props: PipelineProps,
+) {
+    @Volatile
+    private var running = true
+
+    private var lastFlush: Instant = Instant.EPOCH
+
+    fun loop() {
+        while (running) {
+            tick()
+            Thread.sleep(200)
+        }
+    }
+
+    internal fun tick() {
+        val targets = if (props.phase == PipelineProps.Phase.A) Targets.phaseA else Targets.phaseB
+        for (t in targets) {
+            val head = msgEvents.headUnsent(t) ?: continue
+            if (head.state == EventStatus.PENDING && head.nextAttemptAt != null && head.nextAttemptAt > Instant.now()) {
+                // 队头退避未到期：等待，不跳过（保序）；毒丸升级由实装补 attempts/deadline 判定
+                continue
+            }
+            try {
+                deliver(t, head)
+                msgEvents.markSent(head.eventId!!)
+                if (t == Targets.ES_FLIGHT_HTS && props.phase == PipelineProps.Phase.B) {
+                    // 定案 2：ES 投递成功 → 同线程同步 enqueue 删除事件（不轮询 ack）
+                    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,
+                    )
+                }
+            }
+        }
+        if (Duration.between(lastFlush, Instant.now()) >= props.schd.flushPeriod) {
+            lastFlush = Instant.now()
+            flushSchd()
+        }
+    }
+
+    private fun deliver(target: String, e: MsgEvent) = when (target) {
+        Targets.KAFKA_MSG -> port.sendKafka("msg", e.payloadJson)
+        Targets.KAFKA_SCHD -> Unit          // schd 走 flushSchd 批量路径
+        Targets.ES_FLIGHT_HTS -> port.indexFlightHts(e.payloadJson)
+        Targets.REDIS_FLIGHT_INFO -> port.projectRedis(e.payloadJson)
+        else -> error("unknown target $target")
+    }
+
+    private fun deleteOf(e: MsgEvent) = """{"op":"delete","refs":${e.partitionKey ?: ""}}"""
+
+    /** 流程 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
+        val payload = SchdAggregation.latestPerFlight(pending).joinToString(",", "[", "]")
+        port.sendKafka("schd", payload)             // 失败整批退避（at-least-once）
+        msgEvents.markAllSent(pending.mapNotNull { it.eventId })
+    }
+}
@@ -0,0 +1,22 @@
+package com.gzzn.omms.msgexchange.nextgen.delivery
+
+import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent
+import com.gzzn.omms.msgexchange.nextgen.domain.SchdPush
+
+/**
+ * ACMA-8 流程 3 flushSchd 聚合（纯函数，可单测）：
+ * 按 PARTITION_KEY(=FLID) 分组、组内取 EVENT_ID 最大——v4 显式 max(EVENT_ID)，
+ * 不依赖集合遍历序（FIX：现役 buffer 无去重会重复投递旧值）。
+ */
+object SchdAggregation {
+    fun latestPerFlight(pending: List<MsgEvent>): List<String> =
+        pending
+            .groupBy { it.partitionKey ?: "" }
+            .map { (_, evs) -> evs.maxBy { it.eventId ?: 0 }.payloadJson }
+
+    /** 由 Decision 直接构造时的等价聚合（测试与 dispatcher 共用语义）。 */
+    fun latestPerFlightOfPush(pushes: List<SchdPush>): List<String> =
+        pushes
+            .groupBy { it.flid }
+            .map { (_, ps) -> ps.maxBy { it.eventSeq }.fltrJson }
+}
@@ -0,0 +1,40 @@
+package com.gzzn.omms.msgexchange.nextgen.domain
+
+/**
+ * ACMA-8 流程 2：Handler 决策（纯函数）产物——状态与报文进，变更与事件出，
+ * 不直接触碰 Redis/Kafka。
+ */
+data class Decision(
+    val flightChanges: List<FlightChange> = emptyList(),
+    val msgNotifies: List<NotifyPayload> = emptyList(),      // → MSG_EVENT(KAFKA:msg)
+    val schdPush: List<SchdPush> = emptyList(),              // → MSG_EVENT(KAFKA:schd)，PARTITION_KEY=FLID
+    val outboundIntents: List<OutboundIntent> = emptyList(), // → COUTMSGS（沿用既有列语义）
+    val refUpserts: List<RefUpsert> = emptyList(),           // → REF_DATA
+)
+
+/** 航班状态变更（阶段 A 由主泵线程 redisApply；阶段 B 落 FLIGHT_STATE 同事务）。 */
+data class FlightChange(
+    val flid: String,
+    val payloadJson: String,
+    val maid: String? = null,
+)
+
+data class NotifyPayload(val payloadJson: String)
+
+data class SchdPush(
+    val flid: String,
+    val fltrJson: String,
+    val eventSeq: Long = 0,            // 由事务插入时赋 EVENT_ID 语义序，聚合取 max
+)
+
+data class OutboundIntent(
+    val coutmsgsXml: String,
+    val ackReqd: Boolean = true,
+)
+
+data class RefUpsert(
+    val rtype: String,
+    val rkey: String,
+    val payloadJson: String,
+    val source: String,
+)
@@ -0,0 +1,34 @@
+package com.gzzn.omms.msgexchange.nextgen.domain
+
+/**
+ * 解码后的入站报文（统一内部模型）。META 字段实名 SNDR/SEQN/DTTM（legacy META.java，I3 幂等键来源）。
+ */
+data class MetaFields(
+    val sndr: String,
+    val type: String,
+    val styp: String,
+    val seqn: Long,
+    val dttm: Long,
+)
+
+/** 消息分派（sealed + 穷尽 when，ACMA-6 选型；取代 legacy 反射 get{TYPE}()）。 */
+sealed interface MsgKind {
+    data class Schd(val subtype: SchdSubtype) : MsgKind
+    data class Flop(val subtype: String) : MsgKind   // 29 类 STYP，阶段 2/3 逐类翻译
+
+    enum class SchdSubtype { RESP, DNLD, ADFT }
+}
+
+data class DecodedMessage(
+    val meta: MetaFields,
+    val kind: MsgKind,
+    val rawXml: String,
+    /** 各类型载荷（vendor POJO 或 Kotlin 模型，阶段 1 后续统一）。 */
+    val body: Any? = null,
+) {
+    val typeTag: String
+        get() = when (val k = kind) {
+            is MsgKind.Schd -> "SCHD-${k.subtype.name}"
+            is MsgKind.Flop -> "FLOP-${k.subtype}"
+        }
+}
@@ -0,0 +1,33 @@
+package com.gzzn.omms.msgexchange.nextgen.domain
+
+/**
+ * ACMA-8 MSG_EVENT（统一投递事件 / outbox）。
+ * TARGET 阶段化：KAFKA:msg、KAFKA:schd 自阶段 A；ES:flight_hts、REDIS:flightInfo 仅阶段 B 投影期。
+ */
+object Targets {
+    const val KAFKA_MSG = "KAFKA:msg"
+    const val KAFKA_SCHD = "KAFKA:schd"
+    const val ES_FLIGHT_HTS = "ES:flight_hts"
+    const val REDIS_FLIGHT_INFO = "REDIS:flightInfo"
+
+    /** 阶段 A 投递目标（仅 Kafka）——I5：Delivery 阶段 A 不写 Redis。 */
+    val phaseA: List<String> = listOf(KAFKA_MSG, KAFKA_SCHD)
+
+    /** 阶段 B 追加投影目标。 */
+    val phaseB: List<String> = phaseA + listOf(ES_FLIGHT_HTS, REDIS_FLIGHT_INFO)
+}
+
+enum class EventStatus { PENDING, SENT, DEAD }
+
+data class MsgEvent(
+    val eventId: Long? = null,
+    val target: String,
+    val partitionKey: String? = null,    // schd 事件恒为 FLID（v4 显式声明）
+    val payloadJson: String,
+    val state: EventStatus = EventStatus.PENDING,
+    val attempts: Int = 0,
+    val nextAttemptAt: java.time.Instant? = null,
+    val errorClass: ErrorClass? = null,
+    val lastError: String? = null,
+    val createdAt: java.time.Instant = java.time.Instant.now(),
+)
@@ -0,0 +1,25 @@
+package com.gzzn.omms.msgexchange.nextgen.domain
+
+/**
+ * ACMA-8 数据模型 / PROC_STATE 状态机（六迁移表）。
+ */
+enum class ProcStatus { PENDING, FAILED, SUCCEEDED, SKIPPED, DEAD }
+
+enum class ErrorClass { MALFORMED, CODEC_ERROR, EXHAUSTED, INFRA }
+
+data class ProcState(
+    val cminmsgsId: Long,
+    val state: ProcStatus,
+    val identityKey: String? = null,     // decode 后首次绑定；FAILED 重试不重绑（I3）
+    val attempts: Int = 0,
+    val nextAttemptAt: java.time.Instant? = null,
+    val errorClass: ErrorClass? = null,
+    val lastError: String? = null,
+    val updatedAt: java.time.Instant = java.time.Instant.now(),
+) {
+    val isTerminal: Boolean
+        get() = state == ProcStatus.SUCCEEDED || state == ProcStatus.SKIPPED || state == ProcStatus.DEAD
+
+    /** 终态皆可归档（矩阵 #12：SUCCEEDED ∪ SKIPPED ∪ DEAD）；PENDING/FAILED 不迁。 */
+    val archivable: Boolean get() = isTerminal
+}
@@ -0,0 +1,120 @@
+package com.gzzn.omms.msgexchange.nextgen.infra.persistence
+
+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.ProcState
+import com.gzzn.omms.msgexchange.nextgen.domain.ProcStatus
+import java.time.Instant
+
+/**
+ * ACMA-8 v4 仓储接口（接口驱动，主泵/调度循环可单测；Micronaut Data JDBC 实装属阶段 1 后续）。
+ */
+interface ProcStateRepository {
+    fun insert(cminmsgsId: Long, state: ProcStatus = ProcStatus.PENDING)
+
+    /** I1：严格 FIFO 队头（最小未完成 CMINMSGS_ID）。 */
+    fun headUnfinished(): ProcState?
+
+    /** I3：identity 首次绑定；返回 false = 另一条消息已持有该键。 */
+    fun tryBindIdentity(cminmsgsId: Long, identityKey: String): Boolean
+
+    fun ownerOfIdentity(identityKey: String): Long?
+
+    fun update(
+        cminmsgsId: Long,
+        state: ProcStatus,
+        nextAttemptAt: Instant? = null,
+        attempts: Int? = null,
+        errorClass: ErrorClass? = null,
+        lastError: String? = null,
+    )
+}
+
+interface MsgEventRepository {
+    fun insertAll(events: List<MsgEvent>): List<Long>
+
+    /** I1 双层同策略：每 target 严格 FIFO 队头。 */
+    fun headUnsent(target: String): MsgEvent?
+
+    fun claimBatch(target: String, limit: Int): List<MsgEvent>   // ORDER BY EVENT_ID ASC
+
+    fun markSent(eventId: Long)
+
+    fun markAllSent(eventIds: List<Long>)
+
+    fun scheduleRetry(eventId: Long, nextAttemptAt: Instant, attempts: Int)
+
+    fun markDead(eventId: Long, errorClass: ErrorClass, lastError: String)
+
+    /** 阶段 B（定案 2）：Delivery 同线程在 ES 投递成功后同步 enqueue 删除事件。 */
+    fun insertSync(events: List<MsgEvent>)
+}
+
+interface RefDataRepository {
+    data class GenMeta(val flids: List<String>, val version: Long)
+
+    fun getGen(day: String): GenMeta?
+
+    /** 流程 4：版本 CAS（expected 未变才写，重放 no-op，不二次自增）。 */
+    fun putGenIfVersion(day: String, expected: Long, new: GenMeta): Boolean
+
+    fun upsertAll(rows: List<Triple<String, String, String>>)   // (rtype, rkey, payloadJson)
+}
+
+interface ReqTrackRepository {
+    data class Req(
+        val reqId: Long,
+        val reqType: String,
+        val state: String,               // REGISTERED/SENT/WAITING/DONE/EXPIRED
+        val sentAt: Instant? = null,
+    )
+
+    fun findOpenByKind(kind: String): Req?
+
+    fun forceExpireOpenOf(kind: String)
+
+    fun insert(kind: String, paramsJson: String): Long
+
+    fun linkCoutmsgs(reqId: Long, coutmsgsId: Long)
+
+    fun markSent(reqId: Long, sentAt: Instant)
+
+    fun expireIfWaiting(reqId: Long)
+
+    fun markDone(reqId: Long)
+}
+
+interface PumpJobRepository {
+    data class Job(val jobId: Long, val kind: String)    // ARCHIVE/HISTORY_SWEEP/PROJECTION_REBUILD
+
+    fun enqueue(kind: String)
+
+    fun headQueued(): Job?
+
+    fun markRunning(jobId: Long)
+
+    fun markDone(jobId: Long)
+
+    fun markFailed(jobId: Long, lastError: String)
+}
+
+interface FlightStateRepository {
+    /** 阶段 B 权威；replaceDay = 单事务删差集+写新代+版本提升。 */
+    fun replaceDay(day: String, flights: List<Pair<String, String>>)
+
+    fun findByDay(day: String): List<Pair<String, String>>
+}
+
+interface CminmsgInboxRepository {
+    /** 事务 1：原文落库（沿用 SUBSYSTEM_* 列；仅 CLOB/DATE_RECEIVED，I3：无接收唯一约束）。 */
+    fun insertRaw(rawXml: String): Long
+
+    fun rawOf(cminmsgsId: Long): String?
+
+    /**
+     * v4 回滚兼容关键：SUCCEEDED 时回填 SUBSYSTEM_*（FIX）+ DATE_PROCESSED/STATUS，
+     * 旧系统按自身 processed 语义可无缝接管（Runbook 第 7 步）。
+     */
+    fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long)
+}
@@ -0,0 +1,26 @@
+package com.gzzn.omms.msgexchange.nextgen.infra.redis
+
+/**
+ * ACMA-8 流程 4 / I4 / I5：Redis Lua 脚本装载与执行入口。
+ * 阶段 A 全部 flightInfo 写均经此处，且仅由主泵线程调用（I5）。
+ */
+enum class RedisScript(val classpathLocation: String) {
+    /** 同一 hash 原子“覆盖新代 + 按代差删”（setArg 为 N 对 field/value，delArg 为差集）。 */
+    SNAPSHOT_REPLACE("lua/snapshot_replace.lua"),
+
+    /** 3:30 清场批量删除（仅 ES 归档成功集）。 */
+    BATCH_DELETE("lua/batch_delete.lua"),
+}
+
+interface FlightRedisClient {
+    /**
+     * 执行脚本。SNAPSHOT_REPLACE：setPairs 为新代全量 field/value，delFields 为按代差集；
+     * BATCH_DELETE：delFields 为待删 FLID 集。
+     */
+    fun eval(script: RedisScript, setPairs: List<Pair<String, String>> = emptyList(), delFields: List<String> = emptyList())
+
+    /** 阶段 A 权威读（处理决策 loadState、3:30 清场 findAll）。 */
+    fun hgetAllFlightInfo(): Map<String, String>
+}
+
+// TODO(阶段1后续): 基于 micronaut-redis-lettuce 的实装（脚本自 classpath 装载并缓存 SHA）。
@@ -0,0 +1,24 @@
+package com.gzzn.omms.msgexchange.nextgen.ingress
+
+import io.micronaut.http.HttpResponse
+import io.micronaut.http.MediaType
+import io.micronaut.http.annotation.Body
+import io.micronaut.http.annotation.Controller
+import io.micronaut.http.annotation.Post
+import io.micronaut.http.annotation.Produces
+
+/**
+ * ACMA-8 契约冻结：端点路径与响应语义不变。现役 Controller 返回记录 ID（原文落库后）。
+ */
+@Controller
+class InboxController(private val inbox: InboxService) {
+
+    @Post("/cminmsgs/send")
+    @Produces(MediaType.TEXT_PLAIN)
+    fun send(@Body rawXml: String): HttpResponse<String> {
+        val receipt = inbox.accept(rawXml)
+        return HttpResponse.ok(receipt.cminmsgsId.toString())   // TODO: 与现役响应体逐字对拍后固化
+    }
+
+    // TODO(阶段2): /schd/sync、/all/flights、/kafka/topics/{name}/msgs 按契约冻结清单补齐。
+}
@@ -0,0 +1,26 @@
+package com.gzzn.omms.msgexchange.nextgen.ingress
+
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.CminmsgInboxRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ProcStateRepository
+import jakarta.inject.Singleton
+import java.time.Instant
+
+/** ACMA-8 流程 1：入站接收（事务 1）。响应语义 = “已持久化”（与现役一致）；不解析报文（I3）。 */
+@Singleton
+class InboxService(
+    private val inbox: CminmsgInboxRepository,
+    private val procState: ProcStateRepository,
+    // TODO(阶段1后续): 事务边界（@Transactional）随 Micronaut Data 实装补齐；
+    // pump 唤醒仅加速，崩溃后主泵 1s 轮询兜底。
+) {
+    data class Receipt(val cminmsgsId: Long, val receivedAt: Instant)
+
+    fun accept(rawXml: String): Receipt {
+        val id = inbox.insertRaw(rawXml)
+        procState.insert(id)                 // 同事务（实装后）；接收层无唯一约束（I3）
+        wakePump()
+        return Receipt(id, Instant.now())
+    }
+
+    private fun wakePump() = Unit           // TODO: 主泵唤醒（仅加速）
+}
@@ -0,0 +1,80 @@
+package com.gzzn.omms.msgexchange.nextgen.jobs
+
+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.persistence.FlightStateRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.MsgEventRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.PumpJobRepository
+import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent
+import com.gzzn.omms.msgexchange.nextgen.domain.Targets
+import jakarta.inject.Singleton
+
+/**
+ * ACMA-8：泵作业执行器。cron 触发入队（PUMP_JOB），主泵 FIFO 执行（决策 1）——
+ * 产物不绕过队头顺序；阶段 A 的清场同步链与归档在此落地（I4/I5）。
+ */
+@Singleton
+class JobExecutor(
+    private val historySweep: HistorySweepJob,
+    private val archive: ArchiveJob,
+    private val projectionRebuild: ProjectionRebuildJob,
+) {
+    fun execute(job: PumpJobRepository.Job) = when (job.kind) {
+        "HISTORY_SWEEP" -> historySweep.run()
+        "ARCHIVE" -> archive.run()
+        "PROJECTION_REBUILD" -> projectionRebuild.run()
+        else -> error("unknown job ${job.kind}")
+    }
+}
+
+/**
+ * 流程 4 runJob(HISTORY_SWEEP)：3:30 清场——主泵作业内同步链（I4，KEEP 现役
+ * FlightHisScheduled 同步语义）：判史 → 同步写 ES（成功集）→ 仅删成功集。
+ */
+@Singleton
+class HistorySweepJob(
+    private val redis: FlightRedisClient,
+    // TODO(阶段2): esFlightHts.saveSync(history) 返回成功集（同步写，不经事件 ack 回查）
+) {
+    fun run() {
+        val all = redis.hgetAllFlightInfo()
+        val history = pickHistory(all)                         // TODO(阶段2): 沿用现役判史规则（KEEP）
+        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)
+}
+
+/** 流程 5 runJob(ARCHIVE)：3:00 归档——1 天前且仅终态可迁（矩阵 #12）。 */
+@Singleton
+class ArchiveJob(
+    // TODO(阶段1后续): CMINMSGS⇆CMINMSGS_HST 迁移 SQL：JOIN PROC_STATE，STATE ∈ {SUCCEEDED,DEAD,SKIPPED}
+) {
+    fun run() {
+        // TODO: 迁移（DEAD 带 ERROR_CLASS、SKIPPED 带 duplicate-of 审计随行保留）
+    }
+}
+
+/** 流程 7：阶段 B 投影重建——切入阶段 B 时全量重建一次，此后增量走事件。 */
+@Singleton
+class ProjectionRebuildJob(
+    private val flightState: FlightStateRepository,
+    private val msgEvents: MsgEventRepository,
+) {
+    fun run() {
+        for (day in activeDays()) {
+            val flights = flightState.findByDay(day)
+            flights.forEach { (flid, payload) ->
+                msgEvents.insertSync(
+                    listOf(MsgEvent(target = Targets.REDIS_FLIGHT_INFO, partitionKey = flid, payloadJson = payload)),
+                )
+            }
+        }
+        // 重建完成前旧 Redis 视图继续服务（数据同源，仅短暂滞后）
+    }
+
+    private fun activeDays(): List<String> = emptyList()       // TODO(阶段B)
+}
@@ -0,0 +1,34 @@
+package com.gzzn.omms.msgexchange.nextgen.processing
+
+import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage
+import com.gzzn.omms.msgexchange.nextgen.domain.Decision
+import com.gzzn.omms.msgexchange.nextgen.domain.MsgKind
+
+/**
+ * ACMA-8 流程 2：Handler = 纯函数（状态与报文进，Decision 出，不碰 Redis/Kafka）。
+ * 32 个 Handler（flop 29 + schd 3）翻译属阶段 2/3，逐条对照 ACMA-4 基线与兼容矩阵 KEEP/FIX。
+ */
+interface Handler {
+    val kind: MsgKind
+
+    /** 在线状态以只读快照传入（阶段 A 读 Redis 权威、阶段 B 读 FLIGHT_STATE，由调用方装配）。 */
+    fun decide(flightView: Map<String, String>, msg: DecodedMessage): Decision
+}
+
+/**
+ * sealed 穷尽分派（ACMA-6 选型：取代 legacy 反射 get{TYPE}()）。
+ */
+class HandlerRegistry(handlers: List<Handler>) {
+    private val byKind: Map<String, Handler> = handlers.associateBy { keyOf(it.kind) }
+
+    fun dispatcherFor(msg: DecodedMessage): Handler? = byKind[keyOf(msg.kind)]
+
+    companion object {
+        fun keyOf(kind: MsgKind): String = when (kind) {
+            is MsgKind.Schd -> "SCHD-${kind.subtype.name}"
+            is MsgKind.Flop -> "FLOP-${kind.subtype}"
+        }
+    }
+}
+
+// TODO(阶段2/3): 注册 3+29 Handler；本阶段仅含骨架（阶段 2 最小纵向链路先做 1 SCHD(DNLD) + 1 FLOP）。
@@ -0,0 +1,24 @@
+package com.gzzn.omms.msgexchange.nextgen.processing
+
+import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps
+import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage
+import java.time.LocalDate
+
+/**
+ * ACMA-8 I3：幂等键 = SNDR|TYPE|STYP|SEQN。
+ * 计算集中此处唯一入口；“是否含日边界”可配置且默认关闭（CONFIRM 矩阵 #11，
+ * SEQN 重置作用域确认前不改语义——上线后不改幂等键）。
+ */
+object Identity {
+    fun of(
+        msg: DecodedMessage,
+        includeDayBoundary: Boolean,
+        day: LocalDate = LocalDate.now(),
+    ): String {
+        val base = "${msg.meta.sndr}|${msg.meta.type}|${msg.meta.styp}|${msg.meta.seqn}"
+        return if (includeDayBoundary) "$base|${day}" else base
+    }
+
+    fun of(msg: DecodedMessage, props: PipelineProps.Identity): String =
+        of(msg, props.includeDayBoundary)
+}
@@ -0,0 +1,158 @@
+package com.gzzn.omms.msgexchange.nextgen.processing
+
+import com.gzzn.omms.msgexchange.nextgen.config.PipelineProps
+import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass
+import com.gzzn.omms.msgexchange.nextgen.domain.MsgEvent
+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
+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.jobs.JobExecutor
+import jakarta.inject.Singleton
+import java.time.Duration
+import java.time.Instant
+
+/**
+ * ACMA-8 流程 2：处理主泵——严格 FIFO + HOL + 毒丸 DEAD 升级（I1）；
+ * 阶段 A 全部 Redis 写在本线程（I5），且先于事件创建（I2 happens-before）。
+ */
+@Singleton
+class Pump(
+    private val procState: ProcStateRepository,
+    private val pumpJobs: PumpJobRepository,
+    private val inbox: CminmsgInboxRepository,
+    private val processor: MessageProcessor,
+    private val jobExecutor: JobExecutor,
+    private val props: PipelineProps,
+) {
+    @Volatile
+    private var running = true
+
+    fun loop() {
+        while (running) {
+            tick()
+        }
+    }
+
+    internal fun tick() {
+        val job = pumpJobs.headQueued()
+        val head = procState.headUnfinished()
+
+        // 统一 FIFO：JOB 与消息同队列语义（定时任务产物不绕过队头顺序，决策 1）。
+        val nextJob = job?.takeIf { jobBefore(job, head) }
+        when {
+            nextJob != null -> execute(nextJob)
+            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",
+                    )
+                } else {
+                    sleepQuietly(Duration.between(Instant.now(), head.nextAttemptAt))
+                }
+            else -> processor.processOne(head.cminmsgsId)
+        }
+    }
+
+    private fun poisoned(head: com.gzzn.omms.msgexchange.nextgen.domain.ProcState): Boolean =
+        head.attempts >= props.pipeline.maxAttempts ||
+            Duration.between(head.updatedAt, Instant.now()) > props.pipeline.headDeadline
+
+    /** JOB 与队头消息的先后由入队时间近似；实装以统一序号列保证（阶段 1 后续）。 */
+    private fun jobBefore(job: PumpJobRepository.Job, head: com.gzzn.omms.msgexchange.nextgen.domain.ProcState?) =
+        head == null || head.state == ProcStatus.FAILED
+
+    private fun execute(job: PumpJobRepository.Job) {
+        pumpJobs.markRunning(job.jobId)
+        try {
+            jobExecutor.execute(job)
+            pumpJobs.markDone(job.jobId)
+        } catch (e: Exception) {
+            pumpJobs.markFailed(job.jobId, e.message ?: "unknown")
+        }
+    }
+
+    private fun sleepQuietly(d: Duration) {
+        if (!d.isNegative && !d.isZero) Thread.sleep(d.toMillis().coerceAtLeast(1))
+    }
+}
+
+/** ACMA-8 流程 2 processOne：解码→绑定→纯函数决策→提交。 */
+@Singleton
+class MessageProcessor(
+    private val inbox: CminmsgInboxRepository,
+    private val procState: ProcStateRepository,
+    private val msgEvents: MsgEventRepository,
+    private val codecHolder: CodecHolder,
+    private val handlers: HandlerHolder,
+    private val redis: FlightRedisClient,
+    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")
+            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)
+                return
+            }
+        }
+
+        val state = procState.headUnfinished()
+        // I3：identity 仅首次绑定；FAILED 重试不重绑（绑定失败 = 另一条同键消息 → SKIPPED）
+        // TODO(阶段1后续): 以行内 IDENTITY_KEY 判定而非 head 复查（含 FAILED 持久化字段读取）
+        if (state != null && state.cminmsgsId == cminmsgsId && state.identityKey == null) {
+            val identity = Identity.of(decoded, props.identity)
+            if (!procState.tryBindIdentity(cminmsgsId, identity)) {
+                val owner = procState.ownerOfIdentity(identity) ?: -1L
+                procState.update(cminmsgsId, ProcStatus.SKIPPED, lastError = "duplicate-of:$owner")
+                return
+            }
+        }
+
+        // 快照消息走专属流程（流程 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}")
+            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)
+            return
+        }
+
+        val decision = handler.decide(redis.hgetAllFlightInfo(), decoded)   // 纯函数
+
+        if (props.phase == PipelineProps.Phase.A) {
+            // 阶段 A：主泵线程先写 Redis（幂等），先于事件创建（I2）；TODO: redisApply(flightChanges)
+        }
+        // 事务 2：事件 + 出站意图 + REF_DATA + CMINMSGS 回填 + SUCCEEDED（实装后 @Transactional）
+        val events = buildList {
+            decision.msgNotifies.forEach { add(MsgEvent(target = Targets.KAFKA_MSG, payloadJson = it.payloadJson)) }
+            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)
+        // 阶段 B：flightState.apply(decision.flightChanges) 进入同一事务；投影事件（ES/REDIS）追加。
+    }
+}
+
+/** 延迟装配占位（阶段 1 后续以 Micronaut Bean 替换直连构造）。 */
+class CodecHolder(val codec: com.gzzn.omms.msgexchange.nextgen.codec.XmlCodec)
+class HandlerHolder(val registry: HandlerRegistry)
@@ -0,0 +1,63 @@
+package com.gzzn.omms.msgexchange.nextgen.processing
+
+import com.gzzn.omms.msgexchange.nextgen.domain.DecodedMessage
+import com.gzzn.omms.msgexchange.nextgen.domain.ErrorClass
+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
+
+/**
+ * ACMA-8 流程 4：日计划快照（generation，主泵内执行）。
+ * staging（内存瞬态，崩溃从 raw 整包重放）→ Lua 原子“覆盖+按代差删”（I4/I5）
+ * → putGenIfVersion CAS + SUCCEEDED 同一 MySQL 事务（重放幂等，版本不二次自增）。
+ * 整包失败 = 现役等价（KEEP）；跳坏行进 quarantine = CONFIRM（矩阵 #9）。
+ */
+@Singleton
+class SnapshotFlow(
+    private val procState: ProcStateRepository,
+    private val refData: RefDataRepository,
+    private val redis: FlightRedisClient,
+) {
+    fun publishSnapshot(cminmsgsId: Long, 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)
+            return
+        }
+        val normalized = (staged as StageResult.Ok).flights            // (flid, payloadJson)
+
+        // 2) 发布：Lua 原子覆盖 + 按代差删（删除集 = gen.flids − 新代，ADFT 自动存活）
+        val day = staged.day
+        val gen = refData.getGen(day)
+        val newFlids = normalized.map { it.first }
+        val delFields = gen?.flids?.minus(newFlids.toSet()) ?: emptyList()
+        redis.eval(RedisScript.SNAPSHOT_REPLACE, setPairs = normalized, delFields = delFields)
+
+        // 3) 事务：putGen CAS + SUCCEEDED（实装后同 @Transactional；CAS 失败=并发，串行泵下不应发生→告警）
+        val expected = gen?.version ?: 0L
+        if (!refData.putGenIfVersion(day, expected, RefDataRepository.GenMeta(newFlids, expected + 1))) {
+            // 重放路径：version 已是目标值 → no-op 视为成功
+            val again = refData.getGen(day)
+            if (again == null || again.version != expected + 1) {
+                procState.update(cminmsgsId, ProcStatus.FAILED, lastError = "gen-cas-conflict")
+                return
+            }
+        }
+        procState.update(cminmsgsId, ProcStatus.SUCCEEDED)
+    }
+
+    /** staging 结果（骨架）。 */
+    sealed interface StageResult {
+        data class Ok(val day: String, val flights: List<Pair<String, String>>) : StageResult
+        data class Invalid(val reason: String) : StageResult
+
+        companion object {
+            fun stagingOf(msg: DecodedMessage): StageResult =
+                Invalid("staging-not-implemented(${msg.typeTag})")   // TODO(阶段2)
+        }
+    }
+}
@@ -0,0 +1,18 @@
+package com.gzzn.omms.msgexchange.nextgen.reference
+
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository
+import jakarta.inject.Singleton
+
+/**
+ * ACMA-8 决策 6：21 类 admin-api 基础数据同步（阶段 6）。
+ * 只产 REF_DATA 行（权威写唯一入口）；影子实例不运行本模块（v4 影子二分）。
+ */
+@Singleton
+class ReferenceService(
+    private val refData: RefDataRepository,
+    // TODO(阶段6): @Client(id="ADMINAPI") 声明式客户端 + 21 类端点配置化清单
+) {
+    fun refresh(type: String) {
+        // TODO(阶段6): 拉取 → upsertAll(type, key, payload)；失败旧数据可用（REF_DATA 保留旧值）
+    }
+}
@@ -0,0 +1,57 @@
+package com.gzzn.omms.msgexchange.nextgen.reference
+
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.RefDataRepository
+import com.gzzn.omms.msgexchange.nextgen.infra.persistence.ReqTrackRepository
+import jakarta.inject.Singleton
+import java.time.Instant
+
+/**
+ * ACMA-8 流程 6：请求式查询状态机（15 类，决策树版 v4）。
+ * 同类并发=1（注册新请求先强制旧请求 EXPIRED，终态不再被匹配）；
+ * 应答匹配：回显字段 CONFIRM（矩阵 #13）→ 精确匹配；否则退化模式（DTTM ≥ SENT 才应用 + 审计）。
+ */
+@Singleton
+class RequestCoordinator(
+    private val reqTrack: ReqTrackRepository,
+    private val refData: RefDataRepository,
+) {
+    fun request(kind: String, rangeJson: String): Long {
+        reqTrack.forceExpireOpenOf(kind)                       // 防护①
+        val reqId = reqTrack.insert(kind, rangeJson)
+        // 出站走既有 COUTMSGS outbox（TODO(阶段6): codec.encodeRqrd + coutmsgs 落库 + link）
+        reqTrack.markSent(reqId, Instant.now())
+        return reqId
+        // TODO(阶段6): timer.at(kind.timeout) { expireIfWaiting(reqId) }
+    }
+
+    /** 应答处理（决策树）：msg.dttm 与回显字段由 codec 解出后传入。 */
+    fun onResp(
+        kind: String,
+        dttm: Long,
+        echoSeqn: Long?,
+        records: List<Triple<String, String, String>>,         // (rtype, rkey, payloadJson)
+    ): String {
+        val req = reqTrack.findOpenByKind(kind) ?: return audit("late/unknown resp kind=$kind")
+        val echoConfirmed = false                               // CONFIRM（矩阵 #13）待机场方结论
+        return when {
+            echoConfirmed && echoSeqn != null -> {              // ① 精确匹配（TODO: matchBy echoSeqn）
+                applyResp(req.reqId, records); "matched"
+            }
+            dttm < epochMillis(req.sentAt) -> audit("stale resp kind=$kind")   // ② DTTM < SENT → 丢弃
+            else -> {                                           // ③ 退化模式应用；残余跨代错配风险明示接受并审计
+                applyResp(req.reqId, records); "applied-degraded"
+            }
+        }
+    }
+
+    private fun applyResp(reqId: Long, records: List<Triple<String, String, String>>) {
+        refData.upsertAll(records)                              // 大应答复用流程 4 staging 路径（阶段 6）
+        reqTrack.markDone(reqId)
+    }
+
+    private fun audit(reason: String): String = reason.also {
+        // TODO(阶段1后续): 审计表/日志（跨代错配残余风险可观测）
+    }
+
+    private fun epochMillis(i: Instant?): Long = i?.toEpochMilli() ?: 0
+}
@@ -0,0 +1,60 @@
+# =====================================================================
+# msgexchange-nextgen 配置（ACMA-8 v4）
+# 口令/端点全部环境变量外置——零入库（ACMA-8 契约冻结 + 安全基线）
+# =====================================================================
+msgx:
+  phase: A                          # A=Redis 权威（主泵单线程写，I5）；B=FLIGHT_STATE 权威+投影
+  service-name: msgexchangeapi      # 契约冻结；影子实例用 msgexchangeapi-shadow + 独立表/key前缀/topic
+  register-eureka: true             # 影子对拍期置 false 或独立服务名
+  pipeline:                         # —— ACMA-8 参数表初值（阶段 0 基线校准）——
+    poll-interval: 1s               # KEEP 现役 1s 轮询节奏
+    claim-batch: 50
+    max-attempts: 5                 # 处理/投递同值
+    backoff-ms: [1000, 2000, 4000, 8000, 16000]   # 指数退避，单次封顶 60s
+    backoff-cap-ms: 60000
+    head-deadline: 10m              # 队头滞留上界 = 最坏 HOL 时长（毒丸升级）
+  schd:
+    flush-period: 3s                # KEEP 现役推送节律
+    flush-limit: 500                # 批上限，防积压尖峰
+  identity:
+    include-day-boundary: false     # CONFIRM（矩阵 #11）：SEQN 重置作用域确认前保持关闭
+  consistency-check:                # 阶段 A 唯一主动一致性哨兵（必选）
+    on-startup: true
+    daily-sample-ratio: 0.01
+
+micronaut:
+  application:
+    name: msgexchange-nextgen
+  server:
+    port: 8080
+
+datasource:
+  url: ${MSGX_DB_URL}
+  username: ${MSGX_DB_USER}
+  password: ${MSGX_DB_PASSWORD}
+  driver-class-name: com.mysql.cj.jdbc.Driver
+
+flyway:
+  enabled: true
+  locations: classpath:db/migration
+
+redis:
+  uri: ${MSGX_REDIS_URI}            # 阶段 A 权威存储；仅主泵线程写（I5）
+
+kafka:
+  bootstrap:
+    servers: ${MSGX_KAFKA_SERVERS}
+  producers:
+    acks: all
+    enable-idempotence: true
+
+eureka:
+  client:
+    registration:
+      enabled: ${msgx.register-eureka}
+    default-zone: ${MSGX_EUREKA_URL:http://127.0.0.1:8761/eureka}
+
+# logstash TCP JSON 通道保留（字段兼容现网，ACMA-3 部署前提）——见 logback.xml
+logstash:
+  host: ${MSGX_LOGSTASH_HOST:127.0.0.1}
+  port: ${MSGX_LOGSTASH_PORT:5044}
@@ -0,0 +1,78 @@
+-- =====================================================================
+-- ACMA-8 v4 数据模型（六表）· Flyway 基线迁移
+-- 与 issue「综合架构设计」数据模型节逐字一致；影子实例在独立 schema 执行同一脚本。
+-- =====================================================================
+
+-- ① 处理状态伴生表（不动 CMINMSGS 旧列；
+--    回滚兼容：SUCCEEDED 时回填 CMINMSGS.DATE_PROCESSED/STATUS，旧系统可接管续跑）
+CREATE TABLE PROC_STATE (
+  CMINMSGS_ID     BIGINT PRIMARY KEY,
+  STATE           VARCHAR(16)  NOT NULL,   -- PENDING/FAILED/SUCCEEDED/SKIPPED/DEAD
+  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
+  LAST_ERROR      VARCHAR(1000) NULL,
+  UPDATED_AT      TIMESTAMP    NOT NULL,
+  UNIQUE KEY UK_PROC_IDENTITY (IDENTITY_KEY),
+  INDEX IDX_PROC_HEAD (STATE, CMINMSGS_ID)          -- 主泵队头查询（I1）
+);
+
+-- ② 统一投递事件（outbox）
+CREATE TABLE MSG_EVENT (
+  EVENT_ID      BIGINT AUTO_INCREMENT PRIMARY KEY,
+  TARGET        VARCHAR(30) NOT NULL,      -- KAFKA:msg / KAFKA:schd（A起）；ES:flight_hts / REDIS:flightInfo（B起）
+  PARTITION_KEY VARCHAR(64) NULL,          -- schd 事件恒为 FLID
+  PAYLOAD_JSON  MEDIUMTEXT  NOT NULL,
+  STATE         VARCHAR(16) NOT NULL,      -- PENDING/SENT/DEAD
+  ATTEMPTS      INT NOT NULL DEFAULT 0,
+  NEXT_ATTEMPT_AT TIMESTAMP NULL,
+  ERROR_CLASS   VARCHAR(20) NULL,
+  LAST_ERROR    VARCHAR(1000) NULL,
+  CREATED_AT    TIMESTAMP NOT NULL,
+  INDEX IDX_EVT_HEAD (TARGET, STATE, EVENT_ID)       -- 每 target 队头查询（I1 双层同策略）
+);
+
+-- ③ 参考数据 + SCHD generation 元数据（21 类；SCHD_GEN 行 RKEY=日期）
+CREATE TABLE REF_DATA (
+  RTYPE VARCHAR(20) NOT NULL,              -- AIRL/ARPT/... | SCHD_GEN
+  RKEY  VARCHAR(64) NOT NULL,
+  PAYLOAD_JSON MEDIUMTEXT NOT NULL,        -- SCHD_GEN: {"flids":[...],"version":N}
+  VERSION      BIGINT NOT NULL DEFAULT 0,  -- putGenIfVersion CAS 依据（流程 4）
+  SOURCE       VARCHAR(20) NOT NULL,       -- ADMINAPI / AODB / PIPELINE
+  REFRESHED_AT TIMESTAMP NOT NULL,
+  PRIMARY KEY (RTYPE, RKEY)
+);
+
+-- ④ 请求状态机（15 类，流程 6）
+CREATE TABLE REQ_TRACK (
+  REQ_ID BIGINT AUTO_INCREMENT PRIMARY KEY,
+  REQ_TYPE VARCHAR(20) NOT NULL,
+  PARAMS_JSON VARCHAR(500) NOT NULL,
+  STATE VARCHAR(16) NOT NULL,              -- REGISTERED/SENT/WAITING/DONE/EXPIRED
+  COUTMSGS_ID INT NULL,
+  SENT_AT TIMESTAMP NULL,
+  COMPLETED_AT TIMESTAMP NULL,
+  INDEX IDX_REQ_OPEN (REQ_TYPE, STATE)
+);
+
+-- ⑤ 泵作业队列（定时任务统一入口：cron 触发入队，主泵 FIFO 执行）
+CREATE TABLE PUMP_JOB (
+  JOB_ID BIGINT AUTO_INCREMENT PRIMARY KEY,
+  KIND   VARCHAR(20) NOT NULL,             -- ARCHIVE / HISTORY_SWEEP / PROJECTION_REBUILD
+  STATE  VARCHAR(16) NOT NULL,             -- QUEUED/RUNNING/DONE/FAILED
+  CREATED_AT TIMESTAMP NOT NULL,
+  UPDATED_AT TIMESTAMP NOT NULL,
+  LAST_ERROR VARCHAR(1000) NULL
+);
+
+-- ⑥ 航班状态权威（阶段 B 启用；此前 Redis 为权威，I5）
+CREATE TABLE FLIGHT_STATE (
+  FLID VARCHAR(32) PRIMARY KEY,
+  FDAY DATE NOT NULL,
+  VERSION BIGINT NOT NULL,
+  MAID VARCHAR(32) NULL,
+  PAYLOAD_JSON MEDIUMTEXT NOT NULL,        -- 投影兼容格式的 FLTR JSON
+  UPDATED_AT TIMESTAMP NOT NULL,
+  INDEX IDX_FS_DAY (FDAY, VERSION)
+);
@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!-- ACMA-8/ACMA-6：logstash TCP JSON 通道保留（字段兼容现网）+ MDC traceId -->
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level [%X{traceId:-}] %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
+        <destination>${LOGSTASH_HOST:-127.0.0.1}:${LOGSTASH_PORT:-5044}</destination>
+        <encoder class="net.logstash.logback.encoder.LogstashEncoder">
+            <includeMdcKeyName>traceId</includeMdcKeyName>
+        </encoder>
+        <reconnectionDelay>10 seconds</reconnectionDelay>
+    </appender>
+
+    <root level="INFO">
+        <appender-ref ref="STDOUT"/>
+        <appender-ref ref="LOGSTASH"/>
+    </root>
+
+    <logger name="com.gzzn.omms.msgexchange.nextgen" level="DEBUG"/>
+</configuration>
@@ -0,0 +1,9 @@
+-- ACMA-8 流程 4（runJob HISTORY_SWEEP）/ I4：3:30 清场批量删除。
+-- 仅删除“ES 归档成功集”（调用方以 saveSync 成功返回集为参），同一泵线程程序序保证依赖。
+-- KEYS[1] = flightInfo；ARGV = 待删除 FLID 列表
+local n = 0
+for i = 1, #ARGV do
+  redis.call('HDEL', KEYS[1], ARGV[i])
+  n = n + 1
+end
+return n
@@ -0,0 +1,19 @@
+-- ACMA-8 流程 4 / I4 / I5：日计划快照“覆盖新代 + 按代差删”，同一 hash 原子执行。
+-- KEYS[1] = flightInfo
+-- ARGV[1] = set 对数 N，其后为 N 对 field/value；再后为待删除 field（按代差集）
+-- 删除集边界：仅上一代 gen.flids − 新代 flids（按代，非全 hash）→ ADFT 自动存活。
+local key = KEYS[1]
+local nSet = tonumber(ARGV[1])
+local i = 2
+while i <= 1 + nSet * 2 do
+  redis.call('HSET', key, ARGV[i], ARGV[i + 1])
+  i = i + 2
+end
+local j = 2 + nSet * 2
+local deleted = 0
+while j <= #ARGV do
+  redis.call('HDEL', key, ARGV[j])
+  deleted = deleted + 1
+  j = j + 1
+end
+return deleted
@@ -0,0 +1,56 @@
+package com.gzzn.omms.msgexchange.nextgen.delivery
+
+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 org.junit.jupiter.api.Test
+import kotlin.test.assertEquals
+
+/**
+ * ACMA-8 流程 3 / 兼容矩阵 #7（FIX）：同一 FLID 多次变更只发最新一态；
+ * v4 显式“组内取 EVENT_ID 最大”，不依赖集合遍历序。
+ */
+class SchdAggregationTest {
+
+    private fun ev(flid: String, eventId: Long, payload: String) = MsgEvent(
+        eventId = eventId,
+        target = Targets.KAFKA_SCHD,
+        partitionKey = flid,
+        payloadJson = payload,
+        state = EventStatus.PENDING,
+    )
+
+    @Test
+    fun `same flight collapsed to max EVENT_ID`() {
+        val pending = listOf(
+            ev("F1", 1, """{"FLID":"F1","v":"old"}"""),
+            ev("F2", 2, """{"FLID":"F2","v":"only"}"""),
+            ev("F1", 3, """{"FLID":"F1","v":"new"}"""),
+        )
+        val latest = SchdAggregation.latestPerFlight(pending)
+        assertEquals(2, latest.size)
+        assertEquals("""{"FLID":"F1","v":"new"}""", latest.first { it.contains("F1") })
+        assertEquals("""{"FLID":"F2","v":"only"}""", latest.first { it.contains("F2") })
+    }
+
+    @Test
+    fun `order-independent grouping yields same latest per flight`() {
+        val a = listOf(
+            ev("F1", 1, "old"),
+            ev("F1", 2, "new"),
+        )
+        val b = a.reversed()
+        assertEquals(
+            SchdAggregation.latestPerFlight(a).toSet(),
+            SchdAggregation.latestPerFlight(b).toSet(),
+        )
+    }
+
+    @Test
+    fun `wire shape is FLTR JSON array`() {
+        val payload = SchdAggregation
+            .latestPerFlight(listOf(ev("F1", 1, "a"), ev("F2", 2, "b")))
+            .joinToString(",", "[", "]")
+        assertEquals("[a,b]", payload)
+    }
+}
@@ -0,0 +1,45 @@
+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.MetaFields
+import com.gzzn.omms.msgexchange.nextgen.domain.MsgKind
+import org.junit.jupiter.api.Test
+import java.time.LocalDate
+import kotlin.test.assertEquals
+
+/**
+ * ACMA-8 I3：identity = SNDR|TYPE|STYP|SEQN；日边界含否集中可配（CONFIRM 矩阵 #11，默认关）。
+ */
+class IdentityTest {
+
+    private fun msg(seqn: Long, dttm: Long = 20260906120000) = DecodedMessage(
+        meta = MetaFields(sndr = "AODB", type = "FLOP", styp = "DELY", seqn = seqn, dttm = dttm),
+        kind = MsgKind.Flop("DELY"),
+        rawXml = "<MSG/>",
+    )
+
+    @Test
+    fun `identity is pipe-joined SNDR TYPE STYP SEQN`() {
+        val identity = Identity.of(msg(42), includeDayBoundary = false)
+        assertEquals("AODB|FLOP|DELY|42", identity)
+    }
+
+    @Test
+    fun `day boundary included only when configured`() {
+        val day = LocalDate.of(2026, 9, 6)
+        assertEquals(
+            "AODB|FLOP|DELY|42|2026-09-06",
+            Identity.of(msg(42), includeDayBoundary = true, day = day),
+        )
+        assertEquals(
+            "AODB|FLOP|DELY|42",
+            Identity.of(msg(42), includeDayBoundary = false, day = day),
+        )
+    }
+
+    @Test
+    fun `props default keeps day boundary off`() {
+        assertEquals(false, PipelineProps.Identity().includeDayBoundary)
+    }
+}