# -*- coding: utf-8 -*-
"""位置权重矩阵（序标定）——「不同身份，参数偏好不同」。

理论出处（本仓原文，非外部知识）
--------------------------------
`docs/theory/智能的公理化基石.md` §十「信任：对协作者行为的概率估计」（该文档章节用中文序号）给出信任的**定义**；
其**合成式**的出处＝`docs/swarm/蜂群互联_v0.1.md` §4.3（该文档用阿拉伯数字章节号；下文引用的 §4.3 均指它）：
    P_trust = f(一致性, 位置可预测性, 版本对齐度)
其中 `版本对齐度` 已由 `theory` + `links.version_alignment` 提供，
`位置可预测性` 见 `predict`，`一致性` 见 `consistency`。本模块只补一件
此前缺失的东西：**三个分量在不同位置下的偏好序**。

六个位置分为**两类**（由功能身份决定，非人为指定）：

  · 认知视角（viewpoint）：设计者 / 反思 / 验证
    ——「思考的不同视角」，**三分量都涉及**，只标定主导，**不设零**。
  · 功能单元（functional）：记录 / 输出 / 维生
    ——功能需求决定，**身份即排除**：主导 + 明确排除一项（≈0）。

排除项可**推导**（派生律：功能身份 → 缺失分量），不是选出来的：

  · 记录单元**不猜**      → 不做预测 → 排除 位置可预测性
  · 输出单元**不记**      → 不复现对账 → 排除 一致性
  · 维生系统**不对外**    → 无外部协议 → 排除 版本对齐度

设计者定位（本轮裁决）：
  · **不参与日常评估**——它是元视角，思考总体规律；细节由反思/验证承担，
    否则要做全局细节整理，耗时极长。

可推导不变量（`invariants()` 逐条自检）
--------------------------------------
  1. **排除唯一**——每个分量恰被 **1** 个位置排除；因视角「3 项都涉及」，
     故每个分量的参与度 = **5/6**（不是 3/6）；
  2. **排除项只属功能单元**，且 A/B/C 各被排除一次；
  3. **视角主导 ↔ 功能单元排除**一一配对：
     设计者(A)↔维生、反思(B)↔记录、验证(C)↔输出。

诚实边界
--------
· 本模块只声明**序**（谁能压过谁），**不宣称数值**——与 `links.py` 的
  `UP_STEP`/`DECAY_DAYS` 同一纪律（蜂群互联 §4.3：「只声明结构，不宣称权重数值」）。
· `blend()` 里的权重是**占位值**（用序秩当量级），仅供仿真，标定属 v0.3。

零第三方依赖。
"""
from __future__ import annotations

import math
import os
import time

from . import protect                       # #431：条目口径保护单点
from .fsutil import append_jsonl, read_jsonl

# --------------------------------------------------------------------------
# 常量
# --------------------------------------------------------------------------

#: 三个分量（顺序即 A/B/C，稳定，供外部按位引用）
COMPONENTS = ("version_alignment", "predictability", "consistency")

#: 分量中文名（日志/自描述用）
COMPONENT_LABELS = {
    "version_alignment": "版本对齐度",
    "predictability": "位置可预测性",
    "consistency": "一致性",
}

#: 位置类别
VIEWPOINT = "viewpoint"      # 认知视角：3 项都涉及，只标主导，不设零
FUNCTIONAL = "functional"    # 功能单元：主导 + 明确排除一项

#: 位置 → 权重序规格
#:   primary   主导分量（权重最高）
#:   secondary 次主导分量（功能单元的第二个偏好；视角为 None=另两项并列）
#:   excluded  被排除分量（≈0；仅功能单元有；视角为 None=不设零）
SPEC = {
    # ---- 认知视角（不参与日常评估者仅 designer）----
    "designer": {"class": VIEWPOINT, "primary": "version_alignment",
                 "secondary": None, "excluded": None},
    "reflect": {"class": VIEWPOINT, "primary": "predictability",
                "secondary": None, "excluded": None},
    "verify": {"class": VIEWPOINT, "primary": "consistency",
               "secondary": None, "excluded": None},
    # ---- 功能单元（身份即排除）----
    "record": {"class": FUNCTIONAL, "primary": "consistency",
               "secondary": "version_alignment", "excluded": "predictability"},
    "output": {"class": FUNCTIONAL, "primary": "predictability",
               "secondary": "version_alignment", "excluded": "consistency"},
    "sustain": {"class": FUNCTIONAL, "primary": "consistency",
                "secondary": "predictability", "excluded": "version_alignment"},
}

