rust_store_core/schema/

registry.rs

//! Schema 注册表：注册/定位/配置（数据结构与规范化见 [`super::definition`]）。

use std::collections::HashMap;
use std::sync::Arc;

use serde_json::{json, Map, Value};

use crate::types::{is_truthy, str_list, COMPUTE_AGG_OPS};

use super::definition::{
    normalize_fields, validate_vector_fields, ComputeDef, FieldDef, Location, RelationDef, Schema,
};
use super::load::LoadConfig;

use crate::command::WriteLinkPolicy;
use crate::error::{CoreError, ERR_POLICY_VIEW_READONLY};

/// 查询档位：判决唯一在 core（照 [`Registry::require_context`] 既有范式）。
///
/// - [`Profile::Standard`]（默认）：标准调用 —— 跨方言对齐的公共能力集；
///   DB 独有能力可用但须代码注释标注「不建议业务查询」（见执行文档 §4.5）。
/// - [`Profile::Text2Query`]：AI 问数 —— 功能收缩 + 硬性限制（行数/深度/准入全收紧）。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Profile {
    /// 标准调用：跨方言对齐的公共能力集；DB 独有能力可用但须注释标注（不建议业务查询）
    #[default]
    Standard,
    /// AI 问数：功能收缩 + 硬性限制
    Text2Query,
}

impl Profile {
    /// 档位字符串（Host / 绑定层单点取用）
    pub fn as_str(&self) -> &'static str {
        match self {
            Profile::Standard => "standard",
            Profile::Text2Query => "text2query",
        }
    }

    /// 字符串 → 档位；未知值 **Err**（禁静默回落 `Standard`，见执行文档 §7）
    pub fn from_str_or_err(s: &str) -> Result<Profile, String> {
        match s {
            "standard" => Ok(Profile::Standard),
            "text2query" => Ok(Profile::Text2Query),
            other => Err(format!("未知 profile: {other}（仅 standard / text2query）")),
        }
    }
}

/// 一条注册项：一份主结构 + 主/从链路 + 版本（D4/D5/D13）。
#[derive(Debug, Clone)]
struct Entry {
    schema: Schema,
    /// 主 + 从链路；`links[0]` 恒为主落点（= `schema` 的定位投影）
    links: Vec<Location>,
    /// 版本：首次注册 = 1；跨批次 reload 同名 ⇒ +1
    version: u64,
}

/// 批次分组结果（判据单点输出；`register_batch` 与 `schema::load::plan_load` 共用）。
pub(super) struct BatchClass {
    /// 组名（按首次出现顺序）
    pub order: Vec<String>,
    /// name → (主定义, 落点, collection)；每组至多一份主
    pub primaries: HashMap<String, (Value, Location, String)>,
    /// name → 从项（原序）
    pub replicas: HashMap<String, Vec<(Value, Location)>>,
}

/// 批次判据错误（调用方各自格式化文案：注册期与装载期文案口径不同）。
pub(super) enum BatchError {
    /// 解析错误（非对象 / 缺 name）
    Parse(String),
    /// 同名主 ≥2（A1）：组名 + 主项在输入中的下标
    PrimaryDuplicate { name: String, indices: Vec<usize> },
    /// 批内四元组冲突
    LocationConflict(Box<LocationConflict>),
}

/// 批次判据单点：**分组 + 主唯一（≥2 ⇒ `Err`）+ 批内四元组冲突（⇒ `Err`）**。
///
/// **0 主不在此报错**（交由调用方决定：`register_batch` 走既有 entry 追加从链路，
/// `plan_load` 直接拒——「主定义缺失」）。`register_batch` 与 `plan_load` 共用本函数，
/// 禁各写一份（判据单点，防 A1/A3 漂移）。
pub(super) fn classify_batch(items: &[(Value, Location)]) -> Result<BatchClass, BatchError> {
    let mut order: Vec<String> = Vec::new();
    let mut primaries: HashMap<String, (Value, Location, String)> = HashMap::new();
    let mut primary_idx: HashMap<String, usize> = HashMap::new();
    let mut replicas: HashMap<String, Vec<(Value, Location)>> = HashMap::new();

    for (idx, (defn, loc)) in items.iter().enumerate() {
        let obj = defn
            .as_object()
            .ok_or_else(|| BatchError::Parse("schema 定义必须是对象".to_string()))?;
        let name = obj
            .get("name")
            .and_then(|v| v.as_str())
            .ok_or_else(|| BatchError::Parse("schema 缺少 name".to_string()))?
            .to_string();
        if !order.iter().any(|n| n == &name) {
            order.push(name.clone());
        }

        if obj.get("replica").map(is_truthy).unwrap_or(false) {
            replicas
                .entry(name)
                .or_default()
                .push((defn.clone(), loc.clone()));
            continue;
        }

        let collection = obj
            .get("collection")
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
            .map(String::from)
            .unwrap_or_else(|| name.clone());
        if let Some(first) = primary_idx.get(&name) {
            // 同名主 ≥2 ⇒ Err（A1）；携带下标供调用方定位（rel 路径 / 提示）
            return Err(BatchError::PrimaryDuplicate {
                name,
                indices: vec![*first, idx],
            });
        }
        primary_idx.insert(name.clone(), idx);
        primaries.insert(name, (defn.clone(), loc.clone(), collection));
    }

    // 批内四元组冲突：主 + 从全部链路（判据函数 `detect_location_conflict`）
    let mut quads: Vec<(String, String, Location)> = Vec::new();
    for name in &order {
        if let Some((_, loc, coll)) = primaries.get(name) {
            quads.push((name.clone(), coll.clone(), loc.clone()));
        }
        if let Some(rs) = replicas.get(name) {
            for (defn, loc) in rs {
                let coll = defn
                    .get("name")
                    .and_then(|v| v.as_str())
                    .unwrap_or(name)
                    .to_string();
                quads.push((name.clone(), coll, loc.clone()));
            }
        }
    }
    if let Some(c) = detect_location_conflict(&quads) {
        return Err(BatchError::LocationConflict(Box::new(c)));
    }

    Ok(BatchClass {
        order,
        primaries,
        replicas,
    })
}

/// 四元组冲突信息（供调用方格式化文案）
pub(super) struct LocationConflict {
    pub collection: String,
    pub prev: String,
    pub name: String,
    pub source: String,
    pub database: Option<String>,
    pub schema: Option<String>,
}

/// 四元组 `(source, database, schema, collection)` 冲突检测（**判据单点**）：
/// 同一四元组被两个不同 schema 名占用 ⇒ `Some(冲突)`。`register_batch`（含既有 entry）
/// 与 `plan_load`（批内）共用。
pub(super) fn detect_location_conflict(
    entries: &[(String, String, Location)],
) -> Option<LocationConflict> {
    let mut seen: HashMap<(String, Option<String>, Option<String>, String), String> =
        HashMap::new();
    for (name, collection, loc) in entries {
        let key = (
            loc.source.clone(),
            loc.database.clone(),
            loc.schema.clone(),
            collection.clone(),
        );
        if let Some(prev) = seen.get(&key) {
            if prev != name {
                return Some(LocationConflict {
                    collection: collection.clone(),
                    prev: prev.clone(),
                    name: name.clone(),
                    source: loc.source.clone(),
                    database: loc.database.clone(),
                    schema: loc.schema.clone(),
                });
            }
        } else {
            seen.insert(key, name.clone());
        }
    }
    None
}

/// `BatchError` → 注册期文案（保留既有子串：`同名主定义重复` / `定位冲突`）。
fn batch_err_to_msg(e: BatchError) -> String {
    match e {
        BatchError::Parse(m) => m,
        BatchError::PrimaryDuplicate { name, .. } => format!(
            "同名主定义重复: {}（同一装载批次内 name 必须唯一，从定义请用 replica: true）",
            name
        ),
        BatchError::LocationConflict(c) => format!(
            "定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
            c.source, c.database, c.schema, c.collection, c.prev
        ),
    }
}

/// schema 目录快照：全局唯一真源的数据体（注册一次，派生视图共享同一 `Arc`）。
#[derive(Debug, Clone, Default)]
pub struct Catalog {
    entries: HashMap<String, Entry>,
    order: Vec<String>,
}

/// 策略束：六项策略的载体（base 可变；视图为派生时点快照，`set` 只影响自身）。
#[derive(Debug, Clone, Default)]
pub struct PolicyBundle {
    /// 上下文强制开关（默认关闭 = fail-open，与 JS 原版 parity）；
    /// 开启后所有 plan 入口对 `ctx: None` 显式报错（fail-secure，见 `permission` 模块文档）。
    /// 内部调用请传显式系统上下文（JSON `{"internal": true}` / `Context::system()`）。
    pub(crate) require_context: bool,
    /// 查询档位（默认 [`Profile::Standard`]；text2query 由 AI 问数链路显式进入）。
    pub(crate) profile: Profile,
    /// RBAC 动态策略（`None` = 未启用，判决原语直通、行为与现状一致）。见 [`crate::rbac`]
    pub(crate) rbac: Option<crate::rbac::RbacPolicy>,
    /// 角色清单与未配置姿态（豁免 / 拒写 / Open|Closed，默认 []/[]/Open）。见 [`crate::permission::RoleRules`]
    pub(crate) role_rules: crate::permission::RoleRules,
    /// 定义层门禁策略（默认 Open —— 全放行，保持既有 parity）。见 [`crate::permission::MetaPolicy`]
    pub(crate) meta_policy: crate::permission::MetaPolicy,
    /// 跨连接写策略（默认 [`WriteLinkPolicy::Reject`]）。见 [`crate::command::write_links`]
    pub(crate) write_link_policy: WriteLinkPolicy,
}

