fix(delivery): msg 队头失败或退避只暂停同 FLID，不阻塞其他航班（ACM2-95）

- drainKafkaMsg 按分区键维护本轮暂停集：队头失败/退避未到期仅停该 FLID（D2）
- 同一 FLID 内仍严格按 EVENT_ID 保序不越队
- 补三条用例：失败仅停同 FLID、退避未到期仅停同 FLID、同 FLID 跨轮保序（TestClocks 无 sleep）

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

@@ -31,12 +31,13 @@ interface DeliveryPort {
 /**
  * 投递调度：把 outbox（待发事件表）里的事件发给下游。
  *
- * KAFKA_MSG 一条一条按登记顺序发，不插队。KAFKA_SCHD 走 flushSchd：outbox 每个 FLID 只保留
+ * KAFKA_MSG 按 `EVENT_ID` 顺序发，保序以 FLID 为单位（D2）：某航班的队头失败或退避未到期
+ * 只暂停该 FLID，其他航班照常推进。KAFKA_SCHD 走 flushSchd：outbox 每个 FLID 只保留
  * 一行（写入侧单行 upsert），发出后按读取时刻的代次做条件确认；删除通知发 value 为空的 tombstone。
  * 两个主题之间不保证先后顺序。
  *
- * 失败处理：队首的重试时间没到就不取；一批里有发送失败，整批重试次数加一并推后退避，
- * 次数用尽整批转 DEAD 当死信。见 docs/implementation.md「Kafka 与读取」。
+ * 失败处理：某 FLID 队首的重试时间没到就跳过该 FLID；发送失败重试次数加一并推后退避，
+ * 次数用尽转 DEAD 当死信。见 docs/implementation.md「Kafka 与读取」。
  */
 @Singleton
 class Dispatcher(
@@ -87,8 +88,10 @@ class Dispatcher(
     /**
      * 按 `EVENT_ID` 顺序批量投递 `KAFKA:msg`。
      *
-     * 保序规则不变：**同一目标内不越序**——队头失败或退避未到期时立即停止本轮，
-     * 不跳过它去投后面的。批量只用来省掉"每条一次 DB 往返 + 一轮一次 sleep"。
+     * 保序以 **FLID 为单位**（D2）：某航班的队头发送失败或退避未到期时，只暂停该 FLID
+     * 的后续事件（保持 `PENDING`、不越队），其他航班照常推进——队头卡住不再拖住全部航班。
+     * 同一 FLID 内部仍然严格按 `EVENT_ID` 先后发送。批量只用来省掉
+     * "每条一次 DB 往返 + 一轮一次 sleep"。
      *
      * @return 本轮成功发出的条数
      */
@@ -97,6 +100,7 @@ class Dispatcher(
         val maxRounds = props.pipeline.deliveryDrainRounds.coerceAtLeast(1)
         var sent = 0
         var round = 0
+        val paused = HashSet<String>()   // 本轮投递中被暂停的 FLID：队头失败或退避未到期
         while (round < maxRounds) {
             round++
             val batch = try {
@@ -107,9 +111,17 @@ class Dispatcher(
             }
             if (batch.isEmpty()) return sent
             for (e in batch) {
+                val flid = e.partitionKey
+                if (flid in paused) continue                              // 同 FLID 暂停中：不越队
                 val next = e.nextAttemptAt
-                if (next != null && next > scheduler.now()) return sent   // 队头退避未到期：停止推进
-                if (!deliver(Targets.KAFKA_MSG, e)) return sent            // 失败已记账：停止以保序
+                if (next != null && next > scheduler.now()) {             // 队头退避未到期：只停该 FLID
+                    paused += flid
+                    continue
+                }
+                if (!deliver(Targets.KAFKA_MSG, e)) {                     // 失败已记账：只停该 FLID
+                    paused += flid
+                    continue
+                }
                 sent++
             }
             if (batch.size < batchSize) return sent

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

@@ -214,6 +214,82 @@ class DispatcherTickTest {
         d.tick()
         assertEquals(1, port.sent.count { it.topic == "msg" && it.key == "F9" })
     }
+
+    @Test
+    fun `msg head failure pauses only the same FLID while other flights deliver in the same tick`() {
+        val repo = StubMsgEvents()
+        val port = FailingMsgKeyPort("F1")
+        val d = dispatcher(repo, port)
+        repo.insertAll(
+            listOf(
+                ev(1, Targets.KAFKA_MSG, "F1", """{"flid":"F1","v":1}"""),
+                ev(2, Targets.KAFKA_MSG, "F2", """{"flid":"F2","v":1}"""),
+                ev(3, Targets.KAFKA_MSG, "F1", """{"flid":"F1","v":2}"""),
+            ),
+        )
+
+        d.tick()
+
+        // D2：F1 队头失败只暂停 F1（含 id=3 的后续），F2 照常发出
+        assertEquals(1, port.sent.count { it.topic == "msg" && it.key == "F2" })
+        assertEquals(0, port.sent.count { it.topic == "msg" && it.key == "F1" })
+        val f1Rows = repo.rows.values.filter { it.partitionKey == "F1" }
+        assertEquals(2, f1Rows.size)
+        assertTrue(f1Rows.all { it.state == EventStatus.PENDING })
+    }
+
+    @Test
+    fun `msg head backoff not yet due pauses only the same FLID`() {
+        val repo = StubMsgEvents()
+        val port = StubDeliveryPort()
+        val d = dispatcher(repo, port)
+        repo.insertAll(
+            listOf(
+                ev(1, Targets.KAFKA_MSG, "F1", """{"flid":"F1","v":1}"""),
+                ev(2, Targets.KAFKA_MSG, "F2", """{"flid":"F2","v":1}"""),
+            ),
+        )
+        repo.scheduleRetry(1, clock.instant().plusSeconds(60), 1)   // F1 队头退避未到期
+
+        d.tick()
+
+        assertEquals(1, port.sent.count { it.topic == "msg" && it.key == "F2" })
+        val f1 = repo.rows.values.single { it.partitionKey == "F1" }
+        assertEquals(EventStatus.PENDING, f1.state)
+        assertEquals(1, f1.attempts)
+    }
+
+    @Test
+    fun `same FLID msg events keep strict order across ticks`() {
+        val repo = StubMsgEvents()
+        val port = StubDeliveryPort()
+        val d = dispatcher(repo, port)
+        repo.insertAll(
+            listOf(
+                ev(1, Targets.KAFKA_MSG, "F1", """{"v":1}"""),
+                ev(2, Targets.KAFKA_MSG, "F1", """{"v":2}"""),
+                ev(3, Targets.KAFKA_MSG, "F1", """{"v":3}"""),
+            ),
+        )
+
+        d.tick()
+
+        val f1Keys = port.sent.filter { it.topic == "msg" }.map { it.payload }
+        assertEquals(listOf("""{"v":1}""", """{"v":2}""", """{"v":3}"""), f1Keys)   // 同 FLID 不越队
+    }
+}
+
+/** KAFKA:msg 对指定 FLID 的发送永远失败，其他 FLID 照常记录，用来验证按 FLID 暂停。 */
+private class FailingMsgKeyPort(private val failingKey: String) : DeliveryPort {
+    val sent = mutableListOf<StubDeliveryPort.Sent>()
+
+    override fun sendKafka(topic: String, key: String, payloadJson: String) {
+        if (key == failingKey) throw IllegalStateException("broker-down:$key")
+        sent.add(StubDeliveryPort.Sent(topic, key, payloadJson))
+    }
+
+    override fun sendKafkaSchd(topic: String, key: String, payloadJson: String) = Unit
+    override fun sendKafkaNull(topic: String, key: String) = Unit
 }
 
 /** KAFKA_SCHD 发送永远失败的投递端口，用来验证退避重试和最终转死信的闭环。 */