POSITION_ORDER = ("designer", "record", "reflect", "verify", "output", "sustain")

#: 参与**日常评估**的位置（设计者是元参照系，不入日常）
DAILY = ("record", "reflect", "verify", "output", "sustain")

#: 序秩（占位量级，仅仿真用）：主导 2 / 次主导 1 / 排除 0；视角另两项 = 1
RANK_PRIMARY = 2.0
RANK_SECONDARY = 1.0
RANK_EXCLUDED = 0.0


class WeightError(Exception):
    """未知位置 / 未知分量。"""


# --------------------------------------------------------------------------
# 基础查询
# --------------------------------------------------------------------------

# 生效条件：position 经 str(position or "").strip() 作为 SPEC 的键，命中时返回对应 spec dict，未命中（含 position 为 None/空串/未知）抛 WeightError；
def _spec(position: str) -> dict:
    s = SPEC.get(str(position or "").strip())
    if s is None:
        raise WeightError(f"未知位置：{position!r}（允许：{POSITION_ORDER}）")
    return s


# 生效条件：component 经 str(component or "").strip() 后若结果在 COMPONENTS 中返回该规范分量字符串，否则抛 WeightError；
def _comp(component: str) -> str:
    c = str(component or "").strip()
    if c not in COMPONENTS:
        raise WeightError(f"未知分量：{component!r}（允许：{COMPONENTS}）")
    return c


# 生效条件：position 可被 _spec 解析且其结果 "class" 等于 VIEWPOINT 时返回 True，非 VIEWPOINT 返回 False，未知 position 抛 WeightError；
def is_viewpoint(position: str) -> bool:
    """是否认知视角（3 项都涉及，不设零）。"""
    return _spec(position)["class"] == VIEWPOINT


# 生效条件：position 可被 _spec 解析且其结果 "class" 等于 FUNCTIONAL 时返回 True，非 FUNCTIONAL 返回 False，未知 position 抛 WeightError；
def is_functional(position: str) -> bool:
    """是否功能单元（身份即排除）。"""
    return _spec(position)["class"] == FUNCTIONAL


# 生效条件：position 经 _spec(position) 验证可解析后，返回原始 position 是否在模块级 DAILY 中；带首尾空白等未 strip 的值会因不在 DAILY 返回 False，未知 position 抛 WeightError；
def in_daily_eval(position: str) -> bool:
    """是否参与日常评估（设计者=False，元参照系）。"""
    _spec(position)
    return position in DAILY


# 生效条件：position 可被 _spec 解析且其 spec 含 "primary" 键时返回该值，未知 position 抛 WeightError；
def dominant(position: str) -> str:
    """主导分量（权重最高者）。"""
    return _spec(position)["primary"]


# 生效条件：position 可被 _spec 解析且其 spec 含 "secondary" 键时返回该值（视角可为 None），未知 position 抛 WeightError；
def secondary(position: str):
    """次主导分量；视角返回 None（另两项并列、均 >0）。"""
    return _spec(position)["secondary"]


# 生效条件：position 可被 _spec 解析且其 spec 含 "excluded" 键时返回该值（视角可为 None），未知 position 抛 WeightError；
def excluded(position: str):
    """被排除分量（≈0）；视角返回 None（不设零）。"""
    return _spec(position)["excluded"]


# --------------------------------------------------------------------------
# 序比较（本模块的权威结论：只有序，没有数值）
# --------------------------------------------------------------------------

# 生效条件：position 与 component 分别经 _spec 与 _comp 验证后，若 component 等于该位置 spec["excluded"] 返回 RANK_EXCLUDED，否则若等于 spec["primary"] 返回 RANK_PRIMARY，否则返回 RANK_SECONDARY；
def rank(position: str, component: str) -> float:
    """分量在该位置的序秩（占位量级）：2 主导 / 1 参与 / 0 排除。"""
    s = _spec(position)
    c = _comp(component)
    if s["excluded"] == c:
        return RANK_EXCLUDED
    if s["primary"] == c:
        return RANK_PRIMARY
    return RANK_SECONDARY


