feat(config): 配置值非法无条件拒启，登记启动必需配置清单（ACM2-97）

- 新增 PipelineConfigCheck（@Context）：退避档位/播种取值校验不再等 autostart（CLM-7、OPS-1）
- OperationDayZoneCheck 扩展 cutoff-hour 0-23 校验
- reference.md 新增「启动必需配置与显式开关」：必需清单、缺值路径、显式开关与运行时故障的边界
- PipelinePropsBindingTest 补一致退避档位；补非法值/范围用例

docs/reference.md

@@ -78,6 +78,22 @@
 
 环境变量名见 [`.env.example`](../.env.example)；其余基础设施键见 `src/main/resources/application.yml`。
 
+### 启动必需配置与显式开关（`CLM-7`、`OPS-1`）
+
+**值非法即拒绝启动**（无条件自检，不等 `autostart`）：
+
+| 校验 | 条件 | 依据 |
+|---|---|---|
+| `msgx.operation-day.zone` | 必须是可解析的 IANA 时区 | `PARAM:msgx.operation-day.zone` |
+| `msgx.operation-day.cutoff-hour` | 取值 0–23 | `PARAM:msgx.operation-day.cutoff-hour` |
+| `kafka.producers.default.acks` / `enable-idempotence` / `max-in-flight-requests-per-connection` | 联合满足幂等生产前提（`all`+`true`+`1`；旧 Broker 降级组合见 `D2`） | `D2` |
+| `msgx.pipeline.backoff-ms` | 档位数必须等于 `max-attempts - 1` | 退避表（implementation.md「参数」） |
+| `msgx.pipeline.cutover-watermark` | 仅 `min`/`zero`/`max`/数字 | 切流播种（`G-SCAN-PREDICATE` 前） |
+
+**缺值**：带环境变量占位的必需键（`MSGX_PG_URL`、`MSGX_MAILBOX_URL`、`MSGX_KAFKA_SERVERS` 等）缺失时由 DI 解析失败拒绝启动，无需自检代码。
+
+**显式开关（非配置错误）**：`msgx.stubs`、`datasources.default.enabled`、`mailbox.shared-mysql.enabled`、`msgx.history.history-store-enabled`、`msgx.pipeline.autostart`。默认关闭是"未接入"的显式声明，不判违规；只有 `msgx.pipeline.autostart=true` 才要求依赖齐备。Kafka 运行时不可达不是配置错误，归 `/health` 与投递指标告警（`OPS-2`）。
+
 ## 指标与健康
 
 | 指标 | 含义 | 关注信号 |

src/main/kotlin/com/gzzn/omms/msgexchange/PipelineLifecycle.kt

@@ -1,6 +1,5 @@
 package com.gzzn.omms.msgexchange
 
-import com.gzzn.omms.msgexchange.config.PipelineProps
 import com.gzzn.omms.msgexchange.delivery.Dispatcher
 import com.gzzn.omms.msgexchange.ingress.InboxPoller
 import com.gzzn.omms.msgexchange.processing.Pump