/// Schema 注册表：**共享目录快照**（`Arc<Catalog>`）+ **策略束**（`PolicyBundle`）。
///
/// - `is_base = true`：目录唯一真源，注册 / 清空写路径仅在此开放；
/// - 派生视图（`with_policy` 产出）持同一 `Arc<Catalog>` 快照 + 叠加策略，`is_base = false`。
#[derive(Debug, Clone)]
pub struct Registry {
    catalog: Arc<Catalog>,
    policy: PolicyBundle,
    /// 是否目录唯一真源（`Default` 即 base）；派生视图为 `false`。
    is_base: bool,
}

// `is_base` 不允许 derive 默认（`false` = 视图是错误语义），手写 Default（即 base）。
impl Default for Registry {
    fn default() -> Self {
        Self {
            catalog: Arc::new(Catalog::default()),
            policy: PolicyBundle::default(),
            is_base: true,
        }
    }
}

/// 策略覆盖项（派生视图的差分输入）：`None` = 继承来源视图；
/// `rbac` 的 `Some(None)` = 显式清除（其余字段无「清除」语义）。
#[derive(Debug, Clone, Default)]
pub struct PolicyOverrides {
    require_context: Option<bool>,
    profile: Option<Profile>,
    rbac: Option<Option<crate::rbac::RbacPolicy>>,
    role_rules: Option<crate::permission::RoleRules>,
    meta_policy: Option<crate::permission::MetaPolicy>,
    write_link_policy: Option<WriteLinkPolicy>,
}

impl PolicyOverrides {
    /// JSON → 覆盖项（**唯一策略 JSON 解析点**，02 双绑定共用；禁在别处重复解析策略配置）。
    ///
    /// 已知键（camelCase，与宿主 config 键风格一致）：`profile` / `requireContext` /
    /// `rbac` / `roleRules` / `metaPolicy` / `writeLinkPolicy`；空对象 = 无覆盖。
    /// 未知键 **Err**（禁静默忽略，对齐 `no-error-masking`）。
    pub fn from_value(v: &Value) -> Result<Self, CoreError> {
        let obj = v
            .as_object()
            .ok_or_else(|| CoreError::Other("策略覆盖必须是 JSON 对象".to_string()))?;
        let mut o = PolicyOverrides::default();
        for (k, val) in obj {
            match k.as_str() {
                "profile" => {
                    let s = val
                        .as_str()
                        .ok_or_else(|| CoreError::Other("profile 必须是字符串".to_string()))?;
                    o.profile = Some(Profile::from_str_or_err(s).map_err(CoreError::Other)?);
                }
                "requireContext" => {
                    o.require_context = Some(val.as_bool().ok_or_else(|| {
                        CoreError::Other("requireContext 必须是布尔".to_string())
                    })?);
                }
                "rbac" => {
                    // 对象 = 注入；null = 显式清除（Some(None)，与缺键 None 区分）
                    o.rbac = Some(match val {
                        Value::Null => None,
                        other => Some(
                            crate::rbac::RbacPolicy::from_json(other).map_err(CoreError::Other)?,
                        ),
                    });
                }
                "roleRules" => {
                    o.role_rules = Some(parse_role_rules(val)?);
                }
                "metaPolicy" => {
                    o.meta_policy = Some(parse_meta_policy(val)?);
                }
                "writeLinkPolicy" => {
                    let s = val.as_str().ok_or_else(|| {
                        CoreError::Other("writeLinkPolicy 必须是字符串".to_string())
                    })?;
                    o.write_link_policy = Some(match s {
                        "reject" => WriteLinkPolicy::Reject,
                        "primaryOnly" => WriteLinkPolicy::PrimaryOnly,
                        other => {
                            return Err(CoreError::Other(format!(
                                "未知 writeLinkPolicy: {other}（仅 reject / primaryOnly）"
                            )))
                        }
                    });
                }
                other => {
                    return Err(CoreError::Other(format!(
                        "{ERR_POLICY_VIEW_READONLY} 未知策略覆盖键 `{other}`"
                    )))
                }
            }
        }
        Ok(o)
    }
}

/// `roleRules` 覆盖解析（复用 `UnconfiguredPolicy::from_str_or_err`；未知子键 Err）。
fn parse_role_rules(v: &Value) -> Result<crate::permission::RoleRules, CoreError> {
    let obj = v
        .as_object()
        .ok_or_else(|| CoreError::Other("roleRules 必须是 JSON 对象".to_string()))?;
    let mut rr = crate::permission::RoleRules::default();
    for (k, val) in obj {
        match k.as_str() {
            "exemptRoles" => rr.exempt_roles = str_list_strict(val, "roleRules.exemptRoles")?,
            "denyWriteRoles" => {
                rr.deny_write_roles = str_list_strict(val, "roleRules.denyWriteRoles")?
            }
            "unconfigured" => {
                let s = val.as_str().ok_or_else(|| {
                    CoreError::Other("roleRules.unconfigured 必须是字符串".to_string())
                })?;
                rr.unconfigured = crate::permission::UnconfiguredPolicy::from_str_or_err(s)
                    .map_err(CoreError::Other)?;
            }
            other => {
                return Err(CoreError::Other(format!(
                    "{ERR_POLICY_VIEW_READONLY} 未知 roleRules 键 `{other}`"
                )))
            }
        }
    }
    Ok(rr)
}

/// `metaPolicy` 覆盖解析（未知子键 Err）。
fn parse_meta_policy(v: &Value) -> Result<crate::permission::MetaPolicy, CoreError> {
    let obj = v
        .as_object()
        .ok_or_else(|| CoreError::Other("metaPolicy 必须是 JSON 对象".to_string()))?;
    let mut mp = crate::permission::MetaPolicy::default();
    for (k, val) in obj {
        match k.as_str() {
            "closed" => {
                mp.closed = val
                    .as_bool()
                    .ok_or_else(|| CoreError::Other("metaPolicy.closed 必须是布尔".to_string()))?;
            }
            "roles" => mp.roles = str_list_strict(val, "metaPolicy.roles")?,
            other => {
                return Err(CoreError::Other(format!(
                    "{ERR_POLICY_VIEW_READONLY} 未知 metaPolicy 键 `{other}`"
                )))
            }
        }
    }
    Ok(mp)
}

/// 严格字符串数组解析（既有 `str_list` 会静默丢弃非字符串元素，此处显式 Err）。
fn str_list_strict(v: &Value, key: &str) -> Result<Vec<String>, CoreError> {
    let arr = v
        .as_array()
        .ok_or_else(|| CoreError::Other(format!("{key} 必须是字符串数组")))?;
    arr.iter()
        .map(|x| {
            x.as_str()
                .map(String::from)
                .ok_or_else(|| CoreError::Other(format!("{key} 的元素必须都是字符串")))
        })
        .collect()
}

impl PolicyBundle {
    /// 叠加覆盖项：`None` 继承自身；`rbac` 特例——`Some(v)`（含 `Some(None)` = 显式清除）。
    fn overridden(&self, o: &PolicyOverrides) -> PolicyBundle {
        PolicyBundle {
            require_context: o.require_context.unwrap_or(self.require_context),
            profile: o.profile.unwrap_or(self.profile),
            rbac: match &o.rbac {
                Some(v) => v.clone(),
                None => self.rbac.clone(),
            },
            role_rules: o
                .role_rules
                .clone()
                .unwrap_or_else(|| self.role_rules.clone()),
            meta_policy: o
                .meta_policy
                .clone()
                .unwrap_or_else(|| self.meta_policy.clone()),
            write_link_policy: o.write_link_policy.unwrap_or(self.write_link_policy),
        }
    }
}

/// 读链路解析结果：已排序链路 + 跳过/降级声明（随 `plan.degraded` 透出，禁静默）。
#[derive(Debug, Clone)]
pub struct ReadLinks {
    /// 已排序链路；`links[0]` = 本次读的选中主链路
    pub links: Vec<Location>,
    /// 跳过/降级声明，随 `plan.degraded` 透出（禁静默）
    pub degraded: Vec<Value>,
}

impl Registry {
    pub fn new() -> Self {
        Self::default()
    }

    /// 派生策略视图（**O(1)**）：共享目录快照 `Arc<Catalog>` + 叠加策略覆盖；派生视图 `is_base = false`。
    ///
    /// 快照语义：视图持**派生时点**目录，base 后续注册经 `Arc::make_mut` 写时复制，对**后续派生**
    /// 视图即时可见，对在途视图不承诺（与 S2 现状口径一致：`store-api/spec/01-routing.md` 运行期新注册不做承诺）。
    pub fn with_policy(&self, overrides: &PolicyOverrides) -> Registry {
        Registry {
            catalog: Arc::clone(&self.catalog),
            policy: self.policy.overridden(overrides),
            is_base: false,
        }
    }

    /// 是否派生视图（`false` = base 目录唯一真源）
    pub fn is_view(&self) -> bool {
        !self.is_base
    }