# 生效条件：position、a、b 经 rank 可得序秩，若 rank(position,a)==rank(position,b) 返回 None，否则返回 ra>rb 的 bool；任一参数未知则抛 WeightError；
def prefers(position: str, a: str, b: str):
    """该位置是否**严格**偏好 a 胜过 b。

    返回 True / False / None（None = 二者并列，无严格序）。
    """
    ra, rb = rank(position, a), rank(position, b)
    if ra == rb:
        return None
    return ra > rb


# 生效条件：position 经 _spec 验证且 COMPONENTS 可用时，按 -rank(position,c) 与 COMPONENTS.index(c) 排序，返回 {"preferred":[rank>0], "excluded":[rank==0], "primary":dominant(position), "secondary":secondary(position)}（无 dominant 键）；
def order(position: str):
    """从高到低排列的分量（并列者按 COMPONENTS 稳定序）；排除项单独列出。"""
    _spec(position)
    ranked = sorted(COMPONENTS, key=lambda c: (-rank(position, c), COMPONENTS.index(c)))
    kept = [c for c in ranked if rank(position, c) > 0]
    exc = [c for c in ranked if rank(position, c) == 0]
    return {"preferred": kept, "excluded": exc,
            "primary": dominant(position), "secondary": secondary(position)}


# 生效条件：position 经 _spec 验证且 COMPONENTS 可用时，将各分量 rank 归一化到 4 位小数返回；若 total 为 0 则按 total=1.0 计算；
def weights(position: str) -> dict:
    """占位数值权重（序秩归一化）；**未标定**，仅供仿真。"""
    _spec(position)
    raw = {c: rank(position, c) for c in COMPONENTS}
    total = sum(raw.values()) or 1.0
    return {c: round(raw[c] / total, 4) for c in COMPONENTS}


# --------------------------------------------------------------------------
# 合成（占位：f(一致性, 位置可预测性, 版本对齐度)）
# --------------------------------------------------------------------------

# 生效条件：position 经 _spec 可解析且 COMPONENTS 可用时，components 对每个分量经 .get(c,0.0) 取值并截到 [0,1]；若 components 无 .get 或该值不能 float 则该分量值记 0.0，仍按 weights(position) 加权求和返回含 score/values/weights/dominant 的 dict；
def blend(position: str, components: dict) -> dict:
    """按位置的权重合成单一信任标量。

    `components` 形如 {"version_alignment": 1.0, "predictability": 0.8,
    "consistency": 0.6}；缺省按 0 计。

    **数值未标定**：这里用序秩当量级，只保证「主导分量对结果影响最大」这
    一方向性事实。标定属路线图 v0.3。
    """
    s = _spec(position)
    w = weights(position)
    vals = {}
    for c in COMPONENTS:
        try:
            vals[c] = max(0.0, min(1.0, float(components.get(c, 0.0))))
        except (AttributeError, TypeError, ValueError):
            vals[c] = 0.0
    score = sum(w[c] * vals[c] for c in COMPONENTS)
    return {"position": position, "class": s["class"],
            "weights": w, "values": vals, "score": round(score, 4),
            "dominant": s["primary"],
            "note": "权重为占位序秩（未标定）；仅方向性结论可信"}


# --------------------------------------------------------------------------
# 不变量自检（把上轮的三条结构约束落成可执行断言）
# --------------------------------------------------------------------------

