refactor(config): drop system-time defaults from domain and entry points

删除会被误用的系统时间默认值：MsgEvent.createdAt、ProcState.updatedAt、InboxPoller.pollOnce(now)、HistorySweepJob.run(now)、Identity.of(day)。Identity 的日期边界改由 MessageProcessor 按 PARAM:msgx.operation-day.zone 算出后传入（边界关闭时身份值不变）；StubProcState/StubInbox 改注入 Clock（测试可传假时钟），全部调用点显式传时间。

验证：./gradlew test 145 tests / 0 fail / 0 skipped。

src/main/kotlin/com/gzzn/omms/msgexchange/domain/MsgEvent.kt

@@ -28,5 +28,5 @@ data class MsgEvent(
     val nextAttemptAt: java.time.Instant? = null,
     val errorClass: ErrorClass? = null,
     val lastError: String? = null,
-    val createdAt: java.time.Instant = java.time.Instant.now(),
+    val createdAt: java.time.Instant,
 )

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

@@ -86,5 +86,5 @@ data class ProcState(
     val backfillAbandonedAt: Instant? = null,
     /** 放弃原因（`MISSING_ROW` / `TRANSIENT_DEADLINE`），供人工对账与恢复判断。 */
     val backfillAbandonedReason: String? = null,
-    val updatedAt: Instant = Instant.now(),
+    val updatedAt: Instant,
 )

src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt

@@ -30,6 +30,7 @@ import com.gzzn.omms.msgexchange.domain.SnapshotLogEntry
 import com.gzzn.omms.msgexchange.infra.persistence.SnapshotLogRepository
 import io.micronaut.context.annotation.Requires
 import jakarta.inject.Singleton
+import java.time.Clock
 import java.time.Instant
 import java.time.LocalDate
 import java.time.ZoneId