    /// 视图只读守卫：非 base 视图禁止写目录（模型唯一真源在 base Registry）。
    ///
    /// 拦截必须显式报错、禁静默降级（对齐 `no-error-masking`）。返回 `Err` 供
    /// 注册类入口 `?` 直通；`clear`（返回 `()`，受「公开签名零改动」约束）调用本
    /// 函数并在 `Err` 时 `panic!`（消息同源，携带 [`ERR_POLICY_VIEW_READONLY`] 前缀）。
    fn ensure_base_catalog(&self) -> Result<(), String> {
        if !self.is_base {
            return Err(format!(
                "{ERR_POLICY_VIEW_READONLY} 非 base 策略视图禁止注册/清空 schema（目录唯一真源在 base Registry）"
            ));
        }
        Ok(())
    }

    /// 注册一个 schema（含自动注册 `<Name>Deleted` 归档表），对应 JS `register`。
    ///
    /// 兼容入口：单条 + 缺省落点（`source="default"`、`database/schema=None`）。
    /// 定义层门禁按当前 [`MetaPolicy`](crate::permission::MetaPolicy) 判决
    /// （默认 Open → 全放行）。需带 ctx 显式过门禁请用 [`Self::register_with_ctx`]。
    pub fn register(&mut self, defn: &Value) -> Result<(), String> {
        self.register_with_ctx(defn, None)
    }

    /// 带 ctx 的注册（单条 + 缺省落点）：内部转 [`Self::register_batch`]。
    pub fn register_with_ctx(
        &mut self,
        defn: &Value,
        ctx: Option<&crate::permission::Context>,
    ) -> Result<(), String> {
        self.register_batch(&[(defn.clone(), Location::default())], ctx)
    }

    /// 带定位的批量注册（D13：唯一性校验落「装载批次集合」）。
    ///
    /// 算法（**先构造后落库，任一步 Err ⇒ 零副作用**）：
    /// 1. 判据单点 [`classify_batch`]（分组 / 主唯一 ≥2 ⇒ `Err` / 批内四元组冲突），与
    ///    `schema::load::plan_load` 同源。
    /// 2. `can_register` 门禁（判决先于 `build_schema`）。
    /// 3. 逐组：有主 ⇒ `build_schema` 为 **主**；0 主 ⇒ 追加既有同名 entry 的从链路，无既有 ⇒ `Err`。
    /// 4. 全批 + 既有 `(source, database, schema, collection)` 跨名冲突 ⇒ `Err`
    ///    （判据函数 [`detect_location_conflict`]，替代旧 `check_location_unique`）。
    /// 5. 提交：`links = [primary.loc] ++ replicas`；命中既有 entry ⇒ `version + 1`
    ///    （`order` 位置不变）、替换结构与链路；新名 ⇒ `version = 1`、`order.push`。
    ///    `Schema.source/database/schema` 由 `primary.loc` 注入（`source == "default"` ⇒ `None`）。
    /// 6. 归档派生：主 `name` 不以 `"Deleted"` 结尾且非 `_isArchive` ⇒ 派生 `<Name>Deleted`
    ///    （`links` 继承主链路，定位同主）。
    pub fn register_batch(
        &mut self,
        items: &[(Value, Location)],
        ctx: Option<&crate::permission::Context>,
    ) -> Result<(), String> {
        self.ensure_base_catalog()?; // 视图只读守卫：目录唯一真源在 base
                                     // ── 1：判据单点（分组 / 主唯一 / 批内四元组冲突），与 plan_load 同源 ──
        let class = classify_batch(items).map_err(batch_err_to_msg)?;

        // 定义层门禁：判决先于 build_schema —— 拒绝即返回，绝不部分写入
        for (defn, _loc) in items {
            let name = defn.get("name").and_then(|v| v.as_str()).unwrap_or("");
            if !crate::permission::can_register(&self.policy.meta_policy, ctx) {
                return Err(format!("ERR_PERMISSION: 无权注册或覆盖定义 {name}"));
            }
        }

        // ── 2：逐组拟定 entry 序列（主 + 归档附表，保序） ──
        struct Planned {
            name: String,
            schema: Schema,
            links: Vec<Location>,
        }
        let mut planned: Vec<Planned> = Vec::new();
        for name in &class.order {
            let rs: Vec<Location> = class
                .replicas
                .get(name)
                .map(|v| v.iter().map(|(_, l)| l.clone()).collect())
                .unwrap_or_default();
            if let Some((raw, loc, collection)) = class.primaries.get(name) {
                let obj = raw.as_object().unwrap();
                let schema = build_schema(obj, name.clone(), collection.clone(), loc)?;
                let is_archive = obj.get("_isArchive").map(is_truthy).unwrap_or(false);
                let mut links = vec![loc.clone()];
                links.extend(rs);
                planned.push(Planned {
                    name: name.clone(),
                    schema: schema.clone(),
                    links: links.clone(),
                });
                // 归档附表紧随主定义派生（`links` 继承主链路，定位同主）；
                // archive:false（漉金Core 接入扩展 02）完全不派生（不进模型表、
                // 不建集合、不占名称冲突检查名额）
                let archive_on = obj.get("archive").map(is_truthy).unwrap_or(true);
                if !is_archive && !name.ends_with("Deleted") && archive_on {
                    let aobj = archive_defn(obj, name, &schema.collection)
                        .as_object()
                        .cloned()
                        .unwrap();
                    let aname = format!("{}Deleted", name);
                    let acoll = format!("{}_deleted", schema.collection);
                    let aschema = build_schema(&aobj, aname.clone(), acoll, &links[0])?;
                    planned.push(Planned {
                        name: aname,
                        schema: aschema,
                        links,
                    });
                }
            } else {
                // 0 主：只能是对既有 entry 追加从链路（跨批演进）
                let Some(existing) = self.catalog.entries.get(name) else {
                    return Err(format!(
                        "未找到主 schema: {}（从定义 `replica` 必须伴随同批主定义，或该 name 已注册）",
                        name
                    ));
                };
                let mut links = existing.links.clone();
                links.extend(rs);
                planned.push(Planned {
                    name: name.clone(),
                    schema: existing.schema.clone(),
                    links,
                });
            }
        }

        // ── 3：四元组唯一性（本批 + 既有，跨名冲突即 Err；判据函数单点） ──
        let batch_names: std::collections::HashSet<&str> =
            planned.iter().map(|p| p.name.as_str()).collect();
        let mut quads: Vec<(String, String, Location)> = Vec::new();
        for (n, e) in &self.catalog.entries {
            if batch_names.contains(n.as_str()) {
                continue; // 本批将整体替换该名，旧链路不参与冲突判定
            }
            for l in &e.links {
                quads.push((n.clone(), e.schema.collection.clone(), l.clone()));
            }
        }
        for p in &planned {
            for l in &p.links {
                quads.push((p.name.clone(), p.schema.collection.clone(), l.clone()));
            }
        }
        if let Some(c) = detect_location_conflict(&quads) {
            return Err(format!(
                "定位冲突: ({}, {:?}, {:?}, {}) 已被 schema `{}` 占用",
                c.source, c.database, c.schema, c.collection, c.prev
            ));
        }

        // ── 4：提交（命中既有 ⇒ version + 1 且 order 位置不变；新名 ⇒ version = 1 + push） ──
        let catalog = Arc::make_mut(&mut self.catalog); // 写时复制（仅 base 走到此）
        for p in planned {
            if let Some(e) = catalog.entries.get_mut(&p.name) {
                e.schema = p.schema;
                e.links = p.links;
                e.version += 1;
            } else {
                catalog.entries.insert(
                    p.name.clone(),
                    Entry {
                        schema: p.schema,
                        links: p.links,
                        version: 1,
                    },
                );
                catalog.order.push(p.name);
            }
        }

        Ok(())
    }

    /// 把 `config.links` 的角色/优先级附着到各 entry 的链路集；声明但缺失的链路按声明序补建。
    ///
    /// 算法（**先校验后落库，任一 `Err` ⇒ 零副作用**，与 [`Self::register_batch`] 同规）：
    /// 1. 逐项取 [`Self::get_entry`]（未注册 ⇒ `Err`，禁猜——该名可能由其它批次注册）。
    /// 2. 每条声明按 [`Location::same_place`] 在 `entry.links` 中匹配来源（只比三元组）：
    ///    命中 ⇒ 写回 `role`/`priority`；未命中 ⇒ 按**声明序**追加 `Location`。
    /// 3. 归档随附表 `<Name>Deleted`（若已注册且未在 `links` 中独立声明）继承主的重算链路。
    /// 4. 未在 `cfg.links` 中出现的 entry / 链路：保持原样（零回归）；`links` 为空 ⇒ no-op。
    pub fn apply_link_decls(&mut self, cfg: &LoadConfig) -> Result<(), String> {
        if cfg.links.is_empty() {
            return Ok(());
        }
        self.ensure_base_catalog()?; // 目录唯一真源在 base（视图只读）

        // ── 1/2：拟定期望链路（全程不落库） ──
        let declared: std::collections::HashSet<&str> =
            cfg.links.iter().map(|(n, _)| n.as_str()).collect();
        let mut planned: Vec<(String, Vec<Location>)> = Vec::new();
        for (name, decls) in &cfg.links {
            let entry = self.get_entry(name).map_err(|_| {
                format!("未注册 schema: {name}（config.links 引用了不存在的 schema）")
            })?;
            let mut links = entry.links.clone();
            for decl in decls {
                let target = Location {
                    source: decl.source.clone(),
                    database: decl.database.clone(),
                    schema: decl.schema.clone(),
                    role: decl.role,
                    priority: decl.priority,
                };
                match links.iter_mut().find(|l| l.same_place(&target)) {
                    Some(hit) => {
                        hit.role = decl.role;
                        hit.priority = decl.priority;
                    }
                    None => links.push(target),
                }
            }
            // 归档随附表继承主的重算链路（主名独立声明同名归档时不覆盖）
            if !name.ends_with("Deleted") {
                let aname = format!("{}Deleted", name);
                if !declared.contains(aname.as_str()) && self.catalog.entries.contains_key(&aname) {
                    planned.push((aname, links.clone()));
                }
            }
            planned.push((name.clone(), links));
        }

        // ── 3：提交（仅替换 links；非重注册 ⇒ version 不变） ──
        let catalog = Arc::make_mut(&mut self.catalog);
        for (name, links) in planned {
            if let Some(e) = catalog.entries.get_mut(&name) {
                e.links = links;
            }
        }
        Ok(())
    }

