#!/usr/bin/env python3 # -*- coding: utf-8 -*- """pbl_compiler.api —— PBL Compiler v1：Blueprint → Game Definition（第9章，M3a/M3b）。 29.6 确定性三要素（可复现证明 = content_fingerprint 相同）： 1) 输入锁定：blueprint_id + blueprint_version_no（版本快照，不读可变主表）； 2) 规则锁定：compiler_version_id（规则集哈希 rules_hash）； 3) 无副作用：不依赖随机数/时间戳/网络；集合一律按稳定键排序后遍历。 落库 world/scene/entity 走 pbl_domain_ext 契约（薄扩展，不改基表结构）。 """ import hashlib import json from pbl_common.api import (PblError, actor_id, crud, json_dump, now_str, sql_exec, sql_rows, sql_scalar, tenant_id) COMPILER_VERSION = 'pbl-compiler-v1.0.0' RULES_HASH_SEED = 'pbl.compiler.rules.v1' GAME_DEF_SCHEMA = 'pbl.game_definition.v1' def _loads(v, d=None): if isinstance(v, (dict, list)): return v try: return json.loads(v) if v else (d or {}) except ValueError: return d or {} def _canon(obj): return json.dumps(obj, sort_keys=True, ensure_ascii=False, separators=(',', ':')) def _sha(obj): return hashlib.sha256(_canon(obj).encode('utf-8')).hexdigest() async def _tid(): return await tenant_id() 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 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 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] 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}), } fp = _sha(definition) definition['meta']['content_fingerprint'] = fp return definition, fp 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') 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)} 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 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} 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} async def pbl_compiler_version_list(**kw): return await crud('pbl_compiler_version', 'pbl', ['code', 'semver', 'rules_hash', 'entrypoint', 'enabled', 'notes'] )['list'](**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]) if kw.get('id'): return await sc['update'](**payload) return await sc['create'](**payload) 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) 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