deliver: 交付收口（引擎代为提交）

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": "失败"
+        }
+      ]
+    }
+  ]
+}

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))

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