    /// 主链路 + 从链路（只读）
    pub fn links_of(&self, name: &str) -> Result<&[Location], String> {
        Ok(&self.get_entry(name)?.links)
    }

    /// 读链路解析：排序（`role` → `priority` → **声明序**）+ 跳过不可达 + 产降级声明。
    ///
    /// 排序用稳定 `sort_by_key`：键为 `(role.map_or(4u8, |r| r.rank()), priority.unwrap_or(i64::MAX))`
    /// ——未声明 `role` 视作最低层（4）、未声明 `priority` 视作最大；稳定排序 ⇒ 同级保声明序，
    /// 无任何 `links` 声明时全为 `(4, MAX)` ⇒ 顺序 = 原序（零回归）。
    ///
    /// `unreachable`：本次调用不可达的 source 列表（由 Host 下发；`None` = 不限制）；
    /// 命中某链路 `source` ⇒ 从结果剔除并 push 降级声明（禁静默）。全部链路被跳过 ⇒
    /// `Err:ERR_READ_NO_REACHABLE_LINK: <name>`（绝不回落主链路）。
    ///
    /// `links[0]` 即读路径的选中落点（替代 `primary_location` 在读路径的角色；
    /// `primary_location` 保留供写路径与兼容入口）。可达过滤判据单点在
    /// [`federation::plan::select::filter_reachable`](crate::federation::plan::select::filter_reachable)，
    /// 本函数**不重复**该判据。
    pub fn resolve_read_links(
        &self,
        name: &str,
        unreachable: Option<&[String]>,
    ) -> Result<ReadLinks, String> {
        let mut links = self.links_of(name)?.to_vec();
        links.sort_by_key(|l| {
            (
                l.role.map_or(4u8, |r| r.rank()),
                l.priority.unwrap_or(i64::MAX),
            )
        });

        let mut degraded: Vec<Value> = Vec::new();
        if let Some(unreachable) = unreachable {
            links.retain(|l| {
                if unreachable.iter().any(|s| s == &l.source) {
                    degraded.push(json!({
                        "code": "readLinkUnreachable",
                        "layer": "schema",
                        "message": format!("schema {} 的链路 {} 不可达，已跳过", name, l.source),
                        "hint": "检查该数据源连接；或调整 store.config.json 的 links 优先级",
                        "schema": name,
                        "source": l.source,
                    }));
                    false
                } else {
                    true
                }
            });
        }

        if links.is_empty() {
            return Err(format!(
                "ERR_READ_NO_REACHABLE_LINK: {} 的全部链路不可达（已按 unreachable 跳过）",
                name
            ));
        }
        Ok(ReadLinks { links, degraded })
    }

    /// 主落点（= `links[0]`）
    pub fn primary_location(&self, name: &str) -> Result<&Location, String> {
        self.get_entry(name)?
            .links
            .first()
            .ok_or_else(|| format!("Schema 无落点链路: {}", name))
    }

    /// 版本（首次 = 1）
    pub fn version_of(&self, name: &str) -> Result<u64, String> {
        Ok(self.get_entry(name)?.version)
    }

    fn get_entry(&self, name: &str) -> Result<&Entry, String> {
        self.catalog
            .entries
            .get(name)
            .ok_or_else(|| format!("Schema 未注册: {}", name))
    }

    /// 按名称获取 schema
    pub fn get(&self, name: &str) -> Result<&Schema, String> {
        self.get_entry(name).map(|e| &e.schema)
    }

    pub fn has(&self, name: &str) -> bool {
        self.catalog.entries.contains_key(name)
    }

    /// 清空 schema 注册表（`schemas` + `order`），对应绑定层的测试隔离 / 动态重建场景。
    ///
    /// 只清 schema，**不动**配置开关（`require_context` / `profile` / `rbac` /
    /// `role_rules`）与回调表（`clear_fns` 对称：各清各的）——开关生命周期属
    /// Registry 配置面，不随 schema 集合重建而丢。
    ///
    /// 非 base 视图调用即 `panic!`（消息携带 [`ERR_POLICY_VIEW_READONLY`] 前缀）——
    /// 本方法返回 `()`（受「core 公开签名零改动」约束，无法回 `Result`），守卫仍显式、不静默。
    pub fn clear(&mut self) {
        if let Err(e) = self.ensure_base_catalog() {
            panic!("{e}");
        }
        let catalog = Arc::make_mut(&mut self.catalog);
        catalog.entries.clear();
        catalog.order.clear();
    }

    /// 开关「上下文强制」（默认关闭 = fail-open，保持 JS parity）。
    /// 开启后：plan 入口遇 `ctx: None` 报 `ERR_NO_CONTEXT`（fail-secure）。
    /// 内部调用须显式传系统上下文（`{"internal": true}`）。
    pub fn set_require_context(&mut self, require: bool) {
        self.policy.require_context = require;
    }

    /// 「上下文强制」开关当前值
    pub fn require_context(&self) -> bool {
        self.policy.require_context
    }

    /// 设置查询档位（`standard` / `text2query`）。
    ///
    /// 档位为**单值状态**（非栈）：由 Host 的 `text2query()` 上下文管理器负责
    /// 进入时设档、退出时恢复（见 `py-store` / `nodejs-store` 门面）。判决一律在 core。
    pub fn set_profile(&mut self, profile: Profile) {
        self.policy.profile = profile;
    }

    /// 当前查询档位
    pub fn profile(&self) -> Profile {
        self.policy.profile
    }

    /// 豁免角色清单：命中者在一切判决环节（静态 + RBAC）直接放行。默认空——无豁免。
    pub fn set_exempt_roles(&mut self, roles: Vec<String>) {
        self.policy.role_rules.exempt_roles = roles;
    }

    /// 拒写角色清单：命中者一切写路径拒绝（读不受影响）。默认空——无拒写。
    pub fn set_deny_write_roles(&mut self, roles: Vec<String>) {
        self.policy.role_rules.deny_write_roles = roles;
    }

    /// schema 白名单缺失/为空时的默认姿态。默认 Open（保持现状语义）。
    pub fn set_unconfigured_policy(&mut self, policy: crate::permission::UnconfiguredPolicy) {
        self.policy.role_rules.unconfigured = policy;
    }

    /// 当前角色规则（静态判决函数与 RBAC decide 的共用取参入口）
    pub fn role_rules(&self) -> &crate::permission::RoleRules {
        &self.policy.role_rules
    }

    /// 定义层门禁策略（`closed=true` 时仅 internal 或 `roles` 白名单可注册/覆盖）。
    /// 默认 Open —— 全放行（保持既有 parity；须在首次业务注册前调用方生效于该次注册）。
    pub fn set_meta_policy(&mut self, closed: bool, roles: Vec<String>) {
        self.policy.meta_policy = crate::permission::MetaPolicy { closed, roles };
    }

    /// 当前定义层门禁策略
    pub fn meta_policy(&self) -> &crate::permission::MetaPolicy {
        &self.policy.meta_policy
    }

    /// 设置跨连接写策略（默认 [`WriteLinkPolicy::Reject`]）。
    ///
    /// 判决唯一在 core（[`crate::command::write_links::resolve_write_links`]）；
    /// 宿主仅需在初始化期把它透传到 registry。
    pub fn set_write_link_policy(&mut self, policy: WriteLinkPolicy) {
        self.policy.write_link_policy = policy;
    }

    /// 当前跨连接写策略
    pub fn write_link_policy(&self) -> WriteLinkPolicy {
        self.policy.write_link_policy
    }

    /// 只读定义层判决（`workflow` 等宿主侧定义面复用同一门禁；判决唯一在 core）。
    ///
    /// 不改变任何注册状态：Open 全放行；Closed 仅 `internal` 或 `roles` 白名单角色。
    /// 语义与 [`Self::register_with_ctx`] 的门禁判据**完全一致**（同一
    /// [`can_register`](crate::permission::can_register)），仅作只读暴露。
    pub fn can_register(&self, ctx: Option<&crate::permission::Context>) -> bool {
        crate::permission::can_register(&self.policy.meta_policy, ctx)
    }

