refactor(query): add admission and normalizer

document/design/2026-08-06-query-service-architecture-upgrade.md

@@ -109,7 +109,7 @@ Policy 不直接接收未经验证的 `Any` DTO。Admission 先执行低成本
 
 - 条件深度、节点数、children 形态；
 - 字段、projection、sort 数量和字符串长度；
-- value、options、RAW payload 的大小上限；
+- value、options，以及未来已绑定 immutable JSON `RAW` payload 的大小上限；
 - 非法分页、负 limit 和明显溢出。
 
 Admission 只防止畸形输入消耗过多资源，不决定业务授权或 Backend 语义。
@@ -628,12 +628,13 @@ mapping `_meta` 保存 mapping version、document kind 和 capability digest。
 
 ### 10.1 当前基线与完成度
 
-以下状态以提交 `9ddeba24f` 为审计基线。状态只根据当前源码和测试判断，不根据设计意图推断：
+以下状态以 stacked Phase 1 分支的当前源码和测试为审计基线。状态只根据可执行代码与验证结果判断，
+不根据设计意图推断：
 
 | Phase | 当前证据 | 状态 |
 |---|---|---|
 | 0 | `QueryHandler` 已形成单一 defer/fail-closed 边界；EventStream factory 已使用 materialized key 与并发缓存；对应同步、异步、partial Flux、cancel、direct handle 和多订阅测试已存在 | 已实现，待 PR #2908 合并 |
-| 1 | P1-A 已在 stacked 分支引入 internal `QueryInvocation`、`NormalizedCondition`、`QueryPlan` 与最小 analytics model；尚无 Normalizer、Planner 或运行时接线 | P1-A 已实现，P1-B/P1-C 未开始 |
+| 1 | P1-A 已引入 internal `QueryInvocation`、语义代数、`QueryPlan` 与最小 analytics model；P1-B 已引入单遍 admission snapshot、全局 value/payload budget、typed rejection 与 43 operator Normalizer；尚无 Planner、Backend compiler 或运行时接线 | P1-A/P1-B 已实现，P1-C 未开始 |
 | 2 | Spring Registrar 仍从 storage `QueryServiceFactory` 直接创建 Bean；WebFlux 仍直接调用 legacy `QueryHandler` | 未开始 |
 | 3 | MongoDB 查询仍由 `AbstractMongoQueryService` 直接执行 `find/countDocuments`，没有 planned compiler、Backend、单操作 page 或 aggregation pipeline | 未开始 |
 | 4 | Elasticsearch 查询仍使用 `from/size` 和 hits/count，没有 field binding、readiness、完整性 validator、PIT 或 composite aggregation | 未开始 |
@@ -716,6 +717,27 @@ Phase 0 提交，不涉及配置、索引或数据。
 - 从 PR #2903 搬运有效 operator fixtures/validator cases，但不复用其 Backend-specific converter 修改；
 - 返回稳定 category/path/code 的 typed rejection，测试不绑定异常文案。
 
