# -*- coding: utf-8 -*- """契约四处同步机械核验（QC 退回 #1/#3/#4/#5 的防复发闸门）。 核验项 ------ 1. 契约函数 10 个：m4a_contract.CONTRACT_FUNCTIONS ↔ __init__.py 导出 ↔ init.py 注册 ↔ wwwroot/api/*.dspy 文件 ↔ scripts/load_path.py PATHS，五处集合完全相等； 2. wwwroot/api/ 下不存在未注册端点，PATHS 不指向不存在文件； 3. 每个 .dspy 的 debug 行必须是 f-string（禁普通字符串占位）； 4. .dspy 禁项审计：无 import / print / uuid； 5. 所有 .py py_compile 通过； 6. 交付文件清单（deliver 用）与磁盘实际存在文件一致——本脚本直接产出真实清单。 用法：python3 scripts/check_contract_sync.py [--json out.json] """ import ast import json import os import re import sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) WORKSPACE_ROOT = os.path.dirname(os.path.dirname(ROOT)) # 机构工作空间根（清单路径基准） sys.path.insert(0, ROOT) sys.path.insert(0, os.path.dirname(ROOT)) API_DIR = os.path.join(ROOT, "wwwroot", "api") PKG_DIR = os.path.join(ROOT, "pbl_agent_runtime") FORBIDDEN_IN_DSPY = ("import ", "print(", "uuid") def _contract_names(): from pbl_agent_runtime.m4a_contract import CONTRACT_FUNCTIONS return list(CONTRACT_FUNCTIONS) def _dspy_files(): if not os.path.isdir(API_DIR): return [] return sorted(f[:-5] for f in os.listdir(API_DIR) if f.endswith(".dspy")) def _load_path_entries(): path = os.path.join(ROOT, "scripts", "load_path.py") with open(path, encoding="utf-8") as handle: tree = ast.parse(handle.read()) for node in tree.body: if isinstance(node, ast.Assign) and getattr(node.targets[0], "id", "") == "PATHS": out = [] for elt in node.value.elts: out.append((elt.elts[0].value, elt.elts[1].value)) return out return [] def _init_exports(): path = os.path.join(PKG_DIR, "__init__.py") with open(path, encoding="utf-8") as handle: tree = ast.parse(handle.read()) exported = set() for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): for alias in node.names: exported.add(alias.asname or alias.name) all_list = set() for node in tree.body: if isinstance(node, ast.Assign) and getattr(node.targets[0], "id", "") == "__all__": all_list = {elt.value for elt in node.value.elts} return exported, all_list def _init_registers(): """从 init.py 静态解析注册名（CONTRACT_FUNCTIONS 循环 + EXTRA_FUNCTIONS）。""" path = os.path.join(PKG_DIR, "init.py") with open(path, encoding="utf-8") as handle: source = handle.read() names = set() match = re.search(r"EXTRA_FUNCTIONS\s*=\s*\(([^)]*)\)", source, re.S) if match: names |= set(re.findall(r"'([^']+)'|\"([^\"]+)\"", match.group(1)) and [a or b for a, b in re.findall(r"'([^']*)'|\"([^\"]*)\"", match.group(1))]) return names, source def main(): problems = [] contracts = _contract_names() dspy = _dspy_files() entries = _load_path_entries() entry_paths = [p for p, _ in entries] entry_dspy = sorted(p.split("/")[-1][:-5] for p in entry_paths if p.endswith(".dspy")) # 1) 集合一致性 if sorted(contracts) != sorted(dspy): problems.append("contract vs dspy mismatch: only_contract=%s only_dspy=%s" % (sorted(set(contracts) - set(dspy)), sorted(set(dspy) - set(contracts)))) if sorted(contracts) != sorted(entry_dspy): problems.append("contract vs load_path mismatch: only_contract=%s only_path=%s" % (sorted(set(contracts) - set(entry_dspy)), sorted(set(entry_dspy) - set(contracts)))) # 2) PATHS 指向的文件必须真实存在 for path, role in entries: # PATHS 是运行时 URL 路径 /{module}/wwwroot相对路径，磁盘落点在 ROOT/wwwroot/ rel = path.lstrip("/") parts = rel.split("/", 1) rel_in_module = parts[1] if len(parts) == 2 and parts[0] == "pbl_agent_runtime" else rel full = os.path.join(ROOT, "wwwroot", rel_in_module) if not os.path.isfile(full): problems.append("load_path points to missing file: %s" % path) if role not in ("any", "user", "admin"): problems.append("load_path bad role %r for %s" % (role, path)) # 3)+4) dspy 内容审计 for name in dspy: full = os.path.join(API_DIR, "%s.dspy" % name) with open(full, encoding="utf-8") as handle: text = handle.read() lines = text.splitlines() debug_lines = [ln for ln in lines if ln.strip().startswith("debug(")] if not debug_lines: problems.append("%s.dspy: no debug line" % name) for ln in debug_lines: if "debug(f'" not in ln and 'debug(f"' not in ln: problems.append("%s.dspy: debug is not f-string -> %s" % (name, ln.strip())) if "params_kw" not in ln and "result" not in ln: problems.append("%s.dspy: debug missing real params -> %s" % (name, ln.strip())) for bad in FORBIDDEN_IN_DSPY: for idx, ln in enumerate(lines, 1): if bad in ln: problems.append("%s.dspy:%d forbidden token %r" % (name, idx, bad)) if not any(ln.strip().startswith("return ") for ln in lines): problems.append("%s.dspy: missing explicit return" % name) if name not in text: problems.append("%s.dspy: does not call contract function %s" % (name, name)) # 5) __init__ 导出 api + 契约对象 exported, all_list = _init_exports() if "api" not in exported or "api" not in all_list: problems.append("__init__.py must export `api` (QC #1)") for name in ("load_pbl_agent_runtime", "self_check"): if name not in exported: problems.append("__init__.py missing export: %s" % name) _, init_source = _init_registers() if "def api" in init_source and "_ApiAccessor" not in init_source: problems.append("init.py api must stay compatible with attribute access") # 6) py_compile 全部 .py compiled = 0 for base, dirs, files in os.walk(ROOT): dirs[:] = [d for d in dirs if d not in ("__pycache__", ".git", "var")] for fname in files: if not fname.endswith(".py"): continue full = os.path.join(base, fname) try: with open(full, encoding="utf-8") as handle: tree = ast.parse(handle.read(), filename=full) compile(tree, full, "exec") if not [n for n in tree.body if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Assign, ast.Import, ast.ImportFrom, ast.If, ast.Try, ast.With, ast.For, ast.While, ast.Expr, ast.AnnAssign))]: problems.append("empty module (no effective statement): %s" % os.path.relpath(full, ROOT)) compiled += 1 except SyntaxError as exc: problems.append("syntax error %s: %s" % (os.path.relpath(full, ROOT), exc)) except Exception as exc: problems.append("compile failed %s: %s" % (os.path.relpath(full, ROOT), exc)) # 7) 真实交付文件清单（存在性过滤，QC #5） # 清单文件自身不计入清单（否则字节数自引用抖动，QC#5 核验永远对不上） manifest_rel = os.path.abspath(os.path.join(ROOT, "scripts", "m4a_manifest.json")) inventory = [] for base, dirs, files in os.walk(ROOT): dirs[:] = [d for d in dirs if d not in ("__pycache__", ".git", "var")] for fname in sorted(files): full = os.path.join(base, fname) if os.path.abspath(full) == manifest_rel: continue if not os.path.isfile(full): continue rel = os.path.relpath(full, WORKSPACE_ROOT) with open(full, encoding="utf-8", errors="ignore") as handle: line_count = sum(1 for _ in handle) inventory.append({"path": rel.replace(os.sep, "/"), "bytes": os.path.getsize(full), "lines": line_count}) inventory.sort(key=lambda item: item["path"]) report = { "success": not problems, "contract_count": len(contracts), "contracts": contracts, "dspy_endpoints": dspy, "load_path_entries": len(entries), "py_compiled": compiled, "problems": problems, "inventory_count": len(inventory), "inventory": inventory, } print(json.dumps({k: v for k, v in report.items() if k != "inventory"}, ensure_ascii=False, indent=2)) if "--json" in sys.argv: out = sys.argv[sys.argv.index("--json") + 1] with open(out, "w", encoding="utf-8") as handle: json.dump(report, handle, ensure_ascii=False, indent=2) print("inventory written ->", out) return 0 if not problems else 1 if __name__ == "__main__": sys.exit(main())