# 生效条件：模块级 POSITION_ORDER、COMPONENTS、SPEC、VIEWPOINT、FUNCTIONAL 等常量齐备时，invariants() 逐条自检并返回 {"ok": all(checks[*].ok), "checks": ...}；
def invariants() -> dict:
    """逐条自检结构约束，返回 {ok, checks{name: {ok, detail}}}。"""
    checks = {}

    # ① 排除唯一：每分量恰被 1 个位置排除 ⇒ 参与度 = 5/6
    part = {c: [p for p in POSITION_ORDER if rank(p, c) > 0] for c in COMPONENTS}
    exc_of = {c: [p for p in POSITION_ORDER if rank(p, c) == 0] for c in COMPONENTS}
    checks["excluded_exactly_once"] = {
        "ok": all(len(exc_of[c]) == 1 for c in COMPONENTS)
              and all(len(part[c]) == len(POSITION_ORDER) - 1 for c in COMPONENTS),
        "detail": {c: {"participants": len(part[c]), "excluded_by": exc_of[c]}
                   for c in COMPONENTS}}

    # ② 排除项只属功能单元，且每分量各被排除一次
    excs = {}
    for p in POSITION_ORDER:
        e = excluded(p)
        if e is None:
            continue
        excs.setdefault(e, []).append(p)
    only_functional = all(is_functional(p)
                          for vs in excs.values() for p in vs)
    both = all(len(excs.get(c) or []) == 1 for c in COMPONENTS)
    checks["exclusion_functional_only_once_each"] = {
        "ok": bool(excs) and only_functional and both,
        "detail": {c: excs.get(c, []) for c in COMPONENTS}}

    # ③ 视角主导 ↔ 功能单元排除 一一配对
    dom_pairs = {}          # 分量 → 视角
    for p in POSITION_ORDER:
        if is_viewpoint(p):
            dom_pairs.setdefault(dominant(p), []).append(p)
    pairing = {}
    for p in POSITION_ORDER:
        if not is_functional(p):
            continue
        e = excluded(p)
        if e and len(dom_pairs.get(e) or []) == 1:
            pairing[dom_pairs[e][0]] = p
    checks["viewpoint_dominant_pairs_unit_exclusion"] = {
        "ok": len(pairing) == 3,
        "detail": pairing}

    # ④ 功能单元恰有一个排除项；视角一个都没有
    checks["functional_excludes_exactly_one"] = {
        "ok": all(excluded(p) is not None for p in POSITION_ORDER
                  if is_functional(p))
              and all(excluded(p) is None for p in POSITION_ORDER
                      if is_viewpoint(p)),
        "detail": {p: excluded(p) for p in POSITION_ORDER}}

    # ⑤ 排除项序秩严格最低
    checks["excluded_is_lowest"] = {
        "ok": all(rank(p, excluded(p)) < rank(p, c)
                  for p in POSITION_ORDER if excluded(p)
                  for c in COMPONENTS if c != excluded(p)),
        "detail": "排除分量序秩 0，严格低于其余分量"}

    return {"ok": all(c["ok"] for c in checks.values()), "checks": checks}


# --------------------------------------------------------------------------
# 自描述
# --------------------------------------------------------------------------

# 生效条件：模块级 COMPONENTS、COMPONENT_LABELS、VIEWPOINT、FUNCTIONAL、SPEC、POSITION_ORDER、DAILY 齐备时，返回位置权重矩阵自描述 dict，其中 components 按 COMPONENT_LABELS、positions 按 SPEC、daily_eval 按 DAILY 展开；
def catalog() -> dict:
    """位置权重矩阵自描述（供 MCP / 文档对照验证）。"""
    return {
        "question": "不同身份，参数偏好不同",
        "components": {c: COMPONENT_LABELS[c] for c in COMPONENTS},
        "classes": {
            VIEWPOINT: "认知视角：3 项都涉及，只标主导，不设零",
            FUNCTIONAL: "功能单元：身份即排除（不猜/不记/不对外）",
        },
        "positions": {p: dict(SPEC[p]) for p in POSITION_ORDER},
        "daily_eval": list(DAILY),
        "meta_reference": [p for p in POSITION_ORDER if p not in DAILY],
        "derivation": {"record": "不猜 → 排除 位置可预测性",
                       "output": "不记 → 排除 一致性",
                       "sustain": "不对外 → 排除 版本对齐度"},
        "formula": "P_trust = f(一致性, 位置可预测性, 版本对齐度)",
        "honest_boundary": "只声明序，不宣称数值；blend() 权重为占位序秩",
    }


# ==========================================================================
# 结构重要性重算（maintain.importance）
# ==========================================================================
#
# 与上面的「位置权重矩阵」是**两件事**，不要混：
#   · 上面回答「不同身份偏好哪些信任分量」（序标定，作用于位置）；
#   · 下面回答「一个节点在认知图结构里该有多大重要性」（重算，作用于节点）。
#
# 三个**可测**分量（对齐计划：按覆盖度、冗余度、验证基底重排）：
#   coverage   覆盖度 —— 有多少节点指向/包含它（入度，结构中枢性）
#   redundancy 冗余度 —— 与同层节点内容哈希撞车的比例（越冗余越该降权）
#   basis      验证基底 —— 外部验证档位（formal_proof/compiler > test > measurement > data > other）
#
# 纪律：只改 `frontmatter.importance` 与索引快照，**不动正文**（减小 diff 与重写成本）；
# 全流程可预演（apply=False）、可留痕（`_maintain.jsonl`）、可回滚（rollback）。
# 性能：dry-run 不读任何节点文件（只用索引快照）；apply 只写「变动超阈值」的节点。

