core/template-console/__init__.py

@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+"""⑤.7 模板配置台（Template Console）内核引擎 —— EPIC #9。
+
+配置台是**跨模板通用的内核能力**：为实施工程师提供一个无代码的配置驱动
+界面，把"模型超参 / RAG / 布局"三类配置 + 点位字典 + 版本发布统一编排，
+并将发布的配置**推送给内核**（edge-gateway / rag-kb / model-framework）。
+
+本包拆为 6 个子模块，对应 6 个 issue（同一 feature 分支承载，单 PR 关联）：
+
+- ``rbac``            (#62) 三级 RBAC 权限（admin / engineer / readonly）；
+- ``point_importer``  (#63) 点位字典 CSV 导入 + 自动校验页面（复用
+                       ``core/edge-gateway/point_dict`` 校验器，增加配置台
+                       级结果聚合 + OPC 节点格式校验 + 模板选择）；
+- ``config_store``    (#64) 配置项 CRUD（模型超参 / RAG / 布局三类，文件系统
+                       版本化 JSON 存储）；
+- ``preview``         (#65) 预览渲染引擎（布局/告警/查询 → 可预览结构化输出，
+                       对齐 iAOP-cockpit-layout-v1 widget 类型）；
+- ``release``         (#66) 版本发布 + 回滚点（基于 config_store 快照，semver）；
+- ``push_channel``    (#67) 配置台↔内核配置推送契约（JSON manifest + 校验和 +
+                       幂等性）。
+
+设计原则（对齐 PRD「可解释可溯源」与既有内核范式）：
+- 纯标准库零运行时依赖（无 pyyaml/numpy/pandas），YAML 子集用内置解析器；
+- dataclass + Enum + 类型注解 + 中文 docstring；
+- 关键决策均带 ``meaning`` / ``reason`` 字段，便于审计与可解释性。
+"""
+from __future__ import annotations
+
+from .rbac import (
+    Action,
+    Permission,
+    Role,
+    RoleKind,
+    User,
+    has_permission,
+)
+
+__all__ = [
+    "Action",
+    "Permission",
+    "Role",
+    "RoleKind",
+    "User",
+    "has_permission",
+]
+
+#: 本包版本（对齐 EPIC #9 模板配置台交付节奏）
+__version__ = "1.0.0"

core/template-console/config_store.py

@@ -0,0 +1,291 @@
+# -*- coding: utf-8 -*-
+"""⑤.7 配置项 CRUD 存储引擎 —— issue #64 / PRD ⑤.7。
+
+配置台要管理三类业务配置：**模型超参 / RAG / 布局**。这些配置是模板交付物的
+"活"部分——实施工程师按现场调参，每次改动都要**可解释、可校验、可版本化**
+（为 #66 发布/回滚提供快照源）。本模块提供基于文件系统的版本化 JSON 存储：
+
+- 三类配置各对应一个 JSON 文件（``model_params.json`` / ``rag_configs.json`` /
+  ``layout.json``），存放在一个 store 根目录下；
+- 每条配置项是一个 ``ConfigItem``（key + value + 含义 + 校验规则）；
+- 提供 ``list / get / upsert / delete`` CRUD，所有写操作都先**校验**再落盘，
+  并记录 ``updated_by`` / ``reason``（对齐 PRD「可解释可溯源」）；
+- 校验规则按类别内置（模型超参的范围/类型、RAG 的来源数、布局的 widget 类型），
+  非法值在 upsert 阶段即被拒绝，避免坏数据进入版本快照。
+
+存储格式（每类一个 JSON，内容为 ``{items: [ConfigItem, ...], schema_version}``）
+刻意简单、人可读，便于实施工程师直接查看/备份。
+
+零运行时依赖：仅用 json / dataclass / Enum / 标准库。
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any, Dict, List, Optional, Tuple
+
+
+# ---------------------------------------------------------------------------
+# 配置类别
+# ---------------------------------------------------------------------------
+
+class ConfigKind(str, Enum):
+    """三类业务配置（对齐 #64 需求）。"""
+
+    MODEL_PARAM = "model_param"   # 模型超参（学习率/迭代数/特征开关…）
+    RAG_CONFIG = "rag_config"     # RAG 知识库配置（top_k/相似度阈值/来源…）
+    LAYOUT = "layout"             # 驾驶舱布局（widget 列表）
+
+
+#: 各类别对应的存储文件名
+KIND_FILENAME: Dict[ConfigKind, str] = {
+    ConfigKind.MODEL_PARAM: "model_params.json",
+    ConfigKind.RAG_CONFIG: "rag_configs.json",
+    ConfigKind.LAYOUT: "layout.json",
+}
+
+#: 存储结构版本（schema 演进时升级，发布快照会带上）
+STORE_SCHEMA_VERSION = 1
+
+#: 驾驶舱布局允许的 widget 类型（对齐 iAOP-cockpit-layout-v1 / resin cockpit）
+ALLOWED_WIDGET_TYPES = {"process_view", "trend", "kpi_card", "alarm_panel", "nl_query"}
+
+
+# ---------------------------------------------------------------------------
+# 配置项数据模型
+# ---------------------------------------------------------------------------
+
+@dataclass
+class ConfigItem:
+    """一条配置项（可解释：带含义、更新人、原因）。"""
+
+    key: str                              # 配置键（类别内唯一，如 learning_rate）
+    value: Any                            # 配置值（标量或结构化）
+    kind: ConfigKind                      # 所属类别
+    meaning: str = ""                     # 业务含义（供配置台展示与审计）
+    updated_by: str = "system"            # 最后修改人（对接 RBAC 用户名）
+    reason: str = ""                      # 本次修改原因（可解释可溯源）
+    updated_at: str = ""                  # ISO8601 时间戳
+
+    def to_dict(self) -> dict:
+        d = asdict(self)
+        d["kind"] = self.kind.value       # 枚举序列化为字符串
+        return d
+
+    @classmethod
+    def from_dict(cls, raw: dict) -> "ConfigItem":
+        return cls(
+            key=raw["key"],
+            value=raw.get("value"),
+            kind=ConfigKind(raw.get("kind")),
+            meaning=raw.get("meaning", ""),
+            updated_by=raw.get("updated_by", "system"),
+            reason=raw.get("reason", ""),
+            updated_at=raw.get("updated_at", ""),
+        )
+
+
+def _now_iso() -> str:
+    """当前 UTC 时间 ISO8601（无时区歧义）。"""
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+# ---------------------------------------------------------------------------
+# 校验（按类别内置规则）
+# ---------------------------------------------------------------------------
+
+@dataclass
+class ValidationResult:
+    """配置项校验结果。"""
+
+    ok: bool
+    errors: List[str] = field(default_factory=list)
+
+    def __bool__(self) -> bool:
+        return self.ok
+
+
+def validate_item(kind: ConfigKind, key: str, value: Any) -> ValidationResult:
+    """按类别校验配置项的 key/value 合法性。
+
+    校验规则（配置台 upsert 前置门禁，防止坏数据进快照）：
+    - 通用：key 非空、匹配 ``[a-z0-9_.-]+``；
+    - model_param：value 为标量（int/float/bool/str）或标量列表；
+    - rag_config：top_k 为 1~50 的正整数、similarity_threshold 为 0~1 浮点、
+      sources 为非空字符串列表；
+    - layout：value 为 widget 列表，每个 widget 有合法 type 与 x/y/w/h。
+    """
+    errors: List[str] = []
+    if not key or not isinstance(key, str):
+        errors.append("key 不能为空")
+    elif not re.match(r"^[a-z0-9_.\-]+$", key):
+        errors.append(f"key '{key}' 仅允许小写字母/数字/._-")
+
+    if kind == ConfigKind.MODEL_PARAM:
+        if not isinstance(value, (int, float, bool, str, list)):
+            errors.append("model_param 的 value 必须为标量或标量列表")
+        elif isinstance(value, list) and any(
+                not isinstance(v, (int, float, bool, str)) for v in value):
+            errors.append("model_param 列表 value 仅允许标量元素")
+        # 常见超参范围提示（软约束，仅对已知键）
+        if key == "learning_rate" and isinstance(value, (int, float)):
+            if not (0 < value < 1):
+                errors.append("learning_rate 应在 (0, 1) 区间")
+        if key == "iterations" and isinstance(value, int):
+            if value <= 0:
+                errors.append("iterations 必须为正整数")
+
+    elif kind == ConfigKind.RAG_CONFIG:
+        if key == "top_k":
+            if not (isinstance(value, int) and 1 <= value <= 50):
+                errors.append("top_k 必须为 1~50 的整数")
+        elif key == "similarity_threshold":
+            if not (isinstance(value, (int, float)) and 0 <= value <= 1):
+                errors.append("similarity_threshold 必须为 0~1 的数")
+        elif key == "sources":
+            if not (isinstance(value, list) and value
+                    and all(isinstance(s, str) and s for s in value)):
+                errors.append("sources 必须为非空字符串列表")
+
+    elif kind == ConfigKind.LAYOUT:
+        if not isinstance(value, list):
+            errors.append("layout 的 value 必须为 widget 列表")
+        else:
+            for i, w in enumerate(value):
+                if not isinstance(w, dict):
+                    errors.append(f"widget[{i}] 必须为对象")
+                    continue
+                wt = w.get("type")
+                if wt not in ALLOWED_WIDGET_TYPES:
+                    errors.append(
+                        f"widget[{i}] 非法 type '{wt}'（合法：{sorted(ALLOWED_WIDGET_TYPES)})")
+                for coord in ("x", "y", "w", "h"):
+                    if not isinstance(w.get(coord), int) or w.get(coord) < 0:
+                        errors.append(f"widget[{i}] {coord} 必须为非负整数")
+
+    return ValidationResult(ok=not errors, errors=errors)
+
+
+# ---------------------------------------------------------------------------
+# 存储引擎
+# ---------------------------------------------------------------------------
+
+class ConfigStore:
+    """基于文件系统的版本化配置存储（三类配置各一 JSON）。
+
+    用法：
+        store = ConfigStore("/path/to/store")
+        store.upsert(ConfigKind.MODEL_PARAM, "learning_rate", 0.001,
+                     meaning="学习率", updated_by="li", reason="首次标定")
+        items = store.list(ConfigKind.MODEL_PARAM)
+    """
+
+    def __init__(self, root: str) -> None:
+        self.root = root
+        os.makedirs(root, exist_ok=True)
+
+    # -- 路径 --
+    def _path(self, kind: ConfigKind) -> str:
+        return os.path.join(self.root, KIND_FILENAME[kind])
+
+    def _read(self, kind: ConfigKind) -> List[ConfigItem]:
+        path = self._path(kind)
+        if not os.path.isfile(path):
+            return []
+        with open(path, "r", encoding="utf-8") as fh:
+            blob = json.load(fh)
+        return [ConfigItem.from_dict(r) for r in blob.get("items", [])]
+
+    def _write(self, kind: ConfigKind, items: List[ConfigItem]) -> None:
+        blob = {
+            "schema_version": STORE_SCHEMA_VERSION,
+            "kind": kind.value,
+            "items": [it.to_dict() for it in items],
+        }
+        path = self._path(kind)
+        # 先写临时文件再替换，避免写一半被读到（原子写）
+        tmp = path + ".tmp"
+        with open(tmp, "w", encoding="utf-8") as fh:
+            json.dump(blob, fh, ensure_ascii=False, indent=2)
+        os.replace(tmp, path)
+
+    # -- 查询 --
+    def list(self, kind: ConfigKind) -> List[ConfigItem]:
+        """列出某类全部配置项。"""
+        return self._read(kind)
+
+    def get(self, kind: ConfigKind, key: str) -> Optional[ConfigItem]:
+        """取单条配置项（不存在返回 None）。"""
+        for it in self._read(kind):
+            if it.key == key:
+                return it
+        return None
+
+    # -- 写 --
+    def upsert(
+        self,
+        kind: ConfigKind,
+        key: str,
+        value: Any,
+        meaning: str = "",
+        updated_by: str = "system",
+        reason: str = "",
+    ) -> ConfigItem:
+        """新增或更新一条配置项（先校验，再落盘）。
+
+        Raises:
+            ValueError: 校验失败（带全部错误明细）。
+        """
+        vr = validate_item(kind, key, value)
+        if not vr:
+            raise ValueError(f"配置项校验失败 [{kind.value}:{key}]：{'; '.join(vr.errors)}")
+        items = self._read(kind)
+        now = _now_iso()
+        existing_idx = next((i for i, it in enumerate(items) if it.key == key), None)
+        item = ConfigItem(
+            key=key, value=value, kind=kind, meaning=meaning,
+            updated_by=updated_by, reason=reason, updated_at=now,
+        )
+        if existing_idx is None:
+            items.append(item)
+        else:
+            items[existing_idx] = item
+        self._write(kind, items)
+        return item
+
+    def delete(self, kind: ConfigKind, key: str) -> bool:
+        """删除一条配置项。返回是否实际删除。"""
+        items = self._read(kind)
+        new_items = [it for it in items if it.key != key]
+        if len(new_items) == len(items):
+            return False
+        self._write(kind, new_items)
+        return True
+
+    # -- 快照（供 #66 release 使用） --
+    def snapshot(self) -> Dict[str, Any]:
+        """全量配置快照（三类聚合，供发布版本固化）。"""
+        return {
+            "schema_version": STORE_SCHEMA_VERSION,
+            "captured_at": _now_iso(),
+            "kinds": {
+                kind.value: [it.to_dict() for it in self._read(kind)]
+                for kind in ConfigKind
+            },
+        }
+
+    def restore(self, snapshot: Dict[str, Any]) -> None:
+        """从快照恢复全部配置（#66 回滚入口）。"""
+        kinds = snapshot.get("kinds", {})
+        for kind in ConfigKind:
+            raw_items = kinds.get(kind.value, [])
+            items = [ConfigItem.from_dict(r) for r in raw_items]
+            self._write(kind, items)
+
+    def item_counts(self) -> Dict[str, int]:
+        """各类配置项数量（配置台仪表盘用）。"""
+        return {kind.value: len(self._read(kind)) for kind in ConfigKind}