@@ -25,7 +24,6 @@ class PipelineLifecycle(
     private val pump: Pump,
     private val dispatcher: Dispatcher,
     private val jobRunner: JobRunner,
-    private val props: PipelineProps,
 ) {
     private val log = org.slf4j.LoggerFactory.getLogger(PipelineLifecycle::class.java)
     private val threads = mutableListOf<Thread>()
@@ -40,9 +38,8 @@ class PipelineLifecycle(
 
     private fun startIfNeeded() {
         if (started) return
-        // 自检放在起线程之前：配置错位（例如退避表档位数与 max-attempts 不匹配、
-        // 有一段退避永远走不到）必须让启动失败，而不是静默按错误参数长期运行。
-        props.pipeline.validate()
+        // 管道参数合法性已由 PipelineConfigCheck 在启动时无条件校验（CLM-7）；
+        // 这里只负责拉起循环线程。
         started = true
         threads += spawn("msgx-inbox-poller", poller::loop)
         threads += spawn("msgx-pump", pump::loop)

src/main/kotlin/com/gzzn/omms/msgexchange/config/OperationDayProps.kt

@@ -32,9 +32,14 @@ class OperationDayProps {
         }
 }
 
-/** 启动自检：时区非法必须在启动时失败，而不是等第一条报文进来才炸；`@Context` 使其急切创建。 */
+/** 启动自检：时区或切日边界非法必须在启动时失败，而不是等第一条报文进来才炸；`@Context` 使其急切创建。 */
 @Singleton
 @Context
 class OperationDayZoneCheck(props: OperationDayProps) {
-    private val zone: ZoneId = props.zoneId()
+    init {
+        props.zoneId()
+        require(props.cutoffHour in 0..23) {
+            "msgx.operation-day.cutoff-hour must be 0-23, got ${props.cutoffHour}"
+        }
+    }
 }

src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineConfigCheck.kt

@@ -0,0 +1,17 @@
+package com.gzzn.omms.msgexchange.config
+
+import io.micronaut.context.annotation.Context
+import jakarta.inject.Singleton
+
+/**
+ * 管道参数启动自检：退避表档位数与 `max-attempts` 的匹配、切流播种取值等，
+ * 值非法时拒绝启动。无条件执行（`CLM-7`）——不再等 `msgx.pipeline.autostart=true`
+ * 才校验：配置值本身非法与依赖是否接入无关，静默放行会让错误参数在开启时才爆出来。
+ */
+@Singleton
+@Context
+class PipelineConfigCheck(props: PipelineProps) {
+    init {
+        props.pipeline.validate()
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/config/OperationDayPropsTest.kt

@@ -28,4 +28,13 @@ class OperationDayPropsTest {
 
         assertTrue(failure.message!!.contains("msgx.operation-day.zone"))
     }
+
+    @Test
+    fun `cutoff hour outside 0-23 is rejected at startup check`() {
+        val props = OperationDayProps().apply { cutoffHour = 24 }
+
+        val failure = assertThrows(IllegalArgumentException::class.java) { OperationDayZoneCheck(props) }
+
+        assertTrue(failure.message!!.contains("msgx.operation-day.cutoff-hour"))
+    }
 }

src/test/kotlin/com/gzzn/omms/msgexchange/config/PipelineConfigCheckTest.kt

@@ -0,0 +1,35 @@
+package com.gzzn.omms.msgexchange.config
+
+import org.junit.jupiter.api.Assertions.assertThrows
+import org.junit.jupiter.api.Test
+
+/**
+ * `CLM-7`/`OPS-1`：管道参数值非法必须在启动自检（无条件 `@Context`）时失败，
+ * 与 `msgx.pipeline.autostart` 是否开启无关。
+ */
+class PipelineConfigCheckTest {
+
+    @Test
+    fun `valid pipeline props pass the unconditional check`() {
+        PipelineConfigCheck(PipelineProps())
+    }
+
+    @Test
+    fun `backoff slot count mismatch rejects startup even without autostart`() {
+        val props = PipelineProps().apply {
+            pipeline.maxAttempts = 5
+            pipeline.backoffMs = listOf(1000L, 2000L)   // 期望 4 档
+        }
+
+        assertThrows(IllegalArgumentException::class.java) { PipelineConfigCheck(props) }
+    }
+
+    @Test
+    fun `illegal cutover watermark rejects startup even without autostart`() {
+        val props = PipelineProps().apply {
+            pipeline.cutoverWatermark = "yesterday"
+        }
+
+        assertThrows(IllegalArgumentException::class.java) { PipelineConfigCheck(props) }
+    }
+}

src/test/kotlin/com/gzzn/omms/msgexchange/config/PipelinePropsBindingTest.kt

@@ -24,6 +24,8 @@ class PipelinePropsBindingTest : TestPropertyProvider {
     override fun getProperties(): Map<String, String> = mapOf(
         "msgx.pipeline.poll-interval" to "PT7S",
         "msgx.pipeline.max-attempts" to "9",
+        // 档位数必须与 max-attempts-1 一致（PipelineConfigCheck 无条件校验，CLM-7）
+        "msgx.pipeline.backoff-ms" to "1000,2000,3000,4000,5000,6000,7000,8000",
         "msgx.schd.flush-limit" to "321",
         "msgx.identity.include-day-boundary" to "true",
     )
@@ -35,6 +37,7 @@ class PipelinePropsBindingTest : TestPropertyProvider {
     fun `nested msgx keys bind from test properties`() {
         assertEquals(Duration.ofSeconds(7), props.pipeline.pollInterval)
         assertEquals(9, props.pipeline.maxAttempts)
+        assertEquals((1..8).map { it * 1000L }, props.pipeline.backoffMs)
         assertEquals(321, props.schd.flushLimit)
         assertTrue(props.identity.includeDayBoundary)
     }