diff --git a/docs/contracts/interface-contract.md b/docs/contracts/interface-contract.md
index f0203b5..1b44ddf 100644
--- a/docs/contracts/interface-contract.md
+++ b/docs/contracts/interface-contract.md
@@ -12,13 +12,13 @@
 |---|---|---|---|---|
 | `POST /cminmsgs/send` | 请求体是 XML 原文；接受 `text/xml`、`application/xml`、`text/plain`，默认 UTF-8；仅限内网，网络层限制来源。 | 报文写入 `CMINMSGS` 后返回信箱编号；写入的报文与上游投递走同一条处理路径、效果一致；该响应只证明已落信，不证明业务处理或下游投递（`US-02`）。 | 空报文、超大小上限、非法 XML 不落信并返回错误；写信失败不返回编号；XML 解析禁用外部实体和外部资源访问。 | 大小上限、请求编码与 `Content-Type` 的精确处理规则、HTTP 状态码、成功/失败响应体字段及样例（`Q15`）。 |
 | `POST /schd/sync` | 触发一次 `RQFD` 日计划请求；请求字段尚未定义，登记与落信规则见「在途与作废」。 | 响应内容尚未定义；无论表示登记还是落信，都不代表 AODB 已收到。 | 错误响应尚未定义。 | 请求字段与时间格式、成功响应表示已登记还是已落信（`Q16`）、状态码、响应字段及样例。 |
-| `GET /all/flights` | 无已定义的请求字段；从 Redis 投影读取当前全部动态航班，不含共享航班，与网页客户端同源（`US-12`）。 | 返回查询到的全部航班，不分页；JSON 由消息文档中的 XML 结构转换而来（`Q21`）。 | Redis 异常时返回错误，不能返回空列表伪装成功。 | 状态码和错误响应样例；外层包装是否沿用旧 `ResponseDto`。 |
+| `GET /all/flights` | 无已定义的请求字段；从 Redis 投影读取当前全部动态航班，不含共享航班，与网页客户端同源（`US-12`）。 | HTTP 200；响应体是裸 JSON 数组（不套旧 `ResponseDto`），元素为与 `KAFKA:schd` 同形的航班 JSON；共享航班（标量 `MAID` 非空）不出现在数组里；不分页（`Q21` 暂定）。 | HTTP 503；JSON 对象 `{"error":"FLIGHT_PROJECTION_UNAVAILABLE","reason":"<细节>"}`；Redis 或投影读失败时不得返回 200 空数组。 | 数组元素字段是否与旧 `SCHD.FLTR` 逐字一致的对拍样例。 |
 
 旧系统线索（来源：旧项目用户故事「HTTP 接口清单」「日计划请求」）：
 
 - 三个接口共用 `ResponseDto`，字段为 `is_success`、`err_code`、`err_msg`、`body`。
 - `POST /cminmsgs/send` 的成功 `body` 是 `CMINMSGS_ID`，样例为 `{is_success: true, body: <ID>}`。
-- `GET /all/flights` 的 `body` 是非共享航班的 `SCHD.FLTR` 列表。
+- 旧 `GET /all/flights` 用 `ResponseDto`，`body` 是非共享航班的 `SCHD.FLTR` 列表；新版暂定直接返回该列表对应的 JSON 数组（`Q21`）。
 - `POST /schd/sync` 的请求体是 `{startDate, endDate}`，时间格式 `yyyy-MM-dd hh:mm` 用无 AM/PM 的 12 小时制。
 
 这些字段是否沿用、时间改用何种无歧义格式，都要核对后定稿。新版日计划是 AODB 当前时刻的完整航班列表（`US-07`），出站编码已定（`C-4`），不带日期筛选。
diff --git a/docs/reference.md b/docs/reference.md
index 179b7f5..f89f27f 100644
--- a/docs/reference.md
+++ b/docs/reference.md
@@ -19,6 +19,7 @@
 | `msgx.pipeline.delivery-drain-rounds` | `10` 轮 | 连续领取多少批后，先尝试发送一批 `schd` 消息 | 暂定 |
 | `msgx.schd.flush-period` | `3s` | 每隔多久尝试发送一批 `schd` 消息 | 沿用旧系统 |
 | `msgx.schd.flush-limit` | `500` 条 | 一批 `schd` 消息最多包含多少个航班 | 暂定 |