MAINTAIN_LOG = "_maintain.jsonl"

#: 验证基底 → 可信权重（枚举来自 nodefile.VERIFICATION_BASIS）
#: 文科来源一致性档（textbook/public_kb）低于可复现档、高于 other：
#: 教材/公开知识库是权威二手来源，但不是可复算证据。
BASIS_TRUST = {
    "formal_proof": 1.00,
    "compiler": 0.90,
    "test": 0.80,
    "measurement": 0.65,
    "textbook": 0.60,
    "data": 0.45,
    "public_kb": 0.40,
    "other": 0.25,
}
BASIS_TRUST_MISSING = 0.25

#: 三分量合成权重（占位量级，沿用「只声明序」纪律；分量序主导→次要）
IMPORTANCE_MIX = {"basis": 0.40, "coverage": 0.35, "redundancy": 0.25}
COV_SAT = 12.0              # 入度饱和点：约 12 条入边即视为覆盖满
IMPORTANCE_FLOOR = 0.10
IMPORTANCE_CEIL = 1.00
IMPORTANCE_PROTECT = 0.70   # ≥ 此值：保护下限（已保护/高危节点不被降破）
APPLY_DELTA = 0.05          # 变动小于此值不写盘（防 IO 放大）


# 生效条件：e 为可 .get 的映射时，遍历 e.get("edges")（假值视为空）逐项按 dict 的 to/target/node/id 或标量转 str 追加非假值，再取 e.get("subgraph")，若为 dict 取其 "nodes" 否则取 subgraph 本身，且当其为 list/tuple 时追加其中 dict 的 id 或标量转 str 的非假值，返回 out；
def _targets(e: dict):
    """索引快照里的「出边目标」：edges ∪ subgraph.nodes（兼容 dict/str）。"""
    out = []
    for x in (e.get("edges") or []):
        if isinstance(x, dict):
            t = x.get("to") or x.get("target") or x.get("node") or x.get("id")
        else:
            t = x
        if t:
            out.append(str(t))
    sg = e.get("subgraph")
    raw = sg.get("nodes") if isinstance(sg, dict) else sg
    if isinstance(raw, (list, tuple)):
        for n in raw:
            nid = n.get("id") if isinstance(n, dict) else n
            if nid:
                out.append(str(nid))
    return out


# 生效条件：basis 为 None 时返回 BASIS_TRUST_MISSING；否则把 str(basis).strip().lower() 作为 BASIS_TRUST 的键，命中返回对应权重，未命中（含空串）返回 BASIS_TRUST_MISSING；
def basis_trust(basis):
    """验证基底 → 可信权重；缺失/未知一律给最低档（不假装可信）。"""
    if basis is None:
        return BASIS_TRUST_MISSING
    return BASIS_TRUST.get(str(basis).strip().lower(), BASIS_TRUST_MISSING)


# 生效条件：cg 的 index（getattr(cg,"index",None) or {}）与其 "nodes" 同为真值时，返回以这些节点 id 为键的入度表，仅当 _targets(e) 给出的目标 t 也在 nodes 且 t != nid 时计数 +1；index 或 "nodes" 为假值（None/{}）时 nodes 回落 {}，直接返回空 dict。
def coverage_index(cg) -> dict:
    """入度表：nid → 被多少节点指向（覆盖度，O(N) 免读文件）。"""
    nodes = (getattr(cg, "index", None) or {}).get("nodes") or {}
    indeg = {nid: 0 for nid in list(nodes)}
    for nid, e in list(nodes.items()):
        for t in _targets(e):
            if t in indeg and t != nid:
                indeg[t] += 1
    return indeg