    /// 注入/清除 RBAC 动态策略；`None` = 关闭（判决原语直通）。
    /// 解析失败 Err（fail-fast，禁静默吞配置错误），成功后判决链路即刻生效。
    pub fn set_rbac(&mut self, policy: Option<&Value>) -> Result<(), String> {
        match policy {
            None => {
                self.policy.rbac = None;
                Ok(())
            }
            Some(v) => {
                let p = crate::rbac::RbacPolicy::from_json(v)?;
                self.policy.rbac = Some(p);
                Ok(())
            }
        }
    }

    /// 当前 RBAC 策略（`None` = 未启用）
    pub fn rbac(&self) -> Option<&crate::rbac::RbacPolicy> {
        self.policy.rbac.as_ref()
    }

    /// 按定位四元组精确获取 schema（命令路由的唯一定位入口）
    ///
    /// 命中判据：`collection` 相等 **且** 四元组命中该 entry 的**任一链路**。
    /// **不做任何猜测回落**。
    pub fn get_by_location(
        &self,
        source: &str,
        database: Option<&str>,
        schema: Option<&str>,
        collection: &str,
    ) -> Result<&Schema, String> {
        let db = database.filter(|s| !s.is_empty());
        let sc = schema.filter(|s| !s.is_empty());
        self.catalog
            .entries
            .values()
            .find(|e| {
                e.schema.collection == collection
                    && e.links.iter().any(|l| {
                        l.source == source
                            && l.database.as_deref().filter(|s| !s.is_empty()) == db
                            && l.schema.as_deref().filter(|s| !s.is_empty()) == sc
                    })
            })
            .map(|e| &e.schema)
            .ok_or_else(|| {
                format!(
                    "Schema 未定位（source = {}, database = {:?}, schema = {:?}, collection = {}）",
                    source, db, sc, collection
                )
            })
    }

    /// 命令结构定位（方言翻译用）：优先四元组精确命中；未命中时（`route_override`
    /// 改写落点的多租户场景）回落 `(source, collection)` 定位 —— 结构 schema
    /// 与定位落点正交（见 multi-datasource-routing-plan.md §6：权限/字段校验仍按结构 schema）。
    ///
    /// `(source, collection)` 恰一候选 → 命中；多候选时取 `database`/`schema` 均 `null` 的
    /// 结构声明，无该声明 → 报错（拒绝猜测，铁律 3）。
    pub fn get_for_command(
        &self,
        source: &str,
        database: Option<&str>,
        schema: Option<&str>,
        collection: &str,
    ) -> Result<&Schema, String> {
        if let Ok(s) = self.get_by_location(source, database, schema, collection) {
            return Ok(s);
        }
        let candidates: Vec<&Entry> = self
            .catalog
            .entries
            .values()
            .filter(|e| {
                e.schema.collection == collection && e.links.iter().any(|l| l.source == source)
            })
            .collect();
        match candidates.len() {
            0 => self.get_by_location(source, database, schema, collection),
            1 => Ok(&candidates[0].schema),
            _ => candidates
                .iter()
                .find(|e| e.schema.database().is_none() && e.schema.schema().is_none())
                .map(|e| &e.schema)
                .ok_or_else(|| {
                    format!(
                        "命令定位 ({}, {:?}, {:?}, {}) 存在多个结构 schema（多租户 override 需唯一的 (source, collection) 结构声明）",
                        source, database, schema, collection
                    )
                }),
        }
    }

    pub fn list(&self) -> Vec<String> {
        self.catalog.order.clone()
    }

    /// schema 绑定的数据源名（= 主链路 source；`default` → `None`）
    pub fn schema_datasource(&self, name: &str) -> Result<Option<String>, String> {
        Ok(self.get(name)?.source.clone())
    }

    /// 解析 schema 绑定的数据源（`config` 为 `{ "sources": { name: kind } }`）
    ///
    /// 纯逻辑：不持有连接（连接句柄留在 Host）。`null` / 空配置下等价单源 Mongo，
    /// 保证既有调用方与 parity fixture 零变更。见 [`crate::datasource`]。
    pub fn resolve_datasource(
        &self,
        schema_name: &str,
        config: &Value,
    ) -> Result<crate::datasource::DataSource, String> {
        let declared = self.get(schema_name)?.source.as_deref();
        let cfg = crate::datasource::DataSourceConfig::from_json(config)?;
        cfg.resolve(declared)
    }
}

/// 由原始定义构建 [`Schema`]（字段/关系/计算列/元数据分块解析，保持单一职责）。
/// 落点（`source`/`database`/`schema`）由 `loc` 注入（定义文件零落点）。
fn build_schema(
    obj: &Map<String, Value>,
    name: String,
    collection: String,
    loc: &Location,
) -> Result<Schema, String> {
    let timestamps = parse_timestamps(obj)?;
    let mut fields = normalize_fields(obj.get("fields"))?;
    // 向量字段注册期校验（A1）：任一步 Err ⇒ 零副作用
    validate_vector_fields(&fields)?;
    if timestamps {
        add_timestamp_fields(&mut fields);
    }
    let relations = parse_relations(obj);
    // 计算列的 agg 需要校验「路径首段命中关系或数组字段」→ 必须晚于 relations/fields 解析
    let computes = parse_computes(obj, &relations, &fields)?;
    // 写触发器声明（onFields 校验依赖已归一 fields；非法声明注册期 Err）
    let triggers = super::triggers::parse_triggers(obj, &name, &fields)?;
    // `_score` 是向量检索评分列（04），与 `fields ∪ relations ∪ computes` 同属结果文档命名空间；
    // 字面保留名必须在注册期拒绝，避免用户字段被评分列覆盖或反之。
    // 注意不能用 `RESERVED_KEYS`：其 `norm_key` 会把 `_score` 归一为 `score`，
    // 连带误拒合法字段名 `score`（存量 schema 常见）。
    for (k, ns) in fields
        .keys()
        .map(|k| (k, "fields"))
        .chain(relations.keys().map(|k| (k, "relations")))
        .chain(computes.iter().map(|(k, _)| (k, "computes")))
    {
        if k == "_score" {
            return Err(format!(
                "{}名称 \"_score\" 为向量检索评分保留键，不得在 schema {ns} 中声明",
                crate::naming::ERR_NAME_CONFLICT
            ));
        }
    }
    // 命名冲突检测（A7，分组口径）：`detect_conflicts` 的 `seen` 是调用内局部量，
    // 分组调用天然实现「组内 + 指定跨组」拦截：
    //   ① fields / relations / computes 各自组内归一撞名 ⇒ 报错（不变）；
    //   ② fields ∩ relations、relations ∩ computes ⇒ 报错（不变）；
    //   ③ fields ∩ computes **放行**：computed 值拥有同名键的输出所有权，fields 仍保留
    //      该键的类型声明（写入校验 / DDL / `$inc` 等原生增量更新），两者职责不同层。
    //      nodejs-store 的 schema 注册不做此检测，py-store / go-store 由此对齐同名覆盖语义。
    // `_id` 已在 detect_conflicts 内豁免；`_score` 由上方字面拦截。
    // fields 键空间完全开放、无字段名保留字——`type` 等契约键作字段名合法，
    // 契约键位于字段定义对象内部，与本层标识符不同层无冲突。
    crate::naming::detect_conflicts(fields.keys().map(String::as_str))?;
    crate::naming::detect_conflicts(relations.keys().map(String::as_str))?;
    crate::naming::detect_conflicts(computes.iter().map(|(k, _)| k.as_str()))?;
    crate::naming::detect_conflicts(
        fields
            .keys()
            .map(String::as_str)
            .chain(relations.keys().map(String::as_str)),
    )?;
    crate::naming::detect_conflicts(
        relations
            .keys()
            .map(String::as_str)
            .chain(computes.iter().map(|(k, _)| k.as_str())),
    )?;
    // accessKeys（声明式行级权限）：外层键必须在已声明字段中，read/write/bypassRoles
    // 形态非法即注册期 Err（禁静默回落）。归档模型不继承（与 ownerKey 现状一致）。
    let access_keys = super::access_keys::parse_access_keys(obj, &fields)?;
    // ownerKey（漉金Core 接入扩展 01）：归属键，缺省 = `createdBy`（零破坏）；
    // 存在时必须为非空字符串，非法注册期显式 Err（禁静默回落）。
    // 不要求键必须出现在 fields 中（`_id` 等系统键合法）。
    let owner_key = match obj.get("ownerKey") {
        None | Some(Value::Null) => None,
        Some(Value::String(s)) if !s.is_empty() => Some(s.clone()),
        Some(other) => {
            return Err(format!(
                "schema \"{name}\" 的 ownerKey 必须为非空字符串，实际 {other}"
            ))
        }
    };
    // archive（漉金Core 接入扩展 02）：归档开关，可选 bool，缺省 true（零破坏）；
    // false 时注册期不派生 `<Name>Deleted` 归档模型（plan_remove 按 registry.has
    // 判定，自然跳过归档副作用，删除行为本身不变）。
    match obj.get("archive") {
        None | Some(Value::Null) | Some(Value::Bool(_)) => {}
        Some(other) => {
            return Err(format!(
                "schema \"{name}\" 的 archive 必须为可选布尔值，实际 {other}"
            ))
        }
    }
    Ok(Schema {
        name,
        collection,
        id_prefix: obj
            .get("idPrefix")
            .and_then(|v| v.as_str())
            .unwrap_or("")
            .to_string(),
        timestamps,
        fields,
        relations,
        computes,
        triggers,
        read: str_list(obj.get("read")),
        write: str_list(obj.get("write")),
        indexes: obj
            .get("indexes")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default(),
        source: if loc.source == crate::datasource::DEFAULT_SOURCE {
            None
        } else {
            Some(loc.source.clone())
        },
        database: loc.database.clone().filter(|s| !s.is_empty()),
        schema: loc.schema.clone().filter(|s| !s.is_empty()),
        owner_key,
        access_keys,
    })
}

