# -*- coding: utf-8 -*-
"""md_cg · 进程/权限模型（记忆 OS #2）

关键动机：**灵枢是开源仓库，而记忆很大一部分是私有内容**。因此权限模型必须
支持「公开知识」与「私有记忆」的物理与逻辑隔离：

  ① 物理隔离：租户（tenant）各自一个根目录。public 租户的根可落在开源仓库内，
     private 租户的根必须落在仓库外（如 ~/.mdcg/private/）。
  ② 逻辑隔离：每个节点带 sensitivity（public < internal < private < secret）；
     每个调用方（Principal）带 clearance，只能读写 ≤ clearance 的节点。
  ③ 硬规则：secret 节点对 clearance < secret 的调用方**写入即拒**（对齐 noema
     「secret 候选自动拒绝」）；写入高于自身 clearance 的敏感度 → 拒绝。

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

import json
import os
import sys
import time
import uuid

from .datapath import aux_root
from .fsutil import publish

# 密级阶梯的体系语义（批次 27 表述正名，使用者 2026-09-24 澄清）：
#   private ≠ 个人隐私保密，而是**错误处置标记**——工作区内容因错误相关、
#   验证未过、待排查而被标记私有，目的是**限制向平级/下级扩散**（防止未
#   证实或已知有误的内容被下游消费），不是对错误处置链路保密。
#   可见性是**故意的冲突设计**：平级/下级限读，但错误处置链路必须可读——
#   设计者（designer，secret 上限）、上级节点、验证单元（verify）三者
#   必读。当前令牌矩阵中 record/reflect/verify/sustain/orchestr 的
#   clearance_cap 均为 internal（forbidden 含 private）——验证单元读错误
#   标记节点是本职却受限于 cap，属**已登记的分级缺口**（见安全审计实锚
#   文档「读权限分级」节），未来专项：verify 面提 cap 或走设计者代查通道。
# 密级阶梯的体系语义（批次 27 正名 / 批次 28 分型落地，使用者 2026-09-24）：
#   **restricted = 错误处置标记（新增档）**——内容因错误相关、验证未过、
#   待排查被标记，目的是限制向平级/下级扩散；可见性按**错误处置链路角色**
#   判定（designer/orchestr/verify 必读，record/worker/output/guest 限读），
#   不走纯密级 rank（链路角色与平级同为 internal，rank 无法区分）；
#   **不加密**（不在 crypto.ENCRYPTED_LEVELS——链路要明文读）。
#   private = 隐私/会话绑定档（语义不变：仅归属会话 + 设计者豁免可见，
#   落盘加密）；secret = 最高密级。
#   可见性的故意冲突设计就此消解：错误标记限扩散但处置链路必读，
#   隐私内容写者私有——两种语义不再共用一个密级档。
SENSITIVITY_ORDER = ("public", "internal", "restricted", "private", "secret")
DEFAULT_SENSITIVITY = "internal"


class AccessDenied(Exception):
    """权限拒绝（读/写/管理）。

    hint（可选）：与本仓 70 处失败路径 hint 同口径的「为什么 + 怎么办」可操作
    指引（issue #34：新用户首个错误无可操作指引——guest 只读写入被拒恰是
    三步接入后必然撞上的第一条失败，不得只说「不行」不说「怎么办」）。
    """

    def __init__(self, msg: str, hint: str = None):
        super().__init__(msg)
        self.hint = hint


# 生效条件：形参 level 为模块级常量 SENSITIVITY_ORDER 中的元素时返回其下标，否则抛 AccessDenied。
def _rank(level: str) -> int:
    try:
        return SENSITIVITY_ORDER.index(level)
    except ValueError:
        raise AccessDenied(f"未知敏感度/密级：{level}") from None


# 生效条件：target_sensitivity / incoming_sensitivity 为 SENSITIVITY_ORDER 成员或 None
# （None 按 DEFAULT_SENSITIVITY 处理）时，返回「把 incoming 内容的明文并入 target 节点
# **是否导致该内容的有效密级下降**」＝ incoming 密级**严格高于** target 密级（合并后它按
# target 的低档可见）；任一形参非法（_rank 抛 AccessDenied）时**按降密级处理**
# （fail-closed：判不了就不许合并）。
def merge_downgrades(target_sensitivity: str, incoming_sensitivity: str) -> bool:
    """合并是否**降密级** —— 唯一判据单点（issue #148）。

    病象：声明 private 的短记忆经 CONVERGE（writelimit.converge_into）或 MERGE
    （forgetting.reinforce）**明文并入更早的 internal 节点**，而目标节点的
    sensitivity **不变** ⇒ 该 private 明文就此获得 internal 的可见性（降密级泄漏，
    访客可读）。

    调用点：`writelimit.converge_into` / `forgetting.reinforce` —— 两者**落盘前**
    一律经本函数，判为降密级即 **fail-closed 不写盘**；**不另写第二份判据**。

    ★方向（本判据最易写反处，实测踩到过）：密级阶梯
    public(0) < internal(1) < restricted(2) < private(3) < secret(4)；
    病象是 **private(3) 并入 internal(1)** —— incoming 排名**更高**，合并后它只
    按 internal 档被看见 ⇒ 该内容被**降级**。故判据为 `new > tgt`（不是 `<`）。

    不适用条件：只判「incoming 密级高于目标 ⇒ 被降级」这一件事——同档合并照旧
    放行；**低密级并入高密级节点**（incoming < target，内容变得更不可见）不在本
    判据范围（非泄漏），仍由既有层闸/保护闸管；本函数不评价内容相关性、不替代
    protect.guard_overwrite。
    """
    try:
        tgt = _rank(target_sensitivity or DEFAULT_SENSITIVITY)
        new = _rank(incoming_sensitivity or DEFAULT_SENSITIVITY)
    except AccessDenied:
        return True                      # fail-closed：判不了就不许合并
    return new > tgt


# 生效条件：全部形参均可省略、clearance 默认取模块常量 DEFAULT_SENSITIVITY，构造时先经 _rank(clearance) 校验，随后 session 为假值（含空串）回落为 sess_+uuid 十二位、role 假值回落 "system"、expires_at 为假值（含 0）置 None、layers_allow/ops_allow 为 None 时保持 None 否则转 tuple、theory_ok 经 bool() 转换。
class Principal:
    """调用方身份（一次会话一个）。

    tenant        —— 租户（决定根目录）
    actor         —— 调用者标识（写入审计）
    clearance     —— 最高可读/可写敏感度
    can_write     —— 是否允许写
    can_admin     —— 是否允许管理操作（forget/restore/review_decide）
    session       —— 会话 id（进程/会话模型：每次连接一个）
    harness       —— 承载端标识（dsh/zcode/codebuddy…；仅归因，不参与授权）
    unit          —— 单元分工（record/reflect/verify/output/sustain；仅归因）
                     注：MCP 请求级参数 `as_unit` 会经 `tokens.narrowed_principal`
                     产出一次性 Principal——其 role/layers/ops 取自该单元 spec、unit
                     记为执行单元。那是**单次调用的临时身份**，不改变本字段的归因
                     定位（见 `mcp_server.call_tool`）。

    令牌扩展（由 `tokens.verify_token` 填充；直接构造时为 None = 不限制）：
    role          —— 角色（designer/reflection/verifier/recorder/output/sustain/guest）
    token_id      —— 令牌 id（可追溯到签发记录）
    parent        —— 派生来源令牌 id（权职分离的委派链）
    expires_at    —— 过期时间戳（None = 不过期）
    layers_allow  —— 可写层白名单（None = 不限制；"*" = 全部）
    ops_allow     —— 可执行 op 白名单（同上）
    auth_mode     —— direct（直接构造）/ token / legacy_env / anonymous

    版本层扩展（蜂群互联层0，由 `theory.check` 填充）：
    theory_ok      —— 版本声明是否合法；False 时**全部写/管理操作被拒**（只读降级）
    theory_version —— 当前声明的协议版本（审计与 whoami 用）
    """

# 生效条件：clearance 须为模块级常量 SENSITIVITY_ORDER 成员（否则 _rank 抛 AccessDenied），session 为假值（含 None/空串）时生成 sess_ 随机串并把 session_auto 置 True（未获声明的进程自动随机；写归因三态收口的识别位，消费方 MdCGSecure._attributed_session），session 真值时原样采用并把 session_auto 置 False（显式声明的来源），expires_at 为假值（含 None/0）时存 None 否则 float(expires_at)，layers_allow/ops_allow 为 None 时存 None 否则 tuple 化；sessions_read 为 None 时存 None（=旧行为，不限制会话读面），为集合（含 list/tuple/set）时 tuple 化并去重（=只允许读这些会话，'*' 表示全部，见 session_recall 的 recent 段）。
    def __init__(self, tenant: str = "default", actor: str = "system",
                 clearance: str = DEFAULT_SENSITIVITY, can_write: bool = True,
                 can_admin: bool = False, session: str = None,
                 harness: str = None, unit: str = None,
                 role: str = None, token_id: str = None, parent: str = None,
                 expires_at: float = None, layers_allow=None, ops_allow=None,
                 auth_mode: str = "direct", theory_ok: bool = True,
                 theory_version: str = None, sessions_read=None):
        _rank(clearance)                      # 校验
        self.tenant = tenant
        self.actor = actor
        self.clearance = clearance
        self.can_write = can_write
        self.can_admin = can_admin
        if session:
            self.session = session
            self.session_auto = False
        else:
            self.session = "sess_" + uuid.uuid4().hex[:12]
            # 归因三态（2026-10-07 设计者裁定「会话身份三态」）：本值是「未获
            # 任何声明来源」的进程自动随机——**只标记来源、不改值**（绑定档的
            # 读回判据锚定值本身，见 mdcos._readable；值语义由 test_p45 A1/C5
            # 钉死）。写归因落盘时由 MdCGSecure._attributed_session 把该态收口
            # 为 'unattributed'。**赋值方**须知：凡把 session 改写为声明值处
            # （_apply_attribution / serve 的 env 注入 / tokens.narrowed_principal
            # 的复制传导）必须同步维护本标记，否则收口会把声明值误当自动随机。
            self.session_auto = True
        # 归因维度（嵌套身份）：只入审计（_audit/_recent），不参与权限判定。
        # 权限域仍由令牌记录决定（tokens.ROLE_SPECS），与 harness/unit 无关。
        # 受控例外：MCP 请求级 `as_unit` 收窄（tokens.narrowed_principal）产出的是
        # 一次性 Principal，其 role/allow 取自单元 spec——**只做减法**（与 owner
        # 求交 + 管理权恒 False），故不构成本字段「参与授权」的先例。
        self.harness = harness
        self.unit = unit
        self.role = (role or "system")
        self.token_id = token_id
        # issue #712（2026-10-30 使用者裁定方案 C）：**分档会话读权限**。
        # None  = 旧行为逐位不变（internal/public 事件跨会话可见；私有档仍由 _readable 的
        #         既有口径按 principal.session 绑定，本字段不参与）；
        # 集合  = 只允许读这些会话（'*' 表示全部）——仅在 session_recall 的 recent 段生效；
        # 注意：can_admin 仍可读所有档（issue #35「设计者豁免」定稿），本字段**不削弱**它。
        self.sessions_read = (None if sessions_read is None
                              else tuple(dict.fromkeys(str(x) for x in sessions_read)))
        self.parent = parent
        self.expires_at = float(expires_at) if expires_at else None
        self.layers_allow = None if layers_allow is None else tuple(layers_allow)
        self.ops_allow = None if ops_allow is None else tuple(ops_allow)
        self.auth_mode = auth_mode
        self.theory_ok = bool(theory_ok)
        self.theory_version = theory_version

    # ---------- 基础判定 ----------

# 生效条件：形参 sensitivity 与 self.clearance 均可被 _rank 映射到模块级常量 SENSITIVITY_ORDER 中时，返回前者排名是否不高于后者；任一不在其中则 _rank 抛 AccessDenied。
    def allows(self, sensitivity: str) -> bool:
        """clearance 是否覆盖该敏感度（可读/可写）。"""
        return _rank(sensitivity) <= _rank(self.clearance)

# 生效条件：无 required 形参或模块级常量前置，仅当 self.expires_at（来自 __init__ 的 expires_at）不为 None 且 time.time() 大于它时返回 True，否则返回 False。
    def expired(self) -> bool:
        return self.expires_at is not None and time.time() > self.expires_at

    @staticmethod
# 生效条件：allow 为 None 时返回 True（未声明=不限制），allow 非 None 时返回 "'*' in allow 或 name in allow" 的布尔结果。
    def _in_scope(allow, name: str) -> bool:
        if allow is None:                     # 未声明 = 不限制（兼容直接构造）
            return True
        return "*" in allow or name in allow

# 生效条件：layer 为假值（None/空串）时以 "knowledge" 参与判定，返回 self._in_scope(self.layers_allow, layer or "knowledge") 的结果。
    def allows_layer(self, layer: str) -> bool:
        return self._in_scope(self.layers_allow, layer or "knowledge")

# 生效条件：op 为假值（None/空串）时以 "" 参与，先经 strip().lower() 归一化，返回 self._in_scope(self.ops_allow, (op or "").strip().lower()) 的结果。
    def allows_op(self, op: str) -> bool:
        return self._in_scope(self.ops_allow, (op or "").strip().lower())

    # ---------- 强制校验（越权即 AccessDenied） ----------

# 生效条件：无 required 形参或模块级常量前置，当 self.expired()（来自 __init__ 的 expires_at 与当前时间比较）为 True 时抛 AccessDenied（带重新签发 hint），否则无操作返回 None。
    def _require_live(self):
        if self.expired():
            raise AccessDenied(
                f"actor={self.actor} 令牌已过期",
                hint="令牌过期不是故障：重新签发并更新 MDCG_TOKEN 后重启宿主"
                     "（python -m md_cg.tokens issue …，README「写入凭据」一节）。")

    # issue #34：op 拒绝按成因分指引——guest（无令牌只读访客）引导配凭据；
    # 有令牌但白名单不含该 op 引导换角色/补授权。「不是故障」安抚语气与
    # 本仓其它 70 处 hint 同口径。
    _GUEST_OP_HINT = (
        "当前为只读访客模式（未检测到写入凭据 MDCG_TOKEN）：读/召回/时间线可用，"
        "写入与管理操作不落盘。这不是故障，配置凭据后即恢复。"
        "要真正落盘请签发并配置写入凭据（README「写入凭据」一节）：\n"
        "  python -m md_cg.tokens issue --role recorder --actor <名> "
        "--clearance internal\n"
        "  setx MDCG_TOKEN \"<token>\"   # 然后重启宿主（DSH/IDE）\n"
        "最小权限可用 --role recorder；op 用法可查 cg(op=\"help\", query=\"write\")。")

# 生效条件：先调用 self._require_live()（过期则抛 AccessDenied）；再要求 self.allows_op(op) 为 True，否则抛 AccessDenied（role=guest 带配凭据 hint，其它角色带补授权 hint）。
    def require_op(self, op: str):
        self._require_live()
        if not self.allows_op(op):
            _op_n = (op or "").strip().lower()
            # issue #678：区分两种成因——**角色本身不含**该 op，与**令牌被派生收窄**。
            # 旧文案一律说「角色 X 无权」，于是「designer（ROLE_SPECS 里 ops_allow=["*"]）
            # 执行 forget 被拒」会读成「designer 角色没有 forget 权限」——当事人会去改角色
            # 定义（改错地方）或以为权限系统故障。实际原因是签发时 --ops-allow 收窄了作用域。
            _role_ops = None
            try:
                from .tokens import ROLE_SPECS      # 延迟导入：tokens 依赖 security，避免环
                _role_ops = (ROLE_SPECS.get(self.role) or {}).get("ops_allow")
            except Exception:
                _role_ops = None
            _role_has = False
            if _role_ops is not None:
                _norm = [str(x).strip().lower() for x in _role_ops]
                _role_has = ("*" in _norm) or (_op_n in _norm)
            if self.role == "guest":
                msg = (f"角色 {self.role} 无权执行 op={op}（作用域 "
                       f"{list(self.ops_allow) if self.ops_allow is not None else '不限'}）")
                raise AccessDenied(msg, hint=self._GUEST_OP_HINT)
            if _role_has:
                raise AccessDenied(
                    f"本令牌的 ops 作用域不含 op={op}（角色 {self.role} 本身**含**该 op，"
                    f"是签发时被收窄；本令牌作用域 "
                    f"{list(self.ops_allow) if self.ops_allow is not None else '不限'}）",
                    hint=f"这是正常闸门不是故障：派生令牌只能**收窄**不能扩权。"
                         f"由签发者重新签发且**不要**用 --ops-allow 收窄 {op}"
                         f"（python -m md_cg.tokens issue …），或换用未收窄的 "
                         f"{self.role} 令牌；更新 MDCG_TOKEN 后重启宿主"
                         f"（README「写入凭据」一节）。")
            msg = (f"角色 {self.role} 无权执行 op={op}（作用域 "
                   f"{list(self.ops_allow) if self.ops_allow is not None else '不限'}）")
            raise AccessDenied(
                msg,
                hint=f"当前令牌（actor={self.actor}，角色 {self.role}）的 ops "
                     f"白名单不含 {op}，属正常闸门不是故障。可：①换用含该 op "
                     f"的角色令牌；②由签发者以 --ops-allow 补授权后重新签发"
                     f"（python -m md_cg.tokens issue …，README「写入凭据」一节）。")

# 生效条件：依次要求 self._require_live() 未抛异常、self.theory_ok 为 True、self.can_write 为 True、self.allows(sensitivity) 为 True；任一不满足则抛 AccessDenied。
    def require_write(self, sensitivity: str):
        self._require_live()
        if not self.theory_ok:
            raise AccessDenied(
                "版本层校验未通过（theory_ok=False）：全部写操作被拒，"
                "仅保留 theory 修复入口")
        if not self.can_write:
            raise AccessDenied(
                f"actor={self.actor} 无写权限",
                hint=(self._GUEST_OP_HINT if self.role == "guest" else
                      f"当前令牌（角色 {self.role}）未授予写权限（can_write="
                      f"False），属正常闸门不是故障。请换用可写角色令牌或由"
                      f"签发者重新签发（README「写入凭据」一节）。"))
        if not self.allows(sensitivity):
            raise AccessDenied(
                f"写入敏感度 {sensitivity} 超出 clearance {self.clearance}")

# 生效条件：先要求 self.require_write(sensitivity) 未抛异常；随后将形参 layer 为假值（None/空串）时按 "knowledge" 处理，并要求 self.allows_layer(layer) 为 True；最后按 MDCG_WRITE_GATE 三档过**分层写入严格度闸**（writegate）：off 零介入放行、warn 记录并放行、enforce 对 deny/require_review 抛 AccessDenied。
    def require_layer_write(self, layer: str, sensitivity: str, reviewers=None, *, skip_gate: bool = False):
        """写层校验：密级 + 层白名单 + **分层复核闸**（2026-10-27 使用者裁定）。

        分层严格度（writegate.LAYER_TIERS）：情景/目标/未解/负记忆 = 0 档
        （无任何复核通道时允许直写——裁定的放宽通道）；知识 = 1 档（需反思单元）；
        结构 = 2 档（反思 + 验证）；自我/锚点 = 3 档（反思 + 验证 + 人工裁定）。
        理由：自我人格偏移与失忆是毁灭性的，宁可拒写。

        reviewers 为本次写入已取得的复核单元（缺省取环境声明 MDCG_REVIEWERS），
        只在闸启用（warn/enforce）时有意义；off 档完全不介入，行为与历史逐位相同。

        **skip_gate（issue #709 连带发现，2026-10-28）**：只给**不落记忆盘**的入队动作用
        （当前唯一调用方＝`MdCGSecure.propose`）。理由：propose 只把候选写进 inbox、
        不写业务节点，却按**目标层**要求复核单元 ⇒ enforce 下形成死锁：
        「要写 knowledge 需 reflect → 要 reflect 需裁决 → 要裁决需有提案 → 提案又要 reflect」。
        跳过闸**不等于绕过写入闸**：accept 落盘时仍走 `add()` 的同一道闸，复核不被削弱。
        层白名单与密级检查**照常执行**（这里跳过的只有 write_gate 一档）。
        **禁止用于任何真正落盘的路径**——那等于关闸。
        """
        self.require_write(sensitivity)
        layer = layer or "knowledge"
        if not self.allows_layer(layer):
            raise AccessDenied(
                f"角色 {self.role} 无权写入 {layer} 层"
                f"（可写层 {list(self.layers_allow) if self.layers_allow is not None else '不限'}）")
        if skip_gate:
            return None
        _wg_verdict = self._write_gate_verdict(layer, reviewers)
        if _wg_verdict and _wg_verdict.get("decision") in ("deny", "require_review"):
            if _wg_verdict.get("mode") == "enforce":
                raise AccessDenied(
                    f"写入 {layer} 层被分层复核闸拒绝：{_wg_verdict.get('reason')}",
                    hint=("这是 2026-10-27 使用者裁定的写入模型：无复核通道时只可写情景层；"
                          "知识/结构/自我层依次需要反思、验证与人工裁定。可：①补齐复核单元后重试；"
                          "②设置 MDCG_REVIEW_CHANNELS 声明可用通道（hive/subagent/team）；"
                          "③由使用者显式把 MDCG_WRITE_GATE 降为 warn/off。"))

# 生效条件：layer 传入、reviewers 缺省时从环境 MDCG_REVIEWERS 解析（逗号分隔）；闸为 off 时返回 None（零介入），否则返回 writegate.decide_write 的判定字典并在 warn 档把「本该复核却未过」写进 stderr 一次。
    def _write_gate_verdict(self, layer, reviewers=None):
        """分层写入严格度闸的内部单点（off 返回 None，调用方据此零介入）。"""
        import os as _os
        from . import writegate as _wg
        mode = _wg.gate_mode(_os.environ)
        if mode == "off":
            return None
        if reviewers is None:
            raw = str(_os.environ.get("MDCG_REVIEWERS") or "")
            reviewers = tuple(x.strip().lower() for x in raw.split(",") if x.strip())
        verdict = _wg.decide_write(layer, reviewers=reviewers, mode=mode)
        if mode == "warn" and verdict.get("decision") in ("deny", "require_review"):
            # issue #708：既要 stderr（宿主愿意转发时实时可见），更要**进程内计数**
            # ——后者经 `cg(op=info).write_gate` 可查，不依赖 DSH 是否转发子进程 stderr。
            import sys as _sys
            _sys.stderr.write("[writegate] warn：%s（未阻断，仅记录；计数见 cg(op=info).write_gate）"
                              % verdict.get("reason"))
            try:
                _wg.warn_record(layer, verdict.get("tier"), verdict.get("reason"))
            except Exception:            # noqa: BLE001  诊断面不得冒泡进写入路径
                pass
        return verdict

# 生效条件：要求 self._require_live() 未抛异常、self.theory_ok 为 True、self.can_admin 为 True 时通过；否则抛 AccessDenied（消息用形参 op 标记操作）。
    def require_admin(self, op: str):
        self._require_live()
        if not self.theory_ok:
            raise AccessDenied(
                f"版本层校验未通过（theory_ok=False）：管理操作 {op} 被拒")
        if not self.can_admin:
            raise AccessDenied(f"actor={self.actor} 无管理权限（{op}）")

# 生效条件：无 required 形参或模块级常量前置，返回包含 self 各属性（tenant/actor/clearance/can_write/can_admin/session/harness/unit/role/token_id/parent/auth_mode/expires_at/theory_ok/theory_version 及 layers_allow/ops_allow）的 dict；其中 layers_allow/ops_allow 为 None 时值为 None，否则 list 化。
    def as_dict(self):
        return {"tenant": self.tenant, "actor": self.actor,
                "clearance": self.clearance, "can_write": self.can_write,
                "can_admin": self.can_admin, "session": self.session,
                "harness": self.harness, "unit": self.unit,
                "role": self.role, "token_id": self.token_id,
                "parent": self.parent, "auth_mode": self.auth_mode,
                "expires_at": self.expires_at,
                "theory_ok": self.theory_ok,
                "theory_version": self.theory_version,
                "layers_allow": (None if self.layers_allow is None
                                 else list(self.layers_allow)),
                "ops_allow": (None if self.ops_allow is None
                              else list(self.ops_allow))}

# 生效条件：无前置；仅返回 tenant/actor/role/clearance/write/admin 的短摘要用于日志与排障，不含 token、密钥材料与能力白名单明细；
    def __repr__(self):
        return (f"Principal(tenant={self.tenant!r}, actor={self.actor!r}, "
                f"role={self.role!r}, clearance={self.clearance!r}, "
                f"write={self.can_write}, admin={self.can_admin})")


# 生效条件：path 为 None 时落至 datapath.aux_root()（默认 ~/.mdcg，可经 MDCG_AUX_ROOT 改）下的 _tenants.json，path 非 None（含空串）时按传入值使用，并在构造内以 self._load() 的返回填充 self.data、以 self.load_error 留存加载期损坏留痕（N124，2026-09-25：登记表损坏/不可读必须可被调用方感知，缺文件不算损坏）。
class TenantRegistry:
    """租户注册表：tenant → {root, clearance_cap, description}。

    默认位置：<registry_dir>/_tenants.json（默认 ~/.mdcg/）。
    设计意图：私有租户的 root 指向仓库外目录，开源仓库里只放 public 租户的根。

    load_error（N124，2026-09-25）：_load 遇「文件在但损坏/不可读」（JSON
    解析失败、OSError、顶层非对象）时留存原因字符串，完好/缺文件为 None——
    供 _resolve_root 等调用方把「静默回落空表」开口成 stderr 告警。
    """

# 生效条件：形参 path 为 None 时取 ~/.mdcg/_tenants.json，否则取 path；self.load_error 置 None 后 self.data 初始化为 _load() 结果（self.path 经 os.path.exists 为真且 JSON 解析为 dict 时取该 dict，否则回落 {"schema":1,"tenants":{}}；文件在但解析失败/OSError/顶层非对象时先置 self.load_error 为原因串，缺文件不置）。
    def __init__(self, path: str = None):
        if path is None:
            path = os.path.join(aux_root(), "_tenants.json")
        self.path = path
        self.load_error = None
        self.data = self._load()

# 生效条件：先置 self.load_error=None；当 self.path 经 os.path.exists 为真且内容可解析为 dict 时返回该 dict；解析失败或 OSError 时置 self.load_error=f"{类型名}: {e}"、顶层 JSON 非对象时置 self.load_error="登记表 JSON 顶层不是对象"；否则（os.path.exists 为假）不留痕；各回落分支均返回 {"schema":1,"tenants":{}}。
    def _load(self):
        self.load_error = None
        if os.path.exists(self.path):
            try:
                with open(self.path, encoding="utf-8") as f:
                    d = json.load(f)
                if isinstance(d, dict):
                    return d
                self.load_error = "登记表 JSON 顶层不是对象"
            except (ValueError, OSError) as e:
                self.load_error = f"{type(e).__name__}: {e}"
        return {"schema": 1, "tenants": {}}

# 生效条件：无 required 形参或模块级常量前置，将 self.data 以 JSON 写入 self.path + ".tmp"，随后 publish（带 Windows 短重试的 os.replace）到 self.path；目录名称为空时用 "." 创建。
    def _save(self):
        os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True)
        tmp = self.path + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(self.data, f, ensure_ascii=False, indent=1)
        publish(tmp, self.path)

# 生效条件：形参 clearance_cap 须为模块级常量 SENSITIVITY_ORDER 成员（否则 _rank 抛 AccessDenied）；形参 tenant/root 提供后写入 self.data["tenants"]（要求 self.data 含 "tenants" 键，否则 KeyError），_save 成功则返回新登记项。
    def register(self, tenant: str, root: str, clearance_cap: str = DEFAULT_SENSITIVITY,
                 description: str = ""):
        _rank(clearance_cap)
        self.data["tenants"][tenant] = {
            "root": os.path.abspath(root),
            "clearance_cap": clearance_cap,
            "description": description,
            "registered_at": time.time(),
        }
        self._save()
        return self.data["tenants"][tenant]

# 生效条件：self.data 含 "tenants" 键（否则 KeyError），且该映射中存在形参 tenant 时返回其值，否则返回 None（.get 缺键回落 None，键存在值为 None 也返回 None）。
    def get(self, tenant: str):
        return self.data["tenants"].get(tenant)

# 生效条件：self.get(tenant) 返回真值（非 None/空 dict 等）时返回 t["root"]（若 t 无 "root" 键则 KeyError）；返回假值时返回 None。
    def root_of(self, tenant: str):
        t = self.get(tenant)
        return t["root"] if t else None

# 生效条件：self.get(tenant) 返回真值时返回 t["clearance_cap"]（缺键则 KeyError）；返回假值时返回模块级常量 DEFAULT_SENSITIVITY。
    def cap_of(self, tenant: str):
        t = self.get(tenant)
        return t["clearance_cap"] if t else DEFAULT_SENSITIVITY

# 生效条件：self.data 含 "tenants" 键时返回其浅拷贝 dict(self.data["tenants"])；该键缺失时按 self.data["tenants"] 取值会 KeyError，无默认回落。
    def all(self):
        return dict(self.data["tenants"])

# 生效条件：cap 由 cap_of(tenant) 决定（形参 tenant 未注册时取模块级常量 DEFAULT_SENSITIVITY）；clearance 为假值（含 None/空串）时 want 取 cap，否则先取 clearance，再在 _rank(want) > _rank(cap) 时夹紧为 cap；actor 为假值时取 tenant；返回 Principal(...)。
    def principal_for(self, tenant: str, actor: str = None, clearance: str = None,
                      can_write: bool = True, can_admin: bool = False,
                      session: str = None) -> Principal:
        """按租户上限夹紧 clearance（调用方不能超过租户上限）。"""
        cap = self.cap_of(tenant)
        want = clearance or cap
        if _rank(want) > _rank(cap):
            want = cap
        return Principal(tenant=tenant, actor=actor or tenant, clearance=want,
                         can_write=can_write, can_admin=can_admin, session=session)


# P1-4（批次 26，外部审查报告）：工具路径参数的根白名单——ingest/export/link
# 的 path/out 是模型可控输入，原样透传 = 任意文件读（/etc/passwd 进图）与
# 任意文件写（evidence export 的 out）。校验语义：**未配置不再放开**。
#
# 语义（三级回落链）——设计者 2026-10-10 裁定「15 a」原话：
#   「记忆写入当然是用 mdcg 的实际配置库。如果没配置，以工作区优先，并配置。」
#   ① env_var（`MDCG_INGEST_ROOT` / `MDCG_EXPORT_ROOT`）已配置 ⇒ 用它
#      （os.pathsep 多根）；realpath 落根内否则拒（fail-closed）。
#   ② env_var 未配置 ⇒ 回落 **mdcg 实际配置库根** = `datapath.mdcg_root()`
#      （`MDCG_ROOT` → `paths.json` 的 "root" → 用户级数据根）。**只在确有显式
#      配置、或该库根目录已存在时**才认——否则把「已经装着记忆的默认库」弃用、
#      把根改到工作区，正是 `datapath.py` 头注记载的「记忆真源分裂成两处」事故
#      形态（issue #18 相邻问题，数据丢失级）。
#   ③ 连库根也无（空白态：env 与 paths.json 都没配，且默认库根目录不存在）
#      ⇒ **以工作区优先，并把它配置下来**：白名单根取**工作区**（进程 cwd，
#      `os.path.realpath(os.getcwd())`），并写进**用户级** `paths.json` 的
#      "root" 键（`datapath.set_user_root`，不碰 env、不写注册表）——下一轮即由
#      ② 读到它，故这是**一次性**动作（幂等，见 `_persist_workspace_root`）。
# `..` 与绝对路径穿越在 realpath 归一后自然被涵盖。支持 os.pathsep 分隔多根。
#
# ★头注更迭（2026-10-10，设计者裁定 #85 选 A）：本判据**原**为「env 未设置 =
#   放开（默认部署零行为变更，与 2026-09-19 的读放开裁定精神一致）」——该自陈
#   **已被 2026-10-10 裁定取代**（原文见上）。同时被取代的还有
#   `docs/plans/统一配置层_设计_v0.1.md:356` 的「**绝不动 `security.py` 判据**」
#   ——那一行是**登记批次**的自我约束（只登记 + 文档化语义 + 给建议），不是判据
#   真源；设计者裁定优先于该自我约束。判据真源 = 本注 + `_whitelist_roots`。

# 生效条件：无入参；`datapath.mdcg_root_configured()` 为真时返回 `datapath.mdcg_root()`；为假但 `datapath.mdcg_root()` 的目录已存在（已装记忆的默认库）时同样返回它；两者皆不成立（空白态）返回 None；datapath 解析抛异常（如 Windows 保留设备名误配的 ValueError）时原样向上抛，不吞。
def _configured_library_root():
    """②级：mdcg 实际配置库根；空白态返回 None。"""
    from . import datapath
    if datapath.mdcg_root_configured():
        return datapath.mdcg_root()
    p = datapath.mdcg_root()
    return p if os.path.isdir(p) else None


# 生效条件：无入参；恒返回 `os.path.realpath(os.getcwd())`；取不到（OSError）时返回 None。
def _workspace_root():
    """③级：工作区根 = 进程当前工作目录（cwd）。"""
    try:
        return os.path.realpath(os.getcwd())
    except OSError:
        return None


#: 工作区落配置的失败告警只打一次——判据里的写副作用不得变成刷屏源（同「不静默」纪律：失败必须可见，但只需可见一次）。
_persist_warned = [False]


# 生效条件：ws 为非空真值时以 `datapath.set_user_root(ws, "root")` 把工作区写进**用户级** `paths.json` 的 "root" 键并返回 True；写失败（只读盘/权限等）时返回 False，且只在**首次**失败时向 stderr 打一行含 ws 与 paths_file() 的告警（不抛——写配置失败不该让判据失效，白名单本轮仍以 ws 生效）。
def _persist_workspace_root(ws: str) -> bool:
    """③级「并配置」：把工作区写进用户级 `paths.json` 的 "root" 键（不碰 env）。

    为什么写 `paths.json` 的 "root"（= 记忆库根）而不是另开一个键：设计者原话的
    对象就是「mdcg 的实际配置库」，仓内既有的「把记忆根配下来」的唯一机制正是
    `datapath.set_user_root(path, "root")`（永远写用户级新位置，不写会随包更新
    消失的旧包内件）。副作用与替代见 `docs/plans/统一配置层_设计_v0.1.md` 之外的
    裁定留痕（本次实现报告）——**只在空白态触发**，且用户可手改该文件撤销。
    """
    from . import datapath
    try:
        datapath.set_user_root(ws, "root")
        return True
    except Exception as exc:                    # noqa: BLE001 —— 写配置失败不构成拒判理由
        if not _persist_warned[0]:
            _persist_warned[0] = True
            sys.stderr.write(
                "[mdcg] 记忆库根未配置且默认库根不存在，已以工作区为白名单根"
                "（%s），但把它写进 %s 失败：%s\n"
                % (ws, datapath.paths_file(), exc))
        return False


# 生效条件：env_var 环境变量去空白后非空时返回 True（该白名单根被**显式配置**）；否则 False。
# 判据单点（issue #326-①）：`_whitelist_roots` 的「已配」分支与 mdcg_ingest 的 auto 闸共用本函数，
# 调用方**不得各自再读一次 os.environ**——「同一语义两个落点判据」是本仓反复栽过的形态。
def whitelist_root_configured(env_var: str) -> bool:
    """该根白名单是否被**显式配置**（env 去空白后非空）。

    根白名单有两个语义层：①「已显式配置」→ 用它（os.pathsep 多根）；②「未配置」→
    回落记忆库根 / 工作区（#85 选 A：不再放开）。需要区分这两层的调用方（如
    `mdcg_ingest` 的 auto 闸：未配置时 `~/.dsh/sessions` 不在回落根内，无条件过闸
    会把**默认工作流整个拦死**，issue #326-① 实测的回归陷阱）必须经本函数判断。
    """
    return bool((os.environ.get(env_var) or "").strip())


# 生效条件：env_var 环境变量去空白后非空时返回其按 os.pathsep 切分、逐段 strip+expanduser+realpath 后的非空根列表；为空时走三级回落链——②级 `_configured_library_root()` 非 None 即返回 `[realpath(该根)]`；否则取 ③级 `_workspace_root()`，非 None 时先 `_persist_workspace_root(ws)` 再返回 `[ws]`；ws 为 None（连 cwd 都取不到）时抛 PermissionError（fail-closed，绝不回落放开）。
def _whitelist_roots(env_var: str) -> list:
    if whitelist_root_configured(env_var):        # 判据单点（见上），不在此另写一份
        raw = (os.environ.get(env_var) or "").strip()
        out = []
        for p in raw.split(os.pathsep):
            p = p.strip()
            if p:
                out.append(os.path.realpath(os.path.expanduser(p)))
        return out
    lib = _configured_library_root()
    if lib:
        return [os.path.realpath(os.path.expanduser(lib))]
    ws = _workspace_root()
    if not ws:
        raise PermissionError(
            "无法确定白名单根：%s 未配置、mdcg 记忆库根未配置且默认库根不存在、"
            "进程工作区（cwd）也取不到——fail-closed 拒访问"
            "（P1-4；2026-10-10 设计者裁定 #85 选 A）" % env_var)
    _persist_workspace_root(ws)
    return [ws]


# 生效条件：path 为 None/空串时直接返回（无路径参数，如 stat/action——不触任何回落与配置写）；否则拒绝「纯空白串」（issue #540，见 _whitespace_only_reason），再取 `_whitelist_roots(env_var)`（env 已配用它；未配回落 mdcg 记忆库根；连库根也无则取工作区并把它配置下来）后，对 realpath(expanduser(path)) 与每个根做「相等或以根+os.sep 为前缀」判定，任一命中即返回；全部未命中抛 PermissionError（含 what/原 path/realpath/roots 与「显式登记」指路）。
#
# issue #540：为什么不能再用 `str(path).strip() == ""` 判「没给参数」
#   str.strip() 剥的是**全部 Unicode 空白**，而其中若干在该平台上**是合法文件名字符**。
#   实测（Windows）——这一区分是本判据的全部依据：
#     · `'   '`（ASCII 空格）建文件 → OSError「Permission denied」
#     · `'\t'`（TAB）建文件        → OSError「Invalid argument」
#       ⇒ 这两个**无法成为文件名**，当作「无路径参数」不产生可落盘的落点。
#       既有契约 `test_p14_root_fallback.c4_no_path_arg` 正是钉住这一语义
#       （`None`/`''`/`'   '` 不触回落、不写配置）——**本修复保留它**。
#     · `'\u00a0'`（NBSP）建文件   → **成功**
#     · `'\u3000'`（全角空格）建文件 → **成功**
#       ⇒ 这两个**能真的建出文件**，而旧判据把它们当「无参数」早退放行
#       ⇒ 实测 `op=export` + `out="\u00a0"` 把整库正文写到白名单根之外。
#   故判据 = 「全部由空白构成，**且含 ASCII 空格以外的空白**」⇒ 拒绝。
#   **不枚举黑名单**（Unicode 空白远多于常见几个，枚举法天然滞后）。
_ASCII_SPACE = " "


def _whitespace_only_reason(path) -> str:
    """path 是否为「纯空白**且含非 ASCII 空格**」的可疑串；是则返回理由，否则返回 ""。

    为何放行纯 ASCII 空格：见上方注释——该形态无法成为文件名（OS 拒绝），
    且 `test_p14_root_fallback` 的 c4 契约把它定义为「无路径参数」。
    本函数只拦「能被平台当成实际文件名」的那一类空白。
    """
    if not isinstance(path, str) or path == "" or path.strip() != "":
        return ""
    if path.strip(_ASCII_SPACE) == "":            # 仅由 ASCII 空格构成 ⇒ 沿用 c4 契约
        return ""
    return (f"路径全部由空白字符构成，拒绝（疑似绕过根约束）：{path!r}"
            f"——纯空白不是「没给路径参数」；若要表达未提供，请传 None 或空串")


def check_path_root(path, env_var: str, what: str) -> None:
    if path is None or str(path).strip(_ASCII_SPACE) == "":
        return                                   # 无路径参数（如 stat/action）
    _why = _whitespace_only_reason(path)
    if _why:
        raise PermissionError(f"{what} {_why}")
    roots = _whitelist_roots(env_var)
    real = os.path.realpath(os.path.expanduser(str(path)))
    for r in roots:
        if real == r or real.startswith(r.rstrip(os.sep) + os.sep):
            return
    raise PermissionError(
        f"{what} 路径超出 {env_var} 白名单，拒绝访问：{path} "
        f"(realpath={real}, roots={roots})——路径类工具参数的根约束"
        f"（P1-4；2026-10-10 设计者裁定 #85 选 A：未配置时回落 mdcg 记忆库根，"
        f"连库根也无则取工作区并把它配置下来）。库外读写请**显式登记**："
        f"设 {env_var}=<根>（os.pathsep 可多根）")


# 生效条件：path 为 None/空串时直接返回（未指定落点，不触任何判定）；否则按 realpath 判三条——① 落在辅助存储根（datapath.aux_root()，master.key/_signer.key/令牌库缺省位置）内即拒；② 落在记忆库根 lib_root 内时只许**根顶层**且文件名不以 _ / . 开头（与 op=export 缺省落点 <root>/export_*.jsonl 同位；节点只在层目录里，_keys.json/_index.json 以下划线开头）；③ 目标（含悬空符号链接）已存在且 allow_overwrite 为假即拒。全部命中放行。
def check_export_target(path, lib_root, what: str,
                        allow_overwrite: bool = False) -> None:
    """导出落点的**第二道**约束（issue #211）：与白名单无关。

    为什么另立判据：check_path_root 只判「在根内」，而 MDCG_EXPORT_ROOT
    未配置时 _whitelist_roots **回落记忆库根本身**（#85 选 A）——于是
    _keys.json、_index.json、各层节点 .md 全在可写面里。缺省部署下，
    持 sustain 令牌（can_admin=False）的调用方两次 MCP 调用即可把证据包
    整份替换掉数据密钥信封库：此后属主读自己的 private 记忆一律 not_found、
    新 private 写入报 LockedError，且库内无备份（永久数据丢失）。

    本函数**只加拒绝分支、不新增放行**，且不改 evidence.write_pack
    （CLI --out 是本机用户自选落点，行为不变）。

    平台口径：比较用 normcase(realpath(...))——Windows 的盘符大小写与
    长路径前缀形态在 realpath 下不保证归一，只按原串比较即可能被绕过。
    """
    if path is None or str(path).strip(_ASCII_SPACE) == "":
        return                                   # 未指定落点（含纯 ASCII 空格，同 c4 契约）
    # issue #540：同 check_path_root —— 含非 ASCII 空白的纯空白串不是「未指定落点」。
    # 本函数是 #211 新增的第二道落点约束，当时照抄了 `str(path).strip() == ""`，
    # 故同样可被 NBSP/U+3000 等绕过（实测本函数对这批输入亦全部早退放行）。
    _why = _whitespace_only_reason(path)
    if _why:
        raise PermissionError(f"{what} {_why}")
    raw = os.path.expanduser(str(path))

    def _norm(p):
        return os.path.normcase(os.path.realpath(p))

    real = _norm(raw)

    def _inside(root):
        r = _norm(root)
        return real == r or real.startswith(r.rstrip(os.sep) + os.sep)

    from . import datapath
    try:
        _aux = datapath.aux_root()
    except ValueError as exc:
        # aux 根解析失败（典型：末段是 Windows 保留设备名，datapath 的 #57
        # 守卫直判拒绝）⇒ 无法校验落点是否落在密钥/令牌库内 ⇒ fail-closed
        # 拒绝，并把原因说清楚（不把 datapath 的原始 ValueError 直接漏给
        # 导出调用方——那会让人以为导出本身坏了）。
        raise PermissionError(
            f"{what}：辅助存储根无法解析，导出落点不可校验（fail-closed 拒）："
            f"{exc}")
    if _inside(_aux):
        raise PermissionError(
            f"{what}：out 不得落在辅助存储根内（密钥与令牌库所在）：{path}")

    if lib_root and _inside(lib_root):
        lib = _norm(lib_root)
        base = os.path.basename(real)
        if os.path.dirname(real) != lib:
            raise PermissionError(
                f"{what}：out 落在记忆库根内时只许根顶层的普通文件"
                f"（不得进层目录）：{path}")
        if base.startswith(("_", ".")):
            raise PermissionError(
                f"{what}：out 落在记忆库根内时文件名不得以 _ 或 . 开头"
                f"（密钥库/索引/元数据所在）：{path}")

    if (os.path.lexists(raw) or os.path.exists(real)) and not allow_overwrite:
        raise PermissionError(
            f"{what} 目标已存在：{path}（不覆盖已有文件）")


# 错误处置链路角色集（批次 28 分型）：restricted（错误处置标记）的可见性
# 不走密级 rank——链路角色与平级同为 internal，rank 无法区分「处置者」与
# 「扩散面」。designer=全局观测本职；orchestr=上级节点（拆解派发/裁决
# 收口）；verify=验证本职（错误标记节点是验证的对象）。
_RESTRICTED_READER_ROLES = ("designer", "orchestr", "verify")


# 生效条件：p 为 Principal 或 None；p 为 None 返回 False；p.can_admin 为真（设计者）或 p.role 属 _RESTRICTED_READER_ROLES 时返回 True，否则返回 False。
def can_read_restricted(p) -> bool:
    if p is None:
        return False
    if getattr(p, "can_admin", False):
        return True
    return getattr(p, "role", "") in _RESTRICTED_READER_ROLES


# --------------------------------------------------------------------------
# 读可见性可选钩子（跨层唯一实现，批次 68）
# --------------------------------------------------------------------------
# 为什么放在本模块：可见性单点 `_readable` 定义在**子类** `MdCGSecure`
# （mdcos.py:4028），而需要它的读数出口散在**下层/旁路模块**——trust（验证态
# 台账/describe）、provenance（派生边端点摘要与拓扑）、evolution（演化账本
# 自由文本）、forgetting（遗忘留痕 jsonl）。这些模块若 `from .mdcos import` 会
# 成环（mdcos 反向导入它们），故单点放在两侧都依赖的 security.py：
#   · `visible_to(cg, entry)`   —— 条目级判定（口径 = `_readable`）；
#   · `node_visible(cg, node_id)` —— 节点 id 级判定（索引取条目 + 缺席处置）。
# 口径与仓内既有三处可选钩子逐条一致：`mdcos._note_visible`（及本函数的
# 委派）、`backfill._readable_guard`、`linkref._known_ids`、`stg._scan/_preview`
# ——「无钩子（纯 MdCGOS，无身份/密级模型）= 不限制；有钩子 = 按判据；
# 判据异常 = 不可见（fail-closed）；宁可少读，不可 fail-open 泄漏」。
# 生效条件：cg 有可调用的 _readable 时返回 bool(判定结果)，判定抛异常返回 False；无该属性/不可调用（纯 MdCGOS）时返回 True。
def visible_to(cg, entry) -> bool:
    """条目级读可见性（cg 无 `_readable` 钩子时不设限）。"""
    fn = getattr(cg, "_readable", None)
    if not callable(fn):
        return True
    try:
        return bool(fn(entry))
    except Exception:                          # noqa: BLE001 —— 判据异常=不可见
        return False


# 生效条件：cg 无可调用 _readable 时返回 True；有钩子且索引条目为 dict 时返回 bool(visible_to(cg, 条目))（判定异常 False）；索引无该条目/条目非 dict 时返回 bool(principal.can_admin)（有身份模型但不可证可见→fail-closed，设计者不受限，否则删除后的治理留痕反被清空）。
def node_visible(cg, node_id) -> bool:
    """节点 id 级读可见性（索引取条目；**缺席 = 不可证可见 → fail-closed**）。

    缺席的两种实情：① 节点从未存在/拼错 id；② 节点已被 forget（索引无条目）。
    有身份模型时两者都不得被「当作可见」——否则 `_forgetting.jsonl` /
    `trust` 台账 / 派生边摘要就成了「已删内容」的旁路读出（N205/N226/N229）。
    例外只给 can_admin（设计者=全局观测与治理本职）：否则删除后的裁决留痕
    对唯一需要它的人也不可见（与 `_readable` 的 can_admin 豁免同款）。
    """
    fn = getattr(cg, "_readable", None)
    if not callable(fn):
        return True                            # 无身份模型：无「越权」可言
    e = ((getattr(cg, "index", None) or {}).get("nodes") or {}).get(node_id)
    if isinstance(e, dict):
        return visible_to(cg, e)
    p = getattr(cg, "principal", None)
    return bool(p is not None and getattr(p, "can_admin", False))