# ============================================================ # 可重生模块注册表（moduleId: core-registry, 可重生 ✅） # plan.md §9.4/§6.10.3：每个源文件头部的 "moduleId: xxx, 可重生" 声明 # 即注册表数据源——本模块扫描源码目录，产出重生中心的模块清单。 # ============================================================ from __future__ import annotations # 前向类型引用 import os # 目录遍历 import re # 模块头解析 from typing import Any # 类型标注 # 扫描根（相对仓库根）：后端 / 前端源码 / 契约 _SCAN_ROOTS = ["server", "apps/web/src", "shared"] # 参与扫描的扩展名 _EXTS = {".py", ".ts", ".tsx", ".css", ".json", ".html"} # 模块头声明的匹配模式：moduleId: xxx（可选"可重生"标记） _HEAD_RE = re.compile(r"moduleId:\s*([\w-]+)(.{0,20}?可重生)?") def _layer_of(path: str) -> str: """按路径猜测模块所属层（注册表的分组维度）。""" p = path.replace("\\", "/") # 统一分隔符 if p.startswith("server/engines"): return "engine" # 引擎层 if p.startswith("server/agent_core"): return "agent-core" # 智能体核心 if p.startswith("server/aps_domain"): return "domain" # 领域层 if p.startswith("server/state"): return "state" # 数据层 if p.startswith("server/gateway") or p.endswith("server/main.py"): return "gateway" # 接入层 if p.startswith("apps/web"): return "ui" # 表现层 if p.startswith("shared"): return "contract" # 契约层 return "other" # 兜底 def scan_modules(repo_root: str = ".") -> list[dict[str, Any]]: """扫描源码，收集全部带 moduleId 声明的模块（P0 只读；重生中心数据源）。""" modules: list[dict[str, Any]] = [] # 结果集 for root in _SCAN_ROOTS: # 逐扫描根 base = os.path.join(repo_root, root) # 绝对化 if not os.path.isdir(base): # 目录缺失跳过 continue for dirpath, dirnames, filenames in os.walk(base): # 递归遍历 dirnames[:] = [d for d in dirnames if d not in ("node_modules", "dist", "__pycache__", "data")] # 剪枝 for fn in filenames: # 逐文件 if os.path.splitext(fn)[1] not in _EXTS: # 非目标扩展名跳过 continue path = os.path.join(dirpath, fn) # 文件路径 try: with open(path, "r", encoding="utf-8") as f: # 只读文件头 head = f.read(2048) # 模块头声明约定在前 2KB 内 except OSError: # 读失败跳过（不致命） continue m = _HEAD_RE.search(head) # 匹配声明 if not m: # 无声明 → 非注册模块 continue rel = os.path.relpath(path, repo_root).replace("\\", "/") # 相对路径 modules.append({ # 注册表条目（ModuleManifest 的扫描态子集） "moduleId": m.group(1), # 模块 ID "path": rel, # 源文件路径 "layer": _layer_of(rel), # 所属层 "regenerable": bool(m.group(2)), # 是否声明可重生 }) modules.sort(key=lambda x: (x["layer"], x["moduleId"])) # 稳定排序（分组展示友好） return modules # 返回清单 def verify_audit_chain(events: list[dict[str, Any]]) -> dict[str, Any]: """校验审计哈希链完整性（门禁管理台"防线状态"的校验按钮，§6.10.2）。 重算每个事件的哈希并核对前驱衔接；返回 {ok, checked, brokenAt}。 """ import hashlib, json # 局部导入（仅校验用） prev = "GENESIS" # 链头哨兵（与 write_audit 一致） for i, ev in enumerate(events): # 顺序遍历 if ev.get("prevHash") != prev: # 前驱衔接断裂 return {"ok": False, "checked": i, "brokenAt": ev.get("id")} body = {k: v for k, v in ev.items() if k != "hash"} # 剥离哈希字段还原签名体 payload = prev + json.dumps(body, ensure_ascii=False, sort_keys=True) # 重建签名载荷 if hashlib.sha256(payload.encode("utf-8")).hexdigest() != ev.get("hash"): # 哈希不符 → 被篡改 return {"ok": False, "checked": i, "brokenAt": ev.get("id")} prev = ev["hash"] # 推进链头 return {"ok": True, "checked": len(events), "brokenAt": None} # 全链完好 # ============================================================ # 可重生流水线注册表（§9.4/§6.10.3，矩阵 91）： # 候选版本登记 + 当前激活版本指针 + 可回退基线； # 与 scan_modules 扫描出的 moduleId 衔接（world["rebuildRegistry"] 持久化）。 # ============================================================ REBUILD_REGISTRY_KEY = "rebuildRegistry" def rebuild_registry(world: dict[str, Any]) -> dict[str, Any]: """获取（缺省创建）可重生流水线注册表分区。""" reg = world.setdefault(REBUILD_REGISTRY_KEY, {}) reg.setdefault("candidates", []) reg.setdefault("activeVersions", {}) reg.setdefault("baselines", {}) return reg def list_rebuild_candidates(world: dict[str, Any]) -> list[dict[str, Any]]: """全部候选版本记录（按登记顺序）。""" return list(rebuild_registry(world)["candidates"]) def get_rebuild_candidate(world: dict[str, Any], module_id: str) -> dict[str, Any] | None: """该 moduleId 的最新候选版本（无则 None）。""" matches = [c for c in rebuild_registry(world)["candidates"] if c.get("moduleId") == module_id] return matches[-1] if matches else None def register_rebuild_candidate(world: dict[str, Any], module_id: str, candidate: dict[str, Any]) -> dict[str, Any]: """登记一个候选版本，并初始化该模块基线（首次 = 当前激活指针或现场源码 current）。""" reg = rebuild_registry(world) reg["candidates"].append(candidate) reg["baselines"].setdefault(module_id, reg["activeVersions"].get(module_id) or "current") return candidate def active_version(world: dict[str, Any], module_id: str) -> str | None: """当前激活版本指针（None=现场源码未登记版本）。""" return rebuild_registry(world)["activeVersions"].get(module_id) def baseline_version(world: dict[str, Any], module_id: str) -> str: """可回退基线（缺省 current=现场源码）。""" return rebuild_registry(world)["baselines"].get(module_id) or "current" def set_active_version(world: dict[str, Any], module_id: str, version: str) -> None: """更新激活版本指针（灰度转正 / 回滚恢复）。""" rebuild_registry(world)["activeVersions"][module_id] = version def set_baseline_version(world: dict[str, Any], module_id: str, version: str) -> None: """更新可回退基线。""" rebuild_registry(world)["baselines"][module_id] = version def rollback_registry(world: dict[str, Any], module_id: str, reason: str = "auto") -> dict[str, Any]: """回滚：激活指针恢复至上一条可用基线，候选标记 ROLLED_BACK（注册表恢复）。""" reg = rebuild_registry(world) baseline = reg["baselines"].get(module_id, "current") previous = reg["activeVersions"].get(module_id) reg["activeVersions"][module_id] = baseline candidate = get_rebuild_candidate(world, module_id) if candidate is not None: candidate["status"] = "ROLLED_BACK" candidate["rollbackReason"] = reason return {"moduleId": module_id, "restoredTo": baseline, "previousActive": previous, "reason": reason}