test: GTDT e2e, v2 write-path guards, and PG test env support

Add JacksonXmlCodec/GtdtHandler/Pipeline tests, PgTestSupport for MSGX_PG_*,
and update smoke/poller expectations for real decode semantics.

src/test/kotlin/com/gzzn/omms/msgexchange/PipelineSmokeTest.kt

@@ -57,33 +57,43 @@ class PipelineSmokeTest {
         assertNotNull(controller)
     }
 
+    companion object {
+        /** 合法 META + FLOP，但未注册 Handler → FAILED(UNSUPPORTED) */
+        val UNSUPPORTED_FLOP_XML = """
+            <MSG>
+              <META><SNDR>AODB</SNDR><SEQN>1</SEQN><DTTM>20260908120000</DTTM><TYPE>FLOP</TYPE><STYP>DELY</STYP></META>
+              <FLOP><FLID>F1</FLID></FLOP>
+            </MSG>
+        """.trimIndent()
+    }
+
     @Test
-    fun `accept then pump tick transitions message to FAILED CODEC_ERROR with backoff`() {
-        val receipt = controller.send("<MSG/>")
+    fun `accept then pump tick transitions message to FAILED UNSUPPORTED with backoff`() {
+        val receipt = controller.send(UNSUPPORTED_FLOP_XML)
         assertNotNull(receipt.body())                       // 受理 ID
         val id = receipt.body()!!.toLong()
 
-        pump.tick()                                         // 主泵领取 → stub codec 未实装 → FAILED(CODEC_ERROR)
+        pump.tick()                                         // 解码成功但无 DELY Handler → FAILED(UNSUPPORTED)
 
         val stub = ctx.getBean(StubProcState::class.java)
         val s = stub.snapshotOf(id)
         assertNotNull(s)
         assertEquals(ProcStatus.FAILED, s!!.state)
-        assertEquals(ErrorClass.CODEC_ERROR, s.errorClass)
+        assertEquals(ErrorClass.UNSUPPORTED, s.errorClass)
         assertEquals(1, s.attempts)
         assertNotNull(s.nextAttemptAt)
     }
 
     @Test
-    fun `replay reopens failed CODEC_ERROR row to PENDING`() {
-        val receipt = controller.send("<MSG/>")
+    fun `replay reopens failed UNSUPPORTED row to PENDING`() {
+        val receipt = controller.send(UNSUPPORTED_FLOP_XML)
         val id = receipt.body()!!.toLong()
         pump.tick()
         val stub = ctx.getBean(StubProcState::class.java)
         assertEquals(ProcStatus.FAILED, stub.snapshotOf(id)!!.state)
 
         val n = ctx.getBean(com.gzzn.omms.msgexchange.infra.retry.ReplayService::class.java)
-            .replay(listOf(ErrorClass.CODEC_ERROR))
+            .replay(listOf(ErrorClass.UNSUPPORTED))
 
         assertEquals(1, n)
         val reopened = stub.snapshotOf(id)!!

src/test/kotlin/com/gzzn/omms/msgexchange/codec/JacksonXmlCodecTest.kt

@@ -0,0 +1,94 @@
+package com.gzzn.omms.msgexchange.codec
+
+import com.gzzn.omms.msgexchange.domain.MsgKind
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertInstanceOf
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+class JacksonXmlCodecTest {
+
+    private val codec = JacksonXmlCodec()
+
+    @Test
+    fun `decode AODBGTDT sample extracts META and three gates`() {
+        val raw = """
+            <?xml version="1.0" encoding="UTF-8"?>
+            <MSG>
+              <META>
+                <SNDR>AODB</SNDR>
+                <SEQN>1243</SEQN>
+                <DTTM>20021010090311</DTTM>
+                <TYPE>FLOP</TYPE>
+                <STYP>GTDT</STYP>
+              </META>
+              <FLOP>
+                <FLID>121112312</FLID>
+                <FFID>CA-CA101-A-12DEC031345-D</FFID>
+                <GTDT GTNO="1">
+                  <GATE>G28</GATE>
+                  <PGOT>15DEC031805</PGOT>
+                  <PGCT>15DEC031925</PGCT>
+                  <GOTM>15DEC031825</GOTM>
+                  <GTYP>D</GTYP>
+                </GTDT>
+                <GTDT GTNO="2">
+                  <GATE>G33</GATE>
+                  <PGOT>15DEC031805</PGOT>
+                  <PGCT>15DEC031925</PGCT>
+                  <GTYP>I</GTYP>
+                </GTDT>
+                <GTDT GTNO="3">
+                  <GATE>G23</GATE>
+                  <PGOT>15DEC031805</PGOT>
+                  <PGCT>15DEC031925</PGCT>
+                  <GTYP>D</GTYP>
+                </GTDT>
+              </FLOP>
+            </MSG>
+        """.trimIndent()
+
+        val result = codec.decode(raw)
+        assertInstanceOf(DecodeResult.Ok::class.java, result)
+        val msg = (result as DecodeResult.Ok).message
+        assertEquals(MsgKind.Flop("GTDT"), msg.kind)
+        assertEquals("AODB", msg.meta.sndr)
+        assertEquals(1243L, msg.meta.seqn)
+
+        val body = msg.body as FlopPayload
+        assertEquals("121112312", body.flid)
+        assertEquals("CA-CA101-A-12DEC031345-D", body.scalars["FFID"])
+        assertEquals(3, body.collections["GTDT"]!!.size)
+        assertEquals("G28", body.collections["GTDT"]!![0]["GATE"])
+        assertEquals("1", body.collections["GTDT"]!![0]["GTNO"])
+        assertEquals("G33", body.collections["GTDT"]!![1]["GATE"])
+        assertEquals("3", body.collections["GTDT"]!![2]["GTNO"])
+    }
+
+    @Test
+    fun `decode GTNO zero marks gate clear snapshot`() {
+        val raw = """
+            <MSG>
+              <META>
+                <SNDR>AODB</SNDR><SEQN>1</SEQN><DTTM>20021010090311</DTTM>
+                <TYPE>FLOP</TYPE><STYP>GTDT</STYP>
+              </META>
+              <FLOP>
+                <FLID>F_CLEAR</FLID>
+                <GTDT GTNO="0"></GTDT>
+              </FLOP>
+            </MSG>
+        """.trimIndent()
+
+        val msg = (codec.decode(raw) as DecodeResult.Ok).message
+        val body = msg.body as FlopPayload
+        assertEquals(listOf(mapOf("GTNO" to "0")), body.collections["GTDT"])
+    }
+
+    @Test
+    fun `malformed xml returns MALFORMED not CODEC_ERROR`() {
+        val err = codec.decode("not-xml") as DecodeResult.Err
+        assertEquals(com.gzzn.omms.msgexchange.domain.ErrorClass.MALFORMED, err.failure.errorClass)
+        assertTrue(err.failure.detail.contains("empty-or-non-xml"))
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt

@@ -0,0 +1,109 @@
+package com.gzzn.omms.msgexchange.domain.flight
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+class FlightStateEngineTest {
+
+    @Test
+    fun `scalar unchanged does not modify existing value`() {
+        val current = FlightNextState("F1", mapOf("FLNO" to "CA100"), emptyMap(), 1L, "m1")
+        val next = FlightStateEngine.apply(
+            current,
+            FlightFieldCommands("F1", scalars = emptyMap()),
+            "m2",
+            bumpVersion = true,
+        )
+        assertEquals("CA100", next.scalars["FLNO"])
+        assertEquals(2L, next.stateVersion)
+    }
+
+    @Test
+    fun `collection replace preserves order and source sequence`() {
+        val items = listOf(
+            mapOf("GTNO" to "1", "GATE" to "A1"),
+            mapOf("GTNO" to "3", "GATE" to "B2"),
+            mapOf("GTNO" to "7", "GATE" to "C3"),
+        )
+        val next = FlightStateEngine.apply(
+            null,
+            FlightFieldCommands("F1", collections = mapOf("GTDT" to CollectionCommand.Replace(items))),
+            "m1",
+            bumpVersion = true,
+        )
+        assertEquals(3, next.collections["GTDT"]!!.size)
+        assertEquals("3", next.collections["GTDT"]!![1]["GTNO"])
+        assertEquals("C3", next.collections["GTDT"]!![2]["GATE"])
+    }
+
+    @Test
+    fun `collection clear removes prior entries`() {
+        val current = FlightNextState(
+            "F1",
+            emptyMap(),
+            mapOf("GTDT" to listOf(mapOf("GTNO" to "1", "GATE" to "A1"))),
+            1L,
+            "m1",
+        )
+        val next = FlightStateEngine.apply(
+            current,
+            FlightFieldCommands("F1", collections = mapOf("GTDT" to CollectionCommand.Clear)),
+            "m2",
+            bumpVersion = true,
+        )
+        assertFalse(next.collections.containsKey("GTDT"))
+    }
+
+    @Test
+    fun `same gate number different attributes both retained on replace`() {
+        val items = listOf(
+            mapOf("GTNO" to "1", "GATE" to "A1", "GTYP" to "D"),
+            mapOf("GTNO" to "1", "GATE" to "A1", "GTYP" to "I"),
+        )
+        val next = FlightStateEngine.apply(
+            null,
+            FlightFieldCommands("F1", collections = mapOf("GTDT" to CollectionCommand.Replace(items))),
+            "m1",
+            bumpVersion = true,
+        )
+        assertEquals("D", next.collections["GTDT"]!![0]["GTYP"])
+        assertEquals("I", next.collections["GTDT"]!![1]["GTYP"])
+    }
+
+    @Test
+    fun `incremental apply bumps state version from persisted value`() {
+        val current = FlightNextState("F1", mapOf("FLNO" to "CA100"), emptyMap(), 5L, "m1")
+        val next = FlightStateEngine.apply(
+            current,
+            FlightFieldCommands("F1", scalars = mapOf("STAT" to ScalarCommand.Set("DEP"))),
+            "m2",
+            bumpVersion = true,
+        )
+        assertEquals(6L, next.stateVersion)
+    }
+
+    @Test
+    fun `apply by source sequence updates single row`() {
+        val current = FlightNextState(
+            "F1",
+            emptyMap(),
+            mapOf("GTDT" to listOf(mapOf("GTNO" to "2", "GATE" to "OLD"))),
+            1L,
+            "m1",
+        )
+        val next = FlightStateEngine.apply(
+            current,
+            FlightFieldCommands(
+                "F1",
+                collections = mapOf(
+                    "GTDT" to CollectionCommand.Apply(mapOf("GTNO" to "2", "GATE" to "NEW"), "2"),
+                ),
+            ),
+            "m2",
+            bumpVersion = true,
+        )
+        assertEquals("NEW", next.collections["GTDT"]!![0]["GATE"])
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightSchdJdbcPgTest.kt

@@ -11,6 +11,7 @@ import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Assumptions.assumeTrue
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
+import com.gzzn.omms.msgexchange.support.PgTestSupport
 import com.zaxxer.hikari.HikariDataSource
 import java.sql.DriverManager
 import java.time.Instant
@@ -37,22 +38,12 @@ class FlightSchdJdbcPgTest {
 
     @BeforeEach
     fun setUp() {
-        val url = "jdbc:postgresql://localhost:5432/msgx"
-        val user = "msgx_dev"
-        val pass = "msgx_dev_pass"
-
-        // 仅当本地 PostgreSQL 容器可用时执行
-        val canConnect = try {
-            DriverManager.getConnection(url, user, pass).use { true }
-        } catch (_: Exception) {
-            false
-        }
-        assumeTrue(canConnect, "Local PostgreSQL on port 5432 is not accessible, skipping PG dialect integration tests")
+        assumeTrue(PgTestSupport.canConnect(), PgTestSupport.skipMessage())
 
         ds = HikariDataSource().apply {
-            jdbcUrl = url
-            username = user
-            this.password = pass
+            jdbcUrl = PgTestSupport.jdbcUrl
+            username = PgTestSupport.user
+            password = PgTestSupport.password
             driverClassName = "org.postgresql.Driver"
             maximumPoolSize = 2
         }
@@ -244,7 +235,7 @@ class FlightSchdJdbcPgTest {
 
         // 持锁事务内：另一连接 NOWAIT 获取同行锁必须立即失败（互斥生效）
         assertEquals("held", txManager.inTransaction {
-            DriverManager.getConnection("jdbc:postgresql://localhost:5432/msgx", "msgx_dev", "msgx_dev_pass").use { other ->
+            DriverManager.getConnection(PgTestSupport.jdbcUrl, PgTestSupport.user, PgTestSupport.password).use { other ->
                 other.autoCommit = false
                 try {
                     other.prepareStatement(

src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlywayMigrationTest.kt

@@ -1,5 +1,6 @@
 package com.gzzn.omms.msgexchange.infra.persistence.jdbc
 
+import com.gzzn.omms.msgexchange.support.PgTestSupport
 import org.flywaydb.core.Flyway
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertTrue
@@ -18,16 +19,10 @@ class FlywayMigrationTest {
 
     @Test
     fun `Flyway automatically migrates V1_0_0 and V1_1_0 onto real PostgreSQL`() {
-        val url = "jdbc:postgresql://localhost:5432/msgx"
-        val user = "msgx_dev"
-        val pass = "msgx_dev_pass"
-
-        val canConnect = try {
-            DriverManager.getConnection(url, user, pass).use { true }
-        } catch (_: Exception) {
-            false
-        }
-        assumeTrue(canConnect, "PostgreSQL on port 5432 not accessible, skipping Flyway test")
+        assumeTrue(PgTestSupport.canConnect(), PgTestSupport.skipMessage())
+        val url = PgTestSupport.jdbcUrl
+        val user = PgTestSupport.user
+        val pass = PgTestSupport.password
 
         val flyway = Flyway.configure()
             .dataSource(url, user, pass)

src/test/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPollerTest.kt

@@ -1,5 +1,6 @@
 package com.gzzn.omms.msgexchange.ingress
 
+import com.gzzn.omms.msgexchange.domain.ErrorClass
 import com.gzzn.omms.msgexchange.domain.ProcStatus
 import com.gzzn.omms.msgexchange.infra.stub.StubInbox
 import com.gzzn.omms.msgexchange.infra.stub.StubProcState
@@ -40,6 +41,7 @@ class InboxPollerTest {
         poller.pollOnce()
         assertEquals(0, poller.pollOnce())
         pump.tick()
-        assertEquals(ProcStatus.FAILED, stubProc.snapshotOf(id)!!.state)
+        assertEquals(ProcStatus.DEAD, stubProc.snapshotOf(id)!!.state)
+        assertEquals(ErrorClass.MALFORMED, stubProc.snapshotOf(id)!!.errorClass)
     }
 }

src/test/kotlin/com/gzzn/omms/msgexchange/processing/GtdtPipelineTest.kt

@@ -0,0 +1,82 @@
+package com.gzzn.omms.msgexchange.processing
+
+import com.gzzn.omms.msgexchange.MutableClock
+import com.gzzn.omms.msgexchange.codec.DecodeResult
+import com.gzzn.omms.msgexchange.codec.JacksonXmlCodec
+import com.gzzn.omms.msgexchange.config.PipelineProps
+import com.gzzn.omms.msgexchange.domain.ProcState
+import com.gzzn.omms.msgexchange.domain.ProcStatus
+import com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine
+import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
+import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager
+import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler
+import com.gzzn.omms.msgexchange.infra.retry.ProcFailure
+import com.gzzn.omms.msgexchange.infra.stub.StubFlightSchd
+import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents
+import com.gzzn.omms.msgexchange.infra.stub.StubProcState
+import com.gzzn.omms.msgexchange.infra.stub.StubReqTrack
+import com.gzzn.omms.msgexchange.processing.handlers.GtdtHandler
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+/** P2-0-1：GTDT 报文端到端 SUCCEEDED + 三门读回（真实 XmlCodec + GtdtHandler）。 */
+class GtdtPipelineTest {
+
+    @Test
+    fun `GTDT XML end-to-end SUCCEEDED with three gates persisted`() {
+        val raw = """
+            <?xml version="1.0" encoding="UTF-8"?>
+            <MSG>
+              <META>
+                <SNDR>AODB</SNDR><SEQN>9001</SEQN><DTTM>20260908120000</DTTM>
+                <TYPE>FLOP</TYPE><STYP>GTDT</STYP>
+              </META>
+              <FLOP>
+                <FLID>FL_GTDT_E2E</FLID>
+                <GTDT GTNO="1"><GATE>A1</GATE><GTYP>D</GTYP></GTDT>
+                <GTDT GTNO="3"><GATE>B2</GATE><GTYP>I</GTYP></GTDT>
+                <GTDT GTNO="7"><GATE>C3</GATE><GTYP>D</GTYP></GTDT>
+              </FLOP>
+            </MSG>
+        """.trimIndent()
+
+        val procState = StubProcState()
+        val flightSchd = StubFlightSchd()
+        val msgEvents = StubMsgEvents()
+        val reqTrack = StubReqTrack()
+        val txManager = object : PipelineTransactionManager {
+            override fun <T> inTransaction(block: () -> T): T = block()
+        }
+        val inbox = object : CminmsgInboxRepository {
+            override fun insertRaw(rawXml: String) = 1L
+            override fun rawOf(cminmsgsId: Long) = raw
+            override fun pollUnprocessed(afterId: Long, limit: Int) = emptyList<Long>()
+            override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) = Unit
+        }
+        val clock = MutableClock(MutableClock.BASE)
+        val procFailure = ProcFailure(procState, FailureScheduler(PipelineProps(), clock))
+        val codec = JacksonXmlCodec()
+        val handler = GtdtHandler()
+        val snapshot = SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox)
+        val processor = MessageProcessor(
+            inbox, procState, msgEvents,
+            CodecHolder(codec),
+            HandlerHolder(HandlerRegistry(listOf(handler))),
+            flightSchd, snapshot, procFailure, PipelineProps(), txManager,
+        )
+
+        val headId = 42L
+        procState.insert(headId, ProcStatus.PENDING)
+        processor.processOne(ProcState(headId, ProcStatus.PENDING))
+
+        assertEquals(ProcStatus.SUCCEEDED, procState.snapshotOf(headId)!!.state)
+        val next = flightSchd.findNextStateByFlid("FL_GTDT_E2E")!!
+        assertEquals(3, next.collections["GTDT"]!!.size)
+        assertEquals("3", next.collections["GTDT"]!![1]["GTNO"])
+        assertEquals("C3", next.collections["GTDT"]!![2]["GATE"])
+        assertEquals(1L, next.stateVersion)
+
+        val roundtrip = FlightStateEngine.fromFlightFields("FL_GTDT_E2E", flightSchd.findByFlid("FL_GTDT_E2E")!!)
+        assertEquals(3, roundtrip.collections["GTDT"]!!.size)
+    }
+}

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

@@ -175,6 +175,22 @@ class MessageProcessorTest {
             return true
         }
         override fun deleteGenBefore(cutoffDay: String): Int = 0
+        override fun findNextStateByFlid(flid: String): com.gzzn.omms.msgexchange.domain.flight.FlightNextState? =
+            flights[flid]?.let { com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine.fromFlightFields(flid, it) }
+        override fun persistNextStates(
+            day: String?,
+            states: List<com.gzzn.omms.msgexchange.domain.flight.FlightNextState>,
+            snapshotReplace: Boolean,
+            now: Instant,
+        ) {
+            states.forEach { state ->
+                val scalarFields = linkedMapOf("FLID" to state.flid)
+                state.scalars.forEach { (k, v) -> scalarFields[k] = v }
+                flights[state.flid] = scalarFields + state.collections.mapValues { (_, items) ->
+                    com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(items)
+                }
+            }
+        }
     }
 
     private class FakeTxManager : PipelineTransactionManager {

src/test/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlowBackfillTest.kt

@@ -0,0 +1,110 @@
+package com.gzzn.omms.msgexchange.processing
+
+import com.gzzn.omms.msgexchange.domain.DecodedMessage
+import com.gzzn.omms.msgexchange.domain.MetaFields
+import com.gzzn.omms.msgexchange.domain.MsgKind
+import com.gzzn.omms.msgexchange.domain.ProcState
+import com.gzzn.omms.msgexchange.domain.ProcStatus
+import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
+import com.gzzn.omms.msgexchange.infra.persistence.FlightSchdRepository
+import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
+import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager
+import com.gzzn.omms.msgexchange.infra.persistence.ProcStateRepository
+import com.gzzn.omms.msgexchange.infra.persistence.ReqTrackRepository
+import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler
+import com.gzzn.omms.msgexchange.infra.retry.ProcFailure
+import com.gzzn.omms.msgexchange.config.PipelineProps
+import com.gzzn.omms.msgexchange.MutableClock
+import com.gzzn.omms.msgexchange.infra.stub.StubFlightSchd
+import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents
+import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTransactionManager
+import com.gzzn.omms.msgexchange.infra.stub.StubProcState
+import com.gzzn.omms.msgexchange.infra.stub.StubReqTrack
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+/** v2 §2.6 / §5：提交后回填失败不得回滚 PROC_STATE。 */
+class SnapshotFlowBackfillTest {
+
+    private class FailingInbox : CminmsgInboxRepository {
+        override fun insertRaw(rawXml: String): Long = 1L
+        override fun rawOf(cminmsgsId: Long): String? = null
+        override fun pollUnprocessed(afterId: Long, limit: Int): List<Long> = emptyList()
+        override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) {
+            throw RuntimeException("mysql-down")
+        }
+    }
+
+    @Test
+    fun `backfill failure after commit leaves PROC_STATE SUCCEEDED`() {
+        val flightSchd = StubFlightSchd()
+        val procState = StubProcState()
+        val msgEvents = StubMsgEvents()
+        val reqTrack = StubReqTrack()
+        val txManager = StubPipelineTransactionManager()
+        val inbox = FailingInbox()
+        val clock = MutableClock(MutableClock.BASE)
+        val procFailure = ProcFailure(procState, FailureScheduler(PipelineProps(), clock))
+
+        val headId = 901L
+        procState.insert(headId, ProcStatus.PENDING)
+        val day = "2026-09-07"
+        val meta = MetaFields("AODB", "SCHD", "DNLD", 9L, 20260907090000L)
+        val decoded = DecodedMessage(meta, MsgKind.Schd(MsgKind.SchdSubtype.DNLD), "<SCHD/>")
+        SnapshotFlow.StageResult.parser = {
+            SnapshotFlow.StageResult.Ok(day, listOf("FL_X" to mapOf("FLID" to "FL_X")))
+        }
+
+        val flow = SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox)
+        flow.publishSnapshot(ProcState(headId, ProcStatus.PENDING), decoded)
+
+        assertEquals(ProcStatus.SUCCEEDED, procState.snapshotOf(headId)!!.state)
+        assertTrue(flightSchd.findByFlid("FL_X") != null)
+    }
+
+    @Test
+    fun `CAS replay requires matching lastMessageId not version alone`() {
+        val base = StubFlightSchd()
+        val procState = StubProcState()
+        val msgEvents = StubMsgEvents()
+        val reqTrack = StubReqTrack()
+        val txManager = StubPipelineTransactionManager()
+        val inbox = object : CminmsgInboxRepository {
+            override fun insertRaw(rawXml: String) = 1L
+            override fun rawOf(cminmsgsId: Long) = null
+            override fun pollUnprocessed(afterId: Long, limit: Int): List<Long> = emptyList()
+            override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) = Unit
+        }
+        val clock = MutableClock(MutableClock.BASE)
+        val procFailure = ProcFailure(procState, FailureScheduler(PipelineProps(), clock))
+
+        val headId = 902L
+        val day = "2026-09-07"
+        procState.insert(headId, ProcStatus.PENDING)
+
+        val conflictingGen = FlightSchdRepository.GenMeta(day, 2L, setOf("OTHER"), lastMessageId = "999")
+        base.putGenIfVersion(day, 0L, FlightSchdRepository.GenMeta(day, 1L, setOf("OTHER"), lastMessageId = "999"))
+        val mockFlightSchd = object : FlightSchdRepository by base {
+            override fun putGenIfVersion(
+                day: String,
+                expected: Long,
+                newGen: FlightSchdRepository.GenMeta,
+                now: java.time.Instant,
+            ): Boolean {
+                base.putGenIfVersion(day, 1L, conflictingGen)
+                return false
+            }
+        }
+
+        SnapshotFlow.StageResult.parser = {
+            SnapshotFlow.StageResult.Ok(day, listOf("FL_Y" to mapOf("FLID" to "FL_Y")))
+        }
+        val meta = MetaFields("AODB", "SCHD", "DNLD", 10L, 20260907100000L)
+        val decoded = DecodedMessage(meta, MsgKind.Schd(MsgKind.SchdSubtype.DNLD), "<SCHD/>")
+        val flow = SnapshotFlow(procState, mockFlightSchd, msgEvents, reqTrack, procFailure, txManager, inbox)
+        flow.publishSnapshot(ProcState(headId, ProcStatus.PENDING), decoded)
+
+        assertEquals(ProcStatus.FAILED, procState.snapshotOf(headId)!!.state)
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/processing/V2PipelineWritePathContractTest.kt

@@ -0,0 +1,115 @@
+package com.gzzn.omms.msgexchange.processing
+
+import com.gzzn.omms.msgexchange.MutableClock
+import com.gzzn.omms.msgexchange.config.PipelineProps
+import com.gzzn.omms.msgexchange.domain.DecodedMessage
+import com.gzzn.omms.msgexchange.domain.Decision
+import com.gzzn.omms.msgexchange.domain.FlightChange
+import com.gzzn.omms.msgexchange.domain.MetaFields
+import com.gzzn.omms.msgexchange.domain.MsgKind
+import com.gzzn.omms.msgexchange.domain.ProcState
+import com.gzzn.omms.msgexchange.domain.ProcStatus
+import com.gzzn.omms.msgexchange.infra.persistence.CminmsgInboxRepository
+import com.gzzn.omms.msgexchange.infra.retry.FailureScheduler
+import com.gzzn.omms.msgexchange.infra.retry.ProcFailure
+import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents
+import com.gzzn.omms.msgexchange.infra.stub.StubPipelineTransactionManager
+import com.gzzn.omms.msgexchange.infra.stub.StubProcState
+import com.gzzn.omms.msgexchange.infra.stub.StubReqTrack
+import com.gzzn.omms.msgexchange.codec.JacksonXmlCodec
+import com.gzzn.omms.msgexchange.processing.handlers.GtdtHandler
+import com.gzzn.omms.msgexchange.support.GuardedFlightSchd
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import java.nio.file.Paths
+import kotlin.io.path.readText
+
+class V2PipelineWritePathContractTest {
+
+    @Test
+    fun `SnapshotFlow and MessageProcessor sources must not call legacy upsert helpers`() {
+        val processingDir = Paths.get("src/main/kotlin/com/gzzn/omms/msgexchange/processing")
+        val snapshotSource = processingDir.resolve("SnapshotFlow.kt").readText()
+        val pumpSource = processingDir.resolve("Pump.kt").readText()
+
+        assertFalse(snapshotSource.contains("upsertSnapshotBatch"))
+        assertFalse(snapshotSource.contains("upsertIncremental"))
+        assertTrue(snapshotSource.contains("persistNextStates"))
+
+        assertFalse(pumpSource.contains("upsertSnapshotBatch"))
+        assertFalse(pumpSource.contains("upsertIncremental"))
+        assertTrue(pumpSource.contains("persistNextStates"))
+    }
+
+    @Test
+    fun `SnapshotFlow DNLD path calls persistNextStates not legacy upsert helpers`() {
+        val flightSchd = GuardedFlightSchd()
+        val procState = StubProcState()
+        val msgEvents = StubMsgEvents()
+        val reqTrack = StubReqTrack()
+        val txManager = StubPipelineTransactionManager()
+        val inbox = object : CminmsgInboxRepository {
+            override fun insertRaw(rawXml: String) = 1L
+            override fun rawOf(cminmsgsId: Long) = null
+            override fun pollUnprocessed(afterId: Long, limit: Int): List<Long> = emptyList()
+            override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) = Unit
+        }
+        val procFailure = ProcFailure(procState, FailureScheduler(PipelineProps(), MutableClock(MutableClock.BASE)))
+
+        val headId = 1001L
+        procState.insert(headId, ProcStatus.PENDING)
+        val day = "2026-09-07"
+        SnapshotFlow.StageResult.parser = {
+            SnapshotFlow.StageResult.Ok(day, listOf("FL_DNLD" to mapOf("FLID" to "FL_DNLD", "FLNO" to "CA100", "GTDT" to """[{"GTNO":"1","GATE":"A1"}]""")))
+        }
+        val decoded = DecodedMessage(MetaFields("AODB", "SCHD", "DNLD", 11L, 20260907110000L), MsgKind.Schd(MsgKind.SchdSubtype.DNLD), "<SCHD/>")
+        SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox)
+            .publishSnapshot(ProcState(headId, ProcStatus.PENDING), decoded)
+
+        assertEquals(ProcStatus.SUCCEEDED, procState.snapshotOf(headId)!!.state)
+        assertEquals(1, flightSchd.persistNextStatesCalls)
+        assertEquals(0, flightSchd.legacySnapshotCalls)
+        assertEquals(0, flightSchd.legacyIncrementalCalls)
+    }
+
+    @Test
+    fun `MessageProcessor FLOP path calls persistNextStates not legacy upsert helpers`() {
+        val flightSchd = GuardedFlightSchd()
+        val procState = StubProcState()
+        val msgEvents = StubMsgEvents()
+        val reqTrack = StubReqTrack()
+        val txManager = StubPipelineTransactionManager()
+        val raws = mutableMapOf<Long, String>()
+        val inbox = object : CminmsgInboxRepository {
+            override fun insertRaw(rawXml: String) = 1L
+            override fun rawOf(cminmsgsId: Long) = raws[cminmsgsId]
+            override fun pollUnprocessed(afterId: Long, limit: Int): List<Long> = emptyList()
+            override fun backfillOnSuccess(cminmsgsId: Long, sndr: String, type: String, styp: String, seqn: Long) = Unit
+        }
+        val procFailure = ProcFailure(procState, FailureScheduler(PipelineProps(), MutableClock(MutableClock.BASE)))
+        val handler = GtdtHandler()
+        val snapshot = SnapshotFlow(procState, flightSchd, msgEvents, reqTrack, procFailure, txManager, inbox)
+        val processor = MessageProcessor(
+            inbox, procState, msgEvents,
+            CodecHolder(JacksonXmlCodec()),
+            HandlerHolder(HandlerRegistry(listOf(handler))),
+            flightSchd, snapshot, procFailure, PipelineProps(), txManager,
+        )
+        val headId = 1002L
+        procState.insert(headId, ProcStatus.PENDING)
+        raws[headId] = """
+            <MSG>
+              <META><SNDR>AODB</SNDR><SEQN>1</SEQN><DTTM>20260907120000</DTTM><TYPE>FLOP</TYPE><STYP>GTDT</STYP></META>
+              <FLOP><FLID>FL_FLOP</FLID><GTDT GTNO="2"><GATE>B9</GATE></GTDT></FLOP>
+            </MSG>
+        """.trimIndent()
+        processor.processOne(ProcState(headId, ProcStatus.PENDING))
+
+        assertEquals(ProcStatus.SUCCEEDED, procState.snapshotOf(headId)!!.state)
+        assertEquals(1, flightSchd.persistNextStatesCalls)
+        assertEquals(0, flightSchd.legacySnapshotCalls)
+        assertEquals(0, flightSchd.legacyIncrementalCalls)
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/processing/handlers/GtdtHandlerTest.kt

@@ -0,0 +1,41 @@
+package com.gzzn.omms.msgexchange.processing.handlers
+
+import com.gzzn.omms.msgexchange.codec.FlopPayload
+import com.gzzn.omms.msgexchange.domain.DecodedMessage
+import com.gzzn.omms.msgexchange.domain.MetaFields
+import com.gzzn.omms.msgexchange.domain.MsgKind
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+class GtdtHandlerTest {
+
+    private val handler = GtdtHandler()
+
+    @Test
+    fun `decide replaces GTDT collection for three gates`() {
+        val msg = DecodedMessage(
+            meta = MetaFields("AODB", "FLOP", "GTDT", 1L, 20260908120000L),
+            kind = MsgKind.Flop("GTDT"),
+            rawXml = "<MSG/>",
+            body = FlopPayload(
+                flid = "121112312",
+                scalars = mapOf("FFID" to "CA-CA101-A-12DEC031345-D"),
+                collections = mapOf(
+                    "GTDT" to listOf(
+                        mapOf("GTNO" to "1", "GATE" to "G28", "GTYP" to "D"),
+                        mapOf("GTNO" to "2", "GATE" to "G33", "GTYP" to "I"),
+                        mapOf("GTNO" to "3", "GATE" to "G23", "GTYP" to "D"),
+                    ),
+                ),
+            ),
+        )
+
+        val decision = handler.decide(emptyMap(), msg)
+        assertEquals(1, decision.flightChanges.size)
+        assertEquals("121112312", decision.flightChanges[0].flid)
+        val gtdtJson = decision.flightChanges[0].fields["GTDT"]!!
+        assertEquals(true, gtdtJson.contains("\"G28\""))
+        assertEquals(1, decision.schdPush.size)
+        assertEquals("121112312", decision.schdPush[0].flid)
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/support/PgTestSupport.kt

@@ -0,0 +1,30 @@
+package com.gzzn.omms.msgexchange.support
+
+/**
+ * JDBC 集成测试 PostgreSQL 连接参数（环境变量化，解决本机 5432 端口冲突）。
+ *
+ * | 变量 | 默认 |
+ * |---|---|
+ * | MSGX_PG_URL | jdbc:postgresql://${MSGX_PG_HOST:127.0.0.1}:${MSGX_PG_PORT:5432}/${MSGX_PG_NAME:msgx} |
+ * | MSGX_PG_USER | msgx_dev |
+ * | MSGX_PG_PASSWORD | msgx_dev_pass |
+ */
+object PgTestSupport {
+    private val host = System.getenv("MSGX_PG_HOST") ?: "127.0.0.1"
+    private val port = System.getenv("MSGX_PG_PORT") ?: "5432"
+    private val name = System.getenv("MSGX_PG_NAME") ?: "msgx"
+
+    val jdbcUrl: String = System.getenv("MSGX_PG_URL")
+        ?: "jdbc:postgresql://$host:$port/$name"
+    val user: String = System.getenv("MSGX_PG_USER") ?: "msgx_dev"
+    val password: String = System.getenv("MSGX_PG_PASSWORD") ?: "msgx_dev_pass"
+
+    fun canConnect(): Boolean = try {
+        java.sql.DriverManager.getConnection(jdbcUrl, user, password).use { true }
+    } catch (_: Exception) {
+        false
+    }
+
+    fun skipMessage(): String =
+        "PostgreSQL not accessible at $jdbcUrl (set MSGX_PG_URL / MSGX_PG_PORT to avoid port conflicts)"
+}

src/test/kotlin/com/gzzn/omms/msgexchange/support/V2WritePathGuard.kt

@@ -0,0 +1,60 @@
+package com.gzzn.omms.msgexchange.support
+
+import com.gzzn.omms.msgexchange.domain.FlightChange
+import com.gzzn.omms.msgexchange.domain.flight.FlightNextState
+import com.gzzn.omms.msgexchange.infra.persistence.FlightFields
+import com.gzzn.omms.msgexchange.infra.persistence.FlightSchdRepository
+import com.gzzn.omms.msgexchange.infra.stub.StubFlightSchd
+import java.time.Instant
+
+/** V1.2.0 宽表集合存储列；v2 persistNextStates 不得写入。 */
+val LEGACY_COLLECTION_STORAGE_COLUMNS: List<String> = listOf(
+    "gate1", "pgot1", "pgct1", "gotm1", "gctm1", "gtyp1",
+    "gate2", "pgot2", "pgct2", "gotm2", "gctm2", "gtyp2",
+    "chkc1", "ccls1", "pcot1", "pcct1", "cotm1", "cctm1", "ctyp1",
+    "chkc2", "ccls2", "pcot2", "pcct2", "cotm2", "cctm2", "ctyp2",
+    "chkc3", "ccls3", "pcot3", "pcct3", "cotm3", "cctm3", "ctyp3",
+    "belt1", "bcls1", "bpcot1", "bpcct1", "fbag1", "lbag1", "btyp1",
+    "belt2", "bcls2", "bpcot2", "bpcct2", "fbag2", "lbag2", "btyp2",
+    "psst1", "stst1", "stet1", "psst2", "stst2", "stet2",
+    "chut1", "chcls1", "pcbt1", "pcet1", "cbtm1", "cetm1", "chtyp1",
+    "chut2", "chcls2", "pcbt2", "pcet2", "cbtm2", "cetm2", "chtyp2",
+    "dely_code", "dely_strt", "dely_dura", "dely_remc",
+    "abtm_a", "abtm_d", "chot_on", "chot_off",
+    "rout_path", "erut_path",
+)
+
+class GuardedFlightSchd(
+    private val inner: FlightSchdRepository = StubFlightSchd(),
+) : FlightSchdRepository {
+    var legacySnapshotCalls = 0
+    var legacyIncrementalCalls = 0
+    var persistNextStatesCalls = 0
+
+    override fun upsertSnapshotBatch(day: String, flights: List<Pair<String, FlightFields>>, now: Instant) {
+        legacySnapshotCalls++
+        throw AssertionError("v2 path must not call upsertSnapshotBatch")
+    }
+
+    override fun upsertIncremental(changes: List<FlightChange>, now: Instant) {
+        legacyIncrementalCalls++
+        throw AssertionError("v2 path must not call upsertIncremental")
+    }
+
+    override fun persistNextStates(day: String?, states: List<FlightNextState>, snapshotReplace: Boolean, now: Instant) {
+        persistNextStatesCalls++
+        inner.persistNextStates(day, states, snapshotReplace, now)
+    }
+
+    override fun deleteDiffByDay(day: String, delFlids: Collection<String>): Int = inner.deleteDiffByDay(day, delFlids)
+    override fun findByFlid(flid: String): FlightFields? = inner.findByFlid(flid)
+    override fun findNextStateByFlid(flid: String): FlightNextState? = inner.findNextStateByFlid(flid)
+    override fun findByFlids(flids: Collection<String>): Map<String, FlightFields> = inner.findByFlids(flids)
+    override fun findByDay(day: String): List<Pair<String, FlightFields>> = inner.findByDay(day)
+    override fun findAll(): Map<String, FlightFields> = inner.findAll()
+    override fun deleteByFlids(flids: Set<String>): Int = inner.deleteByFlids(flids)
+    override fun getGen(day: String): FlightSchdRepository.GenMeta? = inner.getGen(day)
+    override fun putGenIfVersion(day: String, expected: Long, newGen: FlightSchdRepository.GenMeta, now: Instant): Boolean =
+        inner.putGenIfVersion(day, expected, newGen, now)
+    override fun deleteGenBefore(cutoffDay: String): Int = inner.deleteGenBefore(cutoffDay)
+}