/// `timestamps` 仅接受 true/false/'ms'/'s'；单位换算在 Host（core 无时钟，now 由 Host 传入）
fn parse_timestamps(obj: &Map<String, Value>) -> Result<bool, String> {
    match obj.get("timestamps") {
        None | Some(Value::Null) | Some(Value::Bool(true)) => Ok(true),
        Some(Value::Bool(false)) => Ok(false),
        Some(Value::String(s)) if s == "ms" || s == "s" => Ok(true),
        Some(Value::String(s)) => Err(format!(
            "timestamps 仅支持 true/false/'ms'/'s'，实际 {:?}",
            s
        )),
        Some(other) => Err(format!(
            "timestamps 仅支持 true/false/'ms'/'s'，实际 {}",
            other
        )),
    }
}

/// 自动补时间戳字段（`timestamps !== false` 时；已声明则不覆盖）
fn add_timestamp_fields(fields: &mut HashMap<String, FieldDef>) {
    for key in ["createdAt", "updatedAt"] {
        fields.entry(key.to_string()).or_insert_with(|| FieldDef {
            field_type: "number".to_string(),
            required: false,
            default: None,
            read: None,
            write: None,
            fields: None,
            strategy: None,
            dim: None,
            metric: None,
        });
    }
}

fn parse_relations(obj: &Map<String, Value>) -> HashMap<String, RelationDef> {
    let mut relations = HashMap::new();
    if let Some(Value::Object(rm)) = obj.get("relations") {
        for (key, val) in rm {
            let vo = val.as_object();
            let get_str = |k: &str| -> Option<String> {
                vo.and_then(|o| o.get(k))
                    .and_then(|v| v.as_str())
                    .filter(|s| !s.is_empty())
                    .map(String::from)
            };
            relations.insert(
                key.clone(),
                RelationDef {
                    model: get_str("model").unwrap_or_default(),
                    rel_type: get_str("type").unwrap_or_else(|| "many".to_string()),
                    local_field: get_str("localField").unwrap_or_else(|| "_id".to_string()),
                    foreign_field: get_str("foreignField").unwrap_or_else(|| key.clone()),
                    read: str_list(vo.and_then(|o| o.get("read"))),
                },
            );
        }
    }
    relations
}

fn parse_computes(
    obj: &Map<String, Value>,
    relations: &HashMap<String, RelationDef>,
    fields: &HashMap<String, FieldDef>,
) -> Result<Vec<(String, ComputeDef)>, String> {
    let mut computes = Vec::new();
    if let Some(Value::Object(cm)) = obj.get("computes") {
        for (key, val) in cm {
            let vo = val.as_object();
            let get = |k: &str| vo.and_then(|o| o.get(k)).cloned();
            let has_fn = get("fn").map(|v| is_truthy(&v)).unwrap_or(false);
            let has_async_fn = get("asyncFn").map(|v| is_truthy(&v)).unwrap_or(false);
            let agg = parse_agg(
                key,
                get("agg").as_ref(),
                relations,
                fields,
                str_list(get("depends").as_ref())
                    .unwrap_or_default()
                    .as_slice(),
                has_fn || has_async_fn,
            )?;
            computes.push((
                key.clone(),
                ComputeDef {
                    comp_type: get("type")
                        .and_then(|v| v.as_str().map(String::from))
                        .filter(|s| !s.is_empty())
                        .unwrap_or_else(|| "any".to_string()),
                    has_fn,
                    has_async_fn,
                    fn_ref: get("fnRef")
                        .and_then(|v| v.as_str().map(String::from))
                        .filter(|s| !s.is_empty()),
                    agg,
                    depends: str_list(get("depends").as_ref()).unwrap_or_default(),
                    read: str_list(get("read").as_ref()),
                },
            ));
        }
    }
    Ok(computes)
}

/// 解析并校验归一聚合计算列（§9.2(2) + 漉金Core 接入扩展 04）。
///
/// - 算子白名单 [`COMPUTE_AGG_OPS`]，不在白名单 → Err（绝不静默）；
/// - agg 路径首段二分：
///   - 命中 **relations** → relation-agg（现状不变）：`$count` 仅关系整行计数
///     （`lessons`，不接受字段），其余算子必须带字段（`lessons.duration`）；
///   - 命中 **fields 且为数组类型** → inline-array agg：`$size` 仅数组本身
///     （`bidders`，不接受元素字段），其余算子必须带单级元素字段
///     （`bidders.amount`）；`$count` 不适用（文档内数组计数用 `$size`）；
///     且 `depends` 必须包含路径首段（宿主本地重算依据，缺失报错不兜底）；
///   - 首段命中字段但非数组类型 / 两者皆未命中 → Err；
/// - 单级元素字段即止（`a.b.c` 中 b 为数组的深层再入数组不做，§7）；
/// - 与 `fn` / `asyncFn` 互斥。
fn parse_agg(
    key: &str,
    raw: Option<&Value>,
    relations: &HashMap<String, RelationDef>,
    fields: &HashMap<String, FieldDef>,
    depends: &[String],
    has_callback: bool,
) -> Result<Option<Value>, String> {
    let Some(v) = raw.filter(|v| is_truthy(v)) else {
        return Ok(None);
    };
    if has_callback {
        return Err(format!(
            "计算列 \"{key}\" 不能同时声明 agg 与 fn/asyncFn（二选一）"
        ));
    }
    let obj = v.as_object().ok_or_else(|| {
        format!("计算列 \"{key}\" 的 agg 必须是对象，如 {{\"$count\": \"lessons\"}}")
    })?;
    let mut it = obj.iter();
    let (op, arg) = match (it.next(), it.next()) {
        (Some(kv), None) => kv,
        _ => return Err(format!("计算列 \"{key}\" 的 agg 必须且只能声明一个算子键")),
    };
    if !COMPUTE_AGG_OPS.contains(&op.as_str()) {
        return Err(format!(
            "计算列 \"{key}\" 的 agg 算子 {op} 不在白名单（$count/$size/$sum/$avg/$min/$max/$countDistinct）"
        ));
    }
    let path = arg
        .as_str()
        .filter(|s| !s.is_empty() && *s != "*")
        .ok_or_else(|| {
            format!("计算列 \"{key}\" 的 agg 取值必须是路径字符串（如 \"lessons\" / \"lessons.duration\" / \"bidders.amount\"）")
        })?;
    let (rel, field) = match path.split_once('.') {
        Some((r, f)) => (r, Some(f)),
        None => (path, None),
    };
    if field
        .map(|f| f.is_empty() || f.contains('.'))
        .unwrap_or(false)
    {
        return Err(format!(
            "计算列 \"{key}\" 的 agg 路径 \"{path}\" 非法：仅支持「关系.字段」/「数组.元素字段」单级路径（深层路径首批不支持）"
        ));
    }
    if relations.contains_key(rel) {
        // relation-agg：现状校验原样（04 红线：relation 分支零改动）
        match op.as_str() {
            "$count" if field.is_some() => Err(format!(
                "计算列 \"{key}\" 的 $count 仅支持关系整行计数（如 \"lessons\"），按字段计数待 P5"
            )),
            "$count" => Ok(Some(v.clone())),
            _ if field.is_none() => Err(format!(
                "计算列 \"{key}\" 的 agg 算子 {op} 必须引用关系字段（如 \"lessons.duration\"）"
            )),
            _ => Ok(Some(v.clone())),
        }
    } else {
        match fields.get(rel) {
            Some(f) if f.field_type == "array" => {
                // inline-array agg
                if !depends.iter().any(|d| d == rel) {
                    return Err(format!(
                        "计算列 \"{key}\" 的 depends 必须包含 inline agg 路径首段 \"{rel}\"（宿主本地重算依据）"
                    ));
                }
                match op.as_str() {
                    "$size" if field.is_some() => Err(format!(
                        "计算列 \"{key}\" 的 $size 仅支持数组本身（如 \"{rel}\"），不接受元素字段"
                    )),
                    "$size" => Ok(Some(v.clone())),
                    "$count" => Err(format!(
                        "计算列 \"{key}\" 的 $count 不适用于数组字段：文档内数组计数用 $size（如 {{\"$size\": \"{rel}\"}}）"
                    )),
                    _ if field.is_none() => Err(format!(
                        "计算列 \"{key}\" 的 agg 算子 {op} 必须引用数组元素字段（如 \"{rel}.amount\"）"
                    )),
                    _ => Ok(Some(v.clone())),
                }
            }
            Some(f) => Err(format!(
                "计算列 \"{key}\" 的 agg 路径首段 \"{rel}\" 是字段但类型为 \"{}\"（非数组），inline agg 仅支持数组字段",
                f.field_type
            )),
            None => Err(format!(
                "计算列 \"{key}\" 的 agg 引用的路径首段 \"{rel}\" 既非已定义关系也非数组字段"
            )),
        }
    }
}