core/template-console/point_importer.py

@@ -0,0 +1,318 @@
+# -*- coding: utf-8 -*-
+"""⑤.7 点位字典 CSV 导入 + 自动校验页面 —— issue #63 / PRD ⑤.7。
+
+配置台的"导入页面"要解决：实施工程师拿着 DCS 点表（Excel 转 CSV）粘进配置台，
+**一次性看到所有问题**（表头错/量纲错/重复点号/采样率非正/协议非法/OPC 节点
+格式错），而不是改一条报一条。本模块是导入页面的后端引擎。
+
+**复用而非重造**：点位字典的 schema/加载/校验（量纲/数据类型/采样率/重复点号/
+协议）已由内核 ``core/edge-gateway/point_dict``（loader/validator/schema）实现
+并被边缘网关正式使用。本模块在其基础上增加**配置台专属**校验维度：
+
+1. OPC 节点格式校验（OPC UA 节点须形如 ``ns=<数字>;s=<名>`` 或 PLC 寄存器
+   ``holding:<数字>`` / ``coil:<数字>``，与 simulator/opcua 驱动约定一致）；
+2. 表头列顺序严格对齐（实施工程师照表填列，列序错位是高频错误）；
+3. 行级结果聚合为 ``ImportRowIssue``（行号 + 严重级别 + 问题 + 修复建议），
+   供配置台前端逐行渲染、按严重级别过滤；
+4. 模板选择（resin/ti）：不同行业模板的合法量纲集合不同（如树脂含 rpm/mmol·g⁻¹），
+   导入时按模板切换校验基线。
+
+零运行时依赖：复用 ``point_dict`` 子包（纯标准库 csv/dataclass）。
+"""
+from __future__ import annotations
+
+import os
+import re
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Dict, List, Optional, Tuple
+
+# 复用内核 edge-gateway 的点位字典加载/校验（schema/validator/loader）
+# 在 tests/_bootstrap.py 中已把 core/edge-gateway 加入 sys.path，
+# 故此处以顶层包 point_dict 引用（与 edge-gateway 自身测试一致）。
+from point_dict import (  # noqa: E402
+    CSV_HEADERS,
+    VALID_PROTOCOLS,
+    VALID_UNITS,
+    Point,
+    PointDict,
+    load_point_dict_csv,
+)
+from point_dict import schema as _pd_schema  # noqa: E402
+
+
+class Severity(str, Enum):
+    """问题严重级别（配置台前端据此着色/过滤）。"""
+
+    ERROR = "error"    # 阻断：不修复无法入库
+    WARN = "warn"      # 警告：可入库但建议复核（如 OPC 节点为空）
+
+
+class TemplateKind(str, Enum):
+    """行业模板（决定合法量纲等校验基线）。"""
+
+    TI = "ti"          # 氯化/化工通用（templates/ti-cl4）
+    RESIN = "resin"    # 吸附树脂（templates/resin，含 rpm / mmol/g）
+
+
+# OPC 节点格式（与 simulator/opcua/s7 驱动约定一致）：
+#   ns=2;s=CLF.Temp            —— OPC UA 节点（namespace + 字符串 id）
+#   holding:40010 / coil:1     —— Modbus 寄存器（保持/线圈 + 地址）
+_OPC_UA_RE = re.compile(r"^ns=\d+;s=[^\s,]+$")
+_MODBUS_RE = re.compile(r"^(holding|coil|input|discrete):(\d+)$")
+
+
+def _unit_set(template: TemplateKind) -> set:
+    """按模板返回合法量纲集合（树脂含 rpm/mmol·g⁻¹ 等扩展）。"""
+    base = set(VALID_UNITS)
+    if template == TemplateKind.RESIN:
+        # VALID_UNITS 已含树脂扩展（rpm/mmol/g），直接复用
+        return base
+    # ti 模板：移除树脂专属量纲，避免化工模板误用树脂量纲
+    base.discard("rpm")
+    base.discard("mmol/g")
+    return base
+
+
+@dataclass
+class ImportRowIssue:
+    """导入页一行的问题（行号 + 严重级别 + 问题 + 修复建议，可解释）。"""
+
+    row: int                 # CSV 行号（表头=1，数据从 2 起）
+    severity: Severity
+    code: str                # 错误码（对齐 point_dict.validator 的 code + 本模块扩展）
+    message: str             # 问题描述
+    suggestion: str = ""     # 修复建议（供配置台"一键修复"提示）
+
+    def to_dict(self) -> dict:
+        return {
+            "row": self.row, "severity": self.severity.value,
+            "code": self.code, "message": self.message,
+            "suggestion": self.suggestion,
+        }
+
+
+@dataclass
+class ImportReport:
+    """导入校验报告（配置台导入页面数据模型）。"""
+
+    template: TemplateKind
+    total_rows: int = 0              # 数据行数
+    issues: List[ImportRowIssue] = field(default_factory=list)
+    loaded_points: int = 0           # 成功加载的点数
+    file_path: str = ""
+
+    @property
+    def ok(self) -> bool:
+        """无 ERROR 级问题即可入库（WARN 不阻断）。"""
+        return not any(i.severity == Severity.ERROR for i in self.issues)
+
+    @property
+    def error_count(self) -> int:
+        return sum(1 for i in self.issues if i.severity == Severity.ERROR)
+
+    @property
+    def warn_count(self) -> int:
+        return sum(1 for i in self.issues if i.severity == Severity.WARN)
+
+    def summary(self) -> str:
+        """人类可读汇总（配置台导入结果横幅）。"""
+        status = "通过" if self.ok else "未通过"
+        return (f"导入校验{status}：{self.loaded_points} 点 / "
+                f"{self.total_rows} 行，错误 {self.error_count}，警告 {self.warn_count}")
+
+    def to_dict(self) -> dict:
+        return {
+            "template": self.template.value,
+            "total_rows": self.total_rows,
+            "loaded_points": self.loaded_points,
+            "ok": self.ok,
+            "error_count": self.error_count,
+            "warn_count": self.warn_count,
+            "summary": self.summary(),
+            "issues": [i.to_dict() for i in self.issues],
+        }
+
+
+# ---------------------------------------------------------------------------
+# 校验扩展
+# ---------------------------------------------------------------------------
+
+def _validate_opc_node(point: Point) -> List[ImportRowIssue]:
+    """OPC 节点格式校验（配置台扩展维度）。
+
+    约定（与驱动注册表对齐）：
+    - 协议 opcua：节点须匹配 ``ns=<数字>;s=<名>``；
+    - 协议 modbus：节点须匹配 ``holding/coil/input/discrete:<数字>``；
+    - 协议 simulator：节点可空，或任意上述格式（演示用，宽松）；
+    - 节点为空：WARN（可入库但运行时无法采集，建议补全）。
+    """
+    out: List[ImportRowIssue] = []
+    node = (point.opc_node or "").strip()
+    if not node:
+        out.append(ImportRowIssue(
+            row=point.row_number, severity=Severity.WARN, code="empty_opc_node",
+            message=f"第{point.row_number}行 opcNode 为空",
+            suggestion="运行时无法采集，建议补全 OPC UA 节点或 PLC 寄存器地址",
+        ))
+        return out
+
+    proto = (point.protocol or "").lower()
+    ok_ua = bool(_OPC_UA_RE.match(node))
+    ok_mb = bool(_MODBUS_RE.match(node))
+    if proto == "opcua" and not ok_ua:
+        out.append(ImportRowIssue(
+            row=point.row_number, severity=Severity.ERROR, code="bad_opc_node",
+            message=(f"第{point.row_number}行 opcNode '{node}' 不符合 OPC UA "
+                     f"格式 ns=<ns>;s=<name>"),
+            suggestion="示例：ns=2;s=CLF.Temp",
+        ))
+    elif proto == "modbus" and not ok_mb:
+        out.append(ImportRowIssue(
+            row=point.row_number, severity=Severity.ERROR, code="bad_opc_node",
+            message=(f"第{point.row_number}行 opcNode '{node}' 不符合 Modbus "
+                     f"格式 holding/coil/input/discrete:<addr>"),
+            suggestion="示例：holding:40010",
+        ))
+    elif proto not in ("opcua", "modbus") and not (ok_ua or ok_mb):
+        # simulator/s7/... 节点为空已 WARN；非空但格式都不符则 WARN（宽松）
+        out.append(ImportRowIssue(
+            row=point.row_number, severity=Severity.WARN, code="bad_opc_node",
+            message=(f"第{point.row_number}行 opcNode '{node}' 既非 OPC UA 也非 "
+                     f"Modbus 格式"),
+            suggestion="确认节点格式或清空（演示协议可空）",
+        ))
+    return out
+
+
+def _validate_header_order(headers: List[str]) -> List[ImportRowIssue]:
+    """表头列顺序严格对齐（列序错位是实施工程师高频错误）。"""
+    out: List[ImportRowIssue] = []
+    if not headers:
+        out.append(ImportRowIssue(
+            row=1, severity=Severity.ERROR, code="empty_header",
+            message="CSV 缺少表头行",
+            suggestion=f"表头应为：{','.join(CSV_HEADERS)}",
+        ))
+        return out
+    missing = [h for h in CSV_HEADERS if h not in headers]
+    for h in missing:
+        out.append(ImportRowIssue(
+            row=1, severity=Severity.ERROR, code="missing_column",
+            message=f"表头缺少必填列：{h}",
+            suggestion=f"补列 {h}（完整表头：{','.join(CSV_HEADERS)}）",
+        ))
+    if headers[: len(CSV_HEADERS)] != CSV_HEADERS and not missing:
+        out.append(ImportRowIssue(
+            row=1, severity=Severity.WARN, code="bad_column_order",
+            message=f"表头列顺序与标准不一致：{headers}",
+            suggestion=f"标准顺序：{','.join(CSV_HEADERS)}",
+        ))
+    return out
+
+
+def _convert_validator_issues(report: "object", severity_for: Dict[str, Severity]) -> List[ImportRowIssue]:
+    """把内核 validator.ValidationReport.issues 转成 ImportRowIssue。"""
+    out: List[ImportRowIssue] = []
+    for it in getattr(report, "issues", []):
+        sev = severity_for.get(it.code, Severity.ERROR)
+        out.append(ImportRowIssue(
+            row=it.row, severity=sev, code=it.code, message=it.message,
+        ))
+    return out
+
+
+# 内核 validator 错误码 → 严重级别映射
+_SEVERITY_MAP: Dict[str, Severity] = {
+    "missing_column": Severity.ERROR,
+    "missing_field": Severity.ERROR,
+    "bad_unit": Severity.ERROR,
+    "bad_data_type": Severity.ERROR,
+    "bad_sample_rate": Severity.ERROR,
+    "bad_protocol": Severity.ERROR,
+    "dup_point": Severity.ERROR,
+}
+
+
+# ---------------------------------------------------------------------------
+# 导入入口
+# ---------------------------------------------------------------------------
+
+def import_csv(
+    path: str,
+    template: TemplateKind = TemplateKind.TI,
+    extra_unit_check: bool = True,
+) -> Tuple[PointDict, ImportReport]:
+    """导入并校验点位字典 CSV（配置台导入页面后端入口）。
+
+    Args:
+        path: CSV 文件路径（UTF-8，9 列表头）；
+        template: 行业模板（决定合法量纲集合，resin/ti）；
+        extra_unit_check: 是否按模板收窄量纲集合做额外校验。
+
+    Returns:
+        (PointDict, ImportReport)：加载的点位模型 + 校验报告。
+        报告 ``ok`` 为 True 即可入库；WARN 不阻断。
+    """
+    report = ImportReport(template=template, file_path=path)
+
+    # 1) 表头校验（先读表头行）
+    import csv as _csv
+    with open(path, "r", encoding="utf-8-sig") as fh:
+        reader = _csv.reader(fh)
+        rows = list(reader)
+    headers = [c.strip() for c in rows[0]] if rows else []
+    report.issues.extend(_validate_header_order(headers))
+
+    # 2) 加载 + 内核校验（量纲/数据类型/采样率/重复点号/协议）
+    point_dict = load_point_dict_csv(path)
+    report.loaded_points = len(point_dict)
+    report.total_rows = len(point_dict.points)
+
+    from point_dict.validator import validate_point_dict
+    kernel_report = validate_point_dict(point_dict, headers)
+    report.issues.extend(_convert_validator_issues(kernel_report, _SEVERITY_MAP))
+
+    # 3) 模板级量纲收窄（resin 才允许 rpm/mmol·g⁻¹）
+    if extra_unit_check:
+        allowed_units = _unit_set(template)
+        for p in point_dict.points:
+            if p.unit and p.unit not in allowed_units:
+                # 内核 validator 已按全集校验过；这里只补充模板级差异提示
+                if p.unit in ("rpm", "mmol/g") and template == TemplateKind.TI:
+                    report.issues.append(ImportRowIssue(
+                        row=p.row_number, severity=Severity.ERROR,
+                        code="template_unit_mismatch",
+                        message=(f"第{p.row_number}行 量纲 '{p.unit}' 为树脂模板专属，"
+                                 f"当前导入的是 {template.value} 模板"),
+                        suggestion="切换模板为 resin，或修正量纲",
+                    ))
+
+    # 4) OPC 节点格式校验（配置台扩展）
+    for p in point_dict.points:
+        report.issues.extend(_validate_opc_node(p))
+
+    # 行号排序，便于配置台逐行展示
+    report.issues.sort(key=lambda i: (i.row, i.code))
+    return point_dict, report
+
+
+def import_csv_string(
+    content: str,
+    template: TemplateKind = TemplateKind.TI,
+    encoding: str = "utf-8",
+) -> Tuple[PointDict, ImportReport]:
+    """从 CSV 文本导入（配置台粘贴框场景，落临时文件后复用 import_csv）。"""
+    import tempfile
+    tmp = tempfile.NamedTemporaryFile(
+        mode="w", encoding=encoding, suffix=".csv", delete=False)
+    try:
+        tmp.write(content)
+        tmp.flush()
+        tmp.close()
+        return import_csv(tmp.name, template=template)
+    finally:
+        try:
+            os.unlink(tmp.name)
+        except OSError:
+            pass