# 生效条件：cg.index 的 "nodes" 为 dict 时，layer 真值仅处理 e.get("layer")==layer 的节点，layer 假值不过滤；content_hash 为假值（None/空串/0 等）的节点输出 0.0，同 (e.get("layer"), content_hash) 第二次撞车时当前与首次节点均输出 1.0，否则输出 0.0；
def redundancy_map(cg, layer=None) -> dict:
    """同层内容哈希撞车 → 冗余度 1.0，否则 0.0（精确冗余，O(N) 免读文件）。

    诚实边界：这是**精确冗余**（同一层出现完全相同的正文），不是语义近似冗余。
    语义近似由 `forgetting.redundancy`（bigram 覆盖）在写入侧单独负责。
    """
    nodes = (getattr(cg, "index", None) or {}).get("nodes") or {}
    seen, out = {}, {}
    for nid, e in list(nodes.items()):
        if layer and e.get("layer") != layer:
            continue
        h = e.get("content_hash") or ""
        if not h:
            out[nid] = 0.0
            continue
        key = (e.get("layer"), h)
        if key in seen:
            out[nid] = 1.0
            out[seen[key]] = 1.0
        else:
            seen[key] = nid
            out.setdefault(nid, 0.0)
    return out


# 生效条件：cg.index 的 nodes 为 dict 且 entry 非 None 或 nodes.get(nid) 命中时，e 按 entry 优先否则 nodes.get(nid)；indeg/red 分别仅在为 None 时回落 coverage_index(cg)/redundancy_map(cg, layer=e.get("layer"))（空容器不回落）；按 IMPORTANCE_MIX、COV_SAT 及 e.get("verification_basis") 的 basis_trust 计算 coverage/redundancy/basis 分量，按 bool(e.get("protected")) 选择 IMPORTANCE_PROTECT 或 IMPORTANCE_FLOOR 作下限、IMPORTANCE_CEIL 作上限，返回含 before（e.get("importance",0.5) or 0.5）、after、components、protected 的 dict；entry 与 nodes.get(nid) 均为 None 返回 None；
def node_importance(cg, nid, indeg=None, red=None, entry=None):
    """单节点结构重要性分解（可审计：给出分量而非只给分）。"""
    nodes = (getattr(cg, "index", None) or {}).get("nodes") or {}
    e = entry if entry is not None else nodes.get(nid)
    if e is None:
        return None
    indeg = indeg if indeg is not None else coverage_index(cg)
    red = red if red is not None else redundancy_map(cg, layer=e.get("layer"))
    d = int(indeg.get(nid, 0) or 0)
    cov = min(1.0, math.log1p(d) / math.log1p(COV_SAT))
    rr = max(0.0, min(1.0, float(red.get(nid, 0.0) or 0.0)))
    bt = basis_trust(e.get("verification_basis"))
    score = (IMPORTANCE_MIX["basis"] * bt
             + IMPORTANCE_MIX["coverage"] * cov
             + IMPORTANCE_MIX["redundancy"] * (1.0 - rr))
    # #431：此前只认 `protected` 位（窄：漏层保护与按分保护）⇒ self/anchor 节点被降破。
    # 走条目口径单点 protect.is_protected_entry（与 is_protected 同一个函数）。
    protected = protect.is_protected_entry(e)[0]
    floor = IMPORTANCE_PROTECT if protected else IMPORTANCE_FLOOR
    score = max(floor, min(IMPORTANCE_CEIL, score))
    return {"node_id": nid,
            "before": float(e.get("importance", 0.5) or 0.5),
            "after": round(score, 4),
            "components": {"coverage": round(cov, 4), "redundancy": round(rr, 4),
                           "basis_trust": round(bt, 4), "in_degree": d,
                           "verification_basis": e.get("verification_basis")},
            "protected": protected,
            "protected_floor": protected}