/// 构造 `<Name>Deleted` 归档表 schema 定义（字段 = 原字段 + `deletedAt`）。
/// 归档表与原表同落点 —— 落点不由定义文件携带，改由批次继承主链路（见 `register_batch`）。
fn archive_defn(obj: &Map<String, Value>, name: &str, collection: &str) -> Value {
    let mut arch_fields = obj
        .get("fields")
        .and_then(|v| v.as_object())
        .cloned()
        .unwrap_or_default();
    arch_fields.insert("deletedAt".to_string(), json!({ "type": "number" }));
    // 剔除 `_id` 的自增策略：归档表显式拷贝源行 `_id` 值（upsert-by-_id 幂等），
    // 不走数据库自增；DDL 侧也据此保持归档表 `_id` 为普通主键列
    if let Some(idf) = arch_fields.get_mut("_id") {
        if let Some(o) = idf.as_object_mut() {
            o.remove("strategy");
        }
    }
    let arch = json!({
        "name": format!("{}Deleted", name),
        "collection": format!("{}_deleted", collection),
        "idPrefix": "",
        "_isArchive": true,
        "fields": Value::Object(arch_fields),
        "indexes": obj.get("indexes").cloned().unwrap_or_else(|| json!([])),
        // 归档表继承主表 timestamps 语义（true/false/'ms'/'s'；缺省 = true 同 parse_timestamps）。
        // 此前派生 def 丢失该键 ⇒ 归档表恒按 true 注入 createdAt/updatedAt，主表为
        // timestamps:false 且字段含 created_at/updated_at（introspect 同步 snake_case 存量库）
        // 时归档构建即 ERR_NAME_CONFLICT（既有缺陷，4.9.0 修复）。
        "timestamps": obj.get("timestamps").cloned().unwrap_or(json!(true)),
    });
    arch
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn write_link_policy_defaults_to_reject() {
        let r = Registry::new();
        assert_eq!(r.write_link_policy(), WriteLinkPolicy::Reject);
        let mut r2 = Registry::new();
        assert_eq!(r2.write_link_policy(), WriteLinkPolicy::Reject);
        r2.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);
        assert_eq!(r2.write_link_policy(), WriteLinkPolicy::PrimaryOnly);
    }

    // ─── R2 D1：目录快照 + 策略束 + 派生视图（V4 落地） ──────────────

    use crate::permission::UnconfiguredPolicy;

    /// T1 全覆盖覆盖项（六字段悉数改写；`rbac: null` = 显式清除）
    fn overrides_all() -> PolicyOverrides {
        PolicyOverrides::from_value(&json!({
            "profile": "standard",
            "requireContext": false,
            "rbac": null,
            "roleRules": { "exemptRoles": ["x"], "denyWriteRoles": ["y"], "unconfigured": "open" },
            "metaPolicy": { "closed": false, "roles": [] },
            "writeLinkPolicy": "reject"
        }))
        .expect("合法覆盖 JSON")
    }

    /// T1：with_policy 覆盖矩阵——「缺省继承」与「全覆盖」两态 + base 零污染
    #[test]
    fn with_policy_override_matrix() {
        let mut base = Registry::new();
        base.set_require_context(true);
        base.set_profile(Profile::Text2Query);
        base.set_rbac(Some(&json!({}))).expect("合法 RBAC");
        base.set_exempt_roles(vec!["a".to_string()]);
        base.set_deny_write_roles(vec!["b".to_string()]);
        base.set_unconfigured_policy(UnconfiguredPolicy::Closed);
        base.set_meta_policy(true, vec!["m".to_string()]);
        base.set_write_link_policy(WriteLinkPolicy::PrimaryOnly);

        // 缺省继承态：空覆盖 = 全继承来源视图
        let inherit = base.with_policy(&PolicyOverrides::default());
        assert!(inherit.is_view(), "派生视图 is_view 应为真");
        assert!(inherit.require_context());
        assert_eq!(inherit.profile(), Profile::Text2Query);
        assert!(inherit.rbac().is_some());
        assert_eq!(inherit.role_rules().exempt_roles, vec!["a".to_string()]);
        assert_eq!(inherit.role_rules().deny_write_roles, vec!["b".to_string()]);
        assert_eq!(
            inherit.role_rules().unconfigured,
            UnconfiguredPolicy::Closed
        );
        assert!(inherit.meta_policy().closed);
        assert_eq!(inherit.meta_policy().roles, vec!["m".to_string()]);
        assert_eq!(inherit.write_link_policy(), WriteLinkPolicy::PrimaryOnly);

        // 全覆盖态：六字段全部改写（rbac 走 Some(None) 清除）
        let full = base.with_policy(&overrides_all());
        assert!(!full.require_context());
        assert_eq!(full.profile(), Profile::Standard);
        assert!(full.rbac().is_none(), "rbac:null 应清除");
        assert_eq!(full.role_rules().exempt_roles, vec!["x".to_string()]);
        assert_eq!(full.role_rules().deny_write_roles, vec!["y".to_string()]);
        assert_eq!(full.role_rules().unconfigured, UnconfiguredPolicy::Open);
        assert!(!full.meta_policy().closed);
        assert_eq!(full.write_link_policy(), WriteLinkPolicy::Reject);

        // base 零污染
        assert!(base.require_context());
        assert_eq!(base.profile(), Profile::Text2Query);
        assert!(base.rbac().is_some());
        assert!(base.meta_policy().closed);
    }

    /// T2a：视图注册类方法报 `ERR_POLICY_VIEW_READONLY`（前缀可匹配）
    #[test]
    fn view_register_is_readonly_err() {
        let base = Registry::new();
        let mut view = base.with_policy(&PolicyOverrides::default());
        let err = view
            .register(&json!({ "name": "A", "fields": {} }))
            .expect_err("视图注册应 Err");
        assert!(
            err.starts_with(ERR_POLICY_VIEW_READONLY),
            "前缀可匹配: {err}"
        );
        let err2 = view
            .register_with_ctx(&json!({ "name": "A", "fields": {} }), None)
            .expect_err("视图注册应 Err");
        assert!(
            err2.starts_with(ERR_POLICY_VIEW_READONLY),
            "前缀可匹配: {err2}"
        );
        let err3 = view
            .register_batch(
                &[(json!({ "name": "A", "fields": {} }), Location::default())],
                None,
            )
            .expect_err("视图批量注册应 Err");
        assert!(
            err3.starts_with(ERR_POLICY_VIEW_READONLY),
            "前缀可匹配: {err3}"
        );
    }

    /// T2b：视图 `clear`（返回 `()`，受签名零改动约束）以 panic 显式守卫
    #[test]
    #[should_panic(expected = "ERR_POLICY_VIEW_READONLY")]
    fn view_clear_panics_readonly() {
        let base = Registry::new();
        let mut view = base.with_policy(&PolicyOverrides::default());
        view.clear();
    }

    /// fields 键空间完全开放（原初设计修复回归）：契约键 `type` 作字段名合法，
    /// 类型声明在定义对象内部的 `type` 键——不同对象层，注册成功且定义保真。
    #[test]
    fn field_named_type_is_allowed() {
        let mut r = Registry::new();
        r.register(&json!({
            "name": "Msg", "collection": "msg", "timestamps": false,
            "fields": {
                "_id": { "type": "string" },
                "type": { "type": "string", "default": "text" },
                "content": { "type": "string" }
            },
            "relations": {}
        }))
        .expect("字段名 type 合法，注册应成功");
        let s = r.get("Msg").expect("可取");
        let f = s.fields.get("type").expect("type 字段存在");
        assert_eq!(f.field_type, "string");
        assert_eq!(f.default, Some(json!("text")));
    }

    /// 归档派生继承主 def 的 timestamps 语义（4.9.0 修复回归）：timestamps:false +
    /// snake_case 时间戳列（introspect 同步存量库场景）。修复前派生 def 丢 timestamps
    /// 键 ⇒ 归档表恒按 true 注入 createdAt/updatedAt，与主表 created_at/updated_at
    /// 归一撞名 ERR_NAME_CONFLICT。
    #[test]
    fn archive_derivation_inherits_timestamps() {
        let mut r = Registry::new();
        r.register_batch(
            &[(
                json!({
                    "name": "Vector", "collection": "vector", "timestamps": false,
                    "fields": {
                        "_id": { "type": "string" },
                        "created_at": { "type": "string" },
                        "updated_at": { "type": "string" }
                    }
                }),
                Location::default(),
            )],
            None,
        )
        .expect("timestamps:false + snake_case 时间戳列应注册成功（归档派生不撞名）");
        // 主表：不自动注入
        let s = r.get("Vector").expect("主表可取");
        assert!(!s.timestamps);
        assert!(s.fields.contains_key("created_at"));
        assert!(
            !s.fields.contains_key("createdAt"),
            "timestamps:false 不注入 createdAt"
        );
        // 归档表：继承 timestamps:false，保留主表字段且不自动注入
        let arch = r.get("VectorDeleted").expect("归档表自动派生");
        assert_eq!(arch.collection, "vector_deleted");
        assert!(!arch.timestamps, "归档表应继承主表 timestamps:false");
        assert!(arch.fields.contains_key("created_at"), "归档表保留主表字段");
        assert!(
            !arch.fields.contains_key("createdAt"),
            "归档表继承 timestamps:false，不注入 createdAt"
        );
    }

    /// T3：目录快照——派生视图见 base 已注册目录；快照不承诺 base 后续注册可见
    #[test]
    fn view_sees_catalog_snapshot() {
        let mut base = Registry::new();
        base.register(&json!({ "name": "A", "fields": { "x": { "type": "number" } } }))
            .expect("base 注册成功");
        let view = base.with_policy(&PolicyOverrides::default());
        assert!(view.has("A"), "派生视图应见 base 已注册目录");
        assert_eq!(view.get("A").expect("视图可取").name, "A");
        assert_eq!(view.list(), base.list());

        // 快照语义：在途视图不承诺看到 base 后续注册；新派生视图可见
        let mut base2 = Registry::new();
        base2
            .register(&json!({ "name": "A", "fields": {} }))
            .unwrap();
        let snap = base2.with_policy(&PolicyOverrides::default());
        base2
            .register(&json!({ "name": "B", "fields": {} }))
            .unwrap();
        assert!(snap.has("A"), "在途视图仍持其快照");
        let fresh = base2.with_policy(&PolicyOverrides::default());
        assert!(fresh.has("B"), "新派生视图应见 base 后续注册");
    }

    /// T4：视图 `set_*` 只影响自身（base 与兄弟视图不受影响）
    #[test]
    fn view_policy_set_isolated() {
        let base = Registry::new();
        let mut view = base.with_policy(&PolicyOverrides::default());
        view.set_profile(Profile::Text2Query);
        assert_eq!(view.profile(), Profile::Text2Query);
        assert_eq!(base.profile(), Profile::Standard, "视图 set 不影响 base");
        let sibling = base.with_policy(&PolicyOverrides::default());
        assert_eq!(sibling.profile(), Profile::Standard, "不影响兄弟视图");
    }

    /// T5：`from_value` 未知键 / 非法值 Err；`rbac:null` 清除、缺键继承、空对象合法
    #[test]
    fn from_value_unknown_key_and_rbac_null() {
        let e = PolicyOverrides::from_value(&json!({ "nope": 1 })).expect_err("未知键应 Err");
        assert!(e.to_string().starts_with(ERR_POLICY_VIEW_READONLY), "{e}");
        assert!(
            PolicyOverrides::from_value(&json!({ "profile": "bogus" })).is_err(),
            "非法 profile 应 Err（禁静默回落）"
        );
        assert!(
            PolicyOverrides::from_value(&json!({ "writeLinkPolicy": "bogus" })).is_err(),
            "非法 writeLinkPolicy 应 Err"
        );
        assert!(
            PolicyOverrides::from_value(&json!({})).is_ok(),
            "空对象合法"
        );

        let mut base = Registry::new();
        base.set_rbac(Some(&json!({}))).unwrap();
        let cleared =
            base.with_policy(&PolicyOverrides::from_value(&json!({ "rbac": null })).unwrap());
        assert!(cleared.rbac().is_none(), "rbac:null 应清除");
        let kept = base.with_policy(&PolicyOverrides::from_value(&json!({})).unwrap());
        assert!(kept.rbac().is_some(), "缺键应继承");
        assert!(base.rbac().is_some(), "base 不受影响");
    }

    // ─── 04 inline-array agg：parse_agg 路径首段二分校验 ─────────────

    /// 仅解析 computes（供 parse_agg 各分支直接断言，不经 Registry 全链路）
    fn agg_computes(obj: &Value) -> Result<Vec<(String, ComputeDef)>, String> {
        let o = obj.as_object().expect("obj");
        let rels = parse_relations(o);
        let fields = normalize_fields(o.get("fields"))?;
        parse_computes(o, &rels, &fields)
    }

    /// 七算子 inline 合法形态全通过（$size/$max/$countDistinct/$sum/$avg/$min）
    #[test]
    fn inline_agg_all_ops_ok() {
        let def = json!({
            "name": "Auction", "collection": "auctions", "timestamps": false,
            "fields": { "bidders": { "type": "array" } },
            "relations": {},
            "computes": {
                "biddersCount": { "type": "int", "agg": { "$size": "bidders" }, "depends": ["bidders"] },
                "maxBid": { "type": "number", "agg": { "$max": "bidders.amount" }, "depends": ["bidders"] },
                "bidderCount": { "type": "int", "agg": { "$countDistinct": "bidders.bidderId" }, "depends": ["bidders"] },
                "sumBid": { "type": "number", "agg": { "$sum": "bidders.amount" }, "depends": ["bidders"] },
                "avgBid": { "type": "number", "agg": { "$avg": "bidders.amount" }, "depends": ["bidders"] },
                "minBid": { "type": "number", "agg": { "$min": "bidders.amount" }, "depends": ["bidders"] }
            }
        });
        let computes = agg_computes(&def).expect("六算子 inline agg 均合法");
        assert_eq!(computes.len(), 6);
        for (_, c) in &computes {
            assert!(c.agg.is_some(), "inline agg 应保留");
        }
    }

    /// inline agg 的 depends 必须包含路径首段（宿主本地重算依据）
    #[test]
    fn inline_agg_without_depends_head_is_err() {
        let def = json!({
            "name": "A", "collection": "a", "timestamps": false,
            "fields": { "tags": { "type": "array" } },
            "relations": {},
            "computes": { "n": { "type": "int", "agg": { "$size": "tags" } } }
        });
        let err = agg_computes(&def).expect_err("缺 depends 应 Err");
        assert!(err.contains("depends"), "{err}");
    }

    /// $size 仅支持数组本身，不接受元素字段
    #[test]
    fn inline_agg_size_with_element_field_is_err() {
        let def = json!({
            "name": "A", "collection": "a", "timestamps": false,
            "fields": { "tags": { "type": "array" } },
            "relations": {},
            "computes": { "n": { "type": "int", "agg": { "$size": "tags.x" }, "depends": ["tags"] } }
        });
        let err = agg_computes(&def).expect_err("$size 带元素字段应 Err");
        assert!(err.contains("仅支持数组本身"), "{err}");
    }

    /// 文档内数组计数用 $size，$count 不适用于数组字段
    #[test]
    fn inline_agg_count_is_err() {
        let def = json!({
            "name": "A", "collection": "a", "timestamps": false,
            "fields": { "tags": { "type": "array" } },
            "relations": {},
            "computes": { "n": { "type": "int", "agg": { "$count": "tags" }, "depends": ["tags"] } }
        });
        let err = agg_computes(&def).expect_err("数组 $count 应 Err");
        assert!(err.contains("$size"), "{err}");
    }

    /// $max 等元素算子必须引用数组元素字段
    #[test]
    fn inline_agg_element_op_without_field_is_err() {
        let def = json!({
            "name": "A", "collection": "a", "timestamps": false,
            "fields": { "tags": { "type": "array" } },
            "relations": {},
            "computes": { "n": { "type": "int", "agg": { "$max": "tags" }, "depends": ["tags"] } }
        });
        let err = agg_computes(&def).expect_err("$max 无元素字段应 Err");
        assert!(err.contains("元素字段"), "{err}");
    }

    /// 路径首段是字段但类型非数组 → Err
    #[test]
    fn agg_head_non_array_field_is_err() {
        let def = json!({
            "name": "A", "collection": "a", "timestamps": false,
            "fields": { "state": { "type": "string" } },
            "relations": {},
            "computes": { "n": { "type": "int", "agg": { "$size": "state" }, "depends": ["state"] } }
        });
        let err = agg_computes(&def).expect_err("非数组字段应 Err");
        assert!(err.contains("非数组"), "{err}");
    }

    /// 路径首段既非关系也非字段 → Err
    #[test]
    fn agg_head_unknown_is_err() {
        let def = json!({
            "name": "A", "collection": "a", "timestamps": false,
            "fields": { "state": { "type": "string" } },
            "relations": {},
            "computes": { "n": { "type": "int", "agg": { "$size": "ghost" }, "depends": ["ghost"] } }
        });
        let err = agg_computes(&def).expect_err("未知首段应 Err");
        assert!(err.contains("既非已定义关系也非数组字段"), "{err}");
    }

    /// relation 分支零改动红线：$count 整行 / $sum 字段合法；$count 带字段仍 Err；
    /// relation 不做 inline 的 depends 校验
    #[test]
    fn relation_agg_unchanged_by_inline_extension() {
        let def = json!({
            "name": "Shop", "collection": "shops", "timestamps": false,
            "fields": { "name": { "type": "string" } },
            "relations": { "orders": { "model": "Order", "type": "many", "localField": "_id", "foreignField": "shopId" } },
            "computes": {
                "orderCount": { "type": "int", "agg": { "$count": "orders" } },
                "totalAmount": { "type": "number", "agg": { "$sum": "orders.amount" } }
            }
        });
        let computes = agg_computes(&def).expect("relation agg 不受 04 扩展影响");
        assert_eq!(computes.len(), 2);

        let bad = json!({
            "name": "Shop", "collection": "shops", "timestamps": false,
            "fields": { "name": { "type": "string" } },
            "relations": { "orders": { "model": "Order", "type": "many", "localField": "_id", "foreignField": "shopId" } },
            "computes": { "n": { "type": "int", "agg": { "$count": "orders.orderId" } } }
        });
        let err = agg_computes(&bad).expect_err("relation $count 带字段仍应 Err");
        assert!(err.contains("整行计数"), "{err}");
    }
}