core/template-console/rbac.py

@@ -0,0 +1,310 @@
+# -*- coding: utf-8 -*-
+"""⑤.7 配置台三级 RBAC 权限模型 —— issue #62 / PRD ⑤.7。
+
+配置台面向**多角色协作**：实施工程师配模板，行业工程师调参数，运维/管理者
+发布上线。直接对所有人开放写权限会带来误改与不可溯源风险。本模块用三级
+RBAC（基于角色的访问控制）锁定"谁能对哪类配置做什么"，并把每次权限判定
+的**理由**一并返回，对齐 PRD「可解释可溯源」。
+
+三级角色（由低到高，后者继承前者全部权限）：
+
+- ``readonly``  （只读）：查看配置 / 预览 / 历史版本，不可写；
+- ``engineer`` （行业工程师）：只读权限 + 编辑/校验/导入配置（模型超参 /
+                  RAG / 布局 / 点位字典），但**不能发布与回滚**；
+- ``admin``     （管理员）：工程师权限 + 发布 / 回滚 / 推送内核 / 用户管理。
+
+权限判定核心为 ``has_permission(user, resource, action)``，返回
+``PermissionDecision``（allow + reason），便于配置台前端把"为什么拒绝"
+直接展示给操作者，而不是一个干瘪的 403。
+
+零运行时依赖：仅用 dataclass / Enum / 标准库。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Dict, List, Optional, Set
+
+
+# ---------------------------------------------------------------------------
+# 权限维度：资源 × 动作
+# ---------------------------------------------------------------------------
+
+class Resource(str, Enum):
+    """配置台可管控的资源（对齐 #63~#67 子模块）。"""
+
+    POINT_DICT = "point_dict"      # 点位字典（#63）
+    MODEL_PARAM = "model_param"    # 模型超参配置（#64）
+    RAG_CONFIG = "rag_config"      # RAG 知识库配置（#64）
+    LAYOUT = "layout"              # 驾驶舱布局配置（#64/#65）
+    PREVIEW = "preview"            # 预览（#65）
+    RELEASE = "release"            # 版本发布/回滚（#66）
+    PUSH = "push"                  # 配置推送内核（#67）
+    USER = "user"                  # 用户/角色管理
+
+
+class Action(str, Enum):
+    """对资源可执行的动作。"""
+
+    VIEW = "view"        # 查看 / 预览 / 列表
+    EDIT = "edit"        # 新增 / 修改 / 删除 / 导入 / 校验
+    PUBLISH = "publish"  # 发布版本 / 回滚 / 推送内核
+    MANAGE = "manage"    # 用户与角色管理
+
+
+class RoleKind(str, Enum):
+    """三级角色枚举（值即配置资产中的角色标识）。"""
+
+    READONLY = "readonly"
+    ENGINEER = "engineer"
+    ADMIN = "admin"
+
+
+# 各资源的「写」动作等价集合：EDIT 含新增/修改/删除/导入/校验。
+# PUBLISH 含发布/回滚/推送。这样配置台前端只需关心粗粒度动作。
+_WRITE_ACTIONS: Set[Action] = {Action.EDIT, Action.PUBLISH, Action.MANAGE}
+
+
+# ---------------------------------------------------------------------------
+# 权限模型
+# ---------------------------------------------------------------------------
+
+@dataclass(frozen=True)
+class Permission:
+    """一条权限授予（角色 → 资源 → 动作）。
+
+    ``meaning`` 解释该权限的业务含义，用于审计日志与配置台权限矩阵展示。
+    注意：权限**匹配**基于 ``resource:action``（资源×动作），与授予角色无关——
+    这正是角色继承能生效的关键（admin 继承 engineer 的 edit，匹配键相同）。
+    ``role`` 仅作为审计元数据，记录"是谁授予的"。
+    """
+
+    role: RoleKind
+    resource: Resource
+    action: Action
+    meaning: str = ""
+
+    def key(self) -> str:
+        """权限匹配键（资源:动作）—— 角色继承据此累计。"""
+        return f"{self.resource.value}:{self.action.value}"
+
+    def audit_key(self) -> str:
+        """审计唯一键（角色/资源/动作三元组，含授予者）。"""
+        return f"{self.role.value}:{self.resource.value}:{self.action.value}"
+
+
+@dataclass
+class Role:
+    """一个角色：权限集合 + 继承的父角色。"""
+
+    kind: RoleKind
+    label: str                              # 中文展示名
+    permissions: List[Permission] = field(default_factory=list)
+    inherits: Optional[RoleKind] = None     # 继承的低一级角色
+    description: str = ""                   # 角色职责说明（可解释性）
+
+    def permission_keys(self) -> Set[str]:
+        """本角色直接授予的权限键集合。"""
+        return {p.key() for p in self.permissions}
+
+
+@dataclass
+class User:
+    """配置台用户。"""
+
+    username: str
+    role: RoleKind
+    display_name: str = ""
+    # 可选资源级收窄：即便角色允许，列表中的资源也会被额外限制为只读。
+    # 用于"只允许工程师改某几类配置"的细粒度场景。
+    restricted_to_readonly: List[Resource] = field(default_factory=list)
+
+
+@dataclass
+class PermissionDecision:
+    """``has_permission`` 的判定结果（带理由，可解释）。"""
+
+    allow: bool
+    reason: str            # 人类可读的判定理由（允许/拒绝原因）
+    role: RoleKind
+    resource: Resource
+    action: Action
+    source: str = "explicit"  # explicit（本角色直接授予）/ inherited（继承自父角色）
+
+
+# ---------------------------------------------------------------------------
+# 角色注册表：三级权限矩阵（对齐 PRD ⑤.7「三级 RBAC」）
+# ---------------------------------------------------------------------------
+
+def _build_role_registry() -> Dict[RoleKind, Role]:
+    """构建三级角色及其权限矩阵。
+
+    权限设计依据（PRD ⑤.7）：
+    - readonly：可查看所有配置/预览/历史，但不能改、不能发；
+    - engineer：在 readonly 基础上，可编辑/校验/导入四类业务配置，
+                但**发布/回滚/推送/用户管理仍归 admin**（避免未经评审上线）；
+    - admin：在 engineer 基础上，可发布/回滚/推送 + 管理用户角色。
+    """
+    ro = Role(
+        kind=RoleKind.READONLY,
+        label="只读",
+        description="实施/运维只读角色：查看配置、预览、历史版本，不可写。",
+        permissions=[
+            Permission(RoleKind.READONLY, Resource.POINT_DICT, Action.VIEW,
+                       "查看点位字典与校验报告"),
+            Permission(RoleKind.READONLY, Resource.MODEL_PARAM, Action.VIEW,
+                       "查看模型超参配置"),
+            Permission(RoleKind.READONLY, Resource.RAG_CONFIG, Action.VIEW,
+                       "查看 RAG 知识库配置"),
+            Permission(RoleKind.READONLY, Resource.LAYOUT, Action.VIEW,
+                       "查看驾驶舱布局配置"),
+            Permission(RoleKind.READONLY, Resource.PREVIEW, Action.VIEW,
+                       "查看配置预览"),
+            Permission(RoleKind.READONLY, Resource.RELEASE, Action.VIEW,
+                       "查看历史发布版本"),
+        ],
+    )
+
+    engineer = Role(
+        kind=RoleKind.ENGINEER,
+        label="行业工程师",
+        inherits=RoleKind.READONLY,
+        description="行业工程师：编辑/校验/导入业务配置，但不能发布与推送。",
+        permissions=[
+            Permission(RoleKind.ENGINEER, Resource.POINT_DICT, Action.EDIT,
+                       "导入/编辑/校验点位字典 CSV"),
+            Permission(RoleKind.ENGINEER, Resource.MODEL_PARAM, Action.EDIT,
+                       "调整模型超参配置"),
+            Permission(RoleKind.ENGINEER, Resource.RAG_CONFIG, Action.EDIT,
+                       "编辑 RAG 知识库配置"),
+            Permission(RoleKind.ENGINEER, Resource.LAYOUT, Action.EDIT,
+                       "编辑驾驶舱布局配置"),
+            Permission(RoleKind.ENGINEER, Resource.PREVIEW, Action.VIEW,
+                       "预览配置效果（编辑后必看）"),
+        ],
+    )
+
+    admin = Role(
+        kind=RoleKind.ADMIN,
+        label="管理员",
+        inherits=RoleKind.ENGINEER,
+        description="管理员：在工程师基础上负责发布/回滚/推送与用户管理。",
+        permissions=[
+            Permission(RoleKind.ADMIN, Resource.RELEASE, Action.PUBLISH,
+                       "发布新版本与回滚到历史版本"),
+            Permission(RoleKind.ADMIN, Resource.PUSH, Action.PUBLISH,
+                       "把已发布配置推送给内核"),
+            Permission(RoleKind.ADMIN, Resource.USER, Action.MANAGE,
+                       "管理用户与角色分配"),
+            Permission(RoleKind.ADMIN, Resource.POINT_DICT, Action.PUBLISH,
+                       "确认点位字典上线（审批环节）"),
+            Permission(RoleKind.ADMIN, Resource.MODEL_PARAM, Action.PUBLISH,
+                       "确认模型超参上线"),
+            Permission(RoleKind.ADMIN, Resource.LAYOUT, Action.PUBLISH,
+                       "确认布局上线"),
+        ],
+    )
+
+    return {RoleKind.READONLY: ro, RoleKind.ENGINEER: engineer, RoleKind.ADMIN: admin}
+
+
+_ROLES: Dict[RoleKind, Role] = _build_role_registry()
+
+
+def get_role(kind: RoleKind) -> Role:
+    """获取角色定义。"""
+    return _ROLES[kind]
+
+
+def all_roles() -> List[Role]:
+    """全部角色（按权限由低到高）。"""
+    return [_ROLES[RoleKind.READONLY], _ROLES[RoleKind.ENGINEER], _ROLES[RoleKind.ADMIN]]
+
+
+def effective_permissions(kind: RoleKind) -> Set[str]:
+    """角色有效权限键（含继承链）。
+
+    继承解析：admin 继承 engineer 继承 readonly，递归向上累计权限键。
+    """
+    role = _ROLES[kind]
+    keys: Set[str] = set(role.permission_keys())
+    if role.inherits is not None:
+        keys |= effective_permissions(role.inherits)
+    return keys
+
+
+# ---------------------------------------------------------------------------
+# 判定 API
+# ---------------------------------------------------------------------------
+
+def has_permission(
+    user: User,
+    resource: Resource,
+    action: Action,
+) -> PermissionDecision:
+    """判定用户对某资源执行某动作是否被允许（带理由）。
+
+    判定顺序：
+    1. 计算角色有效权限（含继承），命中即允许并标注来源（本角色/继承）；
+    2. 命中后若该资源在用户 ``restricted_to_readonly`` 列表且动作是写动作，
+       则降级拒绝（细粒度收窄）；
+    3. 未命中则拒绝，理由标注缺失的权限三元组。
+
+    Args:
+        user: 配置台用户；
+        resource: 目标资源；
+        action: 目标动作。
+
+    Returns:
+        PermissionDecision：allow + reason（可直接展示给操作者）。
+    """
+    target = f"{resource.value}:{action.value}"
+    eff = effective_permissions(user.role)
+
+    # 细粒度收窄：即便角色允许，特定资源也被限制为只读
+    if resource in user.restricted_to_readonly and action in _WRITE_ACTIONS:
+        return PermissionDecision(
+            allow=False,
+            reason=(f"用户 '{user.username}' 对资源 '{resource.value}' 被收窄为只读，"
+                    f"禁止执行 '{action.value}' 动作"),
+            role=user.role, resource=resource, action=action, source="restricted",
+        )
+
+    if target in eff:
+        # 判定来源：本角色直接授予 or 继承自父角色
+        own = get_role(user.role).permission_keys()
+        source = "explicit" if target in own else "inherited"
+        src_label = "本角色直接授予" if source == "explicit" else "继承自低级角色"
+        return PermissionDecision(
+            allow=True,
+            reason=(f"用户 '{user.username}'（{get_role(user.role).label}）"
+                    f"允许对 '{resource.value}' 执行 '{action.value}'（{src_label}）"),
+            role=user.role, resource=resource, action=action, source=source,
+        )
+
+    return PermissionDecision(
+        allow=False,
+        reason=(f"用户 '{user.username}'（{get_role(user.role).label}）缺少权限 "
+                f"{user.role.value}:{resource.value}:{action.value}；"
+                f"该动作需更高角色或审批"),
+        role=user.role, resource=resource, action=action, source="denied",
+    )
+
+
+def can_publish(user: User) -> bool:
+    """便捷判定：用户是否具备发布（发布/回滚/推送）能力。"""
+    return has_permission(user, Resource.RELEASE, Action.PUBLISH).allow
+
+
+def user_summary(user: User) -> Dict[str, object]:
+    """用户权限概览（供配置台用户卡片/审计日志展示）。"""
+    role = get_role(user.role)
+    return {
+        "username": user.username,
+        "display_name": user.display_name or user.username,
+        "role": user.role.value,
+        "role_label": role.label,
+        "description": role.description,
+        "effective_permission_count": len(effective_permissions(user.role)),
+        "restricted_to_readonly": [r.value for r in user.restricted_to_readonly],
+    }