# 生效条件：给定 cg 且 nodes = cg.index["nodes"] 时按 layer 过滤、limit 为真值才 ids = ids[:int(limit)] 逐节点重算，abs(after-before) < float(min_delta) 记 unchanged（**预演面不接档位闸**）；仅 apply=True 才在**任何写盘之前**先经 autonomy_modes.e_gate() 档位判定（补强批次 v1.2·E 面接线）——非 ALLOW（plan 档无计划）即经 forbidden_result() fail-closed 早退（零写盘、零审计行），ALLOW（confirm/full）才把 frontmatter.importance/importance_source/importance_components 写回（after >= IMPORTANCE_PROTECT 且未 protected 时补写 protected/protection_reason），并向 cg.root 下 append_jsonl(..., MAINTAIN_LOG) 记 batch 后 rebuild_index。
def recalc(cg, layer=None, limit=None, apply=False, min_delta=APPLY_DELTA,
           actor="maintain", dry_run_samples=10):
    """结构重要性重算：覆盖度 + 冗余度 + 验证基底 → 重排节点重要性。

    apply=False（默认）只出报表，不写盘（对应计划「可预演」）。
    apply=True 逐节点改写 `frontmatter.importance`，每批写一条 `_maintain.jsonl`
    （before/after/分量/actor），并提供 rollback 反向应用。

    **档位闸（补强批次 v1.2·E 面接线）**：`apply=True` 的**写盘前**先经
    `autonomy_modes.e_gate()`（设计 §三 plan 档 E「须命中计划步骤」）——plan 档
    无计划 ⇒ `forbidden_result()` fail-closed 早退（零写盘、零审计行）；confirm/
    full ⇒ ALLOW、**原链原样**（逐位不变）。`apply=False`（预演）不接闸。
    """
    nodes = (getattr(cg, "index", None) or {}).get("nodes") or {}
    indeg = coverage_index(cg)
    red = redundancy_map(cg, layer=layer)
    ids = [nid for nid, e in list(nodes.items())
           if not layer or e.get("layer") == layer]
    ids.sort()
    if limit:
        ids = ids[:int(limit)]
    t0 = time.time()
    planned, changed, unchanged, skipped = [], 0, 0, 0
    before_sum = after_sum = 0.0
    by_layer = {}
    batch = time.strftime("%Y%m%d-%H%M%S")
    samples = []
    for nid in ids:
        info = node_importance(cg, nid, indeg=indeg, red=red)
        if info is None:
            skipped += 1
            continue
        before, after = info["before"], info["after"]
        before_sum += before
        after_sum += after
        lay = (nodes.get(nid) or {}).get("layer") or "?"
        bl = by_layer.setdefault(lay, {"nodes": 0, "changed": 0,
                                       "avg_before": 0.0, "avg_after": 0.0})
        bl["nodes"] += 1
        if abs(after - before) < float(min_delta):
            unchanged += 1
            continue
        changed += 1
        bl["changed"] += 1
        plan = dict(info)
        plan["batch"] = batch
        plan["delta"] = round(after - before, 4)
        planned.append(plan)
        if len(samples) < int(dry_run_samples):
            samples.append(plan)
    for lay, bl in list(by_layer.items()):
        bl["avg_before"] = 0.0
        bl["avg_after"] = 0.0
    written = 0
    if apply:
        # 三档自治（补强批次 v1.2·E 面接线，设计 §三 plan 档 E「须命中计划步骤」）：
        # E 类动作**写盘前**的档位判定单点（同 `freshness.recalc` 口径）——
        # 本函数无独立资格闸，闸口取「进入写盘支路的第一行」＝任何盘面写入之前；
        # ALLOW（confirm/full）原链原样，plan 档无计划 ⇒ fail-closed 早退。
        from . import autonomy_modes as _am
        _dec = _am.e_gate()
        if _dec["decision"] != _am.ALLOW:
            return _am.forbidden_result(_dec, action="importance", dry_run=False,
                                        written=0, batch=batch)
        for plan in planned:
            nid = plan["node_id"]
            node = cg.get(nid)
            if not node:
                continue
            fm = node.get("frontmatter") or {}
            fm["importance"] = plan["after"]
            fm["importance_source"] = "recalc"
            fm["importance_components"] = plan["components"]
            if plan["after"] >= IMPORTANCE_PROTECT and not fm.get("protected"):
                fm["protected"] = True
                fm["protection_reason"] = (f"recalc importance={plan['after']:.2f}"
                                           f"≥{IMPORTANCE_PROTECT}")
            path = os.path.join(cg.root, (nodes.get(nid) or {}).get("path")
                                or node.get("path") or f"{nid}.md")
            cg._write_node(nid, path, fm, node.get("content") or "")
            e = nodes.get(nid)
            if e is not None:
                e["importance"] = plan["after"]
                if fm.get("protected"):
                    e["protected"] = True
                    e["protection_reason"] = fm["protection_reason"]
            rec = {"t": time.time(), "action": "importance", "batch": batch,
                   "id": nid, "layer": (e or {}).get("layer"),
                   "before": plan["before"], "after": plan["after"],
                   "delta": plan["delta"], "components": plan["components"],
                   "actor": actor}
            append_jsonl(os.path.join(cg.root, MAINTAIN_LOG), rec)
            written += 1
        if written and hasattr(cg, "rebuild_index"):
            cg.rebuild_index()          # 索引与盘面重新对齐，防内存/磁盘分叉
    n = max(1, len(ids))
    return {
        "ok": True, "action": "importance", "dry_run": not apply,
        "batch": batch, "nodes_scanned": len(ids), "changed": changed,
        "unchanged": unchanged, "skipped": skipped, "written": written,
        "min_delta": float(min_delta),
        "avg_before": round(before_sum / n, 4),
        "avg_after": round(after_sum / n, 4),
        "by_layer": {k: {"nodes": v["nodes"], "changed": v["changed"]}
                     for k, v in by_layer.items()},
        "samples": samples, "elapsed_ms": int((time.time() - t0) * 1000),
        "log": MAINTAIN_LOG,
        "note": ("dry-run：未写盘；apply=True 才改写 importance"
                 if not apply else
                 f"已改写 {written} 个节点；回滚见 rollback(batch={batch})"),
    }


