aps-agent/server/agent_core/registry.py

# ============================================================
# 可重生模块注册表（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}