fix(processing): stamp outbox events from the injected clock

所有生产 MsgEvent 构造点显式传 createdAt：ScheduleProcessor.snapshotEvents、FdelProcessor 删除事件、eventsFor（新增 createdAt 参数，三个调用点传注入时钟）、HistorySweepJob.tombstone（传 run 的 now）。领域默认保留给测试便利，但生产写入不再依赖系统时钟。

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

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

@@ -78,7 +78,7 @@ class HistorySweepJob(
                 // 从没收到过 FDEL、被这里直接清掉的航班，删除前补发一次通知，否则下游一直以为它还在
                 val preDelete = rechecked.filter { it.wasNeverFdel }
                 if (preDelete.isNotEmpty()) {
-                    msgEvents.insertAll(preDelete.map { tombstone(it) })
+                    msgEvents.insertAll(preDelete.map { tombstone(it, now) })
                 }
                 purged = flightState.purgeArchived(rechecked)
             }
@@ -86,11 +86,12 @@ class HistorySweepJob(
         return SweepOutcome(candidates.size, archivedFlids.size, purged, snapLogPurged)
     }
 
-    private fun tombstone(candidate: HistoryCandidate) = MsgEvent(
+    private fun tombstone(candidate: HistoryCandidate, createdAt: Instant) = MsgEvent(
         target = Targets.KAFKA_SCHD,
         partitionKey = candidate.flid,
         eventType = EventType.TOMBSTONE,
         stateVersion = candidate.stateVersion,
         payloadJson = """{"flid":"${candidate.flid}","stateVersion":${candidate.stateVersion},"deleted":true}""",
+        createdAt = createdAt,
     )
 }

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

@@ -55,7 +55,7 @@ class FlopProcessor(
         val change = MergeChange(flid = payload.flid, scalars = payload.scalars, collections = payload.collections)
         val next = FlightStateEngine.mergedState(current, change)
         flightState.persistFullState(next, msgId = head.msgId, now = clock.instant())
-        msgEvents.insertAll(eventsFor(next, mapper))
+        msgEvents.insertAll(eventsFor(next, mapper, clock.instant()))
         procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant())
         ApplyResult.Succeeded
     }
@@ -83,6 +83,7 @@ class FdelProcessor(
         if (deleted) {
             val current = flightState.loadFullSnapshot(payload.flid)
             // 只有"在用 → 删除"这一步才发删除通知，而且和状态变更写在同一个事务里
+            val createdAt = clock.instant()
             msgEvents.insertAll(
                 listOf(
                     MsgEvent(
@@ -97,6 +98,7 @@ class FdelProcessor(
                                 "deleted" to true,
                             ),
                         ),
+                        createdAt = createdAt,
                     ),
                     MsgEvent(
                         target = Targets.KAFKA_MSG,
@@ -105,6 +107,7 @@ class FdelProcessor(
                         payloadJson = mapper.writeValueAsString(
                             mapOf("flid" to payload.flid, "stateVersion" to (current?.stateVersion ?: 0L), "deleted" to true),
                         ),
+                        createdAt = createdAt,
                     ),
                 ),
             )
@@ -149,7 +152,7 @@ class AdftProcessor(
                         if (current != null) {
                             val next = FlightStateEngine.mergedState(current, setOnly(record))
                             flightState.persistFullState(next, msgId = head.msgId, now = clock.instant())
-                            msgEvents.insertAll(eventsFor(next, mapper))
+                            msgEvents.insertAll(eventsFor(next, mapper, clock.instant()))
                         }
                     } else {
                         // 并发下可能已被别的消息处理掉：不改版本、不重复发事件，但留下痕迹便于对账。
@@ -182,7 +185,7 @@ class AdftProcessor(
                     // 若按 INFRA 抛出去，会被当成暂时性故障白白重试到耗尽，并给出误导的错误类别。
                     throw ProtocolViolation("operation-day guard violated flid=${record.flid}")
                 }
-                msgEvents.insertAll(eventsFor(next, mapper))
+                msgEvents.insertAll(eventsFor(next, mapper, clock.instant()))
                 procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant())
                 ApplyResult.Succeeded
             }
@@ -211,7 +214,7 @@ class AdftProcessor(
 // =====================================================================
 
 /** 航班状态变化后要发的两类事件：KAFKA_SCHD 发完整状态，KAFKA_MSG 发一条变更通知。 */
-internal fun eventsFor(next: FlightSnapshot, mapper: ObjectMapper): List<MsgEvent> {
+internal fun eventsFor(next: FlightSnapshot, mapper: ObjectMapper, createdAt: Instant): List<MsgEvent> {
     val payload = linkedMapOf<String, Any>(
         "flid" to next.flid,
         "stateVersion" to next.stateVersion,
@@ -224,12 +227,14 @@ internal fun eventsFor(next: FlightSnapshot, mapper: ObjectMapper): List<MsgEven
             partitionKey = next.flid,
             stateVersion = next.stateVersion,
             payloadJson = mapper.writeValueAsString(payload),
+            createdAt = createdAt,
         ),
         MsgEvent(
             target = Targets.KAFKA_MSG,
             partitionKey = next.flid,
             stateVersion = next.stateVersion,
             payloadJson = mapper.writeValueAsString(mapOf("flid" to next.flid, "stateVersion" to next.stateVersion)),
+            createdAt = createdAt,
         ),
     )
 }

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

@@ -162,6 +162,7 @@ class ScheduleProcessor(
 
     /** 每次航班状态变化登记两个事件：KAFKA_SCHD 发完整状态，KAFKA_MSG 发一条变更通知。 */
     private fun snapshotEvents(next: FlightSnapshot): List<MsgEvent> {
+        val createdAt = clock.instant()
         val payload = linkedMapOf<String, Any>(
             "flid" to next.flid,
             "stateVersion" to next.stateVersion,
@@ -175,8 +176,15 @@ class ScheduleProcessor(
                 partitionKey = next.flid,
                 stateVersion = next.stateVersion,
                 payloadJson = mapper.writeValueAsString(payload),
+                createdAt = createdAt,
+            ),
+            MsgEvent(
+                target = Targets.KAFKA_MSG,
+                partitionKey = next.flid,
+                stateVersion = next.stateVersion,
+                payloadJson = notify,
+                createdAt = createdAt,
             ),
-            MsgEvent(target = Targets.KAFKA_MSG, partitionKey = next.flid, stateVersion = next.stateVersion, payloadJson = notify),
         )
     }