# 生效条件：cg.root/MAINTAIN_LOG 可读出 action=="importance" 记录后，entry_ids 为真值时按记录 id 是否在 {str(x) for x in entry_ids} 过滤；否则 batch 为真值时按 r.get("batch")==batch 过滤；否则取最后一条记录的 batch 再按其过滤；无 importance 记录返回 {"ok":False,"error":"no_records","reverted":0}，过滤后无记录返回 {"ok":False,"error":"batch_not_found",...}；对命中且 cg.get(nid) 为真的记录写回 rec.get("before") 并计数，若 reverted 且 cg 有 rebuild_index 则调用，最后 append_jsonl 写 actor 并返回 {"ok":True,"batch":batch,"reverted":reverted,"ids":ids}；
def rollback(cg, batch=None, entry_ids=None, actor="maintain"):
    """把重要性重算反向应用（bulk 改写的可回滚兑现）。

    batch 为空则回滚**最近一批**。返回 {ok, reverted, batch, ids}。
    """
    log_path = os.path.join(cg.root, MAINTAIN_LOG)
    records = [r for r in read_jsonl(log_path)
               if r.get("action") == "importance"]
    if not records:
        return {"ok": False, "error": "no_records", "reverted": 0}
    if entry_ids:
        want = {str(x) for x in entry_ids}
        records = [r for r in records if str(r.get("id")) in want]
    elif batch:
        records = [r for r in records if r.get("batch") == batch]
    else:
        batch = records[-1].get("batch")
        records = [r for r in records if r.get("batch") == batch]
    if not records:
        return {"ok": False, "error": "batch_not_found", "batch": batch,
                "reverted": 0}
    nodes = (getattr(cg, "index", None) or {}).get("nodes") or {}
    reverted, ids = 0, []
    # 同一节点多条记录时以最早一条的 before 为准
    for rec in records:
        nid = rec.get("id")
        node = cg.get(nid)
        if not node:
            continue
        fm = node.get("frontmatter") or {}
        fm["importance"] = rec.get("before")
        fm["importance_source"] = "recalc_rollback"
        path = os.path.join(cg.root, (nodes.get(nid) or {}).get("path")
                            or node.get("path") or f"{nid}.md")
        cg._write_node(nid, path, fm, node.get("content") or "")
        e = nodes.get(nid)
        if e is not None:
            e["importance"] = rec.get("before")
        reverted += 1
        ids.append(nid)
    if reverted and hasattr(cg, "rebuild_index"):
        cg.rebuild_index()
    append_jsonl(log_path, {"t": time.time(), "action": "importance_rollback",
                            "batch": batch, "reverted": reverted, "actor": actor})
    return {"ok": True, "batch": batch, "reverted": reverted, "ids": ids}


# 生效条件：cg.root/MAINTAIN_LOG 可读出 JSONL 记录后，action 为真值时按 r.get("action")==action 过滤，假值不过滤；返回 recs[-int(limit):]，其中 limit=0 时 int(0)=0 使切片为 recs[0:] 返回全部而非空；
def history(cg, limit=100, action=None):
    """维护留痕（最近 limit 条），可按 action 过滤。"""
    recs = list(read_jsonl(os.path.join(cg.root, MAINTAIN_LOG)))
    if action:
        recs = [r for r in recs if r.get("action") == action]
    return recs[-int(limit):]