core/template-console/tests/_bootstrap.py

@@ -0,0 +1,26 @@
+# -*- coding: utf-8 -*-
+"""测试引导：把 `core/template-console` 以包名 `template_console` 挂载到 sys.modules。
+
+目录名 `template-console` 含连字符，无法直接以包名 import；挂载后模块内
+相对导入（`from .rbac import ...`）在 unittest 发现机制下可正常解析。
+
+同时把兄弟内核目录 `core/edge-gateway` 加入 sys.path，使 point_importer
+可复用其 `point_dict` 子包（loader/validator/schema），避免重复造轮子。
+"""
+import os
+import sys
+import types
+
+# 1) 挂载 core/template-console 为 template_console 包
+CONSOLE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, CONSOLE_DIR)
+if "template_console" not in sys.modules:
+    pkg = types.ModuleType("template_console")
+    pkg.__path__ = [CONSOLE_DIR]
+    sys.modules["template_console"] = pkg
+
+# 2) 暴露兄弟内核 edge-gateway/point_dict（#63 复用其校验器）
+CORE_DIR = os.path.dirname(CONSOLE_DIR)
+EDGE_GW_DIR = os.path.join(CORE_DIR, "edge-gateway")
+if os.path.isdir(EDGE_GW_DIR) and EDGE_GW_DIR not in sys.path:
+    sys.path.insert(0, EDGE_GW_DIR)

core/template-console/tests/test_config_store.py

@@ -0,0 +1,191 @@
+# -*- coding: utf-8 -*-
+"""配置项 CRUD 存储引擎测试（issue #64）。
+
+覆盖：
+1. 三类配置 CRUD（list/get/upsert/delete）；
+2. 原子写 + 持久化（重开 store 仍在）；
+3. 校验规则（model_param/rag_config/layout，非法值拒绝）；
+4. 快照 snapshot/restore（为 #66 提供基础）；
+5. 可解释字段（meaning/reason/updated_by/updated_at 落盘）。
+"""
+import json
+import os
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import _bootstrap  # noqa: F401
+
+from template_console.config_store import (  # noqa: E402
+    ALLOWED_WIDGET_TYPES,
+    ConfigItem,
+    ConfigKind,
+    ConfigStore,
+    ValidationResult,
+    validate_item,
+)
+
+
+class _TmpStore:
+    def __init__(self):
+        self._tmp = tempfile.mkdtemp()
+        self.store = ConfigStore(self._tmp)
+
+    def cleanup(self):
+        import shutil
+        shutil.rmtree(self._tmp, ignore_errors=True)
+
+
+class ValidationTest(unittest.TestCase):
+    """校验规则。"""
+
+    def test_model_param_scalar_ok(self):
+        self.assertTrue(validate_item(ConfigKind.MODEL_PARAM, "learning_rate", 0.001))
+
+    def test_model_param_learning_rate_range(self):
+        vr = validate_item(ConfigKind.MODEL_PARAM, "learning_rate", 1.5)
+        self.assertFalse(vr)
+        self.assertTrue(any("learning_rate" in e for e in vr.errors))
+
+    def test_model_param_bad_key(self):
+        vr = validate_item(ConfigKind.MODEL_PARAM, "Bad Key!", 1)
+        self.assertFalse(vr)
+
+    def test_rag_top_k_bounds(self):
+        self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "top_k", 0))
+        self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "top_k", 51))
+        self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "top_k", 10))
+
+    def test_rag_similarity_threshold(self):
+        self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "similarity_threshold", 0.5))
+        self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "similarity_threshold", 1.5))
+
+    def test_rag_sources_must_be_nonempty_list(self):
+        self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "sources", []))
+        self.assertFalse(validate_item(ConfigKind.RAG_CONFIG, "sources", ["", "x"]))
+        self.assertTrue(validate_item(ConfigKind.RAG_CONFIG, "sources", ["sop", "gb"]))
+
+    def test_layout_widget_type(self):
+        bad = [{"type": "unknown", "x": 0, "y": 0, "w": 1, "h": 1}]
+        self.assertFalse(validate_item(ConfigKind.LAYOUT, "dashboard", bad))
+        good = [{"type": "trend", "x": 0, "y": 0, "w": 6, "h": 2}]
+        self.assertTrue(validate_item(ConfigKind.LAYOUT, "dashboard", good))
+
+    def test_layout_widget_coords_nonneg_int(self):
+        bad = [{"type": "trend", "x": -1, "y": 0, "w": 1, "h": 1}]
+        vr = validate_item(ConfigKind.LAYOUT, "dashboard", bad)
+        self.assertFalse(vr)
+
+
+class CrudTest(unittest.TestCase):
+    """CRUD + 持久化。"""
+
+    def setUp(self):
+        self.ctx = _TmpStore()
+        self.store = self.ctx.store
+
+    def tearDown(self):
+        self.ctx.cleanup()
+
+    def test_upsert_and_get(self):
+        self.store.upsert(ConfigKind.MODEL_PARAM, "iterations", 100,
+                          meaning="迭代数", updated_by="li", reason="标定")
+        it = self.store.get(ConfigKind.MODEL_PARAM, "iterations")
+        self.assertIsNotNone(it)
+        self.assertEqual(it.value, 100)
+        self.assertEqual(it.updated_by, "li")
+        self.assertEqual(it.reason, "标定")
+        self.assertTrue(it.updated_at)  # 时间戳已写
+
+    def test_upsert_rejects_invalid(self):
+        with self.assertRaises(ValueError):
+            self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 999)
+
+    def test_upsert_overwrites(self):
+        self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.1)
+        self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.01, reason="调小")
+        it = self.store.get(ConfigKind.MODEL_PARAM, "lr")
+        self.assertEqual(it.value, 0.01)
+        self.assertEqual(it.reason, "调小")
+
+    def test_list_and_delete(self):
+        self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 5)
+        self.store.upsert(ConfigKind.RAG_CONFIG, "similarity_threshold", 0.6)
+        self.assertEqual(len(self.store.list(ConfigKind.RAG_CONFIG)), 2)
+        self.assertTrue(self.store.delete(ConfigKind.RAG_CONFIG, "top_k"))
+        self.assertIsNone(self.store.get(ConfigKind.RAG_CONFIG, "top_k"))
+        self.assertFalse(self.store.delete(ConfigKind.RAG_CONFIG, "nope"))
+
+    def test_persistence_across_reopen(self):
+        self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
+        # 重开一个指向同一目录的 store
+        store2 = ConfigStore(self.ctx._tmp)
+        it = store2.get(ConfigKind.MODEL_PARAM, "lr")
+        self.assertIsNotNone(it)
+        self.assertEqual(it.value, 0.001)
+
+    def test_json_file_is_human_readable(self):
+        self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001, meaning="学习率")
+        path = os.path.join(self.ctx._tmp, "model_params.json")
+        with open(path, encoding="utf-8") as fh:
+            blob = json.load(fh)
+        self.assertEqual(blob["schema_version"], 1)
+        self.assertEqual(blob["kind"], "model_param")
+        self.assertEqual(blob["items"][0]["meaning"], "学习率")
+
+
+class SnapshotTest(unittest.TestCase):
+    """快照与恢复（#66 基础）。"""
+
+    def setUp(self):
+        self.ctx = _TmpStore()
+        self.store = self.ctx.store
+
+    def tearDown(self):
+        self.ctx.cleanup()
+
+    def test_snapshot_captures_all_kinds(self):
+        self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
+        self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 8)
+        snap = self.store.snapshot()
+        self.assertIn("captured_at", snap)
+        self.assertEqual(set(snap["kinds"].keys()),
+                         {"model_param", "rag_config", "layout"})
+        self.assertEqual(len(snap["kinds"]["model_param"]), 1)
+
+    def test_restore_replicates_state(self):
+        self.store.upsert(ConfigKind.MODEL_PARAM, "lr", 0.001)
+        self.store.upsert(ConfigKind.RAG_CONFIG, "top_k", 8)
+        snap = self.store.snapshot()
+        # 清空再恢复
+        self.store.delete(ConfigKind.MODEL_PARAM, "lr")
+        self.store.delete(ConfigKind.RAG_CONFIG, "top_k")
+        self.store.restore(snap)
+        self.assertEqual(self.store.get(ConfigKind.MODEL_PARAM, "lr").value, 0.001)
+        self.assertEqual(self.store.get(ConfigKind.RAG_CONFIG, "top_k").value, 8)
+
+    def test_item_counts(self):
+        self.store.upsert(ConfigKind.LAYOUT, "dashboard",
+                          [{"type": "trend", "x": 0, "y": 0, "w": 6, "h": 2}])
+        counts = self.store.item_counts()
+        self.assertEqual(counts["layout"], 1)
+        self.assertEqual(counts["model_param"], 0)
+
+
+class ConfigItemSerializationTest(unittest.TestCase):
+    """ConfigItem 序列化往返。"""
+
+    def test_roundtrip(self):
+        it = ConfigItem(key="lr", value=0.1, kind=ConfigKind.MODEL_PARAM,
+                        meaning="学习率", updated_by="li", reason="init",
+                        updated_at="2026-01-01T00:00:00Z")
+        d = it.to_dict()
+        self.assertEqual(d["kind"], "model_param")
+        it2 = ConfigItem.from_dict(d)
+        self.assertEqual(it2.value, 0.1)
+        self.assertEqual(it2.kind, ConfigKind.MODEL_PARAM)
+
+
+if __name__ == "__main__":
+    unittest.main()

core/template-console/tests/test_point_importer.py

