#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M3a 自检脚本 —— pbl_compiler 确定性编译内核（canonical + gd_builder）。 运行（模块仓库根目录）:: cd modules/pbl_compiler && python3 scripts/test_m3a_selfcheck.py 退出码：0 = 6 组断言全部通过；1 = 有断言失败；2 = 被测模块加载失败。 ``apps/pbls/build.sh`` 第 9 步（py_compile）之后可直接串接本脚本做部署前门禁。 设计约束 -------- * **纯函数级自检**：只加载 ``pbl_compiler/canonical.py`` 与 ``pbl_compiler/gd_builder.py`` 两个纯函数模块，不连数据库、不依赖宿主应用 ``ServerEnv``、不需要 ``ahserver`` （故用 importlib 按文件路径加载，绕开包 ``__init__.py`` 的 ahserver 依赖）。 * **防御式导入**：被测函数缺失时打印 ``SELFCHECK_FAIL: missing `` 并计入失败， 绝不静默跳过（上一轮 QC 退回项 R1：声称的自检脚本必须真实可跑）。 * 6 组断言与 ``README.md`` §6 表格一一对应。 """ from __future__ import annotations import copy import hashlib import importlib.util import json import os import sys import types HERE = os.path.dirname(os.path.abspath(__file__)) MOD_ROOT = os.path.dirname(HERE) PKG_DIR = os.path.join(MOD_ROOT, 'pbl_compiler') PASS = [] FAIL = [] # -------------------------------------------------------------------------- # 输出与断言工具 # -------------------------------------------------------------------------- def _ok(group, msg): PASS.append('[%s] %s' % (group, msg)) print(' PASS [%s] %s' % (group, msg)) def _bad(group, msg): FAIL.append('[%s] %s' % (group, msg)) print(' FAIL [%s] %s' % (group, msg)) def _check(group, cond, msg): """断言：True → PASS，False → FAIL。返回 bool 便于短路后续依赖断言。""" if cond: _ok(group, msg) else: _bad(group, msg) return bool(cond) # -------------------------------------------------------------------------- # 按文件路径加载纯函数模块（避开 pbl_compiler/__init__.py 的 ahserver 依赖） # -------------------------------------------------------------------------- def _load_pure_modules(): """返回 (canonical_mod, gd_builder_mod)；任一失败返回 (None, None)。""" pkg_name = 'pbl_compiler' if pkg_name not in sys.modules: pkg = types.ModuleType(pkg_name) pkg.__path__ = [PKG_DIR] sys.modules[pkg_name] = pkg def _load(sub): path = os.path.join(PKG_DIR, sub + '.py') if not os.path.isfile(path): print('SELFCHECK_FAIL: missing file %s' % path) return None full = '%s.%s' % (pkg_name, sub) spec = importlib.util.spec_from_file_location(full, path) if spec is None or spec.loader is None: print('SELFCHECK_FAIL: cannot build import spec for %s' % path) return None mod = importlib.util.module_from_spec(spec) sys.modules[full] = mod try: spec.loader.exec_module(mod) except Exception as exc: # noqa: BLE001 print('SELFCHECK_FAIL: import %s raised %s: %s' % (full, type(exc).__name__, exc)) return None setattr(sys.modules[pkg_name], sub, mod) return mod canonical = _load('canonical') gd_builder = _load('gd_builder') if canonical is not None else None return canonical, gd_builder def _pick(mod, names, group): """按候选名取可调用对象；全部缺失 → 明确报 missing 并计 FAIL。""" if mod is None: return None for name in names: fn = getattr(mod, name, None) if callable(fn): return fn print('SELFCHECK_FAIL: missing %s in %s' % ('/'.join(names), getattr(mod, '__name__', '?'))) _bad(group, 'missing callable %s' % '/'.join(names)) return None # -------------------------------------------------------------------------- # 参考实现（用于交叉验证「规则本身」，不作为通过依据） # -------------------------------------------------------------------------- def _ref_canonical(obj): return json.dumps(obj, sort_keys=True, separators=(',', ':'), ensure_ascii=False) def _ref_fingerprint(obj): return hashlib.sha256(_ref_canonical(obj).encode('utf-8')).hexdigest() # -------------------------------------------------------------------------- # 测试夹具 # -------------------------------------------------------------------------- def _snapshot(): """最小可用蓝图版本快照（7 类子对象中的 4 类，键名走 M1a 落库形态）。""" return { 'content': { 'subobjects': { 'project': [{'code': 'pj1', 'name': '火星基地', 'title': '火星基地'}], 'problem': [{'code': 'pb1', 'name': '氧气不足', 'driving_question': '如何在 30 天内自给氧气？'}], 'learning_goals': [{'code': 'g1', 'name': '光合作用', 'weight': 0.5}], 'scenes': [{'code': 's1', 'name': '温室'}, {'code': 's2', 'name': '控制室'}], 'roles': [{'code': 'r1', 'name': '生物学家', 'capability_key': 'pbl.role.skill'}], 'missions': [{'code': 'm1', 'name': '建温室', 'seq': 1}], 'events': [{'code': 'e1', 'name': '进入温室', 'trigger': 'enter', 'response': [{'capability': 'pbl.item.consume'}]}], 'artifacts': [{'code': 'a1', 'name': '氧气报告'}], 'evidence_specs': [{'code': 'ev1', 'name': '实验记录'}], 'rubrics': [{'code': 'rb1', 'name': '协作', 'weight': 1.0}], 'assets': [{'code': 'as1', 'name': '温室模型.glb', 'kind': 'model'}], } } } def _ctx(**over): ctx = { 'blueprint_id': 1001, 'blueprint_version_no': 3, 'compiler_version': '1.0.0', 'ruleset_version': 'pbl.rules.v1', 'rules_hash': 'deadbeefdeadbeef', 'created_at': '2026-09-15 09:40:04', 'duration_ms': 7, 'task_no': 'CT-T1-1001-3-000001', } ctx.update(over) return ctx def _registry(): return [ {'capability_key': 'pbl.role.skill', 'category': 'role', 'version_no': 1, 'args_schema_json': '{"level":"int"}', 'permission_required': '', 'is_enabled': True}, {'capability_key': 'pbl.item.consume', 'category': 'item', 'version_no': 1, 'args_schema_json': '{"count":"int"}', 'permission_required': '', 'is_enabled': True}, ] def _build(f_build, snapshot, ctx, registry): """调用 build_game_definition 并归一化返回值为 (gd, fp)。""" out = f_build(snapshot, ctx, registry) if isinstance(out, tuple) and len(out) == 2: return out[0], out[1] if isinstance(out, dict): fp = (out.get('manifest') or {}).get('fingerprint') return out, fp raise AssertionError('build_game_definition 返回类型异常: %r' % type(out).__name__) # -------------------------------------------------------------------------- # 主流程：6 组断言 # -------------------------------------------------------------------------- def main(): print('=== pbl_compiler M3a selfcheck (6 groups) ===') print('module root : %s' % MOD_ROOT) canonical, gd_builder = _load_pure_modules() if canonical is None or gd_builder is None: print('SELFCHECK_FAIL: canonical/gd_builder 加载失败，自检中止') return 2 f_canon = _pick(canonical, ('canonical_json',), 'G1') f_strip = _pick(canonical, ('strip_volatile',), 'G3') f_fp = _pick(canonical, ('sha256_fingerprint', 'fingerprint'), 'G1') f_reghash = _pick(canonical, ('registry_hash',), 'G6') f_build = _pick(gd_builder, ('build_game_definition',), 'G4') top_keys = getattr(gd_builder, 'GD_TOP_KEYS', None) err_cls = getattr(canonical, 'CanonicalError', ValueError) if f_canon is None or f_fp is None or f_build is None: print('SELFCHECK_FAIL: 核心函数缺失，自检中止') return 2 # ---------------- G1 sort_keys：键序无关 + 指纹形态 ---------------- print('-- G1 canonical 键序无关 / 指纹形态 --') a = {'name': 'quest-1', 'meta': {'z': 1, 'a': 2}, 'tags': ['x', 'y']} b = {'tags': ['x', 'y'], 'name': 'quest-1', 'meta': {'a': 2, 'z': 1}} ca, cb = f_canon(a), f_canon(b) _check('G1', ca == cb, '同内容不同键序 → canonical_json 完全相同') _check('G1', f_fp(a) == f_fp(b), '同内容不同键序 → sha256_fingerprint 相同') fp_a = f_fp(a) _check('G1', isinstance(fp_a, str) and len(fp_a) == 64 and all(ch in '0123456789abcdef' for ch in fp_a), '指纹为 64 位小写十六进制 sha256（实测 %s…）' % fp_a[:16]) _check('G1', ca == _ref_canonical(a), 'canonical_json 与参考实现(sort_keys=True + separators=(",",":"))一致') _check('G1', f_fp(a) != f_fp({'name': 'quest-2', 'meta': {'z': 1, 'a': 2}, 'tags': ['x', 'y']}), '内容变更 → 指纹必变') # ---------------- G2 零空白 / 中文不转义 / 浮点定点 / 拒 NaN ---------------- print('-- G2 零空白 / ensure_ascii=False / 浮点定点 / NaN 拒绝 --') nested = {'规则': {'概率': 0.5, 'n': 3, 'list': [{'b': 1.0, 'a': '中文值'}]}} cn = f_canon(nested) _check('G2', not any(ch in cn for ch in ' \n\t\r'), '规范化串不含任何空白字符（零空白分隔符）') _check('G2', '\\u' not in cn, 'ensure_ascii=False：中文原样保留不被转义') _check('G2', f_canon(json.loads(cn)) == cn, '规范化幂等：canonical(json.loads(canonical(x))) == canonical(x)') _check('G2', f_fp({'v': 0.1 + 0.2}) == f_fp({'v': 0.3}), '浮点定点化：0.1+0.2 与 0.3 指纹相同（FLOAT_PRECISION 量化）') _check('G2', f_fp({'v': 1.0}) == f_fp({'v': 1}), '整值浮点与整数指纹相同（定点化去尾零）') try: f_canon({'v': float('nan')}) _check('G2', False, 'NaN 应被拒绝（allow_nan=False）') except err_cls: _check('G2', True, 'NaN 触发 CanonicalError（非有限浮点不可规范化）') except Exception as exc: # noqa: BLE001 _check('G2', False, 'NaN 抛出非预期异常 %s: %s' % (type(exc).__name__, exc)) # ---------------- G3 strip_volatile：易变字段不参与指纹 ---------------- print('-- G3 strip_volatile 易变字段不参与指纹 --') base = {'blueprint_id': 1001, 'gd_version': 1, 'content': {'entities': [{'code': 'e1', 'hp': 100}]}} v1 = dict(base, created_at='2026-09-15 09:40:04', id=11, trace_id='t-1', duration_ms=5, task_no='CT-1') v2 = dict(base, created_at='2030-01-01 00:00:00', id=99, trace_id='t-9', duration_ms=888, task_no='CT-9') _check('G3', f_fp(v1) == f_fp(v2), 'created_at/id/trace_id/duration_ms/task_no 变化 → 指纹不变') _check('G3', f_fp(v1) == f_fp(base), '易变字段整体存在与否不影响指纹') changed = copy.deepcopy(base) changed['content']['entities'][0]['hp'] = 101 _check('G3', f_fp(changed) != f_fp(base), '业务字段(hp)变化 → 指纹必变') if f_strip is not None: stripped = f_strip(v1) _check('G3', not any(k in stripped for k in ('created_at', 'id', 'trace_id', 'duration_ms', 'task_no')), 'strip_volatile 确实剥离全部易变键') _check('G3', stripped.get('content') == base['content'], 'strip_volatile 只剥不改：业务子树原样保留') _check('G3', f_strip({'a': [{'created_at': 'x', 'k': 1}]}) == {'a': [{'k': 1}]}, 'strip_volatile 递归进入 list[dict]') _check('G3', f_fp(base) == _ref_fingerprint(f_strip(v1)) if f_strip else False, '模块指纹 == sha256(参考规范化(strip_volatile(x)))（算法口径一致）') # ---------------- G4 GD 恰好 10 个顶层键 ---------------- print('-- G4 Game Definition 10 顶层键 --') EXPECTED = ('manifest', 'pbl', 'world', 'scenes', 'entities', 'events', 'states', 'capabilities', 'assessment', 'assets') _check('G4', tuple(top_keys or ()) == EXPECTED, 'gd_builder.GD_TOP_KEYS == 10 键契约（实测 %s）' % (top_keys,)) snap = _snapshot() try: gd, fp = _build(f_build, snap, _ctx(), _registry()) except Exception as exc: # noqa: BLE001 gd, fp = None, None _bad('G4', 'build_game_definition 抛异常 %s: %s' % (type(exc).__name__, exc)) if isinstance(gd, dict): keys = tuple(gd.keys()) _check('G4', len(keys) == 10, 'GD 顶层键数量 == 10（实测 %d）' % len(keys)) _check('G4', set(keys) == set(EXPECTED), 'GD 顶层键集合一致；缺=%s 多=%s' % (sorted(set(EXPECTED) - set(keys)), sorted(set(keys) - set(EXPECTED)))) _check('G4', isinstance(fp, str) and len(fp) == 64, 'build_game_definition 同时返回 64 位指纹') _check('G4', gd['manifest'].get('fingerprint') == fp, 'manifest.fingerprint 与返回指纹一致') _check('G4', gd['manifest'].get('schema') == 'pbl.game_definition.v1' and gd['manifest'].get('deterministic') is True and gd['manifest'].get('llmUsed') is False, 'manifest 声明 schema/deterministic=True/llmUsed=False（零 LLM）') # 口径统一（README §8 P1）：gd_builder 以**补齐派生字段后的 # capabilities.items** 计算 registry_hash，故自检主断言对同一口径重算； # 第二条断言验证「派生不丢语义」——对原始注册表行重算必须得到同值。 caps_items = (gd.get('capabilities') or {}).get('items') or [] _check('G4', gd['manifest'].get('registryHash') == f_reghash(caps_items), 'manifest.registryHash == canonical.registry_hash(GD capabilities.items)（统一口径）') # 派生不丢语义：GD items 是「注册表行 ∪ 蓝图引用但未注册的能力」， # 故 items 可能多于原始注册表（多出的标 registered=false，供 M4b # fail-closed 告警）。此处断言**原始注册表每条能力都在 items 中且 # version_no 一致**，即派生过程未丢失/未篡改注册表语义。 raw = {str(r.get('capability_key')): int(r.get('version_no') or 1) for r in _registry()} got = {str(i.get('capability_key')): int(i.get('version_no') or 1) for i in caps_items} missing = sorted(k for k in raw if got.get(k) != raw[k]) _check('G4', not missing, '原始注册表能力全部进入 GD items 且 version_no 一致（缺失/不一致=%s）' % missing) # ---------------- G5 同输入重编译 N 次指纹全等（US-11 / F-CP-03） ---------------- print('-- G5 确定性：同输入重编译指纹全等 --') fps = [] for i in range(3): try: _gd, _fp = _build(f_build, copy.deepcopy(snap), _ctx(), _registry()) fps.append(_fp) except Exception as exc: # noqa: BLE001 _bad('G5', '第 %d 次编译抛异常 %s: %s' % (i + 1, type(exc).__name__, exc)) _check('G5', len(fps) == 3 and len(set(fps)) == 1, '连续 3 次编译指纹全等（%s）' % (fps[0][:16] if fps else 'N/A')) try: _gd_v, fp_v = _build(f_build, copy.deepcopy(snap), _ctx(created_at='2031-12-31 23:59:59', duration_ms=9999, task_no='CT-OTHER-999'), _registry()) _check('G5', fps and fp_v == fps[0], 'ctx 易变字段(created_at/duration_ms/task_no)变化 → 指纹不变') except Exception as exc: # noqa: BLE001 _bad('G5', '易变 ctx 编译抛异常 %s: %s' % (type(exc).__name__, exc)) snap2 = copy.deepcopy(snap) subs = snap2['content']['subobjects'] subs['scenes'] = list(reversed(subs['scenes'])) try: _gd_o, fp_o = _build(f_build, snap2, _ctx(), _registry()) _check('G5', fps and fp_o == fps[0], '子对象入库顺序颠倒 → 指纹不变（集合稳定化排序生效）') except Exception as exc: # noqa: BLE001 _bad('G5', '乱序快照编译抛异常 %s: %s' % (type(exc).__name__, exc)) snap3 = copy.deepcopy(snap) snap3['content']['subobjects']['roles'][0]['name'] = '工程师' try: _gd_c, fp_c = _build(f_build, snap3, _ctx(), _registry()) _check('G5', fps and fp_c != fps[0], '业务内容变更 → 指纹必变') except Exception as exc: # noqa: BLE001 _bad('G5', '变更快照编译抛异常 %s: %s' % (type(exc).__name__, exc)) # ---------------- G6 registry_hash 语义 ---------------- print('-- G6 registry_hash 能力注册表指纹 --') if f_reghash is not None: reg = _registry() h1 = f_reghash(reg) _check('G6', isinstance(h1, str) and len(h1) == 64, 'registry_hash 为 64 位 sha256（%s…）' % h1[:16]) _check('G6', f_reghash(list(reversed(reg))) == h1, 'registry_hash 与能力行顺序无关') reg_more = reg + [{'capability_key': 'pbl.event.broadcast', 'category': 'event', 'version_no': 1, 'args_schema_json': '{}', 'permission_required': '', 'is_enabled': True}] _check('G6', f_reghash(reg_more) != h1, '新增一条能力 → registry_hash 变化') reg_bump = [dict(r) for r in reg] reg_bump[0]['version_no'] = 2 _check('G6', f_reghash(reg_bump) != h1, '能力 version_no 升级 → registry_hash 变化') reg_schema = [dict(r) for r in reg] reg_schema[0]['args_schema_json'] = '{"level":"str"}' _check('G6', f_reghash(reg_schema) != h1, 'args_schema 变更 → registry_hash 变化') reg_ts = [dict(r, created_at='2030-01-01 00:00:00', id=777) for r in reg] _check('G6', f_reghash(reg_ts) == h1, '仅改 created_at/id 等易变字段 → registry_hash 不变') _check('G6', f_reghash([]) != h1 and len(f_reghash([])) == 64, '空注册表有确定哈希且与非空不同') else: _bad('G6', 'canonical.registry_hash 不可用') print('--- summary: pass=%d fail=%d ---' % (len(PASS), len(FAIL))) if FAIL: for item in FAIL: print('SELFCHECK_FAIL %s' % item) return 1 print('SELFCHECK_OK 6 groups all passed (G1..G6)') return 0 if __name__ == '__main__': sys.exit(main())