@@ -56,7 +57,7 @@ class StubPipelineLock : PipelineLockRepository {
 @Singleton
 @Requires(property = "msgx.stubs", value = "true")
 /** 内存版 PROC_STATE：入队幂等，写终态时一并写下回填待办。 */
-class StubProcState : ProcStateRepository {
+class StubProcState(private val clock: Clock = Clock.systemUTC()) : ProcStateRepository {
     val rows = linkedMapOf<Long, ProcState>()
     val bound = linkedMapOf<String, Long>()
 
@@ -74,6 +75,7 @@ class StubProcState : ProcStateRepository {
             msgId, ProcStatus.PENDING,
             receivedAt = receivedAt,
             enqueuedAt = enqueuedAt ?: receivedAt,
+            updatedAt = clock.instant(),
         )
         return true
     }
@@ -89,7 +91,7 @@ class StubProcState : ProcStateRepository {
         val owner = bound[identityKey]
         if (owner != null && owner != msgId) return false
         bound[identityKey] = msgId
-        rows[msgId] = (rows[msgId] ?: ProcState(msgId, ProcStatus.PENDING)).copy(identityKey = identityKey)
+        rows[msgId] = (rows[msgId] ?: ProcState(msgId, ProcStatus.PENDING, updatedAt = clock.instant())).copy(identityKey = identityKey)
         return true
     }
 
@@ -103,14 +105,14 @@ class StubProcState : ProcStateRepository {
         errorClass: ErrorClass?,
         lastError: String?,
     ) {
-        val old = rows[msgId] ?: ProcState(msgId, state)
+        val old = rows[msgId] ?: ProcState(msgId, state, updatedAt = clock.instant())
         rows[msgId] = old.copy(
             state = state,
             nextAttemptAt = nextAttemptAt ?: old.nextAttemptAt,
             attempts = attempts ?: old.attempts,
             errorClass = errorClass ?: old.errorClass,
             lastError = lastError ?: old.lastError,
-            updatedAt = Instant.now(),
+            updatedAt = clock.instant(),
         )
     }
 
@@ -123,7 +125,7 @@ class StubProcState : ProcStateRepository {
         attempts: Int?,
         now: Instant,
     ) {
-        val old = rows[msgId] ?: ProcState(msgId, state)
+        val old = rows[msgId] ?: ProcState(msgId, state, updatedAt = clock.instant())
         rows[msgId] = old.copy(
             state = state,
             errorClass = errorClass,
@@ -451,7 +453,7 @@ class StubReqTrack : ReqTrackRepository {
 @Singleton
 @Requires(property = "msgx.stubs", value = "true")
 /** 内存版共享信箱：可以模拟上游写入、库方清除，并记录哪些行被打上了处理标记。 */
-class StubInbox : CminmsgInboxRepository {
+class StubInbox(private val clock: Clock = Clock.systemUTC()) : CminmsgInboxRepository {
     val raws = linkedMapOf<Long, String>()
     private val received = linkedMapOf<Long, Instant>()
     private val marks = linkedMapOf<Long, String>()
@@ -464,7 +466,7 @@ class StubInbox : CminmsgInboxRepository {
     override fun insertRaw(rawXml: String): Long {
         val id = ids.incrementAndGet()
         raws[id] = rawXml
-        received[id] = Instant.now()
+        received[id] = clock.instant()
         return id
     }
 
@@ -497,7 +499,7 @@ class StubInbox : CminmsgInboxRepository {
     fun simulateExternalWrite(rawXml: String): Long = insertRaw(rawXml)
 
     /** 测试辅助：模拟先分配 ID、稍后才可见的迟提交行。 */
-    fun restoreRow(msgId: Long, rawXml: String, receivedAt: Instant = Instant.now()) {
+    fun restoreRow(msgId: Long, rawXml: String, receivedAt: Instant = clock.instant()) {
         raws[msgId] = rawXml
         received[msgId] = receivedAt
     }

src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxPoller.kt

@@ -48,7 +48,7 @@ class InboxPoller(
     private var running = false
 
     /** @return 本轮新登记的消息条数（已登记过的行不计入，也不影响水位推进）。 */
-    fun pollOnce(now: Instant = Instant.now()): Int {
+    fun pollOnce(now: Instant): Int {
         seedCutoverWatermarkIfConfigured(now)
         detectLateArrivalsIfDue(now)
         val batch = props.pipeline.claimBatch.coerceAtLeast(1)

src/main/kotlin/com/gzzn/omms/msgexchange/jobs/HistorySweepJob.kt

@@ -49,7 +49,7 @@ class HistorySweepJob(
 
     data class SweepOutcome(val selected: Int, val archived: Int, val purged: Int, val snapLogPurged: Int = 0)
 
-    fun run(now: Instant = Instant.now()): SweepOutcome {
+    fun run(now: Instant): SweepOutcome {
         // 留痕是本地可重建记录，清理不需要外部归档确认，不受历史存储开关牵制。
         val snapLogPurged = snapLogPurge?.purgeBefore(now.minus(Duration.ofDays(props.snapLogRetentionDays))) ?: 0
 

src/main/kotlin/com/gzzn/omms/msgexchange/processing/Identity.kt

@@ -16,12 +16,13 @@ object Identity {
     fun of(
         msg: DecodedMessage,
         includeDayBoundary: Boolean,
-        day: LocalDate = LocalDate.now(),
+        day: LocalDate,
     ): 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)
+    /** 日期边界由调用方按机场时区算出后传入；本算法不再自己读系统时间。 */
+    fun of(msg: DecodedMessage, props: PipelineProps.Identity, day: LocalDate): String =
+        of(msg, props.includeDayBoundary, day)
 }

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

@@ -3,6 +3,7 @@ package com.gzzn.omms.msgexchange.processing
 import com.gzzn.omms.msgexchange.codec.FlopPayload
 import com.gzzn.omms.msgexchange.codec.ScheduleBody
 import com.gzzn.omms.msgexchange.codec.XmlCodec
+import com.gzzn.omms.msgexchange.config.OperationDayProps
 import com.gzzn.omms.msgexchange.config.PipelineProps
 import com.gzzn.omms.msgexchange.domain.DecodedMessage
 import com.gzzn.omms.msgexchange.domain.ErrorClass
@@ -18,6 +19,7 @@ import jakarta.inject.Singleton
 import java.time.Clock
 import java.time.Duration
 import java.time.Instant
+import java.time.LocalDate
 import java.util.concurrent.atomic.AtomicLong
 
 /**
@@ -147,6 +149,7 @@ class MessageProcessor(
     private val procFailure: ProcFailure,
     private val props: PipelineProps,
     private val clock: Clock,
+    private val operationDayProps: OperationDayProps,
 ) {
     private val log = org.slf4j.LoggerFactory.getLogger(MessageProcessor::class.java)
 
@@ -186,7 +189,9 @@ class MessageProcessor(
 
         // I3：identity 仅首次绑定（head.identityKey == null）；FAILED 重试不重绑
         if (head.identityKey == null) {
-            val identity = Identity.of(decoded, props.identity)
+            val identity = Identity.of(
+                decoded, props.identity, LocalDate.now(clock.withZone(operationDayProps.zoneId())),
+            )
             if (!procState.tryBindIdentity(head.msgId, identity)) {
                 val owner = procState.ownerOfIdentity(identity) ?: -1L
                 log.info("duplicate-of:{} -> SKIPPED msgId={}", owner, head.msgId)

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

@@ -92,7 +92,7 @@ class PipelineSmokeTest {
         assertNotNull(receipt.body())                       // 受理 ID
         val id = receipt.body()!!.toLong()
         // 兼容入口只保证"已落信 + 已入队"，**不推进水位**；必须先被收报发现（W 追平）才可领取。
-        ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce()
+        ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now())
 
         pump.tick()                                         // 解码成功但无 DELY Handler → FAILED(UNSUPPORTED)
 
@@ -109,7 +109,7 @@ class PipelineSmokeTest {
     fun `replay reopens failed UNSUPPORTED row to PENDING`() {
         val receipt = controller.send(UNSUPPORTED_XML)
         val id = receipt.body()!!.toLong()
-        ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce()
+        ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now())
         pump.tick()
         val stub = ctx.getBean(StubProcState::class.java)
         assertEquals(ProcStatus.FAILED, stub.snapshotOf(id)!!.state)
@@ -135,7 +135,7 @@ class PipelineSmokeTest {
         val inbox = ctx.getBean(com.gzzn.omms.msgexchange.infra.stub.StubInbox::class.java)
         val proc = ctx.getBean(StubProcState::class.java)
         val dead = inbox.simulateExternalWrite("<MSG/>")
-        ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce()
+        ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now())
 
         pump.tick()   // 解码 MALFORMED → DEAD + 回填意图（同一条 UPDATE）
 
@@ -152,7 +152,7 @@ class PipelineSmokeTest {
         assertTrue(inbox.isMarked(dead))
 
         val fresh = inbox.simulateExternalWrite("<MSG/>")
-        ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce()
+        ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now())
         assertNotNull(proc.find(fresh))   // 死信不阻断后续发现
     }
 
@@ -167,7 +167,7 @@ class PipelineSmokeTest {
         val flights = ctx.getBean(com.gzzn.omms.msgexchange.infra.stub.StubFlightState::class.java)
         val events = ctx.getBean(StubMsgEvents::class.java)
         val id = inbox.simulateExternalWrite("<MSG/>")          // 非法报文（无 META）→ MALFORMED
-        ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce()
+        ctx.getBean(com.gzzn.omms.msgexchange.ingress.InboxPoller::class.java).pollOnce(Instant.now())
 
         pump.tick()
 
@@ -260,8 +260,8 @@ class PipelineSmokeTest {
         val events = ctx.getBean(StubMsgEvents::class.java)
         val port = ctx.getBean(StubDeliveryPort::class.java)
         events.insertAll(listOf(
-            MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = "F1", stateVersion = 1, payloadJson = """{"FLID":"F1","v":"old"}"""),
-            MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = "F1", stateVersion = 2, payloadJson = """{"FLID":"F1","v":"new"}"""),
+            MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = "F1", stateVersion = 1, payloadJson = """{"FLID":"F1","v":"old"}""", createdAt = Instant.now()),
+            MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = "F1", stateVersion = 2, payloadJson = """{"FLID":"F1","v":"new"}""", createdAt = Instant.now()),
         ))
 
         dispatcher.flushSchd()

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

@@ -32,7 +32,7 @@ class DispatcherTickTest {
         Dispatcher(repo, port, p, FailureScheduler(p, clock))
 
     private fun ev(id: Long, target: String, key: String, payload: String, version: Long = 0) =
-        MsgEvent(eventId = id, target = target, partitionKey = key, stateVersion = version, payloadJson = payload)
+        MsgEvent(eventId = id, target = target, partitionKey = key, stateVersion = version, payloadJson = payload, createdAt = MutableClock.BASE)
 
     @Test
     fun `first tick delivers msg and starts the independent schd flush`() {
@@ -107,6 +107,7 @@ class DispatcherTickTest {
                 MsgEvent(
                     eventId = 1, target = Targets.KAFKA_SCHD, partitionKey = "F1",
                     eventType = EventType.TOMBSTONE, stateVersion = 3, payloadJson = """{"flid":"F1","deleted":true}""",
+                    createdAt = MutableClock.BASE,
                 ),
             ),
         )
@@ -160,6 +161,7 @@ class DispatcherTickTest {
                 MsgEvent(
                     target = Targets.KAFKA_SCHD, partitionKey = "F1",
                     eventType = EventType.TOMBSTONE, stateVersion = 3, payloadJson = """{"flid":"F1","deleted":true}""",
+                    createdAt = MutableClock.BASE,
                 ),
             ),
         )
@@ -234,7 +236,7 @@ private class ReentrantSchdPort(private val repo: StubMsgEvents) : DeliveryPort
     override fun sendKafkaSchd(topic: String, key: String, payloadJson: String) {
         repo.insertAll(
             listOf(
-                MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = key, stateVersion = 99, payloadJson = """{"v":"new"}"""),
+                MsgEvent(target = Targets.KAFKA_SCHD, partitionKey = key, stateVersion = 99, payloadJson = """{"v":"new"}""", createdAt = MutableClock.BASE),
             ),
         )
     }

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

@@ -22,13 +22,14 @@ class ReplayServiceTest {
         fun seed(id: Long, status: ProcStatus, ec: ErrorClass?) {
             rows[id] = ProcState(
                 id, status, identityKey = "k$id", attempts = 3, errorClass = ec, lastError = "x",
+                updatedAt = Instant.EPOCH,
             )
         }
 
         override fun insertIfAbsent(msgId: Long, receivedAt: Instant?, enqueuedAt: Instant?): Boolean =
             rows.putIfAbsent(
                 msgId,
-                ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt, enqueuedAt = enqueuedAt ?: receivedAt),
+                ProcState(msgId, ProcStatus.PENDING, receivedAt = receivedAt, enqueuedAt = enqueuedAt ?: receivedAt, updatedAt = Instant.EPOCH),
             ) == null
         override fun find(msgId: Long): ProcState? = rows[msgId]
         override fun findSuccessTerminal(msgId: Long): Boolean = rows[msgId]?.state == ProcStatus.SUCCEEDED

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

@@ -33,7 +33,7 @@ class FdelAndAdftProcessorTest {
 
     private val msgId = 7L
 
-    private fun head() = ProcState(msgId, ProcStatus.PENDING)
+    private fun head() = ProcState(msgId, ProcStatus.PENDING, updatedAt = java.time.Instant.EPOCH)
 
     private fun msg() = DecodedMessage(
         meta = MetaFields("AODB", "FDEL", "", 9L, 1L),
@@ -172,6 +172,6 @@ class FdelAndAdftProcessorTest {
 
     @Suppress("unused")
     private fun unused() {
-        MsgEvent(target = "x", partitionKey = "y", payloadJson = "")
+        MsgEvent(target = "x", partitionKey = "y", payloadJson = "", createdAt = java.time.Instant.EPOCH)
     }
 }

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

@@ -22,7 +22,7 @@ class IdentityTest {
 
     @Test
     fun `identity is pipe-joined SNDR TYPE STYP SEQN`() {
-        val identity = Identity.of(msg(42), includeDayBoundary = false)
+        val identity = Identity.of(msg(42), includeDayBoundary = false, day = LocalDate.of(2026, 9, 6))
         assertEquals("AODB|FLOP|DELY|42", identity)
     }
 

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

@@ -39,7 +39,7 @@ class ScheduleProcessorTest {
 
     private val msgId = 42L
 
-    private fun head() = ProcState(msgId, ProcStatus.PENDING)
+    private fun head() = ProcState(msgId, ProcStatus.PENDING, updatedAt = java.time.Instant.EPOCH)
 
     private fun message(body: ScheduleBody) = DecodedMessage(
         meta = MetaFields("AODB", "SCHD", "DNLD", 100L, 1L),