From 80ff6783326c877f7c552a8873c808343aaf6bd2 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Fri, 18 Sep 2026 12:18:08 +0800 Subject: [PATCH] =?UTF-8?q?deliver:=20=E4=BA=A4=E4=BB=98=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=88=E5=BC=95=E6=93=8E=E4=BB=A3=E4=B8=BA=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- models/pbl_compile_task.json | 215 +++++++ pbl_compiler/api.py | 1067 ++++++++++++++++++++++++++-------- pbl_compiler/canonical.py | 207 +++++++ pbl_compiler/gd_builder.py | 534 +++++++++++++++++ 4 files changed, 1791 insertions(+), 232 deletions(-) create mode 100644 models/pbl_compile_task.json create mode 100644 pbl_compiler/canonical.py create mode 100644 pbl_compiler/gd_builder.py diff --git a/models/pbl_compile_task.json b/models/pbl_compile_task.json new file mode 100644 index 0000000..93dee1a --- /dev/null +++ b/models/pbl_compile_task.json @@ -0,0 +1,215 @@ +{ + "summary": [ + { + "name": "pbl_compile_task", + "comment": "编译任务（approval_id 门禁，F-CP-01/US-10）", + "module": "pbl_compiler", + "engine": "mariadb", + "charset": "utf8mb4", + "tenant_scoped": true, + "append_only": false + } + ], + "fields": [ + { + "name": "tenant_id", + "comment": "租户ID(强制打头)", + "null": false, + "type": "str", + "length": 64 + }, + { + "name": "id", + "comment": "主键", + "null": false, + "type": "int", + "length": 20, + "unsigned": true, + "auto_increment": true + }, + { + "name": "task_no", + "comment": "任务流水号(确定性派生:CT+blueprint_id+version+seq)", + "null": false, + "type": "str", + "length": 64 + }, + { + "name": "blueprint_id", + "comment": "蓝图ID", + "null": false, + "type": "int", + "length": 20, + "unsigned": true + }, + { + "name": "blueprint_version", + "comment": "蓝图版本(须已审批)", + "null": false, + "type": "int", + "length": 11 + }, + { + "name": "compiler_version", + "comment": "编译器版本号", + "null": false, + "type": "str", + "length": 32 + }, + { + "name": "ruleset_version", + "comment": "规则集版本", + "null": true, + "type": "str", + "length": 32 + }, + { + "name": "status", + "comment": "appcodes:pbl_compile_status(pending/running/success/failed)", + "null": false, + "type": "str", + "length": 32 + }, + { + "name": "approval_id", + "comment": "关联审批记录(未审批拒绝 F-CP-01)", + "null": true, + "type": "int", + "length": 20, + "unsigned": true + }, + { + "name": "game_def_id", + "comment": "产出的 Game Definition ID(success 时回填)", + "null": true, + "type": "int", + "length": 20, + "unsigned": true + }, + { + "name": "content_fingerprint", + "comment": "产物 SHA-256 指纹(success 时回填,US-11 比对锚点)", + "null": true, + "type": "str", + "length": 64 + }, + { + "name": "error_code", + "comment": "失败错误码(PBL_E_*)", + "null": true, + "type": "str", + "length": 64 + }, + { + "name": "error_msg", + "comment": "失败原因", + "null": true, + "type": "str", + "length": 2048 + }, + { + "name": "triggered_by", + "comment": "触发人", + "null": false, + "type": "str", + "length": 64 + }, + { + "name": "duration_ms", + "comment": "编译耗时(易变字段,不参与指纹)", + "null": false, + "type": "int", + "length": 11 + }, + { + "name": "started_at", + "comment": "开始时间", + "null": true, + "type": "datetime" + }, + { + "name": "finished_at", + "comment": "结束时间", + "null": true, + "type": "datetime" + }, + { + "name": "created_at", + "comment": "创建时间(应用层写入)", + "null": false, + "type": "datetime" + }, + { + "name": "updated_at", + "comment": "更新时间(应用层写入)", + "null": false, + "type": "datetime" + } + ], + "indexes": [ + { + "name": "PRIMARY", + "unique": true, + "fields": [ + "id" + ] + }, + { + "name": "uk_ct_task_no", + "unique": true, + "fields": [ + "tenant_id", + "task_no" + ] + }, + { + "name": "idx_ct_blueprint", + "unique": false, + "fields": [ + "tenant_id", + "blueprint_id", + "blueprint_version" + ] + }, + { + "name": "idx_ct_status", + "unique": false, + "fields": [ + "tenant_id", + "status" + ] + }, + { + "name": "idx_ct_fingerprint", + "unique": false, + "fields": [ + "tenant_id", + "content_fingerprint" + ] + } + ], + "codes": [ + { + "name": "pbl_compile_status", + "comment": "编译任务状态(第29.6章)", + "items": [ + { + "code": "pending", + "name": "待执行" + }, + { + "code": "running", + "name": "编译中" + }, + { + "code": "success", + "name": "成功" + }, + { + "code": "failed", + "name": "失败" + } + ] + } + ] +} diff --git a/pbl_compiler/api.py b/pbl_compiler/api.py index cc6cc63..33942ac 100644 --- a/pbl_compiler/api.py +++ b/pbl_compiler/api.py @@ -1,283 +1,886 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""pbl_compiler.api —— PBL Compiler v1：Blueprint → Game Definition（第9章，M3a/M3b）。 +"""pbl_compiler.api —— PBL Compiler v1 编译核心（M3a：确定性编译 + Game Definition）。 -29.6 确定性三要素（可复现证明 = content_fingerprint 相同）： -1) 输入锁定：blueprint_id + blueprint_version_no（版本快照，不读可变主表）； -2) 规则锁定：compiler_version_id（规则集哈希 rules_hash）； -3) 无副作用：不依赖随机数/时间戳/网络；集合一律按稳定键排序后遍历。 -落库 world/scene/entity 走 pbl_domain_ext 契约（薄扩展，不改基表结构）。 +设计依据：projects/pbls/docs/01-design/modules/pbl_compiler.md §4/§5.1/§5.3、 +data-model.md §E（4 表）、第 9 章（GD 10 顶层键）、第 29.6 章（确定性可复现）。 + +M3a 范围（本文件）： + 1. ``compile``：已审批蓝图版本 + compilerVersion → GD（10 顶层键）→ 指纹落库； + 2. 29.6 确定性：sort_keys / 无时间戳参与指纹 / registry_hash（canonical.py + gd_builder.py）； + 3. ``content_fingerprint`` 唯一约束落库（uk_gd_fingerprint，同指纹幂等命中不重复插）； + 4. ``verify_determinism``：连续 N 次编译指纹必须全等（US-11 / F-CP-03）； + 5. 编译器版本管理：register / get / list / diff（F-CP-05）； + 6. 编译任务：pbl_compile_task（approval_id 门禁，未审批 → PBL_E_STATE_ILLEGAL/403）。 + +M3b 范围（不在本文件）：apply_game_definition / rollback_mapping / list_mappings +（GD → world/scene/entity 落库映射 + 单事务回滚，表 pbl_compile_mapping）。 + +铁律： + - 编译全程 **零 LLM 调用**（6.2）；违例即 PBL_E_STATE_ILLEGAL； + - 所有读写 **tenant_id 强制打头**，缺失 fail-closed； + - GD 表 append-only：只插不改（除 compiler_version_id 回填与 M3b 落库回填）。 """ +from __future__ import annotations + import hashlib +import inspect import json +import time -from pbl_common.api import (PblError, actor_id, crud, json_dump, now_str, sql_exec, - sql_rows, sql_scalar, tenant_id) +from pbl_common.api import ( + PblError, + actor_id, + json_dump, + now_str, + sql_exec, + sql_rows, + tenant_crud, + tenant_id, + write_audit, +) +from pbl_compiler.canonical import ( + CANONICAL_ENCODING, + FLOAT_PRECISION, + GD_SCHEMA, + VOLATILE_KEYS, + canonical_json, + registry_hash, + sha256_fingerprint, + short_hash, +) +from pbl_compiler.gd_builder import GD_TOP_KEYS, build_game_definition -COMPILER_VERSION = 'pbl-compiler-v1.0.0' +__all__ = [ + 'COMPILER_VERSION', 'RULESET_VERSION', 'RULES_HASH_SEED', 'GD_SCHEMA', + 'GD_TOP_KEYS', 'QUALITY_LADDER', 'COMPILE_GATE_MIN_STATE', + 'compile', 'compile_blueprint', 'get_compile_task', 'list_compile_tasks', + 'get_game_definition', 'get_game_definition_by_blueprint', 'verify_determinism', + 'register_compiler_version', 'get_compiler_version', 'list_compiler_versions', + 'diff_compiler_versions', 'is_approved', 'ladder_at_or_above', + 'pbl_compiler_compile', 'pbl_compiler_preview', 'pbl_compiler_compare', + 'pbl_compiler_task_get', 'pbl_compiler_task_list', + 'pbl_game_definition_get', 'pbl_game_definition_get_by_blueprint', + 'pbl_compiler_verify_determinism', 'pbl_compiler_version_register', + 'pbl_compiler_version_get', 'pbl_compiler_version_list', + 'pbl_compiler_version_diff', 'pbl_compiler_version_save', + 'pbl_capability_list', 'pbl_capability_register', +] + +#: 缺省编译器版本 / 规则集版本（无 DB 记录时的兜底，保证编译可追溯） +COMPILER_VERSION = '1.0.0' +RULESET_VERSION = 'pbl.rules.v1' RULES_HASH_SEED = 'pbl.compiler.rules.v1' -GAME_DEF_SCHEMA = 'pbl.game_definition.v1' + +#: 第 15 章质量阶梯（5 级），编译门禁最低档 +QUALITY_LADDER = ('draft', 'needs_work', 'pbl_ready', 'playtest_ready', 'publish_ready') +COMPILE_GATE_MIN_STATE = 'pbl_ready' + +#: 编译任务状态（appcodes:pbl_compile_status） +TASK_PENDING, TASK_RUNNING, TASK_SUCCESS, TASK_FAILED = ( + 'pending', 'running', 'success', 'failed') + +DB = 'pbl' +_T_GD = 'pbl_game_definition' +_T_TASK = 'pbl_compile_task' +_T_CV = 'pbl_compiler_version' +_T_CAP = 'pbl_capability_registry' -def _loads(v, d=None): - if isinstance(v, (dict, list)): - return v +# ========================================================================== +# 基础工具 +# ========================================================================== +async def _await_maybe(value): + """兼容 pbl_common 同步/异步双形态返回（sql_rows 可能返回 coroutine）。""" + if inspect.isawaitable(value): + return await value + return value + + +def _loads(value, default=None): + if isinstance(value, (dict, list)): + return value + if value in (None, '', b''): + return default if default is not None else {} try: - return json.loads(v) if v else (d or {}) - except ValueError: - return d or {} + return json.loads(value) + except (ValueError, TypeError): + return default if default is not None else {} -def _canon(obj): - return json.dumps(obj, sort_keys=True, ensure_ascii=False, separators=(',', ':')) +def _err(code, message): + """统一抛错（PblError 构造签名兼容 (code,msg) / (msg,code=)）。""" + try: + return PblError(code, message) + except TypeError: + exc = PblError(message) + try: + exc.code = code + except Exception: + pass + return exc -def _sha(obj): - return hashlib.sha256(_canon(obj).encode('utf-8')).hexdigest() +def _tid(): + """当前租户（fail-closed：缺失即拒）。""" + tid = tenant_id() + if not tid: + raise _err('PBL_E_TENANT_MISSING', '缺少租户上下文，拒绝编译（fail-closed）') + return str(tid) -async def _tid(): - return await tenant_id() +def _actor(): + try: + return str(actor_id() or 'system') + except Exception: + return 'system' -async def _compiler_version(tid, requested=None): - if requested: - rows = await sql_rows('SELECT * FROM `pbl_compiler_version` WHERE `tenant_id` = ${t}$' - ' AND (`id` = ${r}$ OR `semver` = ${r2}$) LIMIT 1', - {'t': tid, 'r': requested, 'r2': requested}, 'pbl') - if rows: - return rows[0] - rows = await sql_rows('SELECT * FROM `pbl_compiler_version` WHERE `tenant_id` = ${t}$' - ' AND `enabled` = 1 ORDER BY `id` ASC LIMIT 1', {'t': tid}, 'pbl') - if rows: - return rows[0] - cv = {'id': None, 'semver': COMPILER_VERSION, 'code': 'PBL_COMPILER_V1', - 'rules_hash': hashlib.sha256(RULES_HASH_SEED.encode()).hexdigest()[:16], - 'entrypoint': 'pbl_compiler.compile'} - return cv +def _crud(table): + return tenant_crud(table, module='pbl_compiler', db=DB) -async def _snapshot(tid, bp_id, version_no): - rows = await sql_rows('SELECT * FROM `pbl_blueprint_version` WHERE `tenant_id` = ${t}$' - ' AND `blueprint_id` = ${b}$ AND `version_no` = ${v}$ LIMIT 1', - {'t': tid, 'b': bp_id, 'v': version_no}, 'pbl') - if rows: - return _loads(rows[0]['snapshot_json']) - bp = await sql_rows('SELECT * FROM `pbl_blueprint` WHERE `tenant_id` = ${t}$' - ' AND `id` = ${id}$ LIMIT 1', {'t': tid, 'id': bp_id}, 'pbl') - if not bp: - raise PblError('PBL_NOT_FOUND', '蓝图不存在') - return _loads(bp[0]['content_json']) +async def _rows(sql, params): + return list(await _await_maybe(sql_rows(sql, params, DB)) or []) -async def compile_model(model, cv, bp_id, version_no): - """纯函数式编译（无 DB、无时间、无随机）——确定性核心，可单测复算。""" - goals = sorted(_loads(model.get('learning_goals'), []), key=lambda x: _canon(x)) - roles = sorted(_loads(model.get('roles'), []), key=lambda x: _canon(x)) - missions = sorted(_loads(model.get('missions'), []), key=lambda x: _canon(x)) - tasks = sorted(_loads(model.get('tasks'), []), key=lambda x: _canon(x)) - artifacts = sorted(_loads(model.get('artifacts'), []), key=lambda x: _canon(x)) - ev_specs = sorted(_loads(model.get('evidence_specs'), []), key=lambda x: _canon(x)) - rules_hash = cv.get('rules_hash') or hashlib.sha256(RULES_HASH_SEED.encode()).hexdigest()[:16] +async def _one(sql, params): + rows = await _rows(sql, params) + return rows[0] if rows else None - entities, events, unlocks = [], [], [] - for m in missions: - mid = m.get('obj_uid') or m.get('code') or _sha(m)[:12] - entities.append({'entity_id': 'mission:%s' % mid, 'kind': 'mission', - 'capability_key': 'pbl.mission.show', - 'initial_state': {'completed': False, 'order': m.get('order_no', 0)}, - 'server_authoritative': True}) - for t in [x for x in tasks if (x.get('mission_id') == m.get('id') - or x.get('mission_uid') == mid)]: - tid_ = t.get('obj_uid') or t.get('code') or _sha(t)[:12] - entities.append({'entity_id': 'task:%s' % tid_, 'kind': 'task', - 'capability_key': 'pbl.task.track', - 'initial_state': {'done': False, 'parent': mid}, - 'server_authoritative': True}) - events.append({'event_type': 'task.completed', 'source': 'task:%s' % tid_, - 'condition': {'all_of': ['task:%s.done == true' % tid_]}, - 'response': [{'op': 'emit', 'event': 'task.completed'}]}) - unlocks.append({'from_event': 'mission.completed:%s' % mid, 'target': mid, - 'effect': 'pbl.unlockMission'}) - for r in roles: - rid = r.get('obj_uid') or r.get('code') or _sha(r)[:12] - entities.append({'entity_id': 'role:%s' % rid, 'kind': 'role', - 'capability_key': 'pbl.role.assign', - 'initial_state': {'assigned': False}, 'server_authoritative': True}) - for a in artifacts: - aid = a.get('obj_uid') or a.get('code') or _sha(a)[:12] - entities.append({'entity_id': 'artifact:%s' % aid, 'kind': 'artifact_spec', - 'capability_key': 'pbl.artifact.submit', - 'initial_state': {'submitted': False}, 'server_authoritative': True}) - for e in ev_specs: - eid = e.get('obj_uid') or _sha(e)[:12] - events.append({'event_type': 'evidence.required', 'source': 'evidence_spec:%s' % eid, - 'condition': {'min_count': e.get('min_count', 1)}, - 'response': [{'op': 'collect', 'evidence_type': - e.get('evidence_type') or 'artifact'}]}) - definition = { - 'schema': GAME_DEF_SCHEMA, - 'meta': {'source_blueprint_id': bp_id, 'source_blueprint_version': version_no, - 'compiler_version': cv.get('semver') or COMPILER_VERSION, - 'compiler_rules_hash': rules_hash, 'deterministic': True}, - 'world': {'mode': 'shared', 'authority': 'server', 'max_members': 8, - 'heartbeat_seconds': 15, 'poll_fallback_seconds': 3}, - 'learning_goals': goals, - 'entities': sorted(entities, key=lambda x: x['entity_id']), - 'events': sorted(events, key=lambda x: _canon(x)), - 'unlock_graph': sorted(unlocks, key=lambda x: _canon(x)), - 'script_bindings': [{'event': ev['event_type'], 'script_type': 1, - 'rule_json': _canon(ev)} - for ev in sorted(events, key=lambda x: _canon(x))], - 'capabilities': sorted({e['capability_key'] for e in entities}), +async def _exec(sql, params): + return await _await_maybe(sql_exec(sql, params, DB)) + + +async def _audit(action, table, row_id, detail=None): + """审计写入（best-effort：审计失败不阻断编译主流程，但必须尝试）。""" + try: + return await _await_maybe(write_audit( + action=action, table=table, row_id=row_id, + tenant_id=_tid(), operator=_actor(), + detail=json_dump(detail or {}))) + except Exception: + return None + + +def ladder_at_or_above(state): + """质量阶梯：返回 >= state 的全部状态（第 15 章 5 级）。""" + if state not in QUALITY_LADDER: + return list(QUALITY_LADDER) + return list(QUALITY_LADDER[QUALITY_LADDER.index(state):]) + + +LADDER_AT_OR_ABOVE = ladder_at_or_above # 兼容旧脚手架命名 + + +def _rules_hash(ruleset_version, compiler_version): + """规则集哈希（确定性派生，不用随机）。""" + seed = '%s|%s|%s' % (RULES_HASH_SEED, compiler_version or COMPILER_VERSION, + ruleset_version or RULESET_VERSION) + return hashlib.sha256(seed.encode(CANONICAL_ENCODING)).hexdigest()[:16] + + +# ========================================================================== +# 编译器版本（F-CP-05） +# ========================================================================== +async def register_compiler_version(compiler_version=None, ruleset_version=None, + fingerprint_baseline=None, changelog=None, + status='active', **kw): + """登记编译器版本（平台级 tenant '__platform__' 亦可，此处按当前租户落库）。 + + 重复登记同 (tenant, compiler_version) → PBL_E_DUPLICATE。 + """ + tid = _tid() + cv = str(compiler_version or kw.get('semver') or COMPILER_VERSION).strip() + rv = str(ruleset_version or RULESET_VERSION).strip() + if not cv: + raise _err('PBL_E_PARAM', 'compiler_version 必填') + dup = await _one( + 'SELECT `id` FROM `%s` WHERE `tenant_id` = ${t}$ AND `compiler_version` = ${c}$ LIMIT 1' + % _T_CV, {'t': tid, 'c': cv}) + if dup: + raise _err('PBL_E_DUPLICATE', '编译器版本已存在: %s' % cv) + row = { + 'compiler_version': cv, + 'ruleset_version': rv, + 'rules_hash': _rules_hash(rv, cv), + 'fingerprint_baseline': fingerprint_baseline or '', + 'changelog': changelog or '', + 'status': status or 'active', + 'released_at': now_str(), + 'created_at': now_str(), + 'updated_at': now_str(), } - fp = _sha(definition) - definition['meta']['content_fingerprint'] = fp - return definition, fp + res = await _await_maybe(_crud(_T_CV).C(row, tenant_id=tid)) + new_id = (res or {}).get('id') if isinstance(res, dict) else res + await _audit('create', _T_CV, new_id, {'compiler_version': cv, 'ruleset_version': rv}) + return {'ok': True, 'id': new_id, 'compiler_version': cv, 'ruleset_version': rv, + 'rules_hash': row['rules_hash'], 'status': row['status']} -async def pbl_compiler_compile(**kw): - import time - t0 = time.time() - tid = await _tid() - bp_id = kw.get('blueprint_id') - bp = await sql_rows('SELECT * FROM `pbl_blueprint` WHERE `tenant_id` = ${t}$' - ' AND `id` = ${id}$ LIMIT 1', {'t': tid, 'id': bp_id}, 'pbl') +async def get_compiler_version(compiler_version=None, **kw): + """取编译器版本；缺省取当前租户 enabled/active 的最新一条，无则返回内置兜底。""" + tid = _tid() + if compiler_version: + row = await _one( + 'SELECT * FROM `%s` WHERE `tenant_id` = ${t}$ AND (`compiler_version` = ${c}$' + ' OR `id` = ${i}$) LIMIT 1' % _T_CV, + {'t': tid, 'c': str(compiler_version), 'i': str(compiler_version)}) + if row: + return row + raise _err('PBL_E_NOT_FOUND', '编译器版本不存在: %s' % compiler_version) + row = await _one( + 'SELECT * FROM `%s` WHERE `tenant_id` = ${t}$ AND (`status` = ${s}$ OR `enabled` = 1)' + ' ORDER BY `id` DESC LIMIT 1' % _T_CV, {'t': tid, 's': 'active'}) + if row: + return row + return {'id': None, 'compiler_version': COMPILER_VERSION, 'semver': COMPILER_VERSION, + 'ruleset_version': RULESET_VERSION, 'rules_hash': _rules_hash(None, None), + 'fingerprint_baseline': '', 'status': 'builtin', + 'entrypoint': 'pbl_compiler.compile', 'changelog': '内置兜底版本（未登记）'} + + +async def list_compiler_versions(status=None, **kw): + tid = _tid() + sql = 'SELECT * FROM `%s` WHERE `tenant_id` = ${t}$' % _T_CV + params = {'t': tid} + if status: + sql += ' AND `status` = ${s}$' + params['s'] = str(status) + sql += ' ORDER BY `id` DESC' + rows = await _rows(sql, params) + return {'ok': True, 'items': rows, 'total': len(rows)} + + +async def diff_compiler_versions(v1=None, v2=None, **kw): + """两编译器版本差异：规则集版本/规则哈希/指纹基线（US-19 版本历史可查）。""" + a = await get_compiler_version(v1) + b = await get_compiler_version(v2) + fields = ('compiler_version', 'ruleset_version', 'rules_hash', + 'fingerprint_baseline', 'status') + changed = [] + for f in fields: + av, bv = a.get(f), b.get(f) + if (av or '') != (bv or ''): + changed.append({'field': f, 'v1': av, 'v2': bv}) + return {'ok': True, + 'v1': {f: a.get(f) for f in fields}, + 'v2': {f: b.get(f) for f in fields}, + 'ruleset_changed': (a.get('ruleset_version') or '') != (b.get('ruleset_version') or ''), + 'fingerprint_baseline_changed': + (a.get('fingerprint_baseline') or '') != (b.get('fingerprint_baseline') or ''), + 'changed': changed, + 'identical': not changed} + + +# ========================================================================== +# 审批门禁（US-10 / F-CP-01：未审批拒绝编译） +# ========================================================================== +async def is_approved(blueprint_id, version_no=None, **kw): + """蓝图版本是否已审批。 + + 判定顺序（fail-closed，任一命中即视为已审批）： + 1. pbl_blueprint_version.is_approved / approved_at / approval_id； + 2. pbl_approval（status=approved 且 target 指向该蓝图版本）； + 3. pbl_blueprint.quality_state >= COMPILE_GATE_MIN_STATE（第 15 章阶梯兜底）。 + 返回 (approved: bool, approval_id, evidence: str, blueprint_row|None, version_row|None) + """ + tid = _tid() + bp = await _one('SELECT * FROM `pbl_blueprint` WHERE `tenant_id` = ${t}$' + ' AND `id` = ${id}$ LIMIT 1', {'t': tid, 'id': blueprint_id}) if not bp: - raise PblError('PBL_NOT_FOUND', '蓝图不存在') - bp = bp[0] - min_state = kw.get('min_quality_state') or 'pbl_ready' - if bp['quality_state'] not in LADDER_AT_OR_ABOVE(min_state): - raise PblError('PBL_COMPILE_GATE_FAILED', - 'quality_state=%s 未达 %s（第15章门禁）' % (bp['quality_state'], min_state)) - version_no = int(kw.get('blueprint_version_no') or bp['version_no']) - cv = await _compiler_version(tid, kw.get('compiler_version_id')) - model = await _snapshot(tid, bp_id, version_no) - definition, fp = await compile_model(model, cv, bp_id, version_no) - - dup = await sql_rows('SELECT * FROM `pbl_game_definition` WHERE `tenant_id` = ${t}$' - ' AND `content_fingerprint` = ${f}$ LIMIT 1', - {'t': tid, 'f': fp}, 'pbl') - if dup and not kw.get('force_new'): - return {'ok': True, 'idempotent': True, 'id': dup[0]['id'], - 'content_fingerprint': fp, 'compiler_version': cv.get('semver'), - 'note': '同输入+同编译版本 → 同指纹（29.6 可复现）', - 'latency_ms': int((time.time() - t0) * 1000)} - res = await crud('pbl_game_definition', 'pbl', - ['blueprint_id', 'blueprint_version_no', 'compiler_version_id', - 'content_fingerprint', 'definition_json', 'world_id', 'scene_id', - 'entity_count', 'event_count', 'quality_state', 'duration_ms'] - )['create']( - blueprint_id=bp_id, blueprint_version_no=version_no, - compiler_version_id=cv.get('id'), content_fingerprint=fp, - definition_json=json_dump(definition), entity_count=len(definition['entities']), - event_count=len(definition['events']), quality_state=bp['quality_state'], - duration_ms=int((time.time() - t0) * 1000)) - if cv.get('id') is None: - cvr = await crud('pbl_compiler_version', 'pbl', - ['code', 'semver', 'rules_hash', 'entrypoint', 'enabled', 'notes'] - )['create'](code=cv['code'], semver=cv['semver'], - rules_hash=cv['rules_hash'], entrypoint=cv['entrypoint'], - enabled=1, notes='自动登记（编译时确保版本可追溯）') - await sql_exec('UPDATE `pbl_game_definition` SET `compiler_version_id` = ${c}$' - ' WHERE `id` = ${id}$ AND `tenant_id` = ${t}$', - {'c': cvr['id'], 'id': res['id'], 't': tid}, 'pbl') - materialized = None - if kw.get('materialize'): - from pbl_domain_ext.api import pbl_domain_materialize_game_definition - materialized = await pbl_domain_materialize_game_definition(id=res['id']) - await sql_exec('UPDATE `pbl_game_definition` SET `world_id` = ${w}$, `scene_id` = ${s}$' - ' WHERE `id` = ${id}$ AND `tenant_id` = ${t}$', - {'w': materialized.get('world_id'), 's': materialized.get('scene_id'), - 'id': res['id'], 't': tid}, 'pbl') - return {'ok': True, 'id': res['id'], 'content_fingerprint': fp, - 'compiler_version': cv.get('semver') or COMPILER_VERSION, - 'entities': len(definition['entities']), 'events': len(definition['events']), - 'materialized': materialized, 'budget_ms': 5000, - 'latency_ms': int((time.time() - t0) * 1000)} + raise _err('PBL_E_NOT_FOUND', '蓝图不存在: %s' % blueprint_id) + ver_no = int(version_no or bp.get('version_no') or 1) + vrow = await _one('SELECT * FROM `pbl_blueprint_version` WHERE `tenant_id` = ${t}$' + ' AND `blueprint_id` = ${b}$ AND `version_no` = ${v}$ LIMIT 1', + {'t': tid, 'b': blueprint_id, 'v': ver_no}) + if vrow: + for flag_key in ('is_approved', 'approved'): + if str(vrow.get(flag_key, '')).lower() in ('1', 'true', 'yes'): + return True, vrow.get('approval_id') or vrow.get('id'), \ + 'version.%s' % flag_key, bp, vrow + if vrow.get('approved_at'): + return True, vrow.get('approval_id'), 'version.approved_at', bp, vrow + if vrow.get('approval_id'): + return True, vrow.get('approval_id'), 'version.approval_id', bp, vrow + if str(vrow.get('status', '')).lower() in ('approved', 'published'): + return True, vrow.get('approval_id') or vrow.get('id'), \ + 'version.status', bp, vrow + ap = await _one( + 'SELECT * FROM `pbl_approval` WHERE `tenant_id` = ${t}$' + ' AND `target_id` = ${b}$ AND (`target_version` = ${v}$ OR `target_version` IS NULL)' + ' AND `status` = ${s}$ ORDER BY `id` DESC LIMIT 1', + {'t': tid, 'b': blueprint_id, 'v': ver_no, 's': 'approved'}) + if ap: + return True, ap.get('id'), 'pbl_approval.status=approved', bp, vrow + qs = str(bp.get('quality_state') or 'draft') + if qs in ladder_at_or_above(COMPILE_GATE_MIN_STATE) and qs in QUALITY_LADDER: + return True, bp.get('approval_id'), 'quality_state=%s>=%s' % ( + qs, COMPILE_GATE_MIN_STATE), bp, vrow + return False, None, 'no-approval-evidence', bp, vrow -def LADDER_AT_OR_ABOVE(state): - order = ['draft', 'needs_work', 'pbl_ready', 'playtest_ready', 'publish_ready'] - i = order.index(state) if state in order else 0 - return order[i:] +async def _snapshot(tid, blueprint_id, version_no, vrow=None, bp=None): + """取蓝图版本快照（输入锁定：只读版本快照，不读可变主表内容）。""" + if vrow is None: + vrow = await _one('SELECT * FROM `pbl_blueprint_version` WHERE `tenant_id` = ${t}$' + ' AND `blueprint_id` = ${b}$ AND `version_no` = ${v}$ LIMIT 1', + {'t': tid, 'b': blueprint_id, 'v': version_no}) + if vrow: + for key in ('snapshot_json', 'content_json', 'snapshot'): + if vrow.get(key): + return _loads(vrow[key], {}), vrow + if bp is None: + bp = await _one('SELECT * FROM `pbl_blueprint` WHERE `tenant_id` = ${t}$' + ' AND `id` = ${id}$ LIMIT 1', {'t': tid, 'id': blueprint_id}) + if bp and bp.get('content_json'): + return _loads(bp['content_json'], {}), vrow + raise _err('PBL_E_NOT_FOUND', + '蓝图版本快照不存在: blueprint_id=%s version_no=%s' % (blueprint_id, version_no)) + + +async def _capability_rows(tid): + """能力注册表快照（§6.1）——进 GD capabilities + registry_hash。""" + try: + return await _rows('SELECT * FROM `%s` WHERE `tenant_id` = ${t}$' + ' ORDER BY `capability_key` ASC, `version_no` ASC' % _T_CAP, + {'t': tid}) + except Exception: + return [] + + +# ========================================================================== +# 编译核心（M3a） +# ========================================================================== +def _task_no(tid, blueprint_id, version_no, cv, seq): + """任务流水号：确定性派生（29.6 无随机）——CT:{bp}:{ver}:{rules_hash}:{seq}。""" + return 'CT:%s:%s:%s:%s:%05d' % (blueprint_id, version_no, + str(cv.get('compiler_version') + or cv.get('semver') or COMPILER_VERSION), + str(cv.get('rules_hash') or '')[:8], int(seq)) + + +async def _next_task_seq(tid, blueprint_id, version_no): + n = await _one('SELECT COUNT(*) AS `c` FROM `%s` WHERE `tenant_id` = ${t}$' + ' AND `blueprint_id` = ${b}$ AND `blueprint_version` = ${v}$' % _T_TASK, + {'t': tid, 'b': blueprint_id, 'v': version_no}) + try: + return int((n or {}).get('c') or 0) + 1 + except (TypeError, ValueError): + return 1 + + +async def _insert_task(tid, row): + res = await _await_maybe(_crud(_T_TASK).C(row, tenant_id=tid)) + if isinstance(res, dict) and res.get('id'): + return res['id'] + if isinstance(res, (int, str)) and str(res).isdigit(): + return int(res) + got = await _one('SELECT `id` FROM `%s` WHERE `tenant_id` = ${t}$' + ' AND `task_no` = ${n}$ ORDER BY `id` DESC LIMIT 1' % _T_TASK, + {'t': tid, 'n': row['task_no']}) + return (got or {}).get('id') + + +async def _update_task(tid, task_id, fields): + if not task_id: + return None + sets, params = [], {'t': tid, 'id': task_id} + for i, (k, v) in enumerate(sorted(fields.items())): + ph = 'p%d' % i + sets.append('`%s` = ${%s}$' % (k, ph)) + params[ph] = v + sets.append('`updated_at` = ${now}$') + params['now'] = now_str() + try: + return await _exec('UPDATE `%s` SET %s WHERE `tenant_id` = ${t}$' + ' AND `id` = ${id}$' % (_T_TASK, ', '.join(sets)), params) + except Exception: + return None + + +async def compile_blueprint(blueprint_id=None, version_no=None, compiler_version=None, + triggered_by=None, **kw): + """编译主流程（design §5.1 compile）。 + + ① is_approved 门禁（未审批 → PBL_E_STATE_ILLEGAL / HTTP 403） + ② 建 pbl_compile_task(status=running) + ③ get_version 取快照（输入锁定） + ④ 确定性转换生成 GD 10 顶层键（零 LLM，gd_builder 纯函数） + ⑤ canonical_json + sha256 算 content_fingerprint + ⑥ 写 pbl_game_definition（append-only；uk_gd_fingerprint 唯一 → 同指纹幂等命中） + ⑦ task status=success + game_def_id/fingerprint 回填 + ⑧ write_audit + + :return: {task_no, status, game_def_id, fingerprint, idempotent, ...} + """ + t0 = time.time() + tid = _tid() + if blueprint_id in (None, ''): + raise _err('PBL_E_PARAM', 'blueprint_id 必填') + blueprint_id = int(blueprint_id) + operator = str(triggered_by or kw.get('operator') or _actor()) + + # ① 审批门禁（US-10 / F-CP-01） + approved, approval_id, evidence, bp, vrow = await is_approved( + blueprint_id, version_no) + if not approved: + raise _err('PBL_E_STATE_ILLEGAL', + '蓝图版本未审批，拒绝编译（US-10/F-CP-01）: blueprint_id=%s version_no=%s' + % (blueprint_id, version_no or (bp or {}).get('version_no'))) + ver_no = int(version_no or (vrow or {}).get('version_no') + or (bp or {}).get('version_no') or 1) + min_state = str(kw.get('min_quality_state') or COMPILE_GATE_MIN_STATE) + qs = str((bp or {}).get('quality_state') or 'draft') + if qs in QUALITY_LADDER and qs not in ladder_at_or_above(min_state): + raise _err('PBL_E_STATE_ILLEGAL', + 'quality_state=%s 未达 %s（第 15 章门禁）' % (qs, min_state)) + + # 编译器版本（规则锁定） + cv = await get_compiler_version(compiler_version or kw.get('compiler_version_id')) + cv_semver = str(cv.get('compiler_version') or cv.get('semver') or COMPILER_VERSION) + ruleset = str(cv.get('ruleset_version') or RULESET_VERSION) + rules_hash = str(cv.get('rules_hash') or _rules_hash(ruleset, cv_semver)) + + # ② 建编译任务 + seq = await _next_task_seq(tid, blueprint_id, ver_no) + task_no = _task_no(tid, blueprint_id, ver_no, cv, seq) + started = now_str() + task_id = await _insert_task(tid, { + 'task_no': task_no, 'blueprint_id': blueprint_id, 'blueprint_version': ver_no, + 'compiler_version': cv_semver, 'ruleset_version': ruleset, + 'status': TASK_RUNNING, 'approval_id': approval_id, + 'triggered_by': operator, 'duration_ms': 0, + 'started_at': started, 'created_at': started, 'updated_at': started, + }) + + try: + # ③ 快照（输入锁定） + snapshot, vrow2 = await _snapshot(tid, blueprint_id, ver_no, vrow, bp) + # ④⑤ 确定性转换 + 指纹（零 LLM） + caps = await _capability_rows(tid) + ctx = { + 'blueprint_id': blueprint_id, 'blueprint_version_no': ver_no, + 'compiler_version': cv_semver, 'ruleset_version': ruleset, + 'rules_hash': rules_hash, 'created_at': started, + 'duration_ms': 0, 'task_no': task_no, + } + gd, fingerprint = build_game_definition(snapshot, ctx, caps) + missing = [k for k in GD_TOP_KEYS if k not in gd] + if missing: + raise _err('PBL_E_COMPILE', 'GD 缺顶层键: %s' % ','.join(missing)) + + # ⑥ 落库（content_fingerprint 唯一约束 → 幂等命中） + dup = await _one('SELECT * FROM `%s` WHERE `tenant_id` = ${t}$' + ' AND `content_fingerprint` = ${f}$ LIMIT 1' % _T_GD, + {'t': tid, 'f': fingerprint}) + duration_ms = int((time.time() - t0) * 1000) + if dup and not kw.get('force_new'): + gd_id = dup.get('id') + idempotent = True + else: + row = { + 'blueprint_id': blueprint_id, 'blueprint_version_no': ver_no, + 'compiler_version_id': cv.get('id') or 0, + 'compiler_version': cv_semver, 'ruleset_version': ruleset, + 'task_no': task_no, 'content_fingerprint': fingerprint, + 'definition_json': json_dump(gd), + 'gd_schema': gd.get('manifest', {}).get('schema', GD_SCHEMA), + 'registry_hash': gd.get('capabilities', {}).get('registry_hash', ''), + 'world_id': 0, 'scene_id': 0, + 'entity_count': len(gd.get('entities', [])), + 'event_count': len(gd.get('events', [])), + 'scene_count': len(gd.get('scenes', [])), + 'quality_state': qs, 'duration_ms': duration_ms, + 'created_at': started, 'updated_at': started, + } + res = await _await_maybe(_crud(_T_GD).C(row, tenant_id=tid)) + gd_id = (res or {}).get('id') if isinstance(res, dict) else res + if not gd_id: + got = await _one('SELECT `id` FROM `%s` WHERE `tenant_id` = ${t}$' + ' AND `content_fingerprint` = ${f}$' + ' ORDER BY `id` DESC LIMIT 1' % _T_GD, + {'t': tid, 'f': fingerprint}) + gd_id = (got or {}).get('id') + idempotent = False + + # ⑦ 任务收口 + await _update_task(tid, task_id, { + 'status': TASK_SUCCESS, 'game_def_id': gd_id or 0, + 'content_fingerprint': fingerprint, 'duration_ms': duration_ms, + 'finished_at': now_str(), 'error_code': '', 'error_msg': ''}) + # ⑧ 审计 + await _audit('compile', _T_GD, gd_id, { + 'task_no': task_no, 'blueprint_id': blueprint_id, + 'blueprint_version': ver_no, 'compiler_version': cv_semver, + 'content_fingerprint': fingerprint, 'idempotent': idempotent, + 'approval_evidence': evidence}) + return { + 'ok': True, 'task_id': task_id, 'task_no': task_no, + 'status': TASK_SUCCESS, 'game_def_id': gd_id, 'id': gd_id, + 'content_fingerprint': fingerprint, 'fingerprint': fingerprint, + 'compiler_version': cv_semver, 'ruleset_version': ruleset, + 'rules_hash': rules_hash, + 'registry_hash': gd.get('capabilities', {}).get('registry_hash', ''), + 'blueprint_id': blueprint_id, 'blueprint_version_no': ver_no, + 'approval_id': approval_id, 'approval_evidence': evidence, + 'idempotent': idempotent, 'deterministic': True, 'llm_used': False, + 'counts': gd.get('manifest', {}).get('counts', {}), + 'gd_top_keys': list(GD_TOP_KEYS), + 'duration_ms': duration_ms, 'latency_ms': duration_ms, + } + except Exception as exc: + code = getattr(exc, 'code', None) or 'PBL_E_COMPILE' + await _update_task(tid, task_id, { + 'status': TASK_FAILED, 'error_code': str(code), + 'error_msg': str(exc)[:2000], 'finished_at': now_str(), + 'duration_ms': int((time.time() - t0) * 1000)}) + await _audit('compile_failed', _T_TASK, task_id, + {'task_no': task_no, 'error_code': str(code), 'error': str(exc)[:500]}) + raise + + +#: 设计文档契约名（§5.1 ``compile``） +compile = compile_blueprint + + +async def get_compile_task(task_id=None, **kw): + """取编译任务（design §5.1）。""" + tid = _tid() + if task_id in (None, ''): + raise _err('PBL_E_PARAM', 'task_id 必填') + row = await _one('SELECT * FROM `%s` WHERE `tenant_id` = ${t}$' + ' AND (`id` = ${i}$ OR `task_no` = ${n}$) LIMIT 1' % _T_TASK, + {'t': tid, 'i': str(task_id), 'n': str(task_id)}) + if not row: + raise _err('PBL_E_NOT_FOUND', '编译任务不存在: %s' % task_id) + return {'ok': True, 'item': row, **{k: row.get(k) for k in + ('id', 'task_no', 'status', 'blueprint_id', 'blueprint_version', + 'compiler_version', 'content_fingerprint', 'game_def_id', + 'error_code', 'error_msg', 'triggered_by')}} + + +async def list_compile_tasks(blueprint_id=None, page=1, size=20, **kw): + """编译任务分页列表（design §5.1）。""" + tid = _tid() + page = max(1, int(page or 1)) + size = min(200, max(1, int(size or kw.get('rows') or 20))) + where = '`tenant_id` = %s' % json.dumps(tid) + params = {'t': tid} + if blueprint_id not in (None, ''): + where += ' AND `blueprint_id` = ${b}$' + params['b'] = int(blueprint_id) + total_row = await _one('SELECT COUNT(*) AS `c` FROM `%s` WHERE %s' % (_T_TASK, where), + params) + try: + total = int((total_row or {}).get('c') or 0) + except (TypeError, ValueError): + total = 0 + rows = await _rows('SELECT * FROM `%s` WHERE %s ORDER BY `id` DESC' + ' LIMIT %d OFFSET %d' % (_T_TASK, where, size, (page - 1) * size), + params) + return {'ok': True, 'items': rows, 'total': total, 'page': page, 'size': size} + + +async def get_game_definition(game_def_id=None, **kw): + """取 GD 完整正文（design §5.1；M3b/运行时/前端消费源）。""" + tid = _tid() + if game_def_id in (None, ''): + raise _err('PBL_E_PARAM', 'game_def_id 必填') + row = await _one('SELECT * FROM `%s` WHERE `tenant_id` = ${t}$ AND `id` = ${i}$ LIMIT 1' + % _T_GD, {'t': tid, 'i': str(game_def_id)}) + if not row: + raise _err('PBL_E_NOT_FOUND', 'Game Definition 不存在: %s' % game_def_id) + gd = _loads(row.get('definition_json'), {}) + return {'ok': True, 'id': row.get('id'), 'game_def_id': row.get('id'), + 'content_fingerprint': row.get('content_fingerprint'), + 'blueprint_id': row.get('blueprint_id'), + 'blueprint_version_no': row.get('blueprint_version_no'), + 'compiler_version': row.get('compiler_version'), + 'registry_hash': row.get('registry_hash'), + 'definition_json': gd, 'definition': gd, + 'gd_top_keys': sorted(gd.keys()) if isinstance(gd, dict) else [], + 'created_at': row.get('created_at')} + + +async def get_game_definition_by_blueprint(blueprint_id=None, version_no=None, **kw): + """按蓝图版本取 GD（F-CP-04）——取最新一条（append-only 下即最后编译产物）。""" + tid = _tid() + if blueprint_id in (None, ''): + raise _err('PBL_E_PARAM', 'blueprint_id 必填') + sql = ('SELECT * FROM `%s` WHERE `tenant_id` = ${t}$ AND `blueprint_id` = ${b}$' % _T_GD) + params = {'t': tid, 'b': int(blueprint_id)} + if version_no not in (None, ''): + sql += ' AND `blueprint_version_no` = ${v}$' + params['v'] = int(version_no) + sql += ' ORDER BY `id` DESC LIMIT 1' + row = await _one(sql, params) + if not row: + raise _err('PBL_E_NOT_FOUND', '该蓝图版本尚无编译产物: blueprint_id=%s version_no=%s' + % (blueprint_id, version_no)) + return await get_game_definition(row['id']) + + +async def verify_determinism(blueprint_id=None, version_no=None, compiler_version=None, + times=3, **kw): + """US-11 / F-CP-03：连续 N 次编译，指纹必须全等。 + + 实现要点：**不重复落库**——每次都走完整确定性转换（快照→GD→canonical→sha256）， + 只在内存比对指纹；落库仍由 compile 的 uk_gd_fingerprint 唯一约束保证幂等。 + 指纹不等 → PBL_E_STATE_ILLEGAL（确定性被破坏，属重大事故）。 + """ + tid = _tid() + if blueprint_id in (None, ''): + raise _err('PBL_E_PARAM', 'blueprint_id 必填') + times = max(2, min(10, int(times or 3))) + approved, approval_id, evidence, bp, vrow = await is_approved(blueprint_id, version_no) + if not approved: + raise _err('PBL_E_STATE_ILLEGAL', '蓝图版本未审批，拒绝编译校验（US-10）') + ver_no = int(version_no or (vrow or {}).get('version_no') + or (bp or {}).get('version_no') or 1) + cv = await get_compiler_version(compiler_version) + cv_semver = str(cv.get('compiler_version') or cv.get('semver') or COMPILER_VERSION) + ruleset = str(cv.get('ruleset_version') or RULESET_VERSION) + rules_hash = str(cv.get('rules_hash') or _rules_hash(ruleset, cv_semver)) + snapshot, _ = await _snapshot(tid, int(blueprint_id), ver_no, vrow, bp) + caps = await _capability_rows(tid) + + fingerprints, gd_hashes = [], [] + for i in range(times): + ctx = {'blueprint_id': int(blueprint_id), 'blueprint_version_no': ver_no, + 'compiler_version': cv_semver, 'ruleset_version': ruleset, + 'rules_hash': rules_hash, 'created_at': '2000-01-01T00:00:00Z', + 'duration_ms': i * 7 + 1, 'task_no': 'VERIFY-%d' % i} + gd, fp = build_game_definition(snapshot, ctx, caps) + fingerprints.append(fp) + gd_hashes.append(short_hash(canonical_json(gd, strip=False), 16)) + all_equal = len(set(fingerprints)) == 1 + # 易变字段确实被排除：createdAt/durationMs/taskNo 不同但指纹相同 + volatile_excluded = len(set(gd_hashes)) == 1 or all_equal + if not all_equal: + raise _err('PBL_E_STATE_ILLEGAL', + '确定性被破坏（US-11）：%d 次编译指纹不全等 %s' + % (times, json_dump(sorted(set(fingerprints))))) + return {'ok': True, 'deterministic': all_equal, 'all_equal': all_equal, + 'times': times, 'fingerprints': fingerprints, + 'fingerprint': fingerprints[0], + 'unique_fingerprints': sorted(set(fingerprints)), + 'volatile_fields_excluded': sorted( + k for k in ('createdAt', 'durationMs', 'taskNo') if k in VOLATILE_KEYS), + 'volatile_excluded_ok': bool(volatile_excluded), + 'blueprint_id': int(blueprint_id), 'blueprint_version_no': ver_no, + 'compiler_version': cv_semver, 'ruleset_version': ruleset, + 'rules_hash': rules_hash, + 'registry_hash': registry_hash(caps) if caps else '', + 'approval_evidence': evidence, 'llm_used': False, + 'canonical': {'encoding': CANONICAL_ENCODING, 'sort_keys': True, + 'separators': [',', ':'], 'float_precision': FLOAT_PRECISION}} + + +# ========================================================================== +# 预览 / 对比（教师侧，不落库） +# ========================================================================== +async def preview_compile(blueprint_id=None, version_no=None, compiler_version=None, **kw): + """干跑：返回 GD + 指纹，不写任何表（供教师预览与 diff）。""" + tid = _tid() + approved, _, evidence, bp, vrow = await is_approved(blueprint_id, version_no) + ver_no = int(version_no or (vrow or {}).get('version_no') + or (bp or {}).get('version_no') or 1) + cv = await get_compiler_version(compiler_version) + snapshot, _ = await _snapshot(tid, int(blueprint_id), ver_no, vrow, bp) + caps = await _capability_rows(tid) + ctx = {'blueprint_id': int(blueprint_id), 'blueprint_version_no': ver_no, + 'compiler_version': str(cv.get('compiler_version') or cv.get('semver') + or COMPILER_VERSION), + 'ruleset_version': str(cv.get('ruleset_version') or RULESET_VERSION), + 'rules_hash': str(cv.get('rules_hash') or ''), + 'created_at': now_str(), 'duration_ms': 0, 'task_no': 'PREVIEW'} + gd, fp = build_game_definition(snapshot, ctx, caps) + return {'ok': True, 'preview': True, 'approved': bool(approved), + 'approval_evidence': evidence, 'content_fingerprint': fp, + 'fingerprint': fp, 'definition': gd, 'definition_json': gd, + 'gd_top_keys': list(GD_TOP_KEYS), 'would_write_rows': 1, + 'counts': gd.get('manifest', {}).get('counts', {})} + + +async def compare_definitions(a=None, b=None, **kw): + """两 GD 结构化 diff（确定性验证：同输入两次编译指纹必须一致）。""" + tid = _tid() + rows = [] + for key in (a, b): + if key in (None, ''): + continue + row = await _one('SELECT * FROM `%s` WHERE `tenant_id` = ${t}$ AND `id` = ${i}$' + ' LIMIT 1' % _T_GD, {'t': tid, 'i': str(key)}) + if row: + rows.append(row) + if len(rows) < 2: + raise _err('PBL_E_NOT_FOUND', '需两个已编译的 Game Definition') + rows.sort(key=lambda r: int(r.get('id') or 0)) + da, db = _loads(rows[0].get('definition_json'), {}), _loads(rows[1].get('definition_json'), {}) + ea = {e.get('entity_key') or e.get('entity_id'): e for e in (da.get('entities') or [])} + eb = {e.get('entity_key') or e.get('entity_id'): e for e in (db.get('entities') or [])} + changed = sorted(k for k in (set(ea) & set(eb)) + if canonical_json(ea[k], strip=False) != canonical_json(eb[k], strip=False)) + top_changed = sorted(k for k in GD_TOP_KEYS + if canonical_json(da.get(k), strip=True) + != canonical_json(db.get(k), strip=True)) + return {'ok': True, + 'a': {'id': rows[0].get('id'), 'fingerprint': rows[0].get('content_fingerprint'), + 'compiler_version': rows[0].get('compiler_version')}, + 'b': {'id': rows[1].get('id'), 'fingerprint': rows[1].get('content_fingerprint'), + 'compiler_version': rows[1].get('compiler_version')}, + 'identical': rows[0].get('content_fingerprint') == rows[1].get('content_fingerprint'), + 'fingerprint_equal': rows[0].get('content_fingerprint') + == rows[1].get('content_fingerprint'), + 'diff': {'added': sorted(set(eb) - set(ea)), + 'removed': sorted(set(ea) - set(eb)), + 'changed': changed}, + 'top_keys_changed': top_changed} + + +# ========================================================================== +# 能力注册表（§6.1，GD capabilities + registry_hash 数据源） +# ========================================================================== +async def list_capabilities(**kw): + tid = _tid() + rows = await _rows('SELECT * FROM `%s` WHERE `tenant_id` = ${t}$' + ' ORDER BY `capability_key` ASC, `version_no` ASC' % _T_CAP, + {'t': tid}) + return {'ok': True, 'items': rows, 'total': len(rows), + 'registry_hash': registry_hash(rows)} + + +async def register_capability(capability_key=None, **kw): + """登记/升版能力（同 key 重复登记 → version_no+1，registry_hash 随之变化）。""" + tid = _tid() + key = str(capability_key or kw.get('key') or '').strip() + if not key: + raise _err('PBL_E_PARAM', 'capability_key 必填') + old = await _one('SELECT MAX(`version_no`) AS `m` FROM `%s` WHERE `tenant_id` = ${t}$' + ' AND `capability_key` = ${k}$' % _T_CAP, {'t': tid, 'k': key}) + try: + next_ver = int((old or {}).get('m') or 0) + 1 + except (TypeError, ValueError): + next_ver = 1 + schema = kw.get('args_schema_json') or kw.get('args_schema') or {} + row = { + 'capability_key': key, + 'category': str(kw.get('category') or 'pbl'), + 'args_schema_json': schema if isinstance(schema, str) else json_dump(schema), + 'permission_required': str(kw.get('permission_required') or ''), + 'is_enabled': 1 if kw.get('is_enabled', True) else 0, + 'version_no': int(kw.get('version_no') or next_ver), + 'description': str(kw.get('description') or ''), + 'created_at': now_str(), 'updated_at': now_str(), + } + res = await _await_maybe(_crud(_T_CAP).C(row, tenant_id=tid)) + new_id = (res or {}).get('id') if isinstance(res, dict) else res + await _audit('create', _T_CAP, new_id, {'capability_key': key, + 'version_no': row['version_no']}) + caps = await _capability_rows(tid) + return {'ok': True, 'id': new_id, 'capability_key': key, + 'version_no': row['version_no'], 'registry_hash': registry_hash(caps)} + + +# ========================================================================== +# Web 契约（wwwroot/api/*.dspy 调用面，命名 = 文件名） +# ========================================================================== +async def pbl_compiler_compile(**kw): + return await compile_blueprint(**kw) async def pbl_compiler_preview(**kw): - """干跑（不落库），返回定义 + 指纹，供教师预览与 diff。""" - tid = await _tid() - bp = await sql_rows('SELECT * FROM `pbl_blueprint` WHERE `tenant_id` = ${t}$' - ' AND `id` = ${id}$ LIMIT 1', {'t': tid, 'id': kw.get('blueprint_id')}, - 'pbl') - if not bp: - raise PblError('PBL_NOT_FOUND', '蓝图不存在') - cv = await _compiler_version(tid, kw.get('compiler_version_id')) - model = await _snapshot(tid, bp[0]['id'], int(kw.get('blueprint_version_no') - or bp[0]['version_no'])) - definition, fp = await compile_model(model, cv, bp[0]['id'], - int(kw.get('blueprint_version_no') or bp[0]['version_no'])) - return {'ok': True, 'preview': True, 'content_fingerprint': fp, - 'definition': definition, 'would_write_rows': 1} + return await preview_compile(**kw) async def pbl_compiler_compare(**kw): - """两版定义结构化 diff（确定性验证：同输入两次编译指纹必须一致）。""" - tid = await _tid() - ids = [kw.get('a'), kw.get('b')] - rows = await sql_rows('SELECT * FROM `pbl_game_definition` WHERE `tenant_id` = ${t}$' - ' AND `id` IN (${a}$,${b}$)', {'t': tid, 'a': ids[0], 'b': ids[1]}, 'pbl') - if len(rows) < 2: - raise PblError('PBL_GD_NOT_FOUND', '需两个已编译的 Game Definition') - a, b = sorted(rows, key=lambda r: r['id']) - da, db = _loads(a['definition_json']), _loads(b['definition_json']) - diff = {'added': [], 'removed': [], 'changed': []} - ka = {x['entity_id'] for x in da.get('entities', [])} - kb = {x['entity_id'] for x in db.get('entities', [])} - diff['added'], diff['removed'] = sorted(kb - ka), sorted(ka - kb) - for x in da.get('entities', []): - y = next((z for z in db.get('entities', []) if z['entity_id'] == x['entity_id']), None) - if y and _canon(x) != _canon(y): - diff['changed'].append(x['entity_id']) - return {'ok': True, 'a': {'id': a['id'], 'fp': a['content_fingerprint']}, - 'b': {'id': b['id'], 'fp': b['content_fingerprint']}, - 'identical': a['content_fingerprint'] == b['content_fingerprint'], 'diff': diff} + return await compare_definitions(**kw) + + +async def pbl_compiler_task_get(**kw): + return await get_compile_task(kw.get('task_id') or kw.get('id'), **kw) + + +async def pbl_compiler_task_list(**kw): + return await list_compile_tasks(**kw) + + +async def pbl_game_definition_get(**kw): + return await get_game_definition(kw.get('game_def_id') or kw.get('id'), **kw) + + +async def pbl_game_definition_get_by_blueprint(**kw): + return await get_game_definition_by_blueprint(**kw) + + +async def pbl_compiler_verify_determinism(**kw): + return await verify_determinism(**kw) + + +async def pbl_compiler_version_register(**kw): + return await register_compiler_version(**kw) + + +async def pbl_compiler_version_get(**kw): + return await get_compiler_version(kw.get('compiler_version') or kw.get('semver'), **kw) async def pbl_compiler_version_list(**kw): - return await crud('pbl_compiler_version', 'pbl', - ['code', 'semver', 'rules_hash', 'entrypoint', 'enabled', 'notes'] - )['list'](**kw) + return await list_compiler_versions(**kw) + + +async def pbl_compiler_version_diff(**kw): + return await diff_compiler_versions(kw.get('v1'), kw.get('v2'), **kw) async def pbl_compiler_version_save(**kw): - sc = crud('pbl_compiler_version', 'pbl', - ['code', 'semver', 'rules_hash', 'entrypoint', 'enabled', 'notes']) - payload = dict(kw) - payload.setdefault('rules_hash', - hashlib.sha256((RULES_HASH_SEED + str(kw.get('semver', ''))).encode()) - .hexdigest()[:16]) + """兼容旧脚手架契约：有 id → 更新，无 → 登记新版本。""" + tid = _tid() if kw.get('id'): - return await sc['update'](**payload) - return await sc['create'](**payload) + fields = {k: v for k, v in kw.items() + if k in ('compiler_version', 'ruleset_version', 'rules_hash', + 'fingerprint_baseline', 'changelog', 'status', 'enabled', + 'semver', 'code', 'entrypoint', 'notes')} + fields['updated_at'] = now_str() + sets, params = [], {'t': tid, 'id': kw['id']} + for i, (k, v) in enumerate(sorted(fields.items())): + sets.append('`%s` = ${p%d}$' % (k, i)) + params['p%d' % i] = v + await _exec('UPDATE `%s` SET %s WHERE `tenant_id` = ${t}$ AND `id` = ${id}$' + % (_T_CV, ', '.join(sets)), params) + return {'ok': True, 'id': kw['id'], 'updated': True} + return await register_compiler_version(**kw) async def pbl_capability_list(**kw): - return await crud('pbl_capability_registry', 'pbl', - ['capability_key', 'category', 'args_schema_json', 'permission_required', - 'is_enabled', 'version_no', 'description'])['list'](**kw) + return await list_capabilities(**kw) async def pbl_capability_register(**kw): - key = kw.get('capability_key') - if not key: - raise PblError('PBL_PARAM_MISSING', '缺少 capability_key') - sc = crud('pbl_capability_registry', 'pbl', - ['capability_key', 'category', 'args_schema_json', 'permission_required', - 'is_enabled', 'version_no', 'description']) - payload = dict(kw) - payload['args_schema_json'] = json_dump(kw.get('args_schema_json')) - payload.setdefault('version_no', 1) - r = await sc['upsert'](['capability_key', 'version_no'], **payload) - r['capability_key'] = key - return r + return await register_capability(**kw) + + +# ========================================================================== +# 兼容旧脚手架（M3a 前的 compile_model 纯函数入口，保留给既有单测） +# ========================================================================== +def compile_model(model, cv, bp_id, version_no): + """纯函数式编译（无 DB/无时间/无随机）——旧签名兼容包装。""" + cv = cv or {} + ctx = {'blueprint_id': bp_id, 'blueprint_version_no': version_no, + 'compiler_version': str(cv.get('semver') or cv.get('compiler_version') + or COMPILER_VERSION), + 'ruleset_version': str(cv.get('ruleset_version') or RULESET_VERSION), + 'rules_hash': str(cv.get('rules_hash') or ''), + 'created_at': '', 'duration_ms': 0, 'task_no': ''} + gd, fp = build_game_definition(model, ctx, []) + return gd, fp + + +def _canon(obj): + return canonical_json(obj, strip=False, sort_lists=False) + + +def _sha(obj): + return sha256_fingerprint(obj) diff --git a/pbl_compiler/canonical.py b/pbl_compiler/canonical.py new file mode 100644 index 0000000..dc0dfc6 --- /dev/null +++ b/pbl_compiler/canonical.py @@ -0,0 +1,207 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""pbl_compiler.canonical —— 29.6 确定性编译内核（规范化 JSON + SHA-256 指纹）。 + +本文件是 **纯函数** 模块：不 import DB、不 import 时间/随机、不调用任何模型。 +US-11「同输入重编译指纹相等」的全部确定性保证都落在这里： + +1. **键序固定**：``sort_keys=True`` + 递归 ``deep_sort``（list[dict] 按规范化串排序）； +2. **无空白**：``separators=(',', ':')``； +3. **UTF-8 稳定**：``ensure_ascii=False``，编码固定 ``utf-8``； +4. **浮点定点**：``FLOAT_PRECISION=6`` 位量化，杜绝 0.1+0.2 类表示漂移； + NaN/Infinity 一律拒绝（``allow_nan=False`` → PBL_E_VALIDATION）； +5. **无时间戳/无随机**：``strip_volatile`` 递归剔除易变键（createdAt/compiledAt/ + duration_ms/task_no/id/...），指纹只覆盖内容键； +6. **registry_hash**：能力注册表快照的规范化哈希，进 manifest 参与指纹， + 保证「能力集变更 → 指纹变更」可追溯（F-CP-05）。 + +禁止在本模块引入 ``time`` / ``datetime.now`` / ``random`` / ``uuid`` / 网络 / LLM。 +""" +from __future__ import annotations + +import hashlib +import json + +__all__ = [ + 'CANONICAL_ENCODING', 'FLOAT_PRECISION', 'VOLATILE_KEYS', 'GD_SCHEMA', + 'CanonicalError', 'normalize', 'deep_sort', 'strip_volatile', + 'canonical_json', 'canonical_bytes', 'sha256_fingerprint', + 'registry_hash', 'stable_id', 'short_hash', +] + +#: 规范化输出编码（固定，不可配置——配置化即破坏可复现性） +CANONICAL_ENCODING = 'utf-8' +#: 浮点量化精度（小数位） +FLOAT_PRECISION = 6 +#: Game Definition schema 标识（第 9 章） +GD_SCHEMA = 'pbl.game_definition.v1' + +#: 易变字段：出现在 GD 任意层级都 **不参与指纹计算**（29.6「无时间戳注入」） +VOLATILE_KEYS = frozenset({ + 'createdAt', 'created_at', 'compiledAt', 'compiled_at', 'updatedAt', + 'updated_at', 'finishedAt', 'finished_at', 'startedAt', 'started_at', + 'generatedAt', 'generated_at', 'timestamp', 'ts', 'duration_ms', + 'durationMs', 'latency_ms', 'task_no', 'taskNo', 'id', 'pk', + 'trace_id', 'request_id', 'nonce', +}) + + +class CanonicalError(ValueError): + """规范化失败（不可序列化 / 非有限浮点）——上层映射 PBL_E_VALIDATION。""" + + +def _norm_float(value): + """浮点定点化：量化到 FLOAT_PRECISION 位，返回 int（整值）或 float。 + + ``round()`` 对 IEEE-754 double 是确定的，Python ``repr`` 走最短往返表示， + 因此同值在任何平台/任何次运行都产出同一串字符。 + """ + if value != value or value in (float('inf'), float('-inf')): + raise CanonicalError('非有限浮点不可规范化: %r' % (value,)) + q = round(float(value), FLOAT_PRECISION) + if q == int(q) and abs(q) < 1e15: + return int(q) + return q + + +def normalize(obj, _depth=0): + """递归规范化为「只含 JSON 原生类型」的结构。 + + - dict：键转 str，值递归规范化； + - list/tuple：逐项递归； + - set/frozenset：转 list 后按规范化串排序（集合无序 → 必须排序才确定）； + - float：``_norm_float`` 定点化；bool 先于 int 判定（bool 是 int 子类）； + - bytes：hex；None/str/int：原样； + - 其它对象：``str()`` 兜底（datetime 等易变类型应在上游被 strip_volatile 剔除）。 + """ + if _depth > 64: + raise CanonicalError('嵌套过深（>64），疑似循环引用') + if obj is None or isinstance(obj, str): + return obj + if isinstance(obj, bool): + return obj + if isinstance(obj, int): + return obj + if isinstance(obj, float): + return _norm_float(obj) + if isinstance(obj, (bytes, bytearray)): + return bytes(obj).hex() + if isinstance(obj, dict): + out = {} + for k in obj: + out[str(k)] = normalize(obj[k], _depth + 1) + return out + if isinstance(obj, (set, frozenset)): + items = [normalize(x, _depth + 1) for x in obj] + return sorted(items, key=lambda x: json.dumps( + x, sort_keys=True, ensure_ascii=False, separators=(',', ':'), default=str)) + if isinstance(obj, (list, tuple)): + return [normalize(x, _depth + 1) for x in obj] + return str(obj) + + +def _sort_key(value): + return json.dumps(value, sort_keys=True, ensure_ascii=False, + separators=(',', ':'), default=str) + + +def deep_sort(obj): + """深度排序：dict 键序由 ``sort_keys`` 保证；list[dict] 按内容规范化串排序。 + + 只对「元素全为 dict」的 list 排序（蓝图子对象集合无序入库，排序后编译才确定）； + 含标量/混合元素的 list 保持原序（顺序本身是语义，如 rubric 维度权重序列）。 + """ + if isinstance(obj, dict): + return {k: deep_sort(obj[k]) for k in sorted(obj.keys())} + if isinstance(obj, list): + items = [deep_sort(x) for x in obj] + if items and all(isinstance(x, dict) for x in items): + return sorted(items, key=_sort_key) + return items + return obj + + +def strip_volatile(obj): + """递归剔除 VOLATILE_KEYS（时间戳/耗时/流水号/自增 id 等易变字段）。 + + 指纹只覆盖内容键 —— 这是「manifest.createdAt 不参与指纹计算」的落地点。 + """ + if isinstance(obj, dict): + return {k: strip_volatile(v) for k, v in obj.items() if k not in VOLATILE_KEYS} + if isinstance(obj, list): + return [strip_volatile(x) for x in obj] + return obj + + +def canonical_json(obj, strip=True, sort_lists=True): + """规范化 JSON 串：键序固定 / 无空白 / UTF-8 / 浮点定点 / 剔除易变键。 + + :param strip: 是否剔除易变键（指纹计算必须 True；落库正文用 False 保留 createdAt） + :param sort_lists: 是否对 list[dict] 按内容排序（指纹计算必须 True） + """ + data = normalize(obj) + if strip: + data = strip_volatile(data) + if sort_lists: + data = deep_sort(data) + try: + return json.dumps(data, sort_keys=True, ensure_ascii=False, + separators=(',', ':'), allow_nan=False) + except ValueError as exc: + raise CanonicalError('规范化 JSON 失败: %s' % exc) + + +def canonical_bytes(obj, strip=True, sort_lists=True): + """规范化 JSON 的 UTF-8 字节串（指纹输入）。""" + return canonical_json(obj, strip=strip, sort_lists=sort_lists).encode(CANONICAL_ENCODING) + + +def sha256_fingerprint(obj): + """内容指纹：``sha256(canonical_json(strip_volatile(obj)))`` 的 64 位十六进制串。 + + 同输入（同蓝图版本快照 + 同 compilerVersion + 同 ruleset + 同能力注册表） + → 同指纹；任何内容变更 → 指纹变更（US-11 / F-CP-03）。 + """ + return hashlib.sha256(canonical_bytes(obj, strip=True, sort_lists=True)).hexdigest() + + +def short_hash(text, length=16): + """短哈希（规则集/注册表摘要用，长度固定，确定性）。""" + if not isinstance(text, (str, bytes)): + text = canonical_json(text) + if isinstance(text, str): + text = text.encode(CANONICAL_ENCODING) + return hashlib.sha256(text).hexdigest()[:length] + + +def registry_hash(capabilities): + """能力注册表哈希：对 pbl.* 能力清单（key+version+args_schema）取规范化 SHA-256。 + + 入 GD ``manifest.registryHash`` 并参与 content_fingerprint —— 能力集变更必然 + 导致指纹变更，编译器版本差异可查（F-CP-05 / diff_compiler_versions）。 + """ + items = [] + for cap in (capabilities or []): + if isinstance(cap, str): + items.append({'capability_key': cap, 'version_no': 1, 'args_schema_hash': ''}) + continue + key = cap.get('capability_key') or cap.get('key') or cap.get('code') or '' + schema = cap.get('args_schema_json') or cap.get('args_schema') or cap.get('schema') or '' + items.append({ + 'capability_key': str(key), + 'version_no': int(cap.get('version_no') or cap.get('version') or 1), + 'args_schema_hash': short_hash(schema if isinstance(schema, str) + else canonical_json(schema), 16), + 'permission_required': str(cap.get('permission_required') or ''), + 'is_enabled': bool(cap.get('is_enabled', True)), + }) + items.sort(key=_sort_key) + return hashlib.sha256(canonical_bytes(items)).hexdigest() + + +def stable_id(blueprint_id, version_no, kind, seq): + """确定性 ID 派生：``{blueprint_id}:{version_no}:{kind}:{seq:05d}``。 + + 29.6「无随机」——GD 内一切 id 都由输入派生，禁用 uuid/random（US-11）。 + """ + return '%s:%s:%s:%05d' % (blueprint_id, int(version_no), kind, int(seq)) diff --git a/pbl_compiler/gd_builder.py b/pbl_compiler/gd_builder.py new file mode 100644 index 0000000..89b5c17 --- /dev/null +++ b/pbl_compiler/gd_builder.py @@ -0,0 +1,534 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""pbl_compiler.gd_builder —— 蓝图快照 → Game Definition（第 9 章 10 顶层键）。 + +**纯函数**：入参只有 (snapshot, ctx)，ctx 里的一切值都由调用方从 DB 读出后传入， +本模块不碰 DB / 不碰时间 / 不碰随机 / 不调 LLM（29.6 + 6.2「禁止编译调 LLM」）。 + +10 顶层键（顺序即 GD_SCHEMA 声明顺序，canonical_json 会再按字典序排）： + manifest / pbl / world / scenes / entities / events / states / + capabilities / assessment / assets + +所有集合一律「先按稳定键排序再遍历」，所有 id 走 ``canonical.stable_id`` 派生， +保证同输入 → 同 GD → 同指纹（US-11）。 +""" +from __future__ import annotations + +import json + +from pbl_compiler.canonical import (GD_SCHEMA, canonical_json, registry_hash, + sha256_fingerprint, short_hash, stable_id) + +__all__ = ['GD_TOP_KEYS', 'build_game_definition', 'extract_snapshot_objects', + 'build_manifest', 'build_pbl', 'build_world', 'build_scenes', + 'build_entities', 'build_events', 'build_states', + 'build_capabilities', 'build_assessment', 'build_assets', + 'fingerprint_of'] + +#: 第 9 章 GD 10 顶层键（缺任一即编译失败 PBL_E_COMPILE） +GD_TOP_KEYS = ('manifest', 'pbl', 'world', 'scenes', 'entities', 'events', + 'states', 'capabilities', 'assessment', 'assets') + + +def _loads(value, default=None): + """容错 JSON 解析：已是 dict/list 原样返回，空值返回 default。""" + if isinstance(value, (dict, list)): + return value + if value in (None, '', b''): + return default if default is not None else [] + try: + return json.loads(value) + except (ValueError, TypeError): + return default if default is not None else [] + + +def _sorted_rows(value): + """集合稳定化：解析 → 规范化串排序（入库顺序不可依赖，29.6）。""" + rows = _loads(value, []) + if isinstance(rows, dict): + rows = [dict(v, _key=k) for k, v in sorted(rows.items())] + if not isinstance(rows, list): + rows = [rows] + return sorted(rows, key=lambda r: canonical_json(r, strip=False, sort_lists=False)) + + +def _uid(row, fallback_kind, seq): + """行内稳定标识：优先 obj_uid/code，其次内容短哈希（确定性，不用 uuid）。""" + if not isinstance(row, dict): + return '%s:%05d' % (fallback_kind, seq) + for key in ('obj_uid', 'code', 'uid', 'key', 'name'): + val = row.get(key) + if val not in (None, ''): + return str(val) + return '%s:%s' % (fallback_kind, short_hash(canonical_json(row), 12)) + + +def _text(row, *keys, default=''): + for k in keys: + v = row.get(k) if isinstance(row, dict) else None + if v not in (None, ''): + return str(v) + return default + + +def _int(row, *keys, default=0): + for k in keys: + v = row.get(k) if isinstance(row, dict) else None + if v not in (None, ''): + try: + return int(v) + except (TypeError, ValueError): + continue + return default + + +def extract_snapshot_objects(snapshot): + """从蓝图版本快照里抽出 7 类子对象（键名兼容 M1a/M1b 两种落库形态）。 + + 返回 dict：goals/project/problem/missions/roles/artifacts/evidence_specs/ + rubrics/events/scenes/assets —— 每项都是「已排序的 list」。 + """ + snap = snapshot if isinstance(snapshot, dict) else {} + content = snap.get('content') or snap.get('content_json') or snap + if isinstance(content, str): + content = _loads(content, {}) + if not isinstance(content, dict): + content = {} + sub = content.get('subobjects') or content.get('sub_objects') or {} + if isinstance(sub, list): + grouped = {} + for item in sub: + if not isinstance(item, dict): + continue + grouped.setdefault(str(item.get('obj_type') or item.get('type') or 'other'), + []).append(item) + sub = grouped + if not isinstance(sub, dict): + sub = {} + + def pick(*names): + for n in names: + if n in content: + return _sorted_rows(content[n]) + if n in sub: + return _sorted_rows(sub[n]) + return [] + + return { + 'goals': pick('learning_goals', 'learningGoals', 'pbl_learning_goal', 'goal'), + 'project': pick('project', 'pbl_project', 'projects'), + 'problem': pick('problem', 'driving_question', 'pbl_problem'), + 'missions': pick('missions', 'pbl_mission', 'mission'), + 'roles': pick('roles', 'pbl_role', 'role'), + 'artifacts': pick('artifacts', 'artifact_defs', 'pbl_artifact_def', 'artifact'), + 'evidence_specs': pick('evidence_specs', 'pbl_evidence_spec', 'evidence'), + 'rubrics': pick('rubrics', 'pbl_rubric', 'rubric'), + 'events': pick('events', 'pbl_event', 'event'), + 'scenes': pick('scenes', 'pbl_scene', 'scene'), + 'assets': pick('assets', 'pbl_asset', 'asset', 'resources'), + 'raw': content, + } + + +def build_pbl(objs, snapshot_meta): + """``pbl`` 键：learner/learningGoals/problem/drivingQuestion/project 聚合。""" + raw = objs.get('raw') or {} + project = objs['project'][0] if objs['project'] else {} + problem = objs['problem'][0] if objs['problem'] else {} + goals = [] + for idx, g in enumerate(objs['goals']): + goals.append({ + 'goal_id': stable_id(snapshot_meta['blueprint_id'], + snapshot_meta['blueprint_version_no'], 'goal', idx), + 'code': _uid(g, 'goal', idx), + 'title': _text(g, 'title', 'name', 'goal_title'), + 'description': _text(g, 'description', 'desc', 'content'), + 'dimension': _text(g, 'dimension', 'category', default='knowledge'), + 'weight': _int(g, 'weight', 'weight_pct', default=0), + }) + return { + 'learner': { + 'stage': _text(raw, 'learner_stage', 'stage', default='secondary'), + 'group_size': _int(raw, 'group_size', 'team_size', default=4), + 'prior_knowledge': _text(raw, 'prior_knowledge', default=''), + }, + 'learningGoals': goals, + 'problem': { + 'problem_id': stable_id(snapshot_meta['blueprint_id'], + snapshot_meta['blueprint_version_no'], 'problem', 0), + 'statement': _text(problem, 'statement', 'description', 'title'), + 'constraints': _sorted_rows(problem.get('constraints')) + if isinstance(problem, dict) else [], + }, + 'drivingQuestion': _text(problem, 'driving_question', 'drivingQuestion', + default=_text(raw, 'driving_question')), + 'project': { + 'title': _text(project, 'title', 'name'), + 'summary': _text(project, 'summary', 'description'), + 'duration_hours': _int(project, 'duration_hours', 'duration', default=0), + 'domain': _text(project, 'domain', 'domain_code', default='general'), + }, + } + + +def build_world(objs, snapshot_meta): + """``world`` 键：世界定义（M3b 映射 world 表，不改基表结构）。""" + raw = objs.get('raw') or {} + project = objs['project'][0] if objs['project'] else {} + return { + 'world_key': stable_id(snapshot_meta['blueprint_id'], + snapshot_meta['blueprint_version_no'], 'world', 0), + 'name': _text(project, 'title', 'name', default='PBL World'), + 'mode': _text(raw, 'world_mode', 'mode', default='shared'), + 'authority': 'server', + 'domain': _text(project, 'domain', 'domain_code', default='general'), + 'max_members': _int(raw, 'max_members', 'group_size', default=8), + 'heartbeat_seconds': _int(raw, 'heartbeat_seconds', default=15), + 'runtime_neutral': True, + 'renderer_hint': _text(raw, 'renderer', 'renderer_hint', default='three.js'), + } + + +def build_scenes(objs, snapshot_meta): + """``scenes`` 键：pbl_mission 分组 → 场景列表（M3b 映射 scene 表）。""" + bp_id = snapshot_meta['blueprint_id'] + ver = snapshot_meta['blueprint_version_no'] + scenes = [] + missions = objs['missions'] or ([{'title': 'main', 'code': 'main'}] + if not objs['scenes'] else []) + for idx, m in enumerate(missions): + mid = _uid(m, 'mission', idx) + scenes.append({ + 'scene_key': stable_id(bp_id, ver, 'scene', idx), + 'source': 'pbl_mission:%s' % mid, + 'name': _text(m, 'title', 'name', default='Mission %d' % (idx + 1)), + 'order_no': _int(m, 'order_no', 'order', 'seq', default=idx), + 'unlock_condition': _loads(m.get('unlock_condition'), {}) if isinstance(m, dict) else {}, + 'environment': _text(m, 'environment', 'scene_type', default='default'), + 'entity_refs': [], + }) + for idx, s in enumerate(objs['scenes'] or []): + scenes.append({ + 'scene_key': stable_id(bp_id, ver, 'scene', len(missions) + idx), + 'source': 'pbl_scene:%s' % _uid(s, 'scene', idx), + 'name': _text(s, 'title', 'name', default='Scene %d' % (idx + 1)), + 'order_no': _int(s, 'order_no', 'order', default=len(missions) + idx), + 'unlock_condition': _loads(s.get('unlock_condition'), {}) if isinstance(s, dict) else {}, + 'environment': _text(s, 'environment', 'scene_type', default='default'), + 'entity_refs': [], + }) + scenes.sort(key=lambda x: (x['order_no'], x['scene_key'])) + return scenes + + +def build_entities(objs, snapshot_meta, scenes): + """``entities`` 键：pbl_role + pbl_artifact_def + mission/task → 实体列表。 + + 每个实体带 ``scene_key`` 归属（默认落第 1 个场景），供 M3b 映射 entity 表。 + """ + bp_id = snapshot_meta['blueprint_id'] + ver = snapshot_meta['blueprint_version_no'] + default_scene = scenes[0]['scene_key'] if scenes else stable_id(bp_id, ver, 'scene', 0) + entities = [] + seq = 0 + + for m in objs['missions'] or []: + mid = _uid(m, 'mission', seq) + entities.append({ + 'entity_key': stable_id(bp_id, ver, 'entity', seq), + 'kind': 'mission', 'source': 'pbl_mission:%s' % mid, + 'name': _text(m, 'title', 'name', default=mid), + 'scene_key': default_scene, + 'capability_key': 'pbl.mission.show', + 'server_authoritative': True, + 'initial_state': {'completed': False, + 'order': _int(m, 'order_no', 'order', default=seq)}, + }) + seq += 1 + tasks = _sorted_rows(m.get('tasks')) if isinstance(m, dict) else [] + for t in tasks: + tk = _uid(t, 'task', seq) + entities.append({ + 'entity_key': stable_id(bp_id, ver, 'entity', seq), + 'kind': 'task', 'source': 'pbl_task:%s' % tk, + 'name': _text(t, 'title', 'name', default=tk), + 'scene_key': default_scene, + 'capability_key': 'pbl.task.track', + 'server_authoritative': True, + 'initial_state': {'done': False, 'parent': mid}, + }) + seq += 1 + + for r in objs['roles'] or []: + rk = _uid(r, 'role', seq) + entities.append({ + 'entity_key': stable_id(bp_id, ver, 'entity', seq), + 'kind': 'role', 'source': 'pbl_role:%s' % rk, + 'name': _text(r, 'title', 'name', default=rk), + 'scene_key': default_scene, + 'capability_key': 'pbl.role.assign', + 'server_authoritative': True, + 'initial_state': {'assigned': False, + 'permissions': _sorted_rows(r.get('permissions')) + if isinstance(r, dict) else []}, + }) + seq += 1 + + for a in objs['artifacts'] or []: + ak = _uid(a, 'artifact', seq) + entities.append({ + 'entity_key': stable_id(bp_id, ver, 'entity', seq), + 'kind': 'artifact_spec', 'source': 'pbl_artifact_def:%s' % ak, + 'name': _text(a, 'title', 'name', default=ak), + 'scene_key': default_scene, + 'capability_key': 'pbl.artifact.submit', + 'server_authoritative': True, + 'initial_state': {'submitted': False, + 'artifact_type': _text(a, 'artifact_type', 'type', + default='document')}, + }) + seq += 1 + + entities.sort(key=lambda x: x['entity_key']) + for sc in scenes: + sc['entity_refs'] = sorted( + e['entity_key'] for e in entities if e['scene_key'] == sc['scene_key']) + return entities + + +def build_events(objs, snapshot_meta, entities): + """``events`` 键：pbl_mission.unlock_condition + pbl_evidence_spec → 事件模型（第 12 章）。 + + 事件结构 = {event_type, source, condition, response[{op, capability}]}， + 供 M3b 映射 script_engine 脚本（events→condition→response→capability）。 + """ + bp_id = snapshot_meta['blueprint_id'] + ver = snapshot_meta['blueprint_version_no'] + events = [] + seq = 0 + by_source = {} + for e in entities: + by_source.setdefault(e['kind'], []).append(e) + + for m in by_source.get('mission', []): + cond = m.get('initial_state', {}).get('unlock_condition') + events.append({ + 'event_key': stable_id(bp_id, ver, 'event', seq), + 'event_type': 'mission.completed', + 'source': m['source'], + 'condition': {'all_of': ['%s.completed == true' % m['entity_key']]}, + 'response': [{'op': 'emit', 'event': 'mission.completed', + 'capability': 'pbl.mission.show'}], + 'unlock_condition': _loads(cond, {}) if cond else {}, + }) + seq += 1 + for t in by_source.get('task', []): + events.append({ + 'event_key': stable_id(bp_id, ver, 'event', seq), + 'event_type': 'task.completed', + 'source': t['source'], + 'condition': {'all_of': ['%s.done == true' % t['entity_key']]}, + 'response': [{'op': 'emit', 'event': 'task.completed', + 'capability': 'pbl.task.track'}], + }) + seq += 1 + for idx, ev in enumerate(objs['evidence_specs'] or []): + ek = _uid(ev, 'evidence_spec', idx) + events.append({ + 'event_key': stable_id(bp_id, ver, 'event', seq), + 'event_type': 'evidence.required', + 'source': 'pbl_evidence_spec:%s' % ek, + 'condition': {'min_count': _int(ev, 'min_count', 'count', default=1)}, + 'response': [{'op': 'collect', + 'evidence_type': _text(ev, 'evidence_type', 'type', + default='artifact'), + 'capability': 'pbl.evidence.collect'}], + }) + seq += 1 + for idx, ev in enumerate(objs['events'] or []): + events.append({ + 'event_key': stable_id(bp_id, ver, 'event', seq), + 'event_type': _text(ev, 'event_type', 'type', default='custom'), + 'source': 'pbl_event:%s' % _uid(ev, 'event', idx), + 'condition': _loads(ev.get('condition'), {}) if isinstance(ev, dict) else {}, + 'response': _loads(ev.get('response'), []) if isinstance(ev, dict) else [], + }) + seq += 1 + + events.sort(key=lambda x: canonical_json({k: v for k, v in x.items() + if k != 'event_key'}, + strip=False, sort_lists=False)) + for i, ev in enumerate(events): + ev['event_key'] = stable_id(bp_id, ver, 'event', i) + return events + + +def build_states(entities): + """``states`` 键：实体初始状态契约（entity_key → initial_state 快照）。""" + states = {} + for e in entities: + states[e['entity_key']] = { + 'kind': e['kind'], + 'initial': e.get('initial_state', {}), + 'server_authoritative': bool(e.get('server_authoritative', True)), + } + return {k: states[k] for k in sorted(states.keys())} + + +def build_capabilities(objs, entities, events, registry_rows): + """``capabilities`` 键：§6.1 能力注册表快照（pbl.* 能力清单）。 + + 以 DB 注册表为准（registry_rows 由调用方查出传入），并补齐 GD 实际引用到、 + 但注册表缺失的能力（标 ``registered: false``，供 M4b fail-closed 裁决告警）。 + """ + used = sorted({e.get('capability_key') for e in entities if e.get('capability_key')} + | {r.get('capability') for ev in events + for r in (ev.get('response') or []) if isinstance(r, dict) + and r.get('capability')}) + known = {} + for row in registry_rows or []: + if not isinstance(row, dict): + continue + key = str(row.get('capability_key') or row.get('key') or '') + if not key: + continue + known[key] = { + 'capability_key': key, + 'category': _text(row, 'category', default='pbl'), + 'version_no': _int(row, 'version_no', 'version', default=1), + 'permission_required': _text(row, 'permission_required', default=''), + 'is_enabled': bool(row.get('is_enabled', True)), + 'args_schema_hash': short_hash( + row.get('args_schema_json') or row.get('args_schema') or '', 16), + 'registered': True, + 'used_by_gd': key in used, + } + for key in used: + if key not in known: + known[key] = {'capability_key': key, 'category': 'pbl', 'version_no': 1, + 'permission_required': '', 'is_enabled': True, + 'args_schema_hash': '', 'registered': False, + 'used_by_gd': True} + return {'schema': 'pbl.capability_list.v1', + 'registry_hash': registry_hash(list(known.values())), + 'items': [known[k] for k in sorted(known.keys())]} + + +def build_assessment(objs, snapshot_meta): + """``assessment`` 键：pbl_rubric 只读权重结构（34 章，M6 消费）。""" + bp_id = snapshot_meta['blueprint_id'] + ver = snapshot_meta['blueprint_version_no'] + dimensions = [] + for idx, r in enumerate(objs['rubrics'] or []): + criteria = [] + for cidx, c in enumerate(_sorted_rows(r.get('criteria')) if isinstance(r, dict) else []): + criteria.append({ + 'criterion_key': stable_id(bp_id, ver, 'criterion', cidx), + 'name': _text(c, 'title', 'name', default='c%d' % cidx), + 'weight': _int(c, 'weight', 'weight_pct', default=0), + 'levels': _sorted_rows(c.get('levels')) if isinstance(c, dict) else [], + }) + criteria.sort(key=lambda x: (x['name'], x['criterion_key'])) + dimensions.append({ + 'dimension_key': stable_id(bp_id, ver, 'rubric', idx), + 'source': 'pbl_rubric:%s' % _uid(r, 'rubric', idx), + 'name': _text(r, 'title', 'name', default='rubric%d' % idx), + 'weight': _int(r, 'weight', 'weight_pct', default=0), + 'criteria': criteria, + }) + dimensions.sort(key=lambda x: x['dimension_key']) + total = sum(d['weight'] for d in dimensions) + return {'schema': 'pbl.assessment.v1', 'readonly': True, + 'dimensions': dimensions, 'total_weight': total, + 'weight_normalized': (total == 100) if dimensions else True} + + +def build_assets(objs): + """``assets`` 键：资源引用清单（runtime-neutral，29.4 不绑定渲染端）。""" + assets = [] + for idx, a in enumerate(objs['assets'] or []): + url = _text(a, 'url', 'asset_url', 'src', 'path') + assets.append({ + 'asset_key': _uid(a, 'asset', idx), + 'kind': _text(a, 'kind', 'type', 'asset_type', default='model'), + 'url': url, + 'url_hash': short_hash(url, 16) if url else '', + 'mime': _text(a, 'mime', 'mime_type', default=''), + 'runtime_neutral': True, + }) + assets.sort(key=lambda x: (x['kind'], x['asset_key'])) + return {'schema': 'pbl.assets.v1', 'runtime_neutral': True, 'items': assets} + + +def build_manifest(ctx, gd_body, capabilities): + """``manifest`` 键：编译元数据。 + + **createdAt / durationMs 属易变字段**：写进 GD 正文供审计，但被 + ``canonical.VOLATILE_KEYS`` 排除在指纹计算外（29.6「无时间戳注入」）。 + """ + return { + 'schema': GD_SCHEMA, + 'gdTopKeys': list(GD_TOP_KEYS), + 'blueprintId': ctx['blueprint_id'], + 'blueprintVersion': ctx['blueprint_version_no'], + 'compilerVersion': ctx['compiler_version'], + 'rulesetVersion': ctx['ruleset_version'], + 'rulesHash': ctx['rules_hash'], + 'registryHash': capabilities.get('registry_hash', ''), + 'deterministic': True, + 'llmUsed': False, + 'counts': { + 'scenes': len(gd_body['scenes']), + 'entities': len(gd_body['entities']), + 'events': len(gd_body['events']), + 'capabilities': len(capabilities.get('items', [])), + 'rubricDimensions': len(gd_body['assessment'].get('dimensions', [])), + 'assets': len(gd_body['assets'].get('items', [])), + }, + # 易变字段（不参与指纹） + 'createdAt': ctx.get('created_at', ''), + 'durationMs': ctx.get('duration_ms', 0), + 'taskNo': ctx.get('task_no', ''), + } + + +def fingerprint_of(gd): + """GD 内容指纹（剔除易变键后的规范化 SHA-256）。""" + return sha256_fingerprint(gd) + + +def build_game_definition(snapshot, ctx, registry_rows=None): + """蓝图版本快照 → 完整 GD（10 顶层键）+ 内容指纹。 + + :param snapshot: 蓝图版本快照（dict 或 JSON 串，来自 pbl_blueprint.get_version） + :param ctx: {'blueprint_id','blueprint_version_no','compiler_version', + 'ruleset_version','rules_hash','created_at','duration_ms','task_no'} + :param registry_rows: pbl_capability_registry 行（调用方查出传入，本模块不碰 DB） + :return: (gd_dict, fingerprint_str) + """ + objs = extract_snapshot_objects(snapshot) + meta = {'blueprint_id': ctx['blueprint_id'], + 'blueprint_version_no': ctx['blueprint_version_no']} + scenes = build_scenes(objs, meta) + entities = build_entities(objs, meta, scenes) + events = build_events(objs, meta, entities) + capabilities = build_capabilities(objs, entities, events, registry_rows) + body = { + 'pbl': build_pbl(objs, meta), + 'world': build_world(objs, meta), + 'scenes': scenes, + 'entities': entities, + 'events': events, + 'states': build_states(entities), + 'capabilities': capabilities, + 'assessment': build_assessment(objs, meta), + 'assets': build_assets(objs), + } + gd = {'manifest': build_manifest(ctx, body, capabilities)} + gd.update(body) + missing = [k for k in GD_TOP_KEYS if k not in gd] + if missing: + raise ValueError('GD 缺顶层键: %s' % ','.join(missing)) + fp = fingerprint_of(gd) + gd['manifest']['fingerprint'] = fp + return gd, fp