+P1-B 当前实现约束：
+
+- `QueryAdmissionLimits` 同时限制局部容器和整次 admission 的 condition/value node、UTF-8 payload、数字精度；
+  默认值仍是 internal safety baseline，P2-A 接入运行时时再通过配置与真实流量证据校准；
+- legacy `RAW` 不携带 Backend id，admission 不读取、不预算并丢弃原 driver object，只产出 `NativeUnbound` marker，
+  Normalizer 稳定返回 `UNSUPPORTED_FEATURE/NATIVE_BACKEND_UNBOUND`；未来已绑定 immutable JSON Native contract
+  才进入 payload budget，留 P1-C/P5-A；
+- mixed include/exclude 在 Normalizer 保留为 `NormalizedProjection.Mixed`，不在缺失 validation mode 时提前决定
+  compatible/strict policy；P1-C Planner 根据 result shape、validation mode 和 compatibility issue 决策；
+- `RawAdmissionGuard`、`RawValueSnapshotter`、`AdmissionBudget` 与 `QueryNormalizer` 均保持 internal，生产调用链、
+  Spring Bean、现有 wire DTO 和 OpenAPI 不接线。
+
+P1-B 验证证据：
+
+- admission tests 覆盖动态 getter 单读、one-shot iterable、List/Map/ByteArray 防御复制、condition/value cycle、
+  hostile duplicate-key Map、局部与累计预算、稳定 key path、分页 Long offset 和 typed rejection；
+- Normalizer golden tests 覆盖全部 43 个 wire operator、Mongo 空集合常量、数字 canonicalization、system field、
+  多层 `ELEM_MATCH` 相对 scope、literal pattern、projection/sort、一次 Clock、DST-safe 半开时间范围和 RAW 拒绝；
+- `./gradlew :wow-query:check` 与 OpenAPI snapshot test 是该切片的退出验证；P1-C 完成前不宣称 Phase 1 exit gate
+  整体满足。
+
 #### P1-C：Logical Schema 与 Planner
 
 - 引入 `QueryFieldSchema`、logical capability、`RequiredCapabilities`、`SemanticTier`、plan fingerprint；

wow-query/src/main/kotlin/me/ahoo/wow/query/internal/admission/AdmissionBudget.kt

@@ -0,0 +1,86 @@
+/*
+ * Copyright [2021-present] [ahoo wang <ahoowang@qq.com> (https://github.com/Ahoo-Wang)].
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package me.ahoo.wow.query.internal.admission
+
+import me.ahoo.wow.query.internal.rejection.QueryRejectionCategory
+import me.ahoo.wow.query.internal.rejection.QueryRejectionCode
+import me.ahoo.wow.query.internal.rejection.QueryRejectionPath
+import me.ahoo.wow.query.internal.rejection.rejectQuery
+import java.math.BigDecimal
+import java.math.BigInteger
+import java.nio.charset.StandardCharsets
+
+/** Per-admission cumulative budget shared by every condition value, field and option. */
+internal class AdmissionBudget(
+    private val limits: QueryAdmissionLimits,
+) {
+    private var valueNodes: Int = 0
+    private var payloadBytes: Long = 0
+
+    fun enterValue(path: QueryRejectionPath) {
+        if (valueNodes == limits.maxValueNodes) {
+            rejectBudget(path, QueryRejectionCode.VALUE_NODE_LIMIT_EXCEEDED)
+        }
+        valueNodes++
+    }
+
+    fun consumeString(value: String, path: QueryRejectionPath) {
+        if (value.length > limits.maxStringLength) {
+            rejectBudget(path, QueryRejectionCode.STRING_LIMIT_EXCEEDED)
+        }
+        consumeUtf8(value, path)
+    }
+
+    fun consumeUtf8(value: String, path: QueryRejectionPath) {
+        consumePayload(value.toByteArray(StandardCharsets.UTF_8).size.toLong(), path)
+    }
+
+    fun consumeBytes(size: Int, path: QueryRejectionPath) {
+        if (size > limits.maxByteArrayLength) {
+            rejectBudget(path, QueryRejectionCode.BYTE_ARRAY_LIMIT_EXCEEDED)
+        }
+        consumePayload(size.toLong(), path)
+    }
+
+    fun consumeNumber(number: Number, path: QueryRejectionPath): String {
+        when (number) {
+            is BigDecimal -> if (number.precision() > limits.maxNumericPrecision) {
+                rejectBudget(path, QueryRejectionCode.NUMERIC_PRECISION_LIMIT_EXCEEDED)
+            }
+            is BigInteger -> {
+                val maxBits = limits.maxNumericPrecision.toLong() * 4 + 1
+                if (number.abs().bitLength().toLong() > maxBits) {
+                    rejectBudget(path, QueryRejectionCode.NUMERIC_PRECISION_LIMIT_EXCEEDED)
+                }
+            }
+        }
+        val text = number.toString()
+        val precision = text.trimStart('-').count(Char::isDigit)
+        if (precision > limits.maxNumericPrecision) {
+            rejectBudget(path, QueryRejectionCode.NUMERIC_PRECISION_LIMIT_EXCEEDED)
+        }
+        consumeString(text, path)
+        return text
+    }
+
+    private fun consumePayload(size: Long, path: QueryRejectionPath) {
+        if (size > limits.maxValuePayloadBytes - payloadBytes) {
+            rejectBudget(path, QueryRejectionCode.PAYLOAD_LIMIT_EXCEEDED)
+        }
+        payloadBytes += size
+    }
+
+    private fun rejectBudget(path: QueryRejectionPath, code: QueryRejectionCode): Nothing =
+        rejectQuery(QueryRejectionCategory.BUDGET_EXCEEDED, path, code)
+}

