395 lines
19 KiB
Python
395 lines
19 KiB
Python
#!/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 <name>`` 并计入失败，
|
||
绝不静默跳过（上一轮 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())
|