@@ -0,0 +1,228 @@
+# -*- coding: utf-8 -*-
+"""点位字典 CSV 导入 + 自动校验测试（issue #63）。
+
+覆盖：
+1. 合法 CSV 导入通过（ti / resin 两套模板）；
+2. 表头校验（缺失列 / 列序错位）；
+3. 内核校验复用（量纲/数据类型/采样率/重复点号/协议）；
+4. OPC 节点格式校验（opcua/modbus/空）；
+5. 模板级量纲收窄（rpm 仅 resin 允许）；
+6. 报告 ok/汇总/字典化 + 粘贴框入口。
+"""
+import os
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import _bootstrap  # noqa: F401
+
+from template_console.point_importer import (  # noqa: E402
+    ImportReport,
+    ImportRowIssue,
+    Severity,
+    TemplateKind,
+    import_csv,
+    import_csv_string,
+)
+
+GOOD_TI = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
+CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,opcua
+CLF-01,CLF-01.PRES,炉压,kPa,float,1000,true,ns=2;s=CLF.Pres,opcua
+"""
+
+GOOD_RESIN = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
+R-801,R-801.TEMP,反应釜温度,℃,float,1000,true,ns=2;s=R801.Temp,opcua
+R-801,R-801.AGIT,搅拌转速,rpm,float,1000,true,ns=2;s=R801.Agit,opcua
+"""
+
+BAD_MULTI = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
+CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,badnode,opcua
+CLF-01,CLF-01.TEMP,炉压,kPa,badtype,0,true,ns=2;s=CLF.Pres,opcua
+CLF-01,CLF-01.PRES,炉压,kPa,float,500,true,holding:40010,modbus
+"""
+
+
+class _TmpCsv:
+    """临时 CSV 文件助手。"""
+
+    def __init__(self, content):
+        self._tmp = tempfile.mkdtemp()
+        self.path = os.path.join(self._tmp, "points.csv")
+        with open(self.path, "w", encoding="utf-8") as fh:
+            fh.write(content)
+
+    def cleanup(self):
+        import shutil
+        shutil.rmtree(self._tmp, ignore_errors=True)
+
+
+class GoodImportTest(unittest.TestCase):
+    """合法 CSV 导入。"""
+
+    def test_good_ti_imports_ok(self):
+        f = _TmpCsv(GOOD_TI)
+        try:
+            pd, rep = import_csv(f.path, template=TemplateKind.TI)
+            self.assertTrue(rep.ok, rep.summary())
+            self.assertEqual(rep.loaded_points, 2)
+            self.assertEqual(rep.error_count, 0)
+        finally:
+            f.cleanup()
+
+    def test_good_resin_imports_ok_with_rpm(self):
+        f = _TmpCsv(GOOD_RESIN)
+        try:
+            pd, rep = import_csv(f.path, template=TemplateKind.RESIN)
+            self.assertTrue(rep.ok, rep.summary())
+            # rpm 在 resin 模板合法
+            self.assertEqual(rep.error_count, 0)
+        finally:
+            f.cleanup()
+
+    def test_report_summary_and_dict(self):
+        f = _TmpCsv(GOOD_TI)
+        try:
+            _, rep = import_csv(f.path, template=TemplateKind.TI)
+            self.assertIn("通过", rep.summary())
+            d = rep.to_dict()
+            self.assertTrue(d["ok"])
+            self.assertEqual(d["template"], "ti")
+            self.assertEqual(d["loaded_points"], 2)
+        finally:
+            f.cleanup()
+
+
+class HeaderValidationTest(unittest.TestCase):
+    """表头校验。"""
+
+    def test_missing_column_is_error(self):
+        bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp\n"
+        f = _TmpCsv(bad)
+        try:
+            _, rep = import_csv(f.path, template=TemplateKind.TI)
+            self.assertFalse(rep.ok)
+            codes = [i.code for i in rep.issues if i.row == 1]
+            self.assertIn("missing_column", codes)
+        finally:
+            f.cleanup()
+
+    def test_wrong_column_order_is_warn(self):
+        # 列齐全但顺序错（name 提前）→ WARN，不阻断
+        bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,protocol,opcNode\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,opcua,ns=2;s=CLF.Temp\n"
+        f = _TmpCsv(bad)
+        try:
+            _, rep = import_csv(f.path, template=TemplateKind.TI)
+            self.assertIn("bad_column_order", [i.code for i in rep.issues])
+        finally:
+            f.cleanup()
+
+
+class KernelValidationTest(unittest.TestCase):
+    """复用内核校验（量纲/数据类型/采样率/重复点号）。"""
+
+    def test_dup_point_detected(self):
+        bad = GOOD_TI + "CLF-01,CLF-01.TEMP,炉温2,℃,float,1000,true,ns=2;s=CLF.Temp2,opcua\n"
+        f = _TmpCsv(bad)
+        try:
+            _, rep = import_csv(f.path, template=TemplateKind.TI)
+            self.assertFalse(rep.ok)
+            self.assertIn("dup_point", [i.code for i in rep.issues])
+        finally:
+            f.cleanup()
+
+    def test_bad_data_type_and_sample_rate(self):
+        f = _TmpCsv(BAD_MULTI)
+        try:
+            _, rep = import_csv(f.path, template=TemplateKind.TI)
+            codes = [i.code for i in rep.issues]
+            self.assertIn("bad_data_type", codes)
+            self.assertIn("bad_sample_rate", codes)
+        finally:
+            f.cleanup()
+
+    def test_bad_protocol_detected(self):
+        bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,unknownproto\n"
+        f = _TmpCsv(bad)
+        try:
+            _, rep = import_csv(f.path, template=TemplateKind.TI)
+            self.assertIn("bad_protocol", [i.code for i in rep.issues])
+        finally:
+            f.cleanup()
+
+
+class OpcNodeValidationTest(unittest.TestCase):
+    """OPC 节点格式校验（配置台扩展维度）。"""
+
+    def test_bad_opcua_node_is_error(self):
+        # BAD_MULTI 第1行 opcNode=badnode 协议 opcua → ERROR
+        f = _TmpCsv(BAD_MULTI)
+        try:
+            _, rep = import_csv(f.path, template=TemplateKind.TI)
+            opc_issues = [i for i in rep.issues if i.code == "bad_opc_node"]
+            self.assertTrue(any(i.severity == Severity.ERROR for i in opc_issues))
+        finally:
+            f.cleanup()
+
+    def test_valid_modbus_node_ok(self):
+        # BAD_MULTI 第3行 holding:40010 modbus → 不报 bad_opc_node
+        f = _TmpCsv("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.PRES,炉压,kPa,float,500,true,holding:40010,modbus\n")
+        try:
+            _, rep = import_csv(f.path, template=TemplateKind.TI)
+            self.assertNotIn("bad_opc_node", [i.code for i in rep.issues
+                                              if i.severity == Severity.ERROR])
+        finally:
+            f.cleanup()
+
+    def test_empty_opc_node_is_warn(self):
+        f = _TmpCsv("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,,simulator\n")
+        try:
+            _, rep = import_csv(f.path, template=TemplateKind.TI)
+            empties = [i for i in rep.issues if i.code == "empty_opc_node"]
+            self.assertEqual(len(empties), 1)
+            self.assertEqual(empties[0].severity, Severity.WARN)
+            # 警告不阻断
+            self.assertTrue(rep.ok)
+        finally:
+            f.cleanup()
+
+
+class TemplateUnitTest(unittest.TestCase):
+    """模板级量纲收窄。"""
+
+    def test_rpm_rejected_in_ti_template(self):
+        f = _TmpCsv(GOOD_RESIN)
+        try:
+            _, rep = import_csv(f.path, template=TemplateKind.TI)
+            # rpm 是树脂专属，ti 模板应报 template_unit_mismatch
+            self.assertIn("template_unit_mismatch", [i.code for i in rep.issues])
+            self.assertFalse(rep.ok)
+        finally:
+            f.cleanup()
+
+    def test_rpm_allowed_in_resin_template(self):
+        f = _TmpCsv(GOOD_RESIN)
+        try:
+            _, rep = import_csv(f.path, template=TemplateKind.RESIN)
+            self.assertNotIn("template_unit_mismatch", [i.code for i in rep.issues])
+            self.assertTrue(rep.ok, rep.summary())
+        finally:
+            f.cleanup()
+
+
+class ImportStringTest(unittest.TestCase):
+    """粘贴框入口（import_csv_string）。"""
+
+    def test_import_from_string(self):
+        pd, rep = import_csv_string(GOOD_TI, template=TemplateKind.TI)
+        self.assertTrue(rep.ok)
+        self.assertEqual(len(pd), 2)
+
+    def test_import_string_bad_csv(self):
+        bad = "device_id,point_id\nCLF-01,CLF-01.TEMP\n"  # 缺列
+        _, rep = import_csv_string(bad, template=TemplateKind.TI)
+        self.assertFalse(rep.ok)
+
+
+if __name__ == "__main__":
+    unittest.main()

core/template-console/tests/test_rbac.py

@@ -0,0 +1,181 @@
+# -*- coding: utf-8 -*-
+"""三级 RBAC 权限模型测试（issue #62）。
+
+覆盖：
+1. 三级角色权限矩阵正确（readonly/engineer/admin）；
+2. 角色继承（admin 继承 engineer 继承 readonly）；
+3. has_permission 允许/拒绝判定 + 理由可解释；
+4. 细粒度收窄（restricted_to_readonly 把写动作降级拒绝）；
+5. 便捷判定 can_publish / 用户概览。
+"""
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import _bootstrap  # noqa: F401
+
+from template_console.rbac import (  # noqa: E402
+    Action,
+    Permission,
+    Resource,
+    Role,
+    RoleKind,
+    User,
+    all_roles,
+    can_publish,
+    effective_permissions,
+    get_role,
+    has_permission,
+    user_summary,
+)
+
+
+class RoleRegistryTest(unittest.TestCase):
+    """三级角色注册表。"""
+
+    def test_three_roles_present(self):
+        roles = {r.kind for r in all_roles()}
+        self.assertEqual(roles, {RoleKind.READONLY, RoleKind.ENGINEER, RoleKind.ADMIN})
+
+    def test_role_labels_in_chinese(self):
+        self.assertEqual(get_role(RoleKind.READONLY).label, "只读")
+        self.assertEqual(get_role(RoleKind.ENGINEER).label, "行业工程师")
+        self.assertEqual(get_role(RoleKind.ADMIN).label, "管理员")
+
+    def test_role_descriptions_explainable(self):
+        # 可解释性：每个角色都有职责说明
+        for role in all_roles():
+            self.assertTrue(role.description, f"{role.kind} 缺少 description")
+
+    def test_inheritance_chain(self):
+        self.assertEqual(get_role(RoleKind.ADMIN).inherits, RoleKind.ENGINEER)
+        self.assertEqual(get_role(RoleKind.ENGINEER).inherits, RoleKind.READONLY)
+        self.assertIsNone(get_role(RoleKind.READONLY).inherits)
+
+    def test_permission_key_format(self):
+        p = Permission(RoleKind.ADMIN, Resource.USER, Action.MANAGE)
+        # 匹配键为资源:动作（角色无关，便于继承）；审计键含授予角色
+        self.assertEqual(p.key(), "user:manage")
+        self.assertEqual(p.audit_key(), "admin:user:manage")
+
+
+class EffectivePermissionTest(unittest.TestCase):
+    """继承后的有效权限集合。"""
+
+    def test_admin_inherits_engineer_and_readonly(self):
+        eff = effective_permissions(RoleKind.ADMIN)
+        # 匹配键为 resource:action：admin 拥有自身的 user:manage，
+        # 也继承 engineer 的 model_param:edit 与 readonly 的 layout:view
+        self.assertIn("user:manage", eff)
+        self.assertIn("model_param:edit", eff)
+        self.assertIn("layout:view", eff)
+
+    def test_engineer_cannot_publish(self):
+        eff = effective_permissions(RoleKind.ENGINEER)
+        # 工程师不能发布/推送/管用户
+        self.assertNotIn("release:publish", eff)
+        self.assertNotIn("push:publish", eff)
+        self.assertNotIn("user:manage", eff)
+
+    def test_readonly_has_no_write(self):
+        eff = effective_permissions(RoleKind.READONLY)
+        for key in eff:
+            # 只读权限只能以 :view 结尾
+            self.assertTrue(key.endswith(":view"), f"readonly 不应有写/发布权限: {key}")
+
+
+class HasPermissionTest(unittest.TestCase):
+    """has_permission 判定 + 理由。"""
+
+    def setUp(self):
+        self.ro = User("viewer", RoleKind.READONLY, "查看员")
+        self.eng = User("li_engineer", RoleKind.ENGINEER, "李工")
+        self.admin = User("root_admin", RoleKind.ADMIN, "管理员甲")
+
+    def test_readonly_view_allowed(self):
+        d = has_permission(self.ro, Resource.LAYOUT, Action.VIEW)
+        self.assertTrue(d.allow)
+        self.assertEqual(d.source, "explicit")
+
+    def test_readonly_edit_denied(self):
+        d = has_permission(self.ro, Resource.LAYOUT, Action.EDIT)
+        self.assertFalse(d.allow)
+        self.assertIn("缺少权限", d.reason)
+
+    def test_engineer_edit_allowed_inherited_view(self):
+        # 工程师编辑是本角色权限（explicit）
+        d_edit = has_permission(self.eng, Resource.LAYOUT, Action.EDIT)
+        self.assertTrue(d_edit.allow)
+        self.assertEqual(d_edit.source, "explicit")
+        # 工程师查看布局是继承自 readonly（inherited）
+        d_view = has_permission(self.eng, Resource.LAYOUT, Action.VIEW)
+        self.assertTrue(d_view.allow)
+        self.assertEqual(d_view.source, "inherited")
+
+    def test_engineer_publish_denied(self):
+        d = has_permission(self.eng, Resource.RELEASE, Action.PUBLISH)
+        self.assertFalse(d.allow)
+
+    def test_admin_publish_allowed(self):
+        d = has_permission(self.admin, Resource.RELEASE, Action.PUBLISH)
+        self.assertTrue(d.allow)
+        self.assertEqual(d.source, "explicit")
+
+    def test_admin_inherited_engineer_edit(self):
+        d = has_permission(self.admin, Resource.MODEL_PARAM, Action.EDIT)
+        self.assertTrue(d.allow)
+        self.assertEqual(d.source, "inherited")
+
+    def test_decision_carries_reason(self):
+        # 可解释性：无论允许/拒绝，reason 非空且含用户名与资源
+        for user in (self.ro, self.eng, self.admin):
+            d = has_permission(user, Resource.PUSH, Action.PUBLISH)
+            self.assertIn(user.username, d.reason)
+            self.assertIn(Resource.PUSH.value, d.reason)
+
+
+class RestrictedUserTest(unittest.TestCase):
+    """细粒度收窄：restricted_to_readonly。"""
+
+    def test_restricted_engineer_cannot_edit_that_resource(self):
+        # 工程师本可编辑布局，但被收窄为只读后应拒绝
+        u = User("limited", RoleKind.ENGINEER, "受限工程师",
+                 restricted_to_readonly=[Resource.LAYOUT])
+        d = has_permission(u, Resource.LAYOUT, Action.EDIT)
+        self.assertFalse(d.allow)
+        self.assertEqual(d.source, "restricted")
+
+    def test_restricted_engineer_can_still_view(self):
+        u = User("limited", RoleKind.ENGINEER, "受限工程师",
+                 restricted_to_readonly=[Resource.LAYOUT])
+        d = has_permission(u, Resource.LAYOUT, Action.VIEW)
+        self.assertTrue(d.allow)
+
+    def test_restricted_only_affects_named_resource(self):
+        u = User("limited", RoleKind.ENGINEER, "受限工程师",
+                 restricted_to_readonly=[Resource.LAYOUT])
+        # 模型超参未被收窄，仍可编辑
+        d = has_permission(u, Resource.MODEL_PARAM, Action.EDIT)
+        self.assertTrue(d.allow)
+
+
+class ConvenienceTest(unittest.TestCase):
+    """便捷判定与用户概览。"""
+
+    def test_can_publish(self):
+        self.assertFalse(can_publish(User("v", RoleKind.READONLY)))
+        self.assertFalse(can_publish(User("e", RoleKind.ENGINEER)))
+        self.assertTrue(can_publish(User("a", RoleKind.ADMIN)))
+
+    def test_user_summary(self):
+        s = user_summary(User("li", RoleKind.ENGINEER, "李工"))
+        self.assertEqual(s["username"], "li")
+        self.assertEqual(s["role"], "engineer")
+        self.assertEqual(s["role_label"], "行业工程师")
+        self.assertGreater(s["effective_permission_count"], 0)
+        self.assertEqual(s["restricted_to_readonly"], [])
+
+
+if __name__ == "__main__":
+    unittest.main()

templates/ti-cl4/dashboard/README.md

@@ -0,0 +1,37 @@
+# 海绵钛驾驶舱布局资产 + 校验器（Issue #55 / PRD 5.5）
+
+> 父 Issue「⑤ Ti 行业布局模板（四状态流程视图）· 0.5d」
+
+把海绵钛车间驾驶舱布局落为**对齐 iAOP-cockpit-layout-v1 的资产 + 可校验的纯标准库
+校验器**（无 node/前端构建环境，零运行时依赖）。
+
+## 资产：`cockpit.ti.yaml`
+
+四状态工艺流程（PRD 4.2 海绵钛：**氯化 → 精制 → 还原 → 蒸馏**）：
+
+- `process_view` 主视图（首屏立即加载），`stages` 声明四状态覆盖；
+- `trend` 实时趋势（氯化炉温度 `CLF-01.TEMP` / 氯气流量 `CLF-01.CL2`）；
+- `kpi_card` KPI（TiCl₄纯度 `RF-01.PURITY` / 杂质 `RF-01.IMP` / 电耗 `E-01.KWH` / 蒸汽 `ST-01.STEAM`）；
+- `alarm_panel` 告警面板 + `nl_query` NL 查询入口。
+
+所有 `bind` 的 `point_id` 对齐 `templates/ti-cl4/point-dict/point_dict.default.csv`。
+
+## 校验器：`layout_validator.py`
+
+`LayoutValidator(layout_yaml, point_dict_csv).validate()` → `LayoutReport`，校验：
+
+1. **widget 类型合法**：在 `iAOP-cockpit-layout-v1` 允许集合内（process_view/trend/kpi_card/alarm_panel/nl_query）；
+2. **12 列网格不越界**：`0 ≤ x`、`x + w ≤ 12`、`y ≥ 0`、`w/h > 0`；
+3. **bind point_id 在点位字典内**：防模板漂移（trend/kpi_card 的 bind 必须命中点字典）；
+4. **四状态覆盖完整**：process_view 的 stages 必须覆盖氯化/精制/还原/蒸馏，order 单调递增、id 唯一。
+
+零依赖 YAML 子集解析（复制 impurity-forecast 的 `_parse_yaml_subset`，无 pyyaml）。
+
+## 测试
+
+```bash
+python -m unittest discover -s templates/ti-cl4/dashboard/tests -p "test_*.py" -v
+```
+
+覆盖正常 + 边界 + 错误（18 用例）：真实资产端到端通过、非法类型、网格越界/负坐标/零宽、
+bind 漂移、缺 process_view、缺必需状态、order 非单调、stage 重复、$schema 头、CSV 加载、空布局。

templates/ti-cl4/dashboard/__init__.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+"""海绵钛驾驶舱布局资产 + 校验器包（Issue #55）。
+
+对齐 PRD 5.5「⑤ 配置化驾驶舱」布局 JSON Schema（iAOP-cockpit-layout-v1）：
+四状态工艺流程（氯化 → 精制 → 还原 → 蒸馏）。
+"""
+from .layout_validator import (
+    LayoutError,
+    LayoutIssue,
+    LayoutReport,
+    LayoutValidator,
+    WidgetSpec,
+    ALLOWED_WIDGET_TYPES,
+    GRID_COLUMNS,
+    REQUIRED_STAGES,
+)
+
+__all__ = [
+    "LayoutError",
+    "LayoutIssue",
+    "LayoutReport",
+    "LayoutValidator",
+    "WidgetSpec",
+    "ALLOWED_WIDGET_TYPES",
+    "GRID_COLUMNS",
+    "REQUIRED_STAGES",
+]

templates/ti-cl4/dashboard/_sanity_check.py

@@ -0,0 +1,33 @@
+# -*- coding: utf-8 -*-
+"""海绵钛驾驶舱布局冒烟脚本（Issue #55）。
+
+直接运行 ``python _sanity_check.py`` 验证：cockpit.ti.yaml + point_dict.default.csv
+端到端校验通过（widget 类型/网格/bind/四状态全覆盖）。零第三方依赖。
+"""
+import os
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+
+from layout_validator import LayoutValidator  # noqa: E402
+
+LAYOUT_YAML = os.path.join(HERE, "cockpit.ti.yaml")
+POINT_DICT_CSV = os.path.join(
+    HERE, os.pardir, "point-dict", "point_dict.default.csv")
+
+
+def main() -> int:
+    report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate()
+    if not report.passed:
+        print("FAIL")
+        for issue in report.errors:
+            print(f" - [{issue.widget_id}] {issue.field}: {issue.reason}")
+        return 1
+    print(f"OK: {report.widget_count} widgets，"
+          f"类型/网格/bind/四状态校验通过")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

templates/ti-cl4/dashboard/cockpit.ti.yaml

@@ -0,0 +1,108 @@
+# -*- coding: utf-8 -*-
+# 海绵钛（Ti）车间驾驶舱布局资产（iAOP-Template-Ti，EPIC #12）。
+#
+# 对齐 PRD 5.5「⑤ 配置化驾驶舱」布局 JSON Schema（iAOP-cockpit-layout-v1）：
+# 切换行业模板后，驾驶舱按本布局自动重排，无需改前端代码。
+# widget 类型：process_view（工艺流程视图）/ trend（实时趋势）/ kpi_card（KPI卡片）/
+#             alarm_panel（告警面板）/ nl_query（NL查询入口）。
+#
+# 四状态工艺流程（PRD 4.2 海绵钛：氯化 → 精制 → 还原 → 蒸馏）：
+# - 氯化  （CLF-01）：TiO₂ + Cl₂ + C → TiCl₄（沸腾氯化炉）
+# - 精制         ：粗 TiCl₄ → 精 TiCl₄（除钒/除硅，常压精馏）
+# - 还原  （RF-01）：TiCl₄ + Mg → 海绵钛（真空还原，Kroll 法）
+# - 蒸馏         ：海绵钛 + 残余 Mg/MgCl₂ 分离（真空蒸馏）
+# process_view 的 stages 字段声明四状态覆盖，layout_validator 校验完整性。
+#
+# bind 的 point_id 对齐 templates/ti-cl4/point-dict/point_dict.default.csv。
+$schema: iAOP-cockpit-layout-v1
+title: 海绵钛车间驾驶舱
+theme: dark
+widgets:
+  # ---- 四状态工艺流程主视图（首屏立即加载） ---------------------------
+  - type: process_view
+    src: ti_four_state.svg
+    x: 0
+    y: 0
+    w: 12
+    h: 4
+    description: 四状态工艺流程（氯化 → 精制 → 还原 → 蒸馏）
+    stages:
+      - id: chlorination
+        name: 氯化
+        device: CLF-01
+        order: 1
+      - id: purification
+        name: 精制
+        order: 2
+      - id: reduction
+        name: 还原
+        device: RF-01
+        order: 3
+      - id: distillation
+        name: 蒸馏
+        order: 4
+  # ---- 实时趋势：氯化炉温度/氯气流量（工艺核心监控） -------------------
+  - type: trend
+    bind: CLF-01.TEMP
+    x: 0
+    y: 4
+    w: 6
+    h: 2
+    description: 氯化炉温度实时趋势（沸腾氯化炉温 850±50℃）
+  - type: trend
+    bind: CLF-01.CL2
+    x: 6
+    y: 4
+    w: 6
+    h: 2
+    description: 氯气流量实时趋势（流态化监控）
+  # ---- KPI 卡片：还原质量/能耗（海绵钛核心指标） -----------------------
+  - type: kpi_card
+    metric: ticl4_purity
+    bind: RF-01.PURITY
+    label: TiCl₄纯度
+    x: 0
+    y: 6
+    w: 3
+    h: 2
+    description: 还原 TiCl₄ 纯度（%，工艺 ≥ 99.9%）
+  - type: kpi_card
+    metric: ticl4_impurity
+    bind: RF-01.IMP
+    label: 杂质含量
+    x: 3
+    y: 6
+    w: 3
+    h: 2
+    description: 还原杂质含量（%，越低越好）
+  - type: kpi_card
+    metric: energy_per_ton
+    bind: E-01.KWH
+    label: 累计电耗
+    x: 6
+    y: 6
+    w: 3
+    h: 2
+    description: 车间累计电耗（kWh，单吨海绵钛综合能耗输入）
+  - type: kpi_card
+    metric: steam_flow
+    bind: ST-01.STEAM
+    label: 蒸汽流量
+    x: 9
+    y: 6
+    w: 3
+    h: 2
+    description: 蒸汽流量（t/h，公用工程监控）
+  # ---- 告警面板 + NL 查询入口 ------------------------------------------
+  - type: alarm_panel
+    x: 0
+    y: 8
+    w: 9
+    h: 3
+    description: 告警面板（氯化炉温/还原真空度/纯度/杂质异常）
+  - type: nl_query
+    x: 9
+    y: 8
+    w: 3
+    h: 3
+    description: 自然语言查询入口（工艺/质量/能耗问答）

templates/ti-cl4/dashboard/layout_validator.py

@@ -0,0 +1,500 @@
+# -*- coding: utf-8 -*-
+"""海绵钛驾驶舱布局校验器（Issue #55 / PRD 5.5「⑤ 配置化驾驶舱」）。
+
+PRD 5.5：切换行业模板后，驾驶舱按布局资产自动重排，无需改前端代码。
+本模块把布局资产（``cockpit.ti.yaml``）落为**可校验的纯标准库资产 + 校验器**——
+给定布局 YAML + 点位字典 CSV，校验：
+
+1. **widget 类型合法**：在 PRD 5.5 ``iAOP-cockpit-layout-v1`` 允许集合内
+   （process_view/trend/kpi_card/alarm_panel/nl_query）。
+2. **12 列网格不越界**：每个 widget ``0 ≤ x`` 且 ``x + w ≤ 12``，``y ≥ 0``、
+   ``h > 0``；坐标为非负整数，w/h 正整数（网格对齐）。
+3. **bind 的 point_id 在点位字典内**：trend/kpi_card 的 ``bind`` 必须命中
+   ``point_dict.default.csv`` 的 ``point_id`` 列（防模板漂移）。
+4. **四状态视图覆盖完整**：process_view 的 ``stages`` 必须覆盖工艺全流程
+   （氯化/精制/还原/蒸馏），且 order 单调递增、id 唯一。
+
+校验产出 :class:`LayoutReport`（PASS/FAIL + 逐条 :class:`LayoutIssue`，
+每条 issue 带 ``reason`` 可解释）。
+
+设计要点
+--------
+- **零依赖 YAML 子集解析**：复制 impurity-forecast features.py 的
+  ``_parse_yaml_subset``（无 pyyaml），支持 map/list/标量/行内 flow map。
+- **纯标准库**：CSV 用标准库 csv，无 numpy/pyyaml 依赖。
+- **换行业只改资产**：校验器对任何对齐 ``iAOP-cockpit-layout-v1`` 的布局都适用。
+
+用法::
+
+    report = LayoutValidator(layout_yaml, point_dict_csv).validate()
+    if not report.passed:
+        for issue in report.issues:
+            print(issue.severity, issue.widget_id, issue.reason)
+"""
+from __future__ import annotations
+
+import csv
+import os
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Dict, List, Optional, Tuple
+
+#: iAOP-cockpit-layout-v1 允许的 widget 类型集合（对齐 resin _sanity_check）。
+ALLOWED_WIDGET_TYPES = frozenset({
+    "process_view", "trend", "kpi_card", "alarm_panel", "nl_query",
+})
+
+#: 12 列网格（主流前端栅格标准，对齐 cockpit layout v1）。
+GRID_COLUMNS = 12
+
+#: 海绵钛四状态工艺流程（PRD 4.2：氯化 → 精制 → 还原 → 蒸馏）。
+#: process_view 的 stages 必须覆盖这四个 id。
+REQUIRED_STAGES = ("chlorination", "purification", "reduction", "distillation")
+
+
+class LayoutError(ValueError):
+    """布局资产解析/声明错误（YAML 格式错、表头缺字段等）。"""
+
+
+class Severity(str, Enum):
+    """问题严重度。"""
+
+    ERROR = "error"    # 阻断：布局不可用（类型非法/越界/bind 缺失/状态缺失）
+    WARN = "warn"      # 告警：可运行但不规范（重复/顺序乱）
+
+
+@dataclass
+class LayoutIssue:
+    """单条布局校验问题（含 reason 可解释）。"""
+
+    severity: Severity
+    reason: str
+    widget_id: str = ""        # 关联 widget（index 或 src/metric）
+    field: str = ""            # 关联字段（type/x/bind/stages ...）
+
+    @property
+    def is_error(self) -> bool:
+        return self.severity is Severity.ERROR
+
+
+@dataclass
+class LayoutReport:
+    """布局校验报告。"""
+
+    issues: List[LayoutIssue] = field(default_factory=list)
+    widget_count: int = 0
+
+    @property
+    def errors(self) -> List[LayoutIssue]:
+        return [i for i in self.issues if i.is_error]
+
+    @property
+    def passed(self) -> bool:
+        """通过 = 无 ERROR（WARN 不阻断）。"""
+        return not any(i.is_error for i in self.issues)
+
+    def to_dict(self) -> dict:
+        return {
+            "passed": self.passed,
+            "widget_count": self.widget_count,
+            "error_count": len(self.errors),
+            "warn_count": len(self.issues) - len(self.errors),
+            "issues": [
+                {"severity": i.severity.value, "widget_id": i.widget_id,
+                 "field": i.field, "reason": i.reason}
+                for i in self.issues
+            ],
+        }
+
+
+@dataclass
+class WidgetSpec:
+    """单个 widget 的内存模型（从 YAML 解析）。"""
+
+    index: int                       # 在 widgets 列表中的位置（0 起）
+    type: str
+    x: int = 0
+    y: int = 0
+    w: int = 1
+    h: int = 1
+    bind: str = ""                   # trend/kpi_card 绑定的 point_id
+    src: str = ""                    # process_view 的 SVG
+    metric: str = ""                 # kpi_card 的 metric
+    label: str = ""
+    description: str = ""
+    stages: List[Dict[str, object]] = field(default_factory=list)
+
+
+# ---------------------------------------------------------------------------
+# 零依赖 YAML 子集解析（复制自 impurity-forecast features.py，对齐 data-bus）
+# ---------------------------------------------------------------------------
+
+def _parse_scalar(text: str) -> str:
+    """去掉标量两侧引号与行内注释。"""
+    t = text.split(" #", 1)[0].strip()
+    if len(t) >= 2 and t[0] == t[-1] and t[0] in ("'", '"'):
+        return t[1:-1]
+    return t
+
+
+def _parse_flow_value(text: str):
+    """解析 ``key: value`` 右侧值，支持行内 flow map ``{k: v, k: v}``。"""
+    t = text.split(" #", 1)[0].strip()
+    if t.startswith("{") and t.endswith("}"):
+        inner = t[1:-1].strip()
+        out: Dict[str, object] = {}
+        if not inner:
+            return out
+        for part in inner.split(","):
+            if ":" not in part:
+                raise LayoutError(f"flow map 项不是键值对：{part!r}")
+            k, _, v = part.partition(":")
+            out[k.strip()] = _parse_scalar(v)
+        return out
+    return _parse_scalar(text)
+
+
+def _strip_comments(lines: List[str]) -> List[Tuple[str, int]]:
+    out: List[Tuple[str, int]] = []
+    for i, ln in enumerate(lines):
+        s = ln.strip()
+        if not s or s.startswith("#"):
+            continue
+        out.append((ln, i + 1))
+    return out
+
+
+def _parse_node(lines: List[Tuple[str, int]], i: int, indent: int):
+    """递归解析 YAML 节点（map / list / scalar）。返回 (value, next_i)。"""
+    text, _ = lines[i]
+    # ---- list 节点 ----
+    if text.lstrip(" ").startswith("- "):
+        items: List[object] = []
+        while i < len(lines):
+            t, no = lines[i]
+            stripped = t.lstrip(" ")
+            if not stripped.startswith("- "):
+                break
+            lead_j = len(t) - len(t.lstrip(" "))
+            if lead_j != indent:
+                break
+            item_text = stripped[2:].strip()
+            if not item_text:
+                raise LayoutError(f"cockpit.yaml 第 {no} 行：list 项为空")
+            if ":" in item_text:
+                map_indent = len(t) - len(t.lstrip(" ")) + 2
+                lines[i] = (" " * map_indent + item_text, no)
+                v, i = _parse_node(lines, i, map_indent)
+                items.append(v)
+            else:
+                items.append(_parse_flow_value(item_text))
+                i += 1
+        return items, i
+    # ---- map 节点 ----
+    result: Dict[str, object] = {}
+    while i < len(lines):
+        t, no = lines[i]
+        lead_j = len(t) - len(t.lstrip(" "))
+        if lead_j < indent or t.lstrip(" ").startswith("- "):
+            break
+        if lead_j > indent:
+            raise LayoutError(
+                f"cockpit.yaml 第 {no} 行缩进异常（期望 {indent}，实际 {lead_j}）")
+        if ":" not in t:
+            raise LayoutError(f"cockpit.yaml 第 {no} 行不是合法键值对：{t!r}")
+        key, _, rest = t.partition(":")
+        key = key.strip()
+        rest = rest.strip()
+        if rest:
+            result[key] = _parse_flow_value(rest)
+            i += 1
+            continue
+        if i + 1 >= len(lines):
+            raise LayoutError(f"cockpit.yaml 第 {no} 行 {key!r} 缺少值")
+        sub_indent = len(lines[i + 1][0]) - len(lines[i + 1][0].lstrip(" "))
+        if sub_indent <= indent:
+            raise LayoutError(f"cockpit.yaml 第 {no} 行 {key!r} 缺少值（无嵌套）")
+        v, i = _parse_node(lines, i + 1, sub_indent)
+        result[key] = v
+    return result, i
+
+
+def _load_yaml_text(text: str) -> Dict[str, object]:
+    """解析 YAML 文本为 dict（顶层必须是 map）。"""
+    lines = _strip_comments(text.splitlines())
+    if not lines:
+        return {}
+    top_indent = len(lines[0][0]) - len(lines[0][0].lstrip(" "))
+    value, next_i = _parse_node(lines, 0, top_indent)
+    if not isinstance(value, dict):
+        raise LayoutError("cockpit.yaml 顶层必须是 map")
+    if next_i < len(lines):
+        raise LayoutError(
+            f"cockpit.yaml 第 {lines[next_i][1]} 行：顶层存在多个节点")
+    return value
+
+
+# ---------------------------------------------------------------------------
+# 点位字典加载（CSV → point_id 集合）
+# ---------------------------------------------------------------------------
+
+def load_point_ids(csv_path: str) -> List[str]:
+    """从点位字典 CSV 加载全部 point_id（保序，对齐 CSV point_id 列）。
+
+    CSV 表头对齐 core/edge-gateway point_dict schema（第二列 point_id）。
+    """
+    if not os.path.isfile(csv_path):
+        raise LayoutError(f"点位字典 CSV 不存在：{csv_path}")
+    with open(csv_path, "r", encoding="utf-8") as fh:
+        rows = list(csv.reader(fh))
+    if not rows:
+        raise LayoutError(f"点位字典 CSV 为空：{csv_path}")
+    header = [c.strip() for c in rows[0]]
+    if "point_id" not in header:
+        raise LayoutError(
+            f"点位字典 CSV 表头缺 point_id 列：{header}")
+    col = header.index("point_id")
+    ids: List[str] = []
+    for i, row in enumerate(rows[1:], 2):
+        if len(row) <= col:
+            continue
+        pid = row[col].strip()
+        if pid:
+            ids.append(pid)
+    if not ids:
+        raise LayoutError(f"点位字典 CSV 无 point_id 数据行：{csv_path}")
+    return ids
+
+
+# ---------------------------------------------------------------------------
+# 校验器
+# ---------------------------------------------------------------------------
+
+class LayoutValidator:
+    """海绵钛驾驶舱布局校验器。
+
+    Args:
+        layout_yaml_path:  布局资产路径（cockpit.ti.yaml）。
+        point_dict_csv_path: 点位字典 CSV 路径（point_dict.default.csv）。
+        grid_columns:      网格列数（默认 12，对齐 cockpit layout v1）。
+        required_stages:   process_view 必须覆盖的 stage id（默认海绵钛四状态）。
+    """
+
+    def __init__(
+        self,
+        layout_yaml_path: str,
+        point_dict_csv_path: Optional[str] = None,
+        grid_columns: int = GRID_COLUMNS,
+        required_stages: Tuple[str, ...] = REQUIRED_STAGES,
+    ) -> None:
+        if grid_columns <= 0:
+            raise LayoutError(f"grid_columns 必须 > 0，实际 {grid_columns}")
+        self.layout_path = layout_yaml_path
+        self.point_dict_path = point_dict_csv_path
+        self.grid_columns = int(grid_columns)
+        self.required_stages = tuple(required_stages)
+
+    # ------------------------------------------------------------------
+    def validate(self) -> LayoutReport:
+        """执行全部校验，返回报告。"""
+        report = LayoutReport()
+        # 1) 解析布局 YAML
+        try:
+            with open(self.layout_path, "r", encoding="utf-8") as fh:
+                data = _load_yaml_text(fh.read())
+        except LayoutError:
+            raise
+        except OSError as exc:
+            raise LayoutError(f"布局 YAML 读取失败：{self.layout_path} ({exc})") from exc
+
+        # schema 头校验
+        schema = str(data.get("$schema", "")).strip()
+        if schema != "iAOP-cockpit-layout-v1":
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR,
+                field="$schema",
+                reason=f"$schema 应为 'iAOP-cockpit-layout-v1'，实际 {schema!r}",
+            ))
+
+        # 2) 解析 widgets
+        raw_widgets = data.get("widgets") or []
+        if not isinstance(raw_widgets, list):
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, field="widgets",
+                reason=f"widgets 必须是 list，实际 {type(raw_widgets).__name__}"))
+            return report
+        widgets = self._parse_widgets(raw_widgets, report)
+        report.widget_count = len(widgets)
+
+        if not widgets:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, field="widgets",
+                reason="布局无任何 widget"))
+            return report
+
+        # 3) 加载点位字典（bind 校验需要）
+        point_ids: Optional[set] = None
+        if self.point_dict_path:
+            try:
+                point_ids = set(load_point_ids(self.point_dict_path))
+            except LayoutError as exc:
+                report.issues.append(LayoutIssue(
+                    severity=Severity.ERROR, field="point_dict",
+                    reason=str(exc)))
+
+        # 4) 逐 widget 校验
+        for w in widgets:
+            self._check_widget(w, point_ids, report)
+
+        # 5) process_view 四状态覆盖
+        self._check_process_views(widgets, report)
+
+        return report
+
+    # ------------------------------------------------------------------
+    def _parse_widgets(self, raw_widgets: List[object],
+                       report: LayoutReport) -> List[WidgetSpec]:
+        widgets: List[WidgetSpec] = []
+        for idx, item in enumerate(raw_widgets):
+            if not isinstance(item, dict):
+                report.issues.append(LayoutIssue(
+                    severity=Severity.ERROR, widget_id=f"[{idx}]",
+                    field="widgets",
+                    reason=f"widgets[{idx}] 必须是 map，实际 {type(item).__name__}"))
+                continue
+            wtype = str(item.get("type", "")).strip()
+            widgets.append(WidgetSpec(
+                index=idx,
+                type=wtype,
+                x=_to_int(item.get("x"), 0),
+                y=_to_int(item.get("y"), 0),
+                w=_to_int(item.get("w"), 1),
+                h=_to_int(item.get("h"), 1),
+                bind=str(item.get("bind", "")).strip(),
+                src=str(item.get("src", "")).strip(),
+                metric=str(item.get("metric", "")).strip(),
+                label=str(item.get("label", "")).strip(),
+                description=str(item.get("description", "")).strip(),
+                stages=_as_list_of_dict(item.get("stages")),
+            ))
+        return widgets
+
+    # ------------------------------------------------------------------
+    def _check_widget(self, w: WidgetSpec, point_ids: Optional[set],
+                      report: LayoutReport) -> None:
+        wid = f"[{w.index}]({w.type})"
+        # 4a) widget 类型合法
+        if w.type not in ALLOWED_WIDGET_TYPES:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, widget_id=wid, field="type",
+                reason=f"非法 widget 类型 {w.type!r}（允许 {sorted(ALLOWED_WIDGET_TYPES)})"))
+
+        # 4b) 12 列网格不越界（坐标非负整数、x+w ≤ columns、h>0）
+        if w.x < 0 or w.y < 0:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, widget_id=wid, field="grid",
+                reason=f"坐标不能为负：x={w.x} y={w.y}"))
+        if w.w <= 0 or w.h <= 0:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, widget_id=wid, field="grid",
+                reason=f"w/h 必须为正整数：w={w.w} h={w.h}"))
+        if w.x + w.w > self.grid_columns:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, widget_id=wid, field="grid",
+                reason=f"越出 {self.grid_columns} 列网格：x={w.x}+w={w.w}"
+                       f"={w.x + w.w} > {self.grid_columns}"))
+
+        # 4c) bind 的 point_id 必须在点位字典内（trend/kpi_card）
+        if w.bind:
+            if point_ids is not None and w.bind not in point_ids:
+                report.issues.append(LayoutIssue(
+                    severity=Severity.ERROR, widget_id=wid, field="bind",
+                    reason=f"bind point_id {w.bind!r} 不在点位字典内"
+                           f"（防模板漂移，对齐 point_dict.default.csv）"))
+
+    # ------------------------------------------------------------------
+    def _check_process_views(self, widgets: List[WidgetSpec],
+                             report: LayoutReport) -> None:
+        """校验 process_view 的四状态覆盖完整。"""
+        pv = [w for w in widgets if w.type == "process_view"]
+        if not pv:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, field="process_view",
+                reason="布局缺少 process_view（四状态工艺流程主视图必需）"))
+            return
+
+        covered: Dict[str, WidgetSpec] = {}  # stage_id → widget
+        for w in pv:
+            wid = f"[{w.index}](process_view)"
+            if not w.stages:
+                report.issues.append(LayoutIssue(
+                    severity=Severity.ERROR, widget_id=wid, field="stages",
+                    reason="process_view 缺少 stages 声明（四状态覆盖必需）"))
+                continue
+
+            stage_ids: List[str] = []
+            orders: List[int] = []
+            seen: set = set()
+            for st in w.stages:
+                sid = str(st.get("id", "")).strip()
+                sname = str(st.get("name", "")).strip()
+                if not sid:
+                    report.issues.append(LayoutIssue(
+                        severity=Severity.ERROR, widget_id=wid, field="stages",
+                        reason=f"stage 缺少 id（name={sname!r}）"))
+                    continue
+                if sid in seen:
+                    report.issues.append(LayoutIssue(
+                        severity=Severity.WARN, widget_id=wid, field="stages",
+                        reason=f"stage id 重复：{sid!r}"))
+                    continue
+                seen.add(sid)
+                stage_ids.append(sid)
+                covered.setdefault(sid, w)
+                order = st.get("order")
+                if order is not None:
+                    try:
+                        orders.append(int(order))
+                    except (TypeError, ValueError):
+                        report.issues.append(LayoutIssue(
+                            severity=Severity.ERROR, widget_id=wid, field="stages",
+                            reason=f"stage {sid!r} order 非整数：{order!r}"))
+
+            # order 单调递增校验
+            if orders and len(orders) == len(stage_ids):
+                if orders != sorted(orders):
+                    report.issues.append(LayoutIssue(
+                        severity=Severity.ERROR, widget_id=wid, field="stages",
+                        reason=f"stage order 非单调递增：{orders}"))
+
+        # 必需四状态全覆盖
+        missing = [s for s in self.required_stages if s not in covered]
+        if missing:
+            report.issues.append(LayoutIssue(
+                severity=Severity.ERROR, field="stages",
+                reason=f"process_view stages 未覆盖必需四状态：{missing}"
+                       f"（氯化/精制/还原/蒸馏）"))
+
+
+# ---------------------------------------------------------------------------
+# 辅助
+# ---------------------------------------------------------------------------
+
+def _to_int(value: object, default: int) -> int:
+    """把 YAML 解析出的值（可能是 str/int）转为 int；失败返回 default。"""
+    if value is None or value == "":
+        return default
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        raise LayoutError(f"坐标值不是整数：{value!r}")
+
+
+def _as_list_of_dict(value: object) -> List[Dict[str, object]]:
+    if not isinstance(value, list):
+        return []
+    out: List[Dict[str, object]] = []
+    for item in value:
+        if isinstance(item, dict):
+            out.append(item)
+    return out

templates/ti-cl4/dashboard/tests/_bootstrap.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+"""测试引导：把 dashboard 测试根目录加入 sys.path，使 layout_validator 可导入。
+
+dashboard 目录名是合法 Python 标识符，直接作为包导入；本引导把父目录
+（templates/ti-cl4/dashboard）挂到 sys.path，使 ``from layout_validator import ...``
+在 unittest 发现机制下可解析（与 core 模块测试引导同款）。
+"""
+import os
+import sys
+
+PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if PKG_DIR not in sys.path:
+    sys.path.insert(0, PKG_DIR)

templates/ti-cl4/dashboard/tests/test_layout_validator.py

@@ -0,0 +1,309 @@
+# -*- coding: utf-8 -*-
+"""海绵钛驾驶舱布局校验器测试（Issue #55）。
+
+覆盖：
+1. 真实 cockpit.ti.yaml + point_dict.default.csv 全部通过（端到端）；
+2. widget 类型合法集合（非法类型 → ERROR）；
+3. 12 列网格校验（越界/负坐标/非正 w/h）；
+4. bind point_id 在点位字典内（漂移 → ERROR）；
+5. process_view 四状态覆盖（缺 stage id / order 非单调 / 缺必需状态）；
+6. $schema 头校验；
+7. YAML 解析（flow map / 嵌套）；
+8. 点位字典 CSV 加载（缺表头/空文件）；
+9. 空布局 / 边界。
+"""
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import _bootstrap  # noqa: F401  (sys.path 挂载)
+
+from layout_validator import (
+    GRID_COLUMNS,
+    LayoutError,
+    LayoutValidator,
+    REQUIRED_STAGES,
+    Severity,
+    load_point_ids,
+)
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+DASHBOARD_DIR = os.path.dirname(HERE)
+LAYOUT_YAML = os.path.join(DASHBOARD_DIR, "cockpit.ti.yaml")
+POINT_DICT_CSV = os.path.join(
+    DASHBOARD_DIR, os.pardir, "point-dict", "point_dict.default.csv")
+
+
+def _write_layout(tmp_path: str, content: str) -> str:
+    """把布局内容写到临时文件，返回路径。"""
+    path = os.path.join(tmp_path, "cockpit.test.yaml")
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write(content)
+    return path
+
+
+def _write_point_dict(tmp_path: str, ids: list) -> str:
+    """写一个最小点位字典 CSV（仅 point_id 列）。"""
+    path = os.path.join(tmp_path, "points.csv")
+    with open(path, "w", encoding="utf-8") as fh:
+        fh.write("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\n")
+        for i, pid in enumerate(ids):
+            fh.write(f"D{i},{pid},n,u,float,1000,true,n,simulator\n")
+    return path
+
+
+# 最小合法布局模板（便于构造各类变形）
+_VALID_LAYOUT = """\
+$schema: iAOP-cockpit-layout-v1
+title: 测试驾驶舱
+theme: dark
+widgets:
+  - type: process_view
+    src: ti_four_state.svg
+    x: 0
+    y: 0
+    w: 12
+    h: 4
+    description: 四状态工艺流程
+    stages:
+      - id: chlorination
+        name: 氯化
+        order: 1
+      - id: purification
+        name: 精制
+        order: 2
+      - id: reduction
+        name: 还原
+        order: 3
+      - id: distillation
+        name: 蒸馏
+        order: 4
+  - type: trend
+    bind: CLF-01.TEMP
+    x: 0
+    y: 4
+    w: 6
+    h: 2
+    description: 氯化炉温度
+"""
+
+
+class TestEndToEndRealAssets(unittest.TestCase):
+    """真实 cockpit.ti.yaml + point_dict.default.csv 端到端校验。"""
+
+    def test_real_layout_passes(self):
+        report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate()
+        if not report.passed:
+            for issue in report.errors:
+                print("ERROR:", issue.widget_id, issue.field, issue.reason)
+        self.assertTrue(report.passed, "真实布局应通过全部校验")
+        self.assertGreater(report.widget_count, 0)
+
+    def test_real_layout_has_process_view_with_four_stages(self):
+        report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate()
+        # 无 stages 相关 ERROR
+        stage_errors = [i for i in report.errors if i.field == "stages"]
+        self.assertEqual(stage_errors, [])
+
+
+class TestWidgetType(unittest.TestCase):
+    """widget 类型合法性。"""
+
+    def test_invalid_widget_type_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            layout = _VALID_LAYOUT.replace("type: trend", "type: radar_chart")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            errors = [i for i in report.errors if i.field == "type"]
+            self.assertEqual(len(errors), 1)
+            self.assertIn("非法 widget 类型", errors[0].reason)
+
+
+class TestGridBounds(unittest.TestCase):
+    """12 列网格校验。"""
+
+    def test_x_plus_w_exceeds_columns(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            # trend x=10 w=6 → 16 > 12
+            layout = _VALID_LAYOUT.replace(
+                "    bind: CLF-01.TEMP\n    x: 0\n    y: 4\n    w: 6\n    h: 2",
+                "    bind: CLF-01.TEMP\n    x: 10\n    y: 4\n    w: 6\n    h: 2")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            grid_errors = [i for i in report.errors if i.field == "grid"
+                           and "越出" in i.reason]
+            self.assertEqual(len(grid_errors), 1)
+
+    def test_negative_x_rejected(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            layout = _VALID_LAYOUT.replace("    x: 0\n    y: 4\n    w: 6\n    h: 2",
+                                           "    x: -1\n    y: 4\n    w: 6\n    h: 2")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            neg = [i for i in report.errors if "坐标不能为负" in i.reason]
+            self.assertEqual(len(neg), 1)
+
+    def test_zero_width_rejected(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            layout = _VALID_LAYOUT.replace("    x: 0\n    y: 4\n    w: 6\n    h: 2",
+                                           "    x: 0\n    y: 4\n    w: 0\n    h: 2")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            wh = [i for i in report.errors if "w/h 必须为正整数" in i.reason]
+            self.assertEqual(len(wh), 1)
+
+
+class TestBindPointId(unittest.TestCase):
+    """bind point_id 在点位字典内。"""
+
+    def test_bind_not_in_dict_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            path = _write_layout(td, _VALID_LAYOUT)
+            # 点位字典不含 CLF-01.TEMP
+            csv_path = _write_point_dict(td, ["OTHER-01.X"])
+            report = LayoutValidator(path, csv_path).validate()
+            bind_err = [i for i in report.errors if i.field == "bind"]
+            self.assertEqual(len(bind_err), 1)
+            self.assertIn("不在点位字典内", bind_err[0].reason)
+
+    def test_bind_in_dict_passes(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            path = _write_layout(td, _VALID_LAYOUT)
+            csv_path = _write_point_dict(td, ["CLF-01.TEMP"])
+            report = LayoutValidator(path, csv_path).validate()
+            bind_err = [i for i in report.errors if i.field == "bind"]
+            self.assertEqual(bind_err, [])
+
+
+class TestProcessViewStages(unittest.TestCase):
+    """process_view 四状态覆盖。"""
+
+    def test_missing_process_view_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            # 删除 process_view 块（保留 trend）
+            layout = """\
+$schema: iAOP-cockpit-layout-v1
+title: t
+widgets:
+  - type: trend
+    bind: CLF-01.TEMP
+    x: 0
+    y: 0
+    w: 6
+    h: 2
+"""
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            pv_err = [i for i in report.errors if i.field == "process_view"]
+            self.assertEqual(len(pv_err), 1)
+
+    def test_missing_required_stage_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            # 删除 distillation stage
+            layout = _VALID_LAYOUT.replace(
+                "      - id: distillation\n        name: 蒸馏\n        order: 4\n", "")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            missing = [i for i in report.errors if "未覆盖必需四状态" in i.reason]
+            self.assertEqual(len(missing), 1)
+            self.assertIn("distillation", missing[0].reason)
+
+    def test_non_monotonic_order_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            # 把 reduction order 改为 5（> distillation 的 4）→ 非单调
+            layout = _VALID_LAYOUT.replace("      - id: reduction\n        name: 还原\n        order: 3",
+                                           "      - id: reduction\n        name: 还原\n        order: 5")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            order_err = [i for i in report.errors if "order 非单调递增" in i.reason]
+            self.assertEqual(len(order_err), 1)
+
+    def test_duplicate_stage_id_warn(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            # 重复 chlorination（覆盖必需状态校验仍过，但 WARN 重复）
+            layout = _VALID_LAYOUT + """\
+"""
+            # 构造一个有重复 stage 的 process_view（替换 stages 块）
+            dup_layout = _VALID_LAYOUT.replace(
+                "      - id: distillation\n        name: 蒸馏\n        order: 4",
+                "      - id: distillation\n        name: 蒸馏\n        order: 4\n"
+                "      - id: chlorination\n        name: 氯化2\n        order: 5")
+            path = _write_layout(td, dup_layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            dup = [i for i in report.issues if "stage id 重复" in i.reason]
+            self.assertEqual(len(dup), 1)
+
+
+class TestSchemaHeader(unittest.TestCase):
+    """$schema 头校验。"""
+
+    def test_wrong_schema_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            layout = _VALID_LAYOUT.replace("iAOP-cockpit-layout-v1", "some-other-schema")
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            schema_err = [i for i in report.errors if i.field == "$schema"]
+            self.assertEqual(len(schema_err), 1)
+
+
+class TestPointDictLoader(unittest.TestCase):
+    """点位字典 CSV 加载。"""
+
+    def test_load_point_ids(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            csv_path = _write_point_dict(td, ["A.X", "B.Y"])
+            ids = load_point_ids(csv_path)
+            self.assertEqual(ids, ["A.X", "B.Y"])
+
+    def test_missing_csv_raises(self):
+        with self.assertRaises(LayoutError):
+            load_point_ids("/nonexistent/points.csv")
+
+    def test_csv_missing_point_id_column_raises(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            path = os.path.join(td, "bad.csv")
+            with open(path, "w", encoding="utf-8") as fh:
+                fh.write("device_id,name\nD1,n\n")
+            with self.assertRaises(LayoutError):
+                load_point_ids(path)
+
+
+class TestReportExport(unittest.TestCase):
+    """报告序列化 + 边界。"""
+
+    def test_empty_layout_error(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            layout = """\
+$schema: iAOP-cockpit-layout-v1
+title: t
+widgets: []
+"""
+            path = _write_layout(td, layout)
+            report = LayoutValidator(path, _write_point_dict(td, ["CLF-01.TEMP"])).validate()
+            self.assertFalse(report.passed)
+
+    def test_report_to_dict(self):
+        report = LayoutValidator(LAYOUT_YAML, POINT_DICT_CSV).validate()
+        d = report.to_dict()
+        self.assertEqual(d["passed"], True)
+        self.assertIn("widget_count", d)
+        self.assertEqual(d["error_count"], 0)
+
+
+if __name__ == "__main__":
+    unittest.main()