+| `msgx.schd.snapshot-batch` | `200` 个航班 | 日计划一批最多写多少个航班，每批一个事务；缺席清扫按同一上限分批（`US-07` AC4） | 暂定 |
 | `msgx.pipeline.event-retention` | `7d` | 发送成功的消息在数据库保留多久，从 `SENT_AT` 起算 | 暂定 |
 | `msgx.pipeline.terminal-retention` | `90d` | 处理记录（PROC_STATE 终态行）保留多久，从 `UPDATED_AT` 起算；到期且已回填才删除（`US-11`） | 暂定 |
 
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt
index 4183f57..556829e 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/config/PipelineProps.kt
@@ -107,6 +107,12 @@ class PipelineProps {
     class Schd {
         var flushPeriod: Duration = Duration.ofSeconds(3)       // KEEP 现役节律
         var flushLimit: Int = 500
+
+        /**
+         * 一包日计划分多少个航班一批写库，每批一个事务（`US-07` AC4）。
+         * 缺席清扫（`INV-7`）按同一上限分批。整份写完才刷投影、才记终态（`INV-9`）。
+         */
+        var snapshotBatch: Int = 200
     }
 
     @ConfigurationProperties("identity")
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/SnapshotLog.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/SnapshotLog.kt
index bc5a65f..bd2226a 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/SnapshotLog.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/SnapshotLog.kt
@@ -17,9 +17,10 @@ enum class SnapshotResult { COMMITTED, REPLAY_SKIPPED, ROLLED_BACK }
  * EMPTY——空快照，合法但不是常态；
  * RECS_DROP——报头声明的条数和实际收到的对不上；
  * DAY_MISMATCH——有记录算不出运营日，整包拒收；
- * SCHD_REVIVE_CONFLICT——日计划打到了已删除的航班上，航班保持删除不恢复。
+ * SCHD_REVIVE_CONFLICT——日计划打到了已删除的航班上，航班保持删除不恢复；
+ * SCHD_ABSENT_DELETED——覆盖范围内有航班在快照里缺席，已按缺席标删（`INV-7`）。
  */
-enum class SnapshotFlag { EMPTY, RECS_DROP, DAY_MISMATCH, SCHD_REVIVE_CONFLICT }
+enum class SnapshotFlag { EMPTY, RECS_DROP, DAY_MISMATCH, SCHD_REVIVE_CONFLICT, SCHD_ABSENT_DELETED }
 
 data class SnapshotLogEntry(
     val msgId: Long,
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngine.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngine.kt
index 12a7976..4ad1a30 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngine.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngine.kt
@@ -8,8 +8,8 @@ import java.time.LocalDate
  * 航班状态引擎：把报文内容和库里已有的状态合成出新状态，全部在内存里算，不碰数据库和
  * Kafka，方便单独测。三个入口：
  *  - [validateMessage]：整包校验日计划，检查声明条数、航班标识、运营日能不能算出来；
- *  - [snapshotState]：把日计划（DNLD/RESP）合并进当前态——报文里出现的字段覆盖本地值，
- *    没出现的保留，标量给空串表示清空；已删除的航班保持删除，日计划救不回来；
+ *  - [snapshotState]：把日计划（DNLD/RESP）写成新的当前态——报文带的字段写进去，
+ *    没带的字段和集合清掉；已删除的航班保持删除，日计划救不回来；
  *  - [mergedState]：把 FLOP/ADFT 的增量合并进当前态，只改报文明确表达的字段和集合。
  *
  * 合并规则见 docs/implementation.md「SCHD 日计划」「动态运行事件」。
@@ -90,17 +90,18 @@ object FlightStateEngine {
     }
 
     /**
-     * 把一条日计划记录合并进当前态：报文里出现的标量覆盖本地值（给空串表示显式清空），
-     * 没出现的保留；集合一旦出现就整体替换，没出现就保留。
-     *
-     * 新航班（current == null）没有本地值可保留，所以集合按全部键输出空集，保证落库时
-     * 每张明细表都有确定的行集。
+     * 把一条日计划记录写成新的当前态：**以 AODB 下发的这份快照为准**——报文带的字段写进去，
+     * 没带的字段和集合一律清掉（`C-6`、`US-07` AC3）。日计划是覆盖范围内的完整列表，
+     * 不是增量，所以没有"没出现就保留"这回事（那是 FLOP/ADFT 的语义，见 [mergedState]）。
      *
      * keepDeleted = true 时即使收到日计划也保持 DELETED：日计划不能把删掉的航班救回来，
      * 唯一的恢复入口是 ADFT。调用方负责记一条 SCHD_REVIVE_CONFLICT 告警。
+     *
+     * [current] 只用来接上状态与版本，所以取主行就够：没带的字段既然要清掉，
+     * 读回全部明细也没有用处（一包日计划几千条，逐条读明细是白搭的开销）。
      */
     fun snapshotState(
-        current: FlightSnapshot?,
+        current: FlightMainRow?,
         record: ScheduleRecord,
         operationDay: LocalDate,
         keepDeleted: Boolean,
@@ -110,19 +111,10 @@ object FlightStateEngine {
             keepDeleted -> FlightState.DELETED
             else -> current.state
         }
-        val scalars = buildMap {
-            current?.scalars?.let(::putAll)   // 报文没带的标量保持原值
-            putAll(record.scalars)            // 出现 = Set；空串 = 显式清空
-        }
-        val collections = buildMap {
-            if (current == null) {
-                COLLECTION_KEYS.forEach { key -> put(key, emptyList()) }   // 新航班全键输出
-            } else {
-                putAll(current.collections)   // 报文没带的集合保持原值
-            }
-            record.collections.forEach { (key, items) ->
-                if (key in COLLECTION_KEYS) put(key, normalizeCollection(key, items))
-            }
+        val scalars = record.scalars
+        // 全键输出：没带的集合落成空集，明细表因此按"先删后插"清空，不留上一份快照的行
+        val collections = COLLECTION_KEYS.associateWith { key ->
+            record.collections[key]?.let { normalizeCollection(key, it) } ?: emptyList()
         }
         return FlightSnapshot(
             flid = record.flid,
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt
index 62246d4..399c094 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/Repositories.kt
@@ -249,6 +249,12 @@ interface FlightStateRepository {
 
     fun findMainRows(flids: Collection<String>): Map<String, FlightMainRow>
 
+    /**
+     * 查这些运营日下仍在用的航班主行，供日计划做缺席清扫：快照覆盖范围内、报文里没出现的
+     * 航班要标删（`INV-7`）。运营日为空的航班（尚未被日计划收录）不在任何覆盖范围内，不返回。
+     */
+    fun findActiveByOperationDays(operationDays: Collection<LocalDate>): List<FlightMainRow>
+
     /** 读一条航班的完整当前态（主行加全部明细）。要读得一致，得由调用方放在事务里读。 */
     fun loadFullSnapshot(flid: String): FlightSnapshot?
 
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt
index aac979f..eb5b017 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt
@@ -632,6 +632,17 @@ class JdbcFlightStateRepository(
         return out
     }
 
+    override fun findActiveByOperationDays(operationDays: Collection<LocalDate>): List<FlightMainRow> {
+        if (operationDays.isEmpty()) return emptyList()
+        val days = operationDays.distinct()
+        val placeholders = days.joinToString(",") { "?" }
+        return ds.query(
+            SELECT_MAIN + " WHERE state = 'ACTIVE' AND operation_day IN ($placeholders) ORDER BY flid",
+            { ps -> days.forEachIndexed { i, day -> ps.setDate(i + 1, java.sql.Date.valueOf(day)) } },
+            ::mapMainRow,
+        )
+    }
+
     override fun loadFullSnapshot(flid: String): FlightSnapshot? {
         val main = findMainRow(flid) ?: return null
         val scalars = ds.queryOne(
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/projection/FlightProjectionAdapters.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/projection/FlightProjectionAdapters.kt
index c71c5a9..dc82a40 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/projection/FlightProjectionAdapters.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/projection/FlightProjectionAdapters.kt
@@ -28,6 +28,10 @@ class RedisFlightProjectionPort(
         )
     }
 
+    override fun readAll(): List<String> = throw UnsupportedOperationException(
+        "redis client not wired yet (G-REDIS-PROJECTION); key=${props.flightKey}",
+    )
+
     override fun ping(): Boolean = false
 }
 
@@ -46,4 +50,9 @@ class NoopFlightProjectionPort : FlightProjectionPort {
     override fun write(writes: List<FlightProjectionWrite>) {
         log.debug("redis projection disabled, skipping {} write(s) [G-REDIS-PROJECTION]", writes.size)
     }
+
+    /** 没有投影可读时报错而不是回空列表：空列表会被查询方当成"现在没有航班"（`INV-11`）。 */
+    override fun readAll(): List<String> = throw UnsupportedOperationException(
+        "redis projection disabled (msgx.redis.enabled=false)",
+    )
 }
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt
index 05f3508..8066d2d 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubAdapters.kt
@@ -39,8 +39,8 @@ class StubDeliveryPort : DeliveryPort {
  * 航班查询投影的内存假实现，只有配置 msgx.stubs=true 时才装配。
  *
  * [snapshots] 是投影的当前内容（FLID → 整态 JSON），删除会把条目移走，行为对齐 Redis 哈希。
- * [failWith] 用来做故障注入：置上以后每次写都抛这个异常，用于验证"投影写不成功就不算处理完成"
- * （`INV-10`）；置回 null 即恢复。
+ * [failWith] 用来做故障注入：置上以后每次读写都抛这个异常，用于验证"投影写不成功就不算处理完成"
+ * （`INV-10`）与"读不到就报错、不回空列表"（`INV-11`）；置回 null 即恢复。
  */
 @Requires(property = "msgx.stubs", value = "true")
 @Singleton
@@ -55,6 +55,11 @@ class StubFlightProjectionPort : FlightProjectionPort {
         snapshots.clear(); writes.clear(); failWith = null
     }
 
+    override fun readAll(): List<String> {
+        failWith?.let { throw it }
+        return snapshots.values.toList()
+    }
+
     override fun write(writes: List<FlightProjectionWrite>) {
         failWith?.let { throw it }
         this.writes += writes
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt
index 0fc1795..a9b7dfd 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt
@@ -365,6 +365,14 @@ class StubFlightState : FlightStateRepository {
     override fun findMainRows(flids: Collection<String>): Map<String, FlightMainRow> =
         flids.mapNotNull { flid -> mains[flid]?.let { flid to it } }.toMap()
 
+    override fun findActiveByOperationDays(operationDays: Collection<LocalDate>): List<FlightMainRow> {
+        if (operationDays.isEmpty()) return emptyList()
+        val days = operationDays.toSet()
+        return mains.values
+            .filter { it.state == FlightState.ACTIVE && it.operationDay in days }
+            .sortedBy { it.flid }
+    }
+
     override fun loadFullSnapshot(flid: String): FlightSnapshot? = snapshots[flid]
 
     /** 运营日不可变：库里已有运营日且与新值不同时，返回 DAY_GUARD_VIOLATION。 */
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/FlightQueryController.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/FlightQueryController.kt
new file mode 100644
index 0000000..1183736
--- /dev/null
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/FlightQueryController.kt
@@ -0,0 +1,53 @@
+package com.gzzn.omms.msgexchange.ingress
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.gzzn.omms.msgexchange.processing.FlightProjectionPort
+import io.micronaut.http.HttpResponse
+import io.micronaut.http.HttpStatus
+import io.micronaut.http.MediaType
+import io.micronaut.http.annotation.Controller
+import io.micronaut.http.annotation.Get
+import io.micronaut.http.annotation.Produces
+
+/**
+ * `GET /all/flights`：从 Redis 查询投影读当前全部动态航班，不含共享航班（`US-12`、`C-11`）。
+ *
+ * 数据只从投影来，不回落 PG：查询与网页客户端必须读同一份，不然两边会看到不同的航班
+ * （`INV-11`）。投影读不出来就返回错误——空列表会被前端当成"现在没有航班"（`US-12` AC2）。
+ *
+ * 响应口径暂定（`Q21`）：成功是裸 JSON 数组、不套旧系统的 `ResponseDto`；失败是 503 加一个
+ * 错误对象。投影载荷的形状与 `KAFKA:schd` 同一份，尚未按旧系统 `SCHD.FLTR` 逐字对拍。
+ */
+@Controller("/all/flights")
+class FlightQueryController(
+    private val projection: FlightProjectionPort,
+    private val mapper: ObjectMapper,
+) {
+    private val log = org.slf4j.LoggerFactory.getLogger(FlightQueryController::class.java)
+
+    @Get
+    @Produces(MediaType.APPLICATION_JSON)
+    fun all(): HttpResponse<String> =
+        try {
+            HttpResponse.ok(mapper.writeValueAsString(nonSharedFlights()))
+        } catch (e: Exception) {
+            // 读投影失败（Redis 不可用、未接通、载荷读不动）一律报错：这里没有可信的兜底数据源
+            val reason = e.message ?: e.javaClass.simpleName
+            log.error("GET /all/flights unavailable reason={}", reason, e)
+            HttpResponse.status<String>(HttpStatus.SERVICE_UNAVAILABLE)
+                .body(mapper.writeValueAsString(mapOf("error" to PROJECTION_UNAVAILABLE, "reason" to reason)))
+        }
+
+    /**
+     * 共享航班由 `MAID`（主航班 FLID）非空标识，随主航班下发，不单独出现在列表里
+     * （`US-06` AC2 的主/共享关系，见 docs/implementation.md「航班域」）。
+     */
+    private fun nonSharedFlights(): List<com.fasterxml.jackson.databind.JsonNode> =
+        projection.readAll()
+            .map { mapper.readTree(it) }
+            .filter { it.path("scalars").path("MAID").asText("").isBlank() }
+
+    private companion object {
+        private const val PROJECTION_UNAVAILABLE = "FLIGHT_PROJECTION_UNAVAILABLE"
+    }
+}
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxController.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxController.kt
index 4120e0e..8fc7588 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxController.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/ingress/InboxController.kt
@@ -22,5 +22,5 @@ class InboxController(private val inbox: InboxService) {
         return HttpResponse.ok(receipt.msgId.toString())   // TODO: 先返回消息 ID 文本，等跟现役响应体逐字对拍过再定稿
     }
 
-    // TODO(阶段2): 按契约清单补齐 /schd/sync、/all/flights、/kafka/topics/{name}/msgs 这几个接口。
+    // TODO(阶段2): 按契约清单补齐 /schd/sync、/kafka/topics/{name}/msgs 这两个接口。
 }
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt
index a96931c..aa1476e 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/DynamicProcessors.kt
@@ -75,21 +75,7 @@ class FdelProcessor(
         // 重处理开始时已把上一轮没放行的事件清掉，这里不补登就会把这条通知丢了。
         if (deleted || (main.state == FlightState.DELETED && main.lastMsgId == head.msgId)) {
             msgEvents.insertAll(
-                listOf(
-                    // C-9：删除通知只走 KAFKA:msg，schd 不再发 tombstone；
-                    // value 形态（"deleted":true 的 JSON）沿用现状，待 Q5 定稿
-                    MsgEvent(
-                        msgId = head.msgId,
-                        target = Targets.KAFKA_MSG,
-                        partitionKey = payload.flid,
-                        stateVersion = main.stateVersion,
-                        payloadJson = mapper.writeValueAsString(
-                            mapOf("flid" to payload.flid, "stateVersion" to main.stateVersion, "deleted" to true),
-                        ),
-                        state = EventStatus.HELD,
-                        createdAt = clock.instant(),
-                    ),
-                ),
+                listOf(deleteEvent(payload.flid, main.stateVersion, head.msgId, mapper, clock.instant())),
             )
         }
         // INV-8：只要航班是删除态就得从投影里抹掉。重复报文也重删一次——代价是一条幂等命令，
@@ -202,6 +188,28 @@ internal fun flightPayload(next: FlightSnapshot): Map<String, Any> = linkedMapOf
     "collections" to next.collections,
 )
 
+/**
+ * 删除通知：只走 `KAFKA:msg`，`schd` 不发 tombstone（`C-9`）。
+ * FDEL 与日计划覆盖范围内的缺席删除（`INV-7`）共用这一份形状；value 形态待 `Q5` 定稿。
+ */
+internal fun deleteEvent(
+    flid: String,
+    stateVersion: Long,
+    msgId: Long,
+    mapper: ObjectMapper,
+    createdAt: Instant,
+): MsgEvent = MsgEvent(
+    msgId = msgId,
+    target = Targets.KAFKA_MSG,
+    partitionKey = flid,
+    stateVersion = stateVersion,
+    payloadJson = mapper.writeValueAsString(
+        mapOf("flid" to flid, "stateVersion" to stateVersion, "deleted" to true),
+    ),
+    state = EventStatus.HELD,
+    createdAt = createdAt,
+)
+
 /**
  * 航班状态变化后要发的两类事件：KAFKA_SCHD 发完整状态，KAFKA_MSG 发一条变更通知。
  *
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommit.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommit.kt
index da82b07..b3bdafb 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommit.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommit.kt
@@ -22,6 +22,8 @@ data class DomainOutcome<T>(val value: T, val projections: List<FlightProjection
  * 2. 写 Redis 投影：失败就抛出去，这条消息保持未完成，下轮重处理；
  * 3. 终态事务：放行待发事件，写 `SUCCEEDED` 与回填意图。
  *
+ * 日计划要分批写库，第一步因此可以拆成多个事务（[commitBatched]），后两步不变。
+ *
  * 三步整段持 `PIPELINE_LOCK`：行锁提交即释放，盖不住中间的投影写，所以由
  * [PipelineLockRepository.holdAcrossTransactions] 在一条连接上从头持到尾。
  *
@@ -37,21 +39,53 @@ class FlightCommit(
     private val projection: FlightProjectionPort,
     private val clock: Clock,
 ) {
-    fun <T> commit(head: ProcState, domain: () -> DomainOutcome<T>): T = lock.holdAcrossTransactions {
-        val outcome = txManager.inTransaction {
-            lock.lock()
-            msgEvents.discardHeld(head.msgId)   // 清掉上一轮没放行的事件，再登记本轮的
-            domain()
-        }
+    fun <T> commit(head: ProcState, domain: () -> DomainOutcome<T>): T =
+        commitBatched(head) { scope -> scope.batch(domain) }
 
-        writeProjection(head, outcome.projections)
+    /**
+     * 分批版三步提交（`INV-9`、`US-07` AC4）：领域按 [BatchScope.batch] 分成多个事务，
+     * 整份写完之后才刷投影、才写终态。
+     *
+     * 投影写攒到最后一次发出，所以"PG 整份写完且投影按快照刷完"才是处理完成的条件。
+     * 中途失败时已提交的批次不回滚（`US-03` AC3），这条消息保持未完成、下轮整包重来：
+     * 重来时第一批会先清掉上一轮没放行的 `HELD` 事件，不会重复投递。
+     *
+     * 块内也可以直读（读走同一条被钉住的连接），但写一律放进 [BatchScope.batch]。
+     */
+    fun <T> commitBatched(head: ProcState, work: (BatchScope) -> T): T = lock.holdAcrossTransactions {
+        val scope = Batches(head)
+        val value = work(scope)
+
+        writeProjection(head, scope.projections)
 
         txManager.inTransaction {
             lock.lock()
             msgEvents.releaseHeld(head.msgId)
             procState.markTerminal(head.msgId, ProcStatus.SUCCEEDED, now = clock.instant())
         }
-        outcome.value
+        value
+    }
+
+    /** 分批提交的入口：每调用一次 [batch] 就是一个领域事务。 */
+    interface BatchScope {
+        fun <T> batch(domain: () -> DomainOutcome<T>): T
+    }
+
+    private inner class Batches(private val head: ProcState) : BatchScope {
+        val projections = mutableListOf<FlightProjectionWrite>()
+        private var firstBatch = true
+
+        override fun <T> batch(domain: () -> DomainOutcome<T>): T = txManager.inTransaction {
+            lock.lock()
+            // 清残只在第一批做：上一轮没放行的事件要在本轮登记之前清掉，之后的批次不能再清
+            if (firstBatch) {
+                msgEvents.discardHeld(head.msgId)
+                firstBatch = false
+            }
+            val outcome = domain()
+            projections += outcome.projections
+            outcome.value
+        }
     }
 
     private fun writeProjection(head: ProcState, writes: List<FlightProjectionWrite>) {
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightProjection.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightProjection.kt
index f079653..faabe99 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightProjection.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/FlightProjection.kt
@@ -30,6 +30,14 @@ interface FlightProjectionPort {
     /** 一批一起写：同一条消息产生的写要么整批成功，要么抛异常整批重来。 */
     fun write(writes: List<FlightProjectionWrite>)
 
+    /**
+     * 读出投影里当前全部航班的整态 JSON，顺序不承诺。
+     *
+     * 读不到时**必须抛异常**：空列表是"投影里一个航班都没有"的事实，不能用来顶替"读不到"
+     * （`INV-11`、`US-12` AC2）。查询与网页客户端读的是同一份（`C-11`）。
+     */
+    fun readAll(): List<String>
+
     /** 给健康检查用的连通性探测。 */
     fun ping(): Boolean = true
 }
diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt
index 90530d5..f016bbd 100644
--- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt
+++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessor.kt
@@ -3,6 +3,7 @@ package com.gzzn.omms.msgexchange.processing
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.gzzn.omms.msgexchange.codec.ScheduleBody
 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.MsgEvent
 import com.gzzn.omms.msgexchange.domain.ProcState
@@ -44,11 +45,12 @@ class ProtocolViolation(message: String) : RuntimeException(message)
 /**
  * 处理 SCHD 日计划报文（DNLD 和 RESP 走同一条路）：把报文里的航班记录合并进航班当前态。
  *
- * 顺序是：整包校验 → 加锁 → 核对每条航班的运营日 → 逐条合并写入 → 登记待发事件 →
- * 刷投影 → 写终态和回填待办。领域这一段在同一个事务里，任何一步失败整包回滚，
- * 不会留下写了一半的数据；投影与终态按三步提交各走一步（[FlightCommit]）。
+ * 顺序是：整包校验 → 加锁 → 核对每条航班的运营日 → 分批合并写入并登记待发事件 →
+ * 清扫覆盖范围内缺席的航班 → 刷投影 → 写终态和回填待办。
+ * 校验或运营日核对不过就整包拒绝、一条都不落（`INV-4`）；写入阶段按批分事务，
+ * 整份写完且投影刷完才算处理完成，中途失败下轮整包重来（`INV-9`）。
  *
- * 报文里没提到的航班不会被删除——日计划只负责写它带来的那部分。
+ * 日计划是覆盖范围内的完整列表：范围内缺席的航班要标删（`INV-7`），范围外的不受影响。
  */
 @Singleton
 class ScheduleProcessor(
@@ -58,6 +60,7 @@ class ScheduleProcessor(
     private val msgEvents: MsgEventRepository,
     private val snapshotLog: SnapshotLogRepository,
     operationDayProps: OperationDayProps,
+    private val props: PipelineProps,
     private val mapper: ObjectMapper,
     private val clock: Clock,
 ) {
@@ -97,11 +100,13 @@ class ScheduleProcessor(
         }
 
         val flags = linkedSetOf<SnapshotFlag>()
+        val batchSize = props.schd.snapshotBatch
         return try {
-            // 领域这一段在同一个事务里：加锁 → 校验运营日 → 逐条合并写入 → 登记事件；
-            // 投影与终态由 FlightCommit 接着走后两步
-            val upserted = commit.commit(head) {
-                // 一次批量查出这些航班现有的运营日，逐个比对（避免逐条查询）
+            // 分批写：每批一个事务（`US-07` AC4），整份写完才刷投影、才记终态（`INV-9`）。
+            // 覆盖范围内缺席的航班在最后清扫（`INV-7`）。
+            val upserted = commit.commitBatched(head) { batches ->
+                // 一次批量查出这些航班现有的运营日，逐个比对（避免逐条查询）。
+                // 这一比对必须在任何一批写入之前跑完：运营日冲突要整包拒绝、一条都不落（`INV-4`）。
                 val mains = flightState.findMainRows(ok.perRecordDay.keys)
                 ok.perRecordDay.forEach { (flid, day) ->
                     val existing = mains[flid] ?: return@forEach
@@ -112,33 +117,47 @@ class ScheduleProcessor(
                     }
                 }
 
-                var written = 0
-                val events = mutableListOf<MsgEvent>()
-                val projections = mutableListOf<FlightProjectionWrite>()
                 // 一次建索引，避免在航班级循环里反复线性扫描（大日计划下是 O(N²)）。
                 val recordsByFlid = body.records.associateBy { it.flid }
-                ok.perRecordDay.forEach { (flid, day) ->
-                    val record = recordsByFlid.getValue(flid)
-                    val existingMain = mains[flid]
-                    val keepDeleted = existingMain?.state == FlightState.DELETED
-                    if (keepDeleted) flags.add(SnapshotFlag.SCHD_REVIVE_CONFLICT)   // 日计划不会让已删除的航班复活
-                    val current = if (existingMain != null) flightState.loadFullSnapshot(flid) else null
-                    val next = FlightStateEngine.snapshotState(
-                        current = current,
-                        record = record,
-                        operationDay = day,
-                        keepDeleted = keepDeleted,
-                    )
-                    when (flightState.persistFullState(next, msgId = head.msgId, now = clock.instant())) {
-                        PersistOutcome.DAY_GUARD_VIOLATION ->
-                            throw ProtocolViolation("operation-day guard violated flid=$flid")
-                        else -> written++
+                var written = 0
+                ok.perRecordDay.entries.chunked(batchSize).forEach { chunk ->
+                    written += batches.batch {
+                        val events = mutableListOf<MsgEvent>()
+                        val projections = mutableListOf<FlightProjectionWrite>()
+                        var batchWritten = 0
+                        chunk.forEach { (flid, day) ->
+                            val record = recordsByFlid.getValue(flid)
+                            val existingMain = mains[flid]
+                            val keepDeleted = existingMain?.state == FlightState.DELETED
+                            if (keepDeleted) flags.add(SnapshotFlag.SCHD_REVIVE_CONFLICT)   // 日计划不会让已删除的航班复活
+                            val next = FlightStateEngine.snapshotState(
+                                current = existingMain,
+                                record = record,
+                                operationDay = day,
+                                keepDeleted = keepDeleted,
+                            )
+                            when (flightState.persistFullState(next, msgId = head.msgId, now = clock.instant())) {
+                                PersistOutcome.DAY_GUARD_VIOLATION ->
+                                    throw ProtocolViolation("operation-day guard violated flid=$flid")
+                                else -> batchWritten++
+                            }
+                            events += eventsFor(next, head.msgId, mapper, clock.instant())
+                            projections += projectionOf(next, mapper)
+                        }
+                        if (events.isNotEmpty()) msgEvents.insertAll(events)
+                        DomainOutcome(batchWritten, projections)
                     }
-                    events += eventsFor(next, head.msgId, mapper, clock.instant())
-                    projections += projectionOf(next, mapper)
                 }
-                if (events.isNotEmpty()) msgEvents.insertAll(events)
-                DomainOutcome(written, projections)
+
+                sweepAbsent(
+                    head = head,
+                    batches = batches,
+                    coverage = ok.perRecordDay.values.toSet(),
+                    present = ok.perRecordDay.keys,
+                    batchSize = batchSize,
+                    flags = flags,
+                )
+                written
             }
             logSnapshot(head, body, SnapshotResult.COMMITTED, upserted, flags, started)
             ApplyResult.Succeeded
@@ -148,6 +167,41 @@ class ScheduleProcessor(
         }
     }
 
+    /**
+     * 覆盖范围内缺席的航班：标删、登记删除通知、从投影里删掉（`INV-7`、`US-07` AC2/AC5）。
+     *
+     * [coverage] 是这份报文覆盖的运营日，取自报文自身——每条记录的 `SODT` 推出的运营日
+     * （整包校验已保证每条都算得出）。覆盖范围外的航班一条都不碰：前一日延误的航班不在
+     * 今天的日计划窗口里，不能因为缺席就被判为已删除。覆盖窗口的边界口径见 `Q19`。
+     */
+    private fun sweepAbsent(
+        head: ProcState,
+        batches: FlightCommit.BatchScope,
+        coverage: Set<LocalDate>,
+        present: Set<String>,
+        batchSize: Int,
+        flags: MutableSet<SnapshotFlag>,
+    ) {
+        val absent = flightState.findActiveByOperationDays(coverage).filter { it.flid !in present }
+        if (absent.isEmpty()) return
+        flags.add(SnapshotFlag.SCHD_ABSENT_DELETED)
+        absent.chunked(batchSize).forEach { chunk ->
+            batches.batch {
+                val events = mutableListOf<MsgEvent>()
+                val projections = mutableListOf<FlightProjectionWrite>()
+                chunk.forEach { main ->
+                    if (!flightState.markDeleted(main.flid, msgId = head.msgId, now = clock.instant())) return@forEach
+                    // markDeleted 把版本加一；候选是本次读到的 ACTIVE 行，整段持锁期间没有别的写者
+                    val deletedVersion = main.stateVersion + 1
+                    events += deleteEvent(main.flid, deletedVersion, head.msgId, mapper, clock.instant())
+                    projections += FlightProjectionWrite.Delete(main.flid, deletedVersion)
+                }
+                if (events.isNotEmpty()) msgEvents.insertAll(events)
+                DomainOutcome(Unit, projections)
+            }
+        }
+    }
+
     /**
      * 写一条处理留痕（仅供排查和统计，不参与业务判断）。放在事务外做，
      * 写失败也只记日志，不会连累处理结果。scope 是这份报文覆盖的运营日范围。
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index 7b5fdb0..42c034d 100644
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -25,6 +25,7 @@ msgx:
   schd:
     flush-period: 3s                # KEEP 现役推送节律
     flush-limit: 500                # 批上限，防积压尖峰
+    snapshot-batch: 200             # 日计划每批写多少个航班，一批一个事务（US-07 AC4）
   operation-day:                   # SODT + 机场时区 + 切日边界（默认占位 0 点，待业务确认）
     zone: Asia/Shanghai
     cutoff-hour: 0
diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt
index 5c14edc..8398316 100644
--- a/src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt
+++ b/src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt
@@ -11,7 +11,7 @@ import java.time.ZoneId
 
 /**
  * 守着航班状态合并的几条规矩：
- * - 日计划合并：报文里出现的字段覆盖本地值、没出现的保留，标量给空串等于显式清空；
+ * - 日计划：报文带的字段写进去，没带的字段和集合清掉（`C-6`、`US-07` AC3）；
  * - 日计划不能把已删除的航班恢复成在用；
  * - FLOP/ADFT 只改报文表达的字段和集合；
  * - 整包校验（条数、FLID 格式、运营日能不能算出来）有一条不过就整包拒收。
@@ -21,37 +21,31 @@ class FlightStateEngineTest {
     private val opDay = OperationDayCalculator(ZoneId.of("Asia/Shanghai"), 0)
     private val day = LocalDate.of(2026, 12, 15)
 
+    /** 日计划只接状态与版本，所以库里已有的航班用主行表示。 */
+    private fun mainRow(version: Long, state: FlightState = FlightState.ACTIVE) =
+        FlightMainRow("121", day, state, version, lastMsgId = null, updatedAt = java.time.Instant.EPOCH)
+
     @Test
-    fun `day plan merges scalars overwriting overlaps retaining absent and clearing on explicit empty`() {
-        val current = FlightSnapshot(
-            "121", day, FlightState.ACTIVE, 5,
-            scalars = mapOf("FLNO" to "CA001", "REMC" to "old-note"),
-            collections = mapOf("GTDT" to listOf(mapOf("GATE" to "G1"))),
-        )
+    fun `day plan clears fields the snapshot does not carry`() {
         val next = FlightStateEngine.snapshotState(
-            current,
+            mainRow(5),
             ScheduleRecord("121", scalars = mapOf("FLNO" to "CA002", "CNCL" to "")),
             operationDay = day,
             keepDeleted = false,
         )
 
-        assertEquals("CA002", next.scalars["FLNO"])              // 报文里出现的字段覆盖本地值
-        assertEquals("old-note", next.scalars["REMC"])           // 报文没提的标量保留原值
-        assertEquals("", next.scalars["CNCL"])                   // 空串表示显式清空，落库时置成 NULL
-        assertEquals(listOf(mapOf("GATE" to "G1")), next.collections["GTDT"])  // 报文没提的集合保留原值
-        assertEquals(6, next.stateVersion)                       // 每成功合并一次，版本号加一
+        assertEquals("CA002", next.scalars["FLNO"])              // 报文带的字段写进去
+        assertFalse(next.scalars.containsKey("REMC"))            // 没带的标量清掉（AODB 已删掉该值）
+        assertEquals("", next.scalars["CNCL"])                   // 空串同样是清空，落库时置成 NULL
+        assertEquals(emptyList<Map<String, String>>(), next.collections["GTDT"])  // 没带的集合清空
+        assertEquals(6, next.stateVersion)                       // 每成功写入一次，版本号加一
         assertEquals(FlightState.ACTIVE, next.state)
     }
 
     @Test
     fun `day plan replaces collections that are present in the record`() {
-        val current = FlightSnapshot(
-            "121", day, FlightState.ACTIVE, 3,
-            scalars = mapOf("FLNO" to "CA001"),
-            collections = mapOf("GTDT" to listOf(mapOf("GATE" to "G1"))),
-        )
         val next = FlightStateEngine.snapshotState(
-            current,
+            mainRow(3),
             ScheduleRecord(
                 "121",
                 scalars = mapOf("FLNO" to "CA002"),
@@ -85,9 +79,8 @@ class FlightStateEngineTest {
 
     @Test
     fun `day plan keeps DELETED state for revive conflict handling`() {
-        val current = FlightSnapshot("121", day, FlightState.DELETED, 3, mapOf(), emptyMap())
         val next = FlightStateEngine.snapshotState(
-            current, ScheduleRecord("121", mapOf("FLNO" to "CA001")), day, keepDeleted = true,
+            mainRow(3, FlightState.DELETED), ScheduleRecord("121", mapOf("FLNO" to "CA001")), day, keepDeleted = true,
         )
         assertEquals(FlightState.DELETED, next.state)   // 日计划不能把已删除的航班恢复成在用
         assertEquals(4, next.stateVersion)
@@ -113,13 +106,8 @@ class FlightStateEngineTest {
 
     @Test
     fun `unpersisted collections are carried but never merged until their detail tables exist`() {
-        val current = FlightSnapshot(
-            "121", day, FlightState.ACTIVE, 1,
-            scalars = mapOf("FLNO" to "CA001"),
-            collections = mapOf("GTDT" to listOf(mapOf("GATE" to "G1"))),
-        )
         val next = FlightStateEngine.snapshotState(
-            current,
+            mainRow(1),
             ScheduleRecord(
                 "121",
                 scalars = mapOf("FLNO" to "CA002"),
@@ -129,10 +117,9 @@ class FlightStateEngineTest {
             keepDeleted = false,
         )
 
-        // [G-SRVT-VIPF]：wire/domain 保留这两个集合，但合并层既不落库，也不因"出现"清空任何明细
+        // [G-SRVT-VIPF]：wire/domain 保留这两个集合，但合并层不落库、也不进快照
         assertFalse(next.collections.containsKey("SRVT"))
         assertFalse(next.collections.containsKey("VIPF"))
-        assertEquals(listOf(mapOf("GATE" to "G1")), next.collections["GTDT"])
     }
 
     @Test
diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/FlightQueryControllerTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/FlightQueryControllerTest.kt
new file mode 100644
index 0000000..5a137c5
--- /dev/null
+++ b/src/test/kotlin/com/gzzn/omms/msgexchange/ingress/FlightQueryControllerTest.kt
@@ -0,0 +1,47 @@
+package com.gzzn.omms.msgexchange.ingress
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.gzzn.omms.msgexchange.infra.stub.StubFlightProjectionPort
+import io.micronaut.http.HttpStatus
+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
+
+/** `GET /all/flights`：投影读、过滤共享航班、失败时 503（`US-12`、`C-11`、`Q21` 暂定）。 */
+class FlightQueryControllerTest {
+
+    private val mapper = ObjectMapper()
+
+    @Test
+    fun `returns bare JSON array of non-shared flights`() {
+        val projection = StubFlightProjectionPort()
+        projection.snapshots["1"] = """{"flid":"1","scalars":{"MAID":""}}"""
+        projection.snapshots["2"] = """{"flid":"2","scalars":{"MAID":"1"}}"""
+        val controller = FlightQueryController(projection, mapper)
+
+        val response = controller.all()
+
+        assertEquals(HttpStatus.OK, response.status)
+        val root = mapper.readTree(response.body()!!)
+        assertTrue(root.isArray)
+        assertEquals(1, root.size())
+        assertEquals("1", root[0].path("flid").asText())
+    }
+
+    @Test
+    fun `projection failure returns 503 with error body not empty array`() {
+        val projection = StubFlightProjectionPort().apply {
+            failWith = IllegalStateException("redis-down")
+        }
+        val controller = FlightQueryController(projection, mapper)
+
+        val response = controller.all()
+
+        assertEquals(HttpStatus.SERVICE_UNAVAILABLE, response.status)
+        val body = mapper.readTree(response.body()!!)
+        assertEquals("FLIGHT_PROJECTION_UNAVAILABLE", body.path("error").asText())
+        assertTrue(body.path("reason").asText().contains("redis-down"))
+        assertFalse(body.isArray)
+    }
+}
diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt
index 1bc3cec..b33898c 100644
--- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt
+++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/FlightCommitTest.kt
@@ -201,7 +201,7 @@ class FlightCommitTest {
             inbox = inbox,
             procState = f.procState,
             codec = codecReturning(msg()),
-            scheduleProcessor = ScheduleProcessor(commit, f.procState, f.flights, f.events, StubSnapshotLog(), opDay, mapper, clock),
+            scheduleProcessor = ScheduleProcessor(commit, f.procState, f.flights, f.events, StubSnapshotLog(), opDay, props, mapper, clock),
             flopProcessor = FlopProcessor(commit, f.flights, f.events, mapper, clock),
             fdelProcessor = FdelProcessor(commit, f.flights, f.events, mapper, clock),
             adftProcessor = AdftProcessor(commit, f.flights, f.events, opDay, mapper, clock),
@@ -259,5 +259,7 @@ class FlightCommitTest {
         override fun write(writes: List<FlightProjectionWrite>) {
             trace += "projection"
         }
+
+        override fun readAll(): List<String> = throw UnsupportedOperationException("查询侧不参与本用例")
     }
 }
diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt
index 3d6d7f6..3f5c5cf 100644
--- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt
+++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/IgnoreBranchTest.kt
@@ -68,7 +68,7 @@ class IgnoreBranchTest {
             inbox = inbox,
             procState = proc,
             codec = codec,
-            scheduleProcessor = ScheduleProcessor(commit, proc, flights, events, log, opDay, ObjectMapper(), clock),
+            scheduleProcessor = ScheduleProcessor(commit, proc, flights, events, log, opDay, props, ObjectMapper(), clock),
             flopProcessor = FlopProcessor(commit, flights, events, ObjectMapper(), clock),
             fdelProcessor = FdelProcessor(commit, flights, events, ObjectMapper(), clock),
             adftProcessor = AdftProcessor(commit, flights, events, opDay, ObjectMapper(), clock),
diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt
index 0304c18..271cd72 100644
--- a/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt
+++ b/src/test/kotlin/com/gzzn/omms/msgexchange/processing/ScheduleProcessorTest.kt
@@ -3,6 +3,7 @@ package com.gzzn.omms.msgexchange.processing
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.gzzn.omms.msgexchange.codec.ScheduleBody
 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.MetaFields
 import com.gzzn.omms.msgexchange.domain.MsgKind
@@ -16,11 +17,13 @@ import com.gzzn.omms.msgexchange.infra.persistence.FlightStateRepository
 import com.gzzn.omms.msgexchange.infra.persistence.MsgEventRepository
 import com.gzzn.omms.msgexchange.infra.persistence.PersistOutcome
 import com.gzzn.omms.msgexchange.infra.persistence.PipelineTransactionManager
+import com.gzzn.omms.msgexchange.infra.stub.StubFlightProjectionPort
 import com.gzzn.omms.msgexchange.infra.stub.StubFlightState
 import com.gzzn.omms.msgexchange.infra.stub.StubMsgEvents
 import com.gzzn.omms.msgexchange.infra.stub.StubProcState
 import com.gzzn.omms.msgexchange.infra.stub.StubSnapshotLog
 import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
 import org.junit.jupiter.api.Assertions.assertNotNull
 import org.junit.jupiter.api.Assertions.assertNull
 import org.junit.jupiter.api.Assertions.assertTrue
@@ -67,17 +70,20 @@ class ScheduleProcessorTest {
         events: StubMsgEvents = StubMsgEvents(),
         log: StubSnapshotLog = StubSnapshotLog(),
         tx: PipelineTransactionManager? = null,
+        projection: FlightProjectionPort = StubFlightProjectionPort(),
+        snapshotBatch: Int = 200,
     ): ScheduleProcessor {
         val txRunner = tx ?: object : PipelineTransactionManager {
             override fun <T> inTransaction(block: () -> T): T = block()
         }
         return ScheduleProcessor(
-            commit = testCommit(procState = proc, msgEvents = events, txManager = txRunner),
+            commit = testCommit(procState = proc, msgEvents = events, projection = projection, txManager = txRunner),
             procState = proc,
             flightState = flights,
             msgEvents = events,
             snapshotLog = log,
             operationDayProps = OperationDayProps().apply { zone = "Asia/Shanghai"; cutoffHour = 0 },
+            props = PipelineProps().apply { schd.snapshotBatch = snapshotBatch },
             mapper = ObjectMapper(),
             clock = java.time.Clock.systemUTC(),
         )
@@ -197,6 +203,144 @@ class ScheduleProcessorTest {
         assertTrue(log.entries.single().flags.contains(SnapshotFlag.SCHD_REVIVE_CONFLICT))
     }
 
+    /** 覆盖范围内缺席 = AODB 已删掉这个航班（`INV-7`、`US-07` AC2/AC5）。 */
+    @Test
+    fun `absent flights inside the coverage range are deleted, notified and dropped from the projection`() {
+        val proc = StubProcState()
+        proc.insertIfAbsent(msgId, null)
+        val flights = StubFlightState()
+        val events = StubMsgEvents()
+        val log = StubSnapshotLog()
+        val projection = StubFlightProjectionPort()
+        seed(flights, "121", LocalDate.of(2026, 12, 15))
+        seed(flights, "122", LocalDate.of(2026, 12, 15))   // 覆盖范围内，但快照里没带
+        seed(flights, "133", LocalDate.of(2026, 12, 14))   // 覆盖范围外：前一日的航班不受本报文影响
+        projection.snapshots["122"] = """{"flid":"122"}"""
+        projection.snapshots["133"] = """{"flid":"133"}"""
+
+        val result = processor(proc = proc, flights = flights, events = events, log = log, projection = projection)
+            .applyScheduleRecords(head(), message(makeBody("121" to "15DEC261723")))
+
+        assertEquals(ApplyResult.Succeeded, result)
+        assertEquals(FlightState.DELETED, flights.findMainRow("122")!!.state)
+        assertEquals(2L, flights.findMainRow("122")!!.stateVersion)        // 标删推进版本
+        assertFalse(projection.snapshots.containsKey("122"))               // 投影按快照刷掉
+        val delete = events.rows.values.single { it.partitionKey == "122" }
+        assertEquals(Targets.KAFKA_MSG, delete.target)                     // 删除通知只走 msg（C-9）
+        assertTrue(delete.payloadJson!!.contains("\"deleted\":true"))
+        assertEquals(FlightState.ACTIVE, flights.findMainRow("133")!!.state)
+        assertTrue(projection.snapshots.containsKey("133"))
+        assertTrue(log.entries.single().flags.contains(SnapshotFlag.SCHD_ABSENT_DELETED))
+    }
+
+    /** 还没被日计划收录的航班（运营日为空）不属于任何覆盖范围，缺席清扫碰不到它。 */
+    @Test
+    fun `flights without an operation day are out of every coverage range`() {
+        val flights = StubFlightState()
+        val events = StubMsgEvents()
+        flights.persistFullState(
+            com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot(
+                "144", null, FlightState.ACTIVE, 1, mapOf("FLNO" to "CA144"), emptyMap(),
+            ),
+            msgId = 1,
+            now = java.time.Instant.now(),
+        )
+
+        processor(flights = flights, events = events)
+            .applyScheduleRecords(head(), message(makeBody("121" to "15DEC261723")))
+
+        assertEquals(FlightState.ACTIVE, flights.findMainRow("144")!!.state)
+        assertEquals(1L, flights.findMainRow("144")!!.stateVersion)
+    }
+
+    /** `US-07` AC4：分批写，每批一个事务；终态在整份写完、投影刷完之后才落（`INV-9`）。 */
+    @Test
+    fun `each batch commits in its own transaction and the terminal state comes last`() {
+        val proc = StubProcState()
+        proc.insertIfAbsent(msgId, null)
+        val flights = StubFlightState()
+        val tx = TxRunner { false }
+        val body = makeBody("121" to "15DEC261723", "122" to "15DEC261823", "123" to "15DEC261923")
+
+        val result = processor(proc = proc, flights = flights, tx = tx, snapshotBatch = 1)
+            .applyScheduleRecords(head(), message(body))
+
+        assertEquals(ApplyResult.Succeeded, result)
+        assertEquals(4, tx.entered)            // 3 批领域事务 + 1 个终态事务
+        assertEquals(3, flights.mains.size)
+        assertEquals(ProcStatus.SUCCEEDED, proc.find(msgId)!!.state)
+    }
+
+    /** `INV-10`：投影刷不成功就不算处理完成；已提交的批次不回滚（`US-03` AC3）。 */
+    @Test
+    fun `projection failure keeps committed batches and leaves the message unfinished`() {
+        val proc = StubProcState()
+        proc.insertIfAbsent(msgId, null)
+        val flights = StubFlightState()
+        val events = StubMsgEvents()
+        val projection = StubFlightProjectionPort().apply { failWith = IllegalStateException("redis-down") }
+        val body = makeBody("121" to "15DEC261723", "122" to "15DEC261823")
+
+        org.junit.jupiter.api.Assertions.assertThrows(FlightProjectionFailure::class.java) {
+            processor(proc = proc, flights = flights, events = events, projection = projection, snapshotBatch = 1)
+                .applyScheduleRecords(head(), message(body))
+        }
+
+        assertEquals(2, flights.mains.size)                    // 已提交的批次不撤销
+        assertEquals(ProcStatus.PENDING, proc.find(msgId)!!.state)
+        assertNull(proc.find(msgId)!!.backfillNextAt)
+        assertTrue(events.rows.values.all { it.state == com.gzzn.omms.msgexchange.domain.EventStatus.HELD })
+    }
+
+    /** 运营日冲突在任何一批写入之前就查出来：整包拒绝、一条都不落（`INV-4`）。 */
+    @Test
+    fun `operation day conflict rejects the package before any batch is written`() {
+        val flights = StubFlightState()
+        seed(flights, "122", LocalDate.of(2026, 12, 16))        // 库里 122 属于 12-16
+        val body = makeBody("121" to "15DEC261723", "122" to "15DEC261823")
+
+        val result = processor(flights = flights, snapshotBatch = 1)
+            .applyScheduleRecords(head(), message(body))
+
+        assertTrue(result is ApplyResult.DeadProtocol)
+        assertNull(flights.findMainRow("121"))                 // 第一批也没写
+        assertEquals(LocalDate.of(2026, 12, 16), flights.findMainRow("122")!!.operationDay)
+        assertEquals(1L, flights.findMainRow("122")!!.stateVersion)
+    }
+
+    /** `C-6`、`US-07` AC3：日计划没带的字段按"AODB 已删掉该值"清空。 */
+    @Test
+    fun `fields absent from the day plan snapshot are cleared`() {
+        val proc = StubProcState()
+        proc.insertIfAbsent(msgId, null)
+        val flights = StubFlightState()
+        flights.persistFullState(
+            com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot(
+                "121", LocalDate.of(2026, 12, 15), FlightState.ACTIVE, 1,
+                mapOf("SODT" to "15DEC261723", "REMC" to "old-note"),
+                mapOf("GTDT" to listOf(mapOf("GATE" to "G1"))),
+            ),
+            msgId = 1,
+            now = java.time.Instant.now(),
+        )
+
+        processor(proc = proc, flights = flights).applyScheduleRecords(head(), message(makeBody("121" to "15DEC261723")))
+
+        val snapshot = flights.loadFullSnapshot("121")!!
+        assertFalse(snapshot.scalars.containsKey("REMC"))
+        assertEquals(emptyList<Map<String, String>>(), snapshot.collections["GTDT"])
+    }
+
+    private fun seed(flights: StubFlightState, flid: String, day: LocalDate) {
+        flights.persistFullState(
+            com.gzzn.omms.msgexchange.domain.flight.FlightSnapshot(
+                flid, day, FlightState.ACTIVE, 1, mapOf("SODT" to "15DEC261723"), emptyMap(),
+            ),
+            msgId = 1,
+            now = java.time.Instant.now(),
+        )
+    }
+
     @Test
     fun `infra failure inside transaction throws and leaves nothing persisted`() {
         val proc = StubProcState()