wow-query/src/main/kotlin/me/ahoo/wow/query/internal/admission/AdmittedQuery.kt

@@ -0,0 +1,212 @@
+/*
+ * Copyright [2021-present] [ahoo wang <ahoowang@qq.com> (https://github.com/Ahoo-Wang)].
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package me.ahoo.wow.query.internal.admission
+
+import me.ahoo.wow.api.query.DeletionState
+import me.ahoo.wow.api.query.Operator
+import me.ahoo.wow.api.query.Sort
+import me.ahoo.wow.query.internal.analytics.AnalyticsQuery
+import me.ahoo.wow.query.internal.model.QueryOperation
+import me.ahoo.wow.query.internal.model.QueryResultShape
+import me.ahoo.wow.query.internal.model.QueryTarget
+import me.ahoo.wow.query.internal.normalization.CaseSensitivity
+import me.ahoo.wow.query.internal.normalization.NormalizedValue
+import java.time.LocalTime
+import java.time.ZoneId
+import java.time.format.DateTimeFormatter
+import java.util.Collections
+
+internal data class QueryAdmissionLimits(
+    val maxConditionDepth: Int = 32,
+    val maxConditionNodes: Int = 1024,
+    val maxChildrenPerNode: Int = 128,
+    val maxFieldLength: Int = 512,
+    val maxStringLength: Int = 65_536,
+    val maxCollectionSize: Int = 1024,
+    val maxObjectFields: Int = 256,
+    val maxValueDepth: Int = 16,
+    val maxValueNodes: Int = 16_384,
+    val maxNumericPrecision: Int = 1024,
+    val maxByteArrayLength: Int = 65_536,
+    val maxValuePayloadBytes: Long = 4L * 1024 * 1024,
+    val maxProjectionFields: Int = 128,
+    val maxSortFields: Int = 32,
+    val maxOptions: Int = 8,
+) {
+    init {
+        require(
+            listOf(
+                maxConditionDepth,
+                maxConditionNodes,
+                maxChildrenPerNode,
+                maxFieldLength,
+                maxStringLength,
+                maxCollectionSize,
+                maxObjectFields,
+                maxValueDepth,
+                maxValueNodes,
+                maxNumericPrecision,
+                maxByteArrayLength,
+                maxProjectionFields,
+                maxSortFields,
+                maxOptions,
+            ).all { it > 0 },
+        ) {
+            "Query admission limits must be positive."
+        }
+        require(maxValuePayloadBytes > 0) {
+            "Query value payload limit must be positive."
+        }
+    }
+
+    companion object {
+        val DEFAULT: QueryAdmissionLimits = QueryAdmissionLimits()
+    }
+}
+
+internal data class AdmittedQueryInvocation(
+    val target: QueryTarget,
+    val operation: QueryOperation,
+    val resultShape: QueryResultShape,
+    val input: AdmittedQueryInput,
+)
+
+internal sealed interface AdmittedQueryInput {
+    data class Single(val query: AdmittedRecordQuery) : AdmittedQueryInput
+
+    data class Stream(
+        val query: AdmittedRecordQuery,
+        val limit: Int,
+    ) : AdmittedQueryInput
+
+    data class Page(
+        val query: AdmittedRecordQuery,
+        val page: AdmittedPage,
+    ) : AdmittedQueryInput
+
+    data class Count(val condition: AdmittedCondition) : AdmittedQueryInput
+
+    data class Analytics(val query: AnalyticsQuery) : AdmittedQueryInput
+}
+
+internal class AdmittedRecordQuery(
+    val condition: AdmittedCondition,
+    val projection: AdmittedProjection,
+    sort: Iterable<AdmittedSort>,
+) {
+    val sort: List<AdmittedSort> = Collections.unmodifiableList(sort.toList())
+
+    override fun equals(other: Any?): Boolean =
+        this === other ||
+            other is AdmittedRecordQuery &&
+            condition == other.condition &&
+            projection == other.projection &&
+            sort == other.sort
+
+    override fun hashCode(): Int = 31 * (31 * condition.hashCode() + projection.hashCode()) + sort.hashCode()
+}
+
+internal class AdmittedProjection(
+    include: Iterable<String>,
+    exclude: Iterable<String>,
+) {
+    val include: List<String> = Collections.unmodifiableList(include.toList())
+    val exclude: List<String> = Collections.unmodifiableList(exclude.toList())
+
+    override fun equals(other: Any?): Boolean =
+        this === other ||
+            other is AdmittedProjection &&
+            include == other.include &&
+            exclude == other.exclude
+
+    override fun hashCode(): Int = 31 * include.hashCode() + exclude.hashCode()
+}
+
+internal data class AdmittedSort(
+    val field: String,
+    val direction: Sort.Direction,
+)
+
+internal data class AdmittedPage(
+    val index: Int,
+    val size: Int,
+    val offset: Long,
+)
+
+internal sealed interface AdmittedConditionValue {
+    data object Absent : AdmittedConditionValue
+
+    /** Legacy RAW marker; no driver object crosses the admission boundary. */
+    data object NativeUnbound : AdmittedConditionValue
+
+    data class QueryValue(val value: NormalizedValue) : AdmittedConditionValue
+
+    data class TimeOfDay(val value: LocalTime) : AdmittedConditionValue
+
+    data class Deletion(val value: DeletionState) : AdmittedConditionValue
+}
+
+internal data class AdmittedConditionOptions(
+    val caseSensitivity: CaseSensitivity = CaseSensitivity.SENSITIVE,
+    val zoneId: ZoneId? = null,
+    val datePattern: AdmittedDatePattern? = null,
+)
+
+internal class AdmittedDatePattern(
+    val formatter: DateTimeFormatter,
+    descriptor: String,
+) {
+    private val signature: List<Any?> = listOf(
+        descriptor,
+        formatter.locale,
+        formatter.decimalStyle,
+        formatter.resolverStyle,
+        formatter.chronology,
+        formatter.zone,
+        formatter.resolverFields,
+    )
+
+    override fun equals(other: Any?): Boolean =
+        this === other || other is AdmittedDatePattern && signature == other.signature
+
+    override fun hashCode(): Int = signature.hashCode()
+}
+
+internal class AdmittedCondition(
+    val field: String,
+    val operator: Operator,
+    val value: AdmittedConditionValue,
+    children: Iterable<AdmittedCondition>,
+    val options: AdmittedConditionOptions,
+) {
+    val children: List<AdmittedCondition> = Collections.unmodifiableList(children.toList())
+
+    override fun equals(other: Any?): Boolean =
+        this === other ||
+            other is AdmittedCondition &&
+            field == other.field &&
+            operator == other.operator &&
+            value == other.value &&
+            children == other.children &&
+            options == other.options
+
+    override fun hashCode(): Int {
+        var result = field.hashCode()
+        result = 31 * result + operator.hashCode()
+        result = 31 * result + value.hashCode()
+        result = 31 * result + children.hashCode()
+        result = 31 * result + options.hashCode()
+        return result
+    }
+}