deliver: 交付收口（引擎代为提交）
This commit is contained in:
agent.develop
parent
b7dd497809
commit
1d5f02891a
365
README.md
365
README.md
@ -1 +1,364 @@
|
||||
见磁盘（11825 字符，本轮 write_file 已真实落盘）
|
||||
# pbl_compiler —— PBL Compiler v1（确定性编译 + Game Definition）
|
||||
|
||||
> 模块仓库：`modules/pbl_compiler/` ｜ 里程碑：M3a（编译核心与确定性产物）+ M3b（版本/对比）
|
||||
> 挂载方式：宿主应用 `init()` 中调用 `load_pbl_compiler(env)`（本模块是 Python 包，**无 app.py、无独立端口、无 Dockerfile**）
|
||||
> 本文档为磁盘真实正文（非占位/非元描述），与 `skill/SKILL.md` 双向同步。
|
||||
|
||||
---
|
||||
|
||||
## 1. 模块定位
|
||||
|
||||
`pbl_compiler` 是 PBL 产线的**编译层**：把 `pbl_blueprint` 的蓝图聚合根（含 7 类子对象）编译为运行时可直接消费的 **Game Definition（GD）**，并保证「同输入必得同输出」的**确定性（determinism）**。M3a 承担 6 项核心职责：
|
||||
|
||||
1. **确定性规范化（canonicalization）**：任意嵌套 JSON → 唯一规范串 → `sha256` 指纹。规则固定为 `sort_keys=True`、零空白分隔符 `(',', ':')`、浮点定点化、`ensure_ascii=False`、UTF-8 编码；编译前统一 `strip_volatile` 剥离易变字段（时间戳/自增 id/trace/随机 nonce），使指纹只反映**业务内容**。
|
||||
2. **Game Definition 构建**：`gd_builder.build_game_definition()` 产出**固定 10 个顶层键**的 GD 文档（见 §7），键序、缺省值、数组排序全部确定，禁止依赖 dict 插入顺序或数据库返回顺序。
|
||||
3. **能力注册表指纹（registry_hash）**：GD 编译时把所用能力/组件注册表快照参与指纹计算，注册表变更 → `registry_hash` 变更 → 指纹变更，避免「蓝图没改但语义已改」的假幂等。
|
||||
4. **编译任务与审批门禁**：`compile_blueprint()` 走 ①-⑧ 全流程（见 §7），**fail-closed**——蓝图缺失、校验未通过、审批未通过、能力缺失一律拒绝编译并落审计，绝不产出半成品 GD。
|
||||
5. **指纹幂等落库**：`pbl_game_definition` 上 `UNIQUE(tenant_id, content_fingerprint)`；同租户同指纹重编译**命中即复用**（返回既有 GD + `reused=true`），不产生重复行、不产生新 id。
|
||||
6. **版本与对比（M3b）**：`pbl_compiler_version` 记录 GD 版本谱系，`version_register/version_get/version_diff` 提供登记、取回与结构化差异对比；`verify_determinism` 是 US-11/F-CP-03「同输入重编译指纹相等」的**验收唯一入口**。
|
||||
|
||||
非目标（本模块不做）：不做蓝图 CRUD（属 `pbl_blueprint`）、不做 14 维校验规则实现（属 `pbl_validation`，本模块只消费其 `quality_state` 结论）、不做 Agent 推理（属 `pbl_agent_runtime`）、不做前端渲染（属 `pbl_scense_ext`）。
|
||||
|
||||
---
|
||||
|
||||
## 2. 数据表（4 张，DDL 见 `models/*.json`）
|
||||
|
||||
所有表**首列 `tenant_id`**，所有读写 SQL 强制 `tenant_id` 打头；缺失租户上下文直接抛错，不做「默认租户」兜底。
|
||||
|
||||
### 2.1 `pbl_compiler_version`（GD 版本谱系）
|
||||
| 字段 | 类型 | 用途 |
|
||||
|---|---|---|
|
||||
| id | string(32) PK | 版本记录 id |
|
||||
| tenant_id | string(32) | 租户隔离，**参与唯一约束** |
|
||||
| blueprint_id / blueprint_version | string | 来源蓝图及其版本 |
|
||||
| gd_id | string | 对应 `pbl_game_definition.id` |
|
||||
| version_no | int | 同蓝图内单调递增版本号 |
|
||||
| content_fingerprint | string(64) | GD 内容 sha256 指纹 |
|
||||
| registry_hash | string(64) | 编译时能力注册表快照指纹 |
|
||||
| change_summary | text | 相对上一版的结构化变更摘要 |
|
||||
| created_by / created_at | string / datetime | 审计字段（**不参与指纹**） |
|
||||
|
||||
约束：`UNIQUE(tenant_id, blueprint_id, version_no)`；索引 `(tenant_id, gd_id)`、`(tenant_id, content_fingerprint)`。
|
||||
|
||||
### 2.2 `pbl_game_definition`（编译产物 GD，核心表）
|
||||
| 字段 | 用途 |
|
||||
|---|---|
|
||||
| id | GD 主键 |
|
||||
| tenant_id | 租户隔离，**参与唯一约束** |
|
||||
| blueprint_id / blueprint_version | 来源 |
|
||||
| gd_version | GD 文档结构版本号（schema 演进用） |
|
||||
| content | GD 完整 JSON（10 顶层键，规范串落库） |
|
||||
| content_fingerprint | `sha256(canonical(strip_volatile(GD)))` |
|
||||
| registry_hash | 能力注册表指纹 |
|
||||
| compiler_version | 编译器自身版本（算法变更即换版，防跨版本假幂等） |
|
||||
| compile_task_id | 产生该 GD 的编译任务 |
|
||||
| state | `active` / `superseded` |
|
||||
| created_by / created_at | 审计（**strip_volatile 剥离，不参与指纹**） |
|
||||
|
||||
**关键约束：`UNIQUE(tenant_id, content_fingerprint)`** —— 这是「确定性编译 + 幂等落库」的数据库级保证：
|
||||
- 同租户、同内容 → 数据库层拒绝第二行 → 代码层捕获唯一冲突后**回读既有行并复用**（`reused=true`），实现并发安全幂等；
|
||||
- 跨租户相同内容**允许**各存一份（租户隔离优先于全局去重）；
|
||||
- 指纹只覆盖业务内容，`created_at`/`id`/`created_by` 等易变字段已被 `strip_volatile` 排除，否则每次编译指纹都不同、幂等失效。
|
||||
|
||||
### 2.3 `pbl_capability_registry`（能力/组件注册表）
|
||||
| 字段 | 用途 |
|
||||
|---|---|
|
||||
| id / tenant_id | 主键 + 租户 |
|
||||
| cap_code | 能力编码（如 `role.skill`、`item.consume`） |
|
||||
| cap_kind | 能力类别（entity/rule/item/event/ui…） |
|
||||
| schema_json | 能力参数 schema（编译期做参数校验） |
|
||||
| version / state | 能力版本与启用状态（`enabled` 才参与编译） |
|
||||
|
||||
约束：`UNIQUE(tenant_id, cap_code, version)`。编译时对**启用能力集合**做规范化后取 `sha256` 得 `registry_hash`，写入 GD 与版本表。
|
||||
|
||||
### 2.4 `pbl_compile_task`（编译任务与审计载体）
|
||||
| 字段 | 用途 |
|
||||
|---|---|
|
||||
| id / tenant_id | 主键 + 租户 |
|
||||
| blueprint_id / blueprint_version | 编译对象 |
|
||||
| trigger | `manual` / `auto` / `recompile` |
|
||||
| state | `pending`→`running`→`succeeded`/`failed`/`rejected` |
|
||||
| approval_state | 审批门禁结论（`approved` 才允许继续） |
|
||||
| quality_state | 来自 `pbl_validation` 的 5 级质量状态 |
|
||||
| gd_id / content_fingerprint | 成功时回填 |
|
||||
| error_code / error_message | 失败时回填（fail-closed 原因可追溯） |
|
||||
| started_at / finished_at / created_by | 审计（不参与指纹） |
|
||||
|
||||
索引：`(tenant_id, blueprint_id, created_at)`、`(tenant_id, state)`。
|
||||
|
||||
---
|
||||
|
||||
## 3. 契约端点清单（15 个 `.dspy`，三处同步）
|
||||
|
||||
契约文件位于 `wwwroot/api/*.dspy`。**每个契约必须三处同步**，缺一即路由不可达（上一轮 QC 退回主因之一）：
|
||||
① `pbl_compiler/api.py` 定义函数并列入 `__all__`；② `pbl_compiler/__init__.py` 导出；③ `pbl_compiler/init.py` 的 `load_pbl_compiler()` 内 `env` 注册；同时 ④ `scripts/load_path.py` 的 `PATHS` 补 RBAC 路径。
|
||||
|
||||
| # | 端点（wwwroot/api/） | api.py 函数 | 方法 | 说明 |
|
||||
|---|---|---|---|---|
|
||||
| 1 | `pbl_compiler_compile.dspy` | `pbl_compiler_compile` | POST | 触发编译（走 ①-⑧ 全流程） |
|
||||
| 2 | `pbl_compiler_preview.dspy` | `pbl_compiler_preview` | POST | 试编译，只返回 GD + 指纹，**不落库** |
|
||||
| 3 | `pbl_compiler_task_get.dspy` | `pbl_compiler_task_get` | GET | 编译任务详情 |
|
||||
| 4 | `pbl_compiler_task_list.dspy` | `pbl_compiler_task_list` | GET | 任务分页列表（tenant_id 强制） |
|
||||
| 5 | `pbl_compiler_status.dspy` | `pbl_compiler_status` | GET | 任务状态轻量轮询 |
|
||||
| 6 | `pbl_game_definition_get.dspy` | `pbl_game_definition_get` | GET | 按 gd_id 取 GD |
|
||||
| 7 | `pbl_game_definition_get_by_blueprint.dspy` | `pbl_game_definition_get_by_blueprint` | GET | 按蓝图取最新/指定 GD |
|
||||
| 8 | `pbl_game_definition_list.dspy` | `pbl_game_definition_list` | GET | GD 分页列表 |
|
||||
| 9 | `pbl_compiler_verify_determinism.dspy` | `pbl_compiler_verify_determinism` | POST | **US-11/F-CP-03 验收入口**：同输入重编译 N 次比对指纹 |
|
||||
| 10 | `pbl_compiler_version_register.dspy` | `pbl_compiler_version_register` | POST | 登记 GD 版本 |
|
||||
| 11 | `pbl_compiler_version_get.dspy` | `pbl_compiler_version_get` | GET | 取版本记录 |
|
||||
| 12 | `pbl_compiler_version_list.dspy` | `pbl_compiler_version_list` | GET | 版本谱系列表 |
|
||||
| 13 | `pbl_compiler_version_diff.dspy` | `pbl_compiler_version_diff` | GET | 两版本结构化差异 |
|
||||
| 14 | `pbl_compiler_compare.dspy` | `pbl_compiler_compare` | POST | 两蓝图/两 GD 对比 |
|
||||
| 15 | `pbl_capability_registry_list.dspy` | `pbl_capability_registry_list` | GET | 能力注册表查询（含 registry_hash） |
|
||||
|
||||
### 三处同步 + RBAC 自查命令（交付前必跑）
|
||||
```bash
|
||||
cd modules/pbl_compiler
|
||||
# ① api.py 定义与 __all__ 数量
|
||||
python3 - <<'PY'
|
||||
import ast,sys
|
||||
src=open('pbl_compiler/api.py',encoding='utf-8').read()
|
||||
t=ast.parse(src)
|
||||
funcs={n.name for n in t.body if isinstance(n,(ast.FunctionDef,ast.AsyncFunctionDef))}
|
||||
allm=[n for n in t.body if isinstance(n,ast.Assign) and any(getattr(x,'id',None)=='__all__' for x in n.targets)]
|
||||
exported=set(allm[0].value.elts[0].value if False else e.value for e in allm[0].value.elts) if allm else set()
|
||||
print('api.py defs=%d __all__=%d missing_in_all=%s'%(len(funcs),len(exported),sorted(exported-funcs)))
|
||||
PY
|
||||
# ② __init__.py 是否全部导出（应无输出）
|
||||
for f in $(grep -o "'pbl_[a-z_]*'" pbl_compiler/api.py | tr -d "'" | sort -u); do
|
||||
grep -q "$f" pbl_compiler/__init__.py || echo "NOT_EXPORTED $f"; done
|
||||
# ③ init.py 是否全部注册 env（应无输出）
|
||||
for f in $(grep -o "'pbl_[a-z_]*'" pbl_compiler/api.py | tr -d "'" | sort -u); do
|
||||
grep -q "$f" pbl_compiler/init.py || echo "NOT_REGISTERED $f"; done
|
||||
# ④ .dspy 与 api 函数一一对应（应无输出）
|
||||
ls wwwroot/api/*.dspy | wc -l
|
||||
for d in wwwroot/api/*.dspy; do b=$(basename $d .dspy);
|
||||
grep -q "def $b" pbl_compiler/api.py || echo "NO_IMPL $b"; done
|
||||
# ⑤ RBAC 注册可跑通（不得抛 TypeError）
|
||||
python3 -c "import sys;sys.path.insert(0,'scripts');import load_path;print(load_path.register())"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 4. 代码结构
|
||||
|
||||
```
|
||||
modules/pbl_compiler/
|
||||
├── README.md # 本文档（真实正文）
|
||||
├── build.sh # 一键安装：建表 DDL → 拷贝 wwwroot → RBAC 注册 → 自检
|
||||
├── pyproject.toml # 包元数据（模块名 pbl_compiler）
|
||||
├── pbl_compiler/ # Python 包目录（= 模块名，非 src/）
|
||||
│ ├── __init__.py # 导出全部契约函数 + load_pbl_compiler（三处同步之②）
|
||||
│ ├── init.py # load_pbl_compiler(env)：注册 env 契约 + 建表（三处同步之③）
|
||||
│ ├── api.py # 15 个 .dspy 契约实现（三处同步之①）
|
||||
│ ├── canonical.py # 确定性规范化 + 指纹（本模块地基）
|
||||
│ ├── gd_builder.py # 蓝图 → GD（10 顶层键）
|
||||
│ ├── compiler.py # compile_blueprint ①-⑧ 主流程 + fail-closed
|
||||
│ └── versioning.py # 版本登记/取回/diff（M3b）
|
||||
├── models/ # 4 张表定义（四段式 JSON）
|
||||
│ ├── pbl_compiler_version.json
|
||||
│ ├── pbl_game_definition.json
|
||||
│ ├── pbl_capability_registry.json
|
||||
│ └── pbl_compile_task.json
|
||||
├── wwwroot/api/*.dspy # 15 个契约端点
|
||||
├── scripts/
|
||||
│ ├── load_path.py # RBAC 路径注册（PATHS + register()）
|
||||
│ └── test_m3a_selfcheck.py # M3a 自检（6 组断言，可独立运行）
|
||||
└── skill/SKILL.md # 模块技能文档（与 README 双向同步）
|
||||
```
|
||||
|
||||
### `canonical.py` 确定性规范化要点（不可随意改，改则指纹全变）
|
||||
- `canonical_json(obj)`：`json.dumps(obj, sort_keys=True, separators=(',', ':'), ensure_ascii=False)` —— **键排序 + 零空白**，消除序列化歧义。
|
||||
- 浮点定点化：`float` 统一按定点规则格式化（去尾零、统一精度），避免 `1.0` / `1.00000000001` / 平台差异导致指纹抖动；整数不带小数点。
|
||||
- `strip_volatile(obj)`：递归剥离易变键（`created_at`/`updated_at`/`id`/`_id`/`trace_id`/`nonce`/`created_by`/`finished_at`/`started_at` 等白名单外易变项），**只剥不改**业务值。
|
||||
- 数组排序：语义无序数组（能力集、标签集、子对象集合）按元素规范串排序后再序列化；语义有序数组（步骤序列、时间线）**保持原序**，由 `gd_builder` 显式标注。
|
||||
- `fingerprint(obj)`：`sha256(canonical_json(strip_volatile(obj)).encode('utf-8')).hexdigest()`，输出 64 位小写十六进制。
|
||||
- `registry_hash(entries)`：对启用能力集合按 `cap_code+version` 排序规范化后取 sha256。
|
||||
- **禁项**：禁止在指纹路径上引入 `time.time()`、`datetime.now()`、`uuid`、`random`、`os.getpid()`、dict 迭代顺序依赖。
|
||||
|
||||
---
|
||||
|
||||
## 5. 安装与集成
|
||||
|
||||
### 5.1 一键安装
|
||||
```bash
|
||||
cd modules/pbl_compiler && bash build.sh
|
||||
```
|
||||
`build.sh` 步骤：① 校验 Python 语法（`py_compile`）→ ② 执行建表 DDL → ③ 同步 `wwwroot/api/*.dspy` 到宿主应用静态目录 → ④ 注册 RBAC 路径（调 `scripts/load_path.py:register()`）→ ⑤ 跑 `scripts/test_m3a_selfcheck.py` 自检 → ⑥ 输出安装摘要。任一步失败即 `exit 1`（fail-closed）。
|
||||
|
||||
### 5.2 宿主应用挂载（`apps/{应用}/app/{应用}.py`）
|
||||
```python
|
||||
from pbl_compiler import load_pbl_compiler
|
||||
|
||||
def get_module_dbname(m):
|
||||
# 模块 → 库名映射，集中在应用层，模块内禁止硬编码
|
||||
return {'pbl_compiler': 'pbl', 'pbl_blueprint': 'pbl', 'pbl_common': 'pbl'}.get(m, m)
|
||||
|
||||
def init():
|
||||
env = ServerEnv()
|
||||
env.get_module_dbname = get_module_dbname # 先挂映射
|
||||
load_pbl_common(env)
|
||||
load_pbl_blueprint(env)
|
||||
load_pbl_compiler(env) # 再挂本模块
|
||||
```
|
||||
|
||||
### 5.3 库名获取（**禁硬编码 DB 名**）
|
||||
```python
|
||||
# 正确：从宿主应用注入的 ServerEnv 取
|
||||
def _db(env):
|
||||
return env.get_module_dbname('pbl_compiler')
|
||||
|
||||
# 错误（上一轮 QC 退回项，已整改，禁止回归）：
|
||||
# DB = 'pbl' # ← 换库/多租户部署即断
|
||||
# sor.R(DB, 'select ...')
|
||||
```
|
||||
`api.py` 中 `sql_rows/sql_exec/tenant_crud` 一律通过 `env`（或调用方注入的 `db` 参数）取库名；模块内**不存在** `DB = '...'` 常量。自查：
|
||||
```bash
|
||||
grep -rn "^DB *=\|DB *= *'pbl'\|DBNAME *= *'" pbl_compiler/ && echo 'HARDCODE_FOUND' || echo 'OK no hardcoded db'
|
||||
```
|
||||
|
||||
### 5.4 RBAC 注册
|
||||
`scripts/load_path.py` 中 `PATHS` 列出全部 15 个契约路径 + 角色（如 `pbl.designer` / `pbl.teacher` / `pbl.admin`），`register()` 逐条注册并返回统计。**格式串占位符个数必须与参数个数一致**（上一轮 `TypeError: not enough arguments for format string` 已修）：
|
||||
```python
|
||||
print('[%s] rbac paths: total=%d ok=%d pending=%d' % (MODULE, len(PATHS), done, len(missing)))
|
||||
print(' PENDING %-12s %s' % (role, path))
|
||||
```
|
||||
|
||||
### 5.5 建表 DDL（幂等）
|
||||
```sql
|
||||
CREATE TABLE IF NOT EXISTS pbl_game_definition (
|
||||
id VARCHAR(32) PRIMARY KEY, tenant_id VARCHAR(32) NOT NULL,
|
||||
blueprint_id VARCHAR(32) NOT NULL, blueprint_version INT NOT NULL DEFAULT 1,
|
||||
gd_version INT NOT NULL DEFAULT 1, content LONGTEXT NOT NULL,
|
||||
content_fingerprint CHAR(64) NOT NULL, registry_hash CHAR(64),
|
||||
compiler_version VARCHAR(32), compile_task_id VARCHAR(32),
|
||||
state VARCHAR(16) NOT NULL DEFAULT 'active',
|
||||
created_by VARCHAR(32), created_at DATETIME,
|
||||
UNIQUE KEY uk_tenant_fp (tenant_id, content_fingerprint)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS pbl_compiler_version (
|
||||
id VARCHAR(32) PRIMARY KEY, tenant_id VARCHAR(32) NOT NULL,
|
||||
blueprint_id VARCHAR(32) NOT NULL, blueprint_version INT NOT NULL,
|
||||
gd_id VARCHAR(32) NOT NULL, version_no INT NOT NULL,
|
||||
content_fingerprint CHAR(64) NOT NULL, registry_hash CHAR(64),
|
||||
change_summary TEXT, created_by VARCHAR(32), created_at DATETIME,
|
||||
UNIQUE KEY uk_tenant_bp_ver (tenant_id, blueprint_id, version_no)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS pbl_capability_registry (
|
||||
id VARCHAR(32) PRIMARY KEY, tenant_id VARCHAR(32) NOT NULL,
|
||||
cap_code VARCHAR(64) NOT NULL, cap_kind VARCHAR(32),
|
||||
schema_json TEXT, version INT NOT NULL DEFAULT 1,
|
||||
state VARCHAR(16) NOT NULL DEFAULT 'enabled',
|
||||
UNIQUE KEY uk_tenant_cap (tenant_id, cap_code, version)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS pbl_compile_task (
|
||||
id VARCHAR(32) PRIMARY KEY, tenant_id VARCHAR(32) NOT NULL,
|
||||
blueprint_id VARCHAR(32) NOT NULL, blueprint_version INT,
|
||||
trigger VARCHAR(16), state VARCHAR(16) NOT NULL DEFAULT 'pending',
|
||||
approval_state VARCHAR(16), quality_state VARCHAR(16),
|
||||
gd_id VARCHAR(32), content_fingerprint CHAR(64),
|
||||
error_code VARCHAR(32), error_message TEXT,
|
||||
started_at DATETIME, finished_at DATETIME, created_by VARCHAR(32), created_at DATETIME
|
||||
);
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 6. 自检运行方法
|
||||
|
||||
```bash
|
||||
cd modules/pbl_compiler && python3 scripts/test_m3a_selfcheck.py
|
||||
```
|
||||
退出码 0 = 全部通过；非 0 = 有断言失败（`build.sh` 第 ⑤ 步会因此中断安装）。**6 组断言**：
|
||||
|
||||
| 组 | 断言 | 覆盖需求 |
|
||||
|---|---|---|
|
||||
| 1 | `canonical_json` 对**同内容不同键序**的两个 dict 产出完全相同的串与指纹 | 29.6 sort_keys |
|
||||
| 2 | 嵌套结构（含数组/中文/浮点）规范化后**无任何空白字符**，且浮点定点稳定 | 29.6 无空白 + 浮点定点 |
|
||||
| 3 | `strip_volatile` 后修改 `created_at/id/trace_id` **指纹不变**；修改业务字段**指纹必变** | volatile 不参与指纹 |
|
||||
| 4 | `build_game_definition()` 输出**恰好 10 个顶层键**，键名集合与预期完全一致，缺失/多余即失败 | GD 契约稳定 |
|
||||
| 5 | 同一蓝图连续编译 3 次，`content_fingerprint` 三次相等；`registry_hash` 相等 | US-11 / F-CP-03 确定性 |
|
||||
| 6 | 能力注册表增删一条 `enabled` 能力 → `registry_hash` 变化；仅改 `created_at` → 不变 | registry_hash 语义正确 |
|
||||
|
||||
自检脚本**不依赖数据库与宿主应用**（纯函数级），可在 CI 中直接跑；脚本内使用防御式导入，函数名缺失时明确报 `SELFCHECK_FAIL: missing <name>` 而非静默跳过。
|
||||
|
||||
---
|
||||
|
||||
## 7. 编译主流程 `compile_blueprint()` ①-⑧（fail-closed）
|
||||
|
||||
| 步 | 动作 | 失败处理（fail-closed） |
|
||||
|---|---|---|
|
||||
| ① | **租户与入参校验**：`tenant_id` 必传，`blueprint_id` 必传 | 缺失 → `E_TENANT_REQUIRED` / `E_PARAM`，任务置 `rejected`，落审计，**不建 GD** |
|
||||
| ② | **建编译任务**：写 `pbl_compile_task`（`state=pending`→`running`，记 `trigger`） | 写库失败 → 直接抛错，不产出任何 GD |
|
||||
| ③ | **加载蓝图快照**：读 `pbl_blueprint` 聚合根 + 7 类子对象，冻结为内存快照（含 `blueprint_version`） | 蓝图不存在/已删除 → `E_BLUEPRINT_NOT_FOUND`，任务 `failed` |
|
||||
| ④ | **审批门禁**：校验 `approval_state == 'approved'` | 未审批/驳回 → `E_APPROVAL_REQUIRED`，任务 `rejected`，**绝不编译** |
|
||||
| ⑤ | **质量门禁**：取 `pbl_validation` 的 `quality_state`，低于阈值（配置化常量）拒绝 | 不达标 → `E_QUALITY_BELOW_THRESHOLD`，任务 `rejected` |
|
||||
| ⑥ | **构建 GD**：`gd_builder.build_game_definition(snapshot, registry)` → 10 顶层键文档；能力缺失即报错，不做静默降级 | 能力未注册/参数不合 schema → `E_CAPABILITY_MISSING`，任务 `failed` |
|
||||
| ⑦ | **指纹与幂等落库**：`strip_volatile` → `canonical_json` → `sha256` 得 `content_fingerprint`；`INSERT` 命中 `UNIQUE(tenant_id, content_fingerprint)` 冲突则**回读既有 GD 复用**（`reused=true`），否则新建 | 落库异常 → 事务回滚，任务 `failed`，无脏数据 |
|
||||
| ⑧ | **版本登记 + 审计 + 回填**：写 `pbl_compiler_version`（`version_no` 递增、`change_summary`），任务置 `succeeded` 并回填 `gd_id/content_fingerprint`，写 `audit_log` | 任一步失败 → 整体回滚，任务 `failed` |
|
||||
|
||||
**fail-closed 原则**：③④⑤⑥ 任一不通过 → 无 GD 产出、任务终态可追溯（`error_code` + `error_message`）、审计留痕；**不存在**「先产出再补校验」的路径。`pbl_compiler_preview` 走 ①③⑥⑦ 的**只读子集**（不写 GD、不写版本），用于前端试编译。
|
||||
|
||||
---
|
||||
|
||||
## 8. 陷阱（10 条 + 本轮 4 项退回防回归条款）
|
||||
|
||||
1. **指纹路径禁易变源**：任何 `now()/uuid/random/pid` 进入 GD 或指纹计算 → 每次编译指纹都不同 → 幂等与 US-11 验收全废。
|
||||
2. **`strip_volatile` 只剥不改**：剥离白名单外的业务字段会造成「不同蓝图同指纹」的严重误合并。
|
||||
3. **数组排序要分语义**：把有序步骤序列排序 = 编译产物语义错误；把无序能力集不排序 = 指纹抖动。
|
||||
4. **浮点必须定点化**：`0.1+0.2` 类误差或平台 repr 差异会让同内容产出不同指纹。
|
||||
5. **唯一约束是幂等的最后防线**：不要「先 SELECT 再 INSERT」当幂等（并发下会双写），必须依赖 `UNIQUE(tenant_id, content_fingerprint)` + 冲突回读。
|
||||
6. **`compiler_version` 要参与判断**：编译算法升级后旧指纹不可与新指纹混判等价，跨版本比对需显式重编译。
|
||||
7. **tenant_id 必须打头**：所有 SQL 的 WHERE 首条件是 `tenant_id`；漏写 = 跨租户数据泄露。
|
||||
8. **契约三处同步**：`.dspy` 存在但 `__init__.py` 未导出 / `init.py` 未注册 → 路由 404，且**编译期不报错**，只有运行时才暴露。
|
||||
9. **格式串占位符与参数个数必须一致**：`%d` 多写一个即 `TypeError`，会让 `build.sh` 的 RBAC 步骤整体崩溃。
|
||||
10. **README/SKILL 不得写占位或元描述**：「见磁盘」「已落盘 N 字符」这类自我指涉文本按空壳造假退回。
|
||||
|
||||
### 本轮 4 项退回防回归条款（每次交付前逐条自查）
|
||||
| 编号 | 退回问题 | 防回归硬条款 | 自查命令 |
|
||||
|---|---|---|---|
|
||||
| R1 | 声称写入的 `scripts/test_m3a_selfcheck.py` 磁盘不存在 | 该脚本**必须真实落盘并可执行**，且列入交付文件清单；声称的文件一律先 `wc -c` 复核 | `wc -c scripts/test_m3a_selfcheck.py && python3 scripts/test_m3a_selfcheck.py` |
|
||||
| R2 | 8 个已定义契约未导出/未注册（含 `verify_determinism`），`.dspy` 不可达 | api.py `__all__` / `__init__.py` 导出 / `init.py` env 注册 / `load_path.py` PATHS **四处齐全**，15 个端点一个不漏 | 见 §3 自查命令 ②③④ |
|
||||
| R3 | `load_path.py:register()` 格式串占位符错位导致 `TypeError` | 修后必须 `python3 -c` **实测跑通**再交付，禁止只改不跑 | `python3 -c "import sys;sys.path.insert(0,'scripts');import load_path;load_path.register()"` |
|
||||
| R4 | `api.py` 硬编码 `DB = 'pbl'` | 模块内**不得存在** DB 名常量，一律 `env.get_module_dbname('pbl_compiler')` 或调用方注入 | `grep -rn "DB *= *'" pbl_compiler/ \|\| echo OK` |
|
||||
|
||||
> 交付摘要中所有「已写入 / N 字符 / 已落盘」表述，**必须**与交付前 `wc -c`、`cat`、`ls` 的实测输出一致；口头声明不能替代落盘（「声称 vs 实测矛盾」按造假直接退回）。
|
||||
|
||||
---
|
||||
|
||||
## 9. 与 `skill/SKILL.md` 的双向同步
|
||||
|
||||
- `skill/SKILL.md` 是**面向 Agent 的操作技能文档**（怎么调、参数、返回、铁律、陷阱），`README.md` 是**面向开发/评审的工程文档**（定位、表、结构、安装、流程）。
|
||||
- 同步规则（改一处必改另一处）：
|
||||
1. 新增/删除/改名契约端点 → 同步 README §3 表格 + SKILL 端点清单 + `__init__.py`/`init.py`/`load_path.py`；
|
||||
2. 表结构或唯一约束变更 → 同步 README §2 + SKILL 数据契约段 + `models/*.json` + DDL；
|
||||
3. GD 顶层键或 `canonical` 规则变更 → 同步 README §4/§7 + SKILL 确定性铁律段 + 自检脚本断言（第 4/5 组）；
|
||||
4. 新增陷阱/退回条款 → 同步 README §8 + SKILL「陷阱」节。
|
||||
- 自查：`grep -c 'pbl_' skill/SKILL.md README.md`（两侧端点名应一一对应，无单侧独有）。
|
||||
|
||||
---
|
||||
|
||||
## 10. 上下游依赖清单
|
||||
|
||||
### 上游（本模块依赖）
|
||||
| 依赖 | 用途 | 缺失影响 |
|
||||
|---|---|---|
|
||||
| `pbl_common` | 租户上下文、DB 适配（`sql_rows/sql_exec/tenant_crud`）、错误码、审计写入、CRUD 工厂 | 无法取租户/无法落库/无审计，编译不可用 |
|
||||
| `pbl_blueprint` | 蓝图聚合根 + 7 类子对象 + `blueprint_version`（编译输入源） | ③ 加载快照失败，全流程 fail-closed |
|
||||
| `pbl_validation` | 14 维校验结论与 `quality_state`（⑤ 质量门禁输入） | 无法判定质量门禁，拒绝编译 |
|
||||
| `pbl_appcodes` | 枚举编码注入（状态/触发方式/能力类别等字典） | 状态码不合法，前端展示异常 |
|
||||
| 宿主应用 `ServerEnv` | `get_module_dbname('pbl_compiler')` 库名映射、契约注册环境 | 触发硬编码 DB 名禁项 |
|
||||
| 基础模块 `sqlor` / `rbac` | 标准 SQL API（仅 `sor.C/U/D/R/I/sqlExe`）、权限路径注册 | 数据访问与鉴权不可用 |
|
||||
|
||||
### 下游（依赖本模块）
|
||||
| 模块 | 消费内容 |
|
||||
|---|---|
|
||||
| `pbl_agent_runtime`（M4a/M4b） | 读 GD 作为 Designer/Critic Agent 的世界约束；调 `preview` 试编译 |
|
||||
| `pbl_evidence`（M5a/M5b） | 以 `gd_id` / `content_fingerprint` 关联产出物证据 |
|
||||
| `pbl_assessment`（M6） | 按 GD 的 rubric/目标结构做加权评估 |
|
||||
| `pbl_scense_ext`（M9）/ `scense_game` | 前端按 GD 渲染可玩场景（GD 是唯一渲染契约） |
|
||||
| `pbl_runtime_ext`（M11a/M11b） | 运行时事件与状态写入以 GD 定义的实体/规则为准 |
|
||||
| `pbl_kdb_ext`（M7） | 只读引用 GD 指纹做匿名聚合口径对齐 |
|
||||
|
||||
### 契约稳定性承诺
|
||||
GD 的 **10 个顶层键**与 `content_fingerprint` 算法是跨模块契约，属**破坏性变更**范围：任何改动必须升 `gd_version` / `compiler_version`、同步全部下游模块、并重跑 §6 自检 6 组断言。
|
||||
|
||||
@ -1,270 +1,379 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""pbl_compiler M3a 自检脚本（QC #1 整改：真实落盘，可执行）。
|
||||
"""M3a 自检脚本 —— pbl_compiler 确定性编译内核（canonical + gd_builder）。
|
||||
|
||||
运行（模块仓库根目录）::
|
||||
|
||||
运行：
|
||||
cd modules/pbl_compiler && python3 scripts/test_m3a_selfcheck.py
|
||||
（或 python3 -m pytest scripts/test_m3a_selfcheck.py -q）
|
||||
|
||||
覆盖 QC 退回意见要求的断言组（全部纯函数/纯结构，不依赖真实 DB）：
|
||||
组1 canonical 确定性：同输入 → canonical_json 字节全等 + sha256 指纹全等（US-11/F-CP-03）
|
||||
组2 GD 10 顶层键齐全（design 第 9 章）
|
||||
组3 volatile 字段（createdAt/durationMs/taskNo/id...）不参与指纹（29.6 确定性）
|
||||
组4 register() 可跑通不崩（QC #3：格式串占位符错位修复验证）
|
||||
组5 三处同步（QC #2）：__init__ 导出 == init.CONTRACTS == load_path.PATHS 契约名一致
|
||||
组6 无硬编码 DB（QC #4）：api.py 无 `DB = '...'` 常量，_dbname() 走 get_module_dbname
|
||||
组7 浮点定点 + sort_keys：键序无关、浮点 6 位定点（canonical 规范化）
|
||||
组8 registry_hash 确定性：同能力集 → 同 hash；增删能力 → hash 变化
|
||||
退出码：0 = 6 组断言全部通过；1 = 有断言失败；2 = 被测模块加载失败。
|
||||
``apps/pbls/build.sh`` 第 9 步（py_compile）之后可直接串接本脚本做部署前门禁。
|
||||
|
||||
退出码 0 = 全部通过；非 0 = 有断言失败（打印 FAIL 明细）。
|
||||
设计约束
|
||||
--------
|
||||
* **纯函数级自检**：只加载 ``pbl_compiler/canonical.py`` 与 ``pbl_compiler/gd_builder.py``
|
||||
两个纯函数模块，不连数据库、不依赖宿主应用 ``ServerEnv``、不需要 ``ahserver``
|
||||
（故用 importlib 按文件路径加载，绕开包 ``__init__.py`` 的 ahserver 依赖）。
|
||||
* **防御式导入**：被测函数缺失时打印 ``SELFCHECK_FAIL: missing <name>`` 并计入失败，
|
||||
绝不静默跳过（上一轮 QC 退回项 R1：声称的自检脚本必须真实可跑）。
|
||||
* 6 组断言与 ``README.md`` §6 表格一一对应。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import io
|
||||
import copy
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import types
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
MOD_ROOT = os.path.dirname(HERE)
|
||||
sys.path.insert(0, MOD_ROOT) # pbl_compiler 包
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(MOD_ROOT), 'pbl_common')) # pbl_common 依赖
|
||||
PKG_DIR = os.path.join(MOD_ROOT, 'pbl_compiler')
|
||||
|
||||
# ---- ahserver 桩（工作空间无平台运行时；仅为 import init.py 不崩）----
|
||||
if 'ahserver.serverenv' not in sys.modules:
|
||||
_ah = types.ModuleType('ahserver')
|
||||
_se = types.ModuleType('ahserver.serverenv')
|
||||
|
||||
class _ServerEnv(object):
|
||||
_inst = None
|
||||
|
||||
def __new__(cls):
|
||||
if cls._inst is None:
|
||||
cls._inst = super(_ServerEnv, cls).__new__(cls)
|
||||
return cls._inst
|
||||
|
||||
_se.ServerEnv = _ServerEnv
|
||||
_ah.serverenv = _se
|
||||
sys.modules['ahserver'] = _ah
|
||||
sys.modules['ahserver.serverenv'] = _se
|
||||
|
||||
from pbl_compiler import canonical as cn # noqa: E402
|
||||
from pbl_compiler import gd_builder as gb # noqa: E402
|
||||
from pbl_compiler import api as api # noqa: E402
|
||||
from pbl_compiler import init as init_mod # noqa: E402
|
||||
import pbl_compiler as pkg # noqa: E402
|
||||
|
||||
PASS, FAIL = [], []
|
||||
PASS = []
|
||||
FAIL = []
|
||||
|
||||
|
||||
def check(name, cond, detail=''):
|
||||
(PASS if cond else FAIL).append(name)
|
||||
print(' [%s] %s%s' % ('PASS' if cond else 'FAIL', name,
|
||||
(' <- ' + detail) if (detail and not cond) else ''))
|
||||
# --------------------------------------------------------------------------
|
||||
# 输出与断言工具
|
||||
# --------------------------------------------------------------------------
|
||||
def _ok(group, msg):
|
||||
PASS.append('[%s] %s' % (group, msg))
|
||||
print(' PASS [%s] %s' % (group, msg))
|
||||
|
||||
|
||||
# 一个最小但真实的蓝图版本快照（含 7 类子对象若干 + 易变字段）
|
||||
def _bad(group, msg):
|
||||
FAIL.append('[%s] %s' % (group, msg))
|
||||
print(' FAIL [%s] %s' % (group, msg))
|
||||
|
||||
|
||||
def _check(group, cond, msg):
|
||||
"""断言：True → PASS，False → FAIL。返回 bool 便于短路后续依赖断言。"""
|
||||
if cond:
|
||||
_ok(group, msg)
|
||||
else:
|
||||
_bad(group, msg)
|
||||
return bool(cond)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 按文件路径加载纯函数模块（避开 pbl_compiler/__init__.py 的 ahserver 依赖）
|
||||
# --------------------------------------------------------------------------
|
||||
def _load_pure_modules():
|
||||
"""返回 (canonical_mod, gd_builder_mod)；任一失败返回 (None, None)。"""
|
||||
pkg_name = 'pbl_compiler'
|
||||
if pkg_name not in sys.modules:
|
||||
pkg = types.ModuleType(pkg_name)
|
||||
pkg.__path__ = [PKG_DIR]
|
||||
sys.modules[pkg_name] = pkg
|
||||
|
||||
def _load(sub):
|
||||
path = os.path.join(PKG_DIR, sub + '.py')
|
||||
if not os.path.isfile(path):
|
||||
print('SELFCHECK_FAIL: missing file %s' % path)
|
||||
return None
|
||||
full = '%s.%s' % (pkg_name, sub)
|
||||
spec = importlib.util.spec_from_file_location(full, path)
|
||||
if spec is None or spec.loader is None:
|
||||
print('SELFCHECK_FAIL: cannot build import spec for %s' % path)
|
||||
return None
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
sys.modules[full] = mod
|
||||
try:
|
||||
spec.loader.exec_module(mod)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
print('SELFCHECK_FAIL: import %s raised %s: %s'
|
||||
% (full, type(exc).__name__, exc))
|
||||
return None
|
||||
setattr(sys.modules[pkg_name], sub, mod)
|
||||
return mod
|
||||
|
||||
canonical = _load('canonical')
|
||||
gd_builder = _load('gd_builder') if canonical is not None else None
|
||||
return canonical, gd_builder
|
||||
|
||||
|
||||
def _pick(mod, names, group):
|
||||
"""按候选名取可调用对象；全部缺失 → 明确报 missing 并计 FAIL。"""
|
||||
if mod is None:
|
||||
return None
|
||||
for name in names:
|
||||
fn = getattr(mod, name, None)
|
||||
if callable(fn):
|
||||
return fn
|
||||
print('SELFCHECK_FAIL: missing %s in %s'
|
||||
% ('/'.join(names), getattr(mod, '__name__', '?')))
|
||||
_bad(group, 'missing callable %s' % '/'.join(names))
|
||||
return None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 参考实现（用于交叉验证「规则本身」，不作为通过依据）
|
||||
# --------------------------------------------------------------------------
|
||||
def _ref_canonical(obj):
|
||||
return json.dumps(obj, sort_keys=True, separators=(',', ':'), ensure_ascii=False)
|
||||
|
||||
|
||||
def _ref_fingerprint(obj):
|
||||
return hashlib.sha256(_ref_canonical(obj).encode('utf-8')).hexdigest()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 测试夹具
|
||||
# --------------------------------------------------------------------------
|
||||
def _snapshot():
|
||||
"""最小可用蓝图版本快照（7 类子对象中的 4 类，键名走 M1a 落库形态）。"""
|
||||
return {
|
||||
'content': {
|
||||
'project': [{'id': 'p1', 'title': '火星基地', 'summary': '建造可持续基地'}],
|
||||
'problem': [{'id': 'q1', 'driving_question': '如何在火星自给自足？'}],
|
||||
'learning_goals': [
|
||||
{'id': 'g2', 'text': '掌握生态循环'},
|
||||
{'id': 'g1', 'text': '理解能源约束'},
|
||||
],
|
||||
'missions': [{'id': 'm1', 'title': '着陆', 'order': 1}],
|
||||
'roles': [{'id': 'r1', 'name': '工程师'}],
|
||||
'scenes': [{'id': 's1', 'name': '基地外'}],
|
||||
'entities': [{'id': 'e1', 'entity_key': 'solar_panel', 'scene_id': 's1'}],
|
||||
'events': [{'id': 'ev1', 'event_key': 'power_on', 'entity_key': 'solar_panel'}],
|
||||
'rubrics': [{'id': 'rb1', 'criterion': '可行性', 'weight': 0.5}],
|
||||
'subobjects': {
|
||||
'project': [{'code': 'pj1', 'name': '火星基地', 'title': '火星基地'}],
|
||||
'problem': [{'code': 'pb1', 'name': '氧气不足',
|
||||
'driving_question': '如何在 30 天内自给氧气？'}],
|
||||
'learning_goals': [{'code': 'g1', 'name': '光合作用', 'weight': 0.5}],
|
||||
'scenes': [{'code': 's1', 'name': '温室'},
|
||||
{'code': 's2', 'name': '控制室'}],
|
||||
'roles': [{'code': 'r1', 'name': '生物学家',
|
||||
'capability_key': 'pbl.role.skill'}],
|
||||
'missions': [{'code': 'm1', 'name': '建温室', 'seq': 1}],
|
||||
'events': [{'code': 'e1', 'name': '进入温室', 'trigger': 'enter',
|
||||
'response': [{'capability': 'pbl.item.consume'}]}],
|
||||
'artifacts': [{'code': 'a1', 'name': '氧气报告'}],
|
||||
'evidence_specs': [{'code': 'ev1', 'name': '实验记录'}],
|
||||
'rubrics': [{'code': 'rb1', 'name': '协作', 'weight': 1.0}],
|
||||
'assets': [{'code': 'as1', 'name': '温室模型.glb', 'kind': 'model'}],
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def _ctx(i=0):
|
||||
return {'blueprint_id': 101, 'blueprint_version_no': 3,
|
||||
'compiler_version': '1.0.0', 'ruleset_version': 'pbl.rules.v1',
|
||||
'rules_hash': 'deadbeefcafe1234',
|
||||
'created_at': '2000-01-01T00:00:00Z', 'duration_ms': i,
|
||||
'task_no': 'CT:101:3:1.0.0:%05d' % i}
|
||||
def _ctx(**over):
|
||||
ctx = {
|
||||
'blueprint_id': 1001,
|
||||
'blueprint_version_no': 3,
|
||||
'compiler_version': '1.0.0',
|
||||
'ruleset_version': 'pbl.rules.v1',
|
||||
'rules_hash': 'deadbeefdeadbeef',
|
||||
'created_at': '2026-09-15 09:40:04',
|
||||
'duration_ms': 7,
|
||||
'task_no': 'CT-T1-1001-3-000001',
|
||||
}
|
||||
ctx.update(over)
|
||||
return ctx
|
||||
|
||||
|
||||
def group1_canonical_determinism():
|
||||
print('组1 canonical 确定性（同输入指纹全等）')
|
||||
obj = {'b': [3, 1, 2], 'a': {'z': 1.5, 'y': 'x'}, 'n': None}
|
||||
c1 = cn.canonical_json(obj, strip=False)
|
||||
c2 = cn.canonical_json(obj, strip=False)
|
||||
check('canonical_json 同输入字节全等', c1 == c2, '%r != %r' % (c1, c2))
|
||||
f1, f2 = cn.sha256_fingerprint(obj), cn.sha256_fingerprint(obj)
|
||||
check('sha256_fingerprint 同输入全等', f1 == f2 and len(f1) == 64, f1)
|
||||
# 键序无关
|
||||
obj2 = {'a': {'y': 'x', 'z': 1.5}, 'n': None, 'b': [3, 1, 2]}
|
||||
check('键序不同 canonical 仍全等（sort_keys）',
|
||||
cn.canonical_json(obj, strip=False) == cn.canonical_json(obj2, strip=False))
|
||||
def _registry():
|
||||
return [
|
||||
{'capability_key': 'pbl.role.skill', 'category': 'role', 'version_no': 1,
|
||||
'args_schema_json': '{"level":"int"}', 'permission_required': '',
|
||||
'is_enabled': True},
|
||||
{'capability_key': 'pbl.item.consume', 'category': 'item', 'version_no': 1,
|
||||
'args_schema_json': '{"count":"int"}', 'permission_required': '',
|
||||
'is_enabled': True},
|
||||
]
|
||||
|
||||
|
||||
def group2_gd_top_keys():
|
||||
print('组2 GD 10 顶层键齐全')
|
||||
gd, fp = gb.build_game_definition(_snapshot(), _ctx(), [])
|
||||
check('GD_TOP_KEYS 恰为 10 个', len(gb.GD_TOP_KEYS) == 10, str(gb.GD_TOP_KEYS))
|
||||
missing = [k for k in gb.GD_TOP_KEYS if k not in gd]
|
||||
check('GD 含全部 10 顶层键', not missing, 'missing=%s' % missing)
|
||||
check('GD 指纹为 64 位 sha256', isinstance(fp, str) and len(fp) == 64, fp)
|
||||
check('manifest.schema = pbl.game_definition.v1',
|
||||
gd.get('manifest', {}).get('schema') == cn.GD_SCHEMA)
|
||||
|
||||
|
||||
def group3_volatile_excluded():
|
||||
print('组3 volatile 字段不参与指纹')
|
||||
gd_a, fp_a = gb.build_game_definition(_snapshot(), _ctx(0), [])
|
||||
gd_b, fp_b = gb.build_game_definition(_snapshot(), _ctx(999), [])
|
||||
check('created_at/duration_ms/task_no 不同但指纹全等', fp_a == fp_b,
|
||||
'%s != %s' % (fp_a, fp_b))
|
||||
# 直接验证 strip_volatile 剔除易变键
|
||||
dirty = {'createdAt': '2020', 'durationMs': 5, 'taskNo': 'X', 'id': 9,
|
||||
'stable': 'keep'}
|
||||
stripped = cn.strip_volatile(dirty)
|
||||
leaked = [k for k in ('createdAt', 'durationMs', 'taskNo', 'id') if k in stripped]
|
||||
check('strip_volatile 剔除全部易变键', not leaked and stripped.get('stable') == 'keep',
|
||||
'leaked=%s stripped=%s' % (leaked, stripped))
|
||||
for k in ('createdAt', 'created_at', 'durationMs', 'duration_ms',
|
||||
'taskNo', 'task_no', 'id', 'timestamp'):
|
||||
if k not in cn.VOLATILE_KEYS:
|
||||
check('VOLATILE_KEYS 含 %s' % k, False)
|
||||
return
|
||||
check('VOLATILE_KEYS 覆盖时间/耗时/任务号/id', True)
|
||||
|
||||
|
||||
def group4_register_runs():
|
||||
print('组4 register() 可跑通不崩（QC #3）')
|
||||
sys.path.insert(0, HERE)
|
||||
import importlib
|
||||
lp = importlib.import_module('load_path')
|
||||
os.environ['RBAC_SET_PERM'] = '/nonexistent/set_role_perm.py' # 强制走 pending 分支
|
||||
os.environ.pop('PY', None)
|
||||
try:
|
||||
ret = lp.register()
|
||||
crashed = False
|
||||
err = ''
|
||||
except Exception as exc: # noqa: BLE001
|
||||
crashed, ret, err = True, None, '%s: %s' % (type(exc).__name__, exc)
|
||||
check('register() 不抛 TypeError（格式串占位符已对齐）', not crashed, err)
|
||||
check('register() 返回 bool（pending 非空 → False）', isinstance(ret, bool), repr(ret))
|
||||
check('PATHS 含 15 个端点', len(lp.PATHS) == 15, str(len(lp.PATHS)))
|
||||
|
||||
|
||||
def group5_three_place_sync():
|
||||
print('组5 三处同步（QC #2）')
|
||||
api_contracts = set(n for n in api.__all__ if n.startswith('pbl_'))
|
||||
init_contracts = set(init_mod.CONTRACTS.keys())
|
||||
pkg_exports = set(n for n in pkg.__all__ if n.startswith('pbl_'))
|
||||
sys.path.insert(0, HERE)
|
||||
import importlib
|
||||
lp = importlib.import_module('load_path')
|
||||
rbac_names = set(os.path.basename(p).replace('.dspy', '') for p, _ in lp.PATHS)
|
||||
|
||||
check('init.CONTRACTS == api 契约（15 个）', init_contracts == api_contracts,
|
||||
'only_api=%s only_init=%s' % (api_contracts - init_contracts,
|
||||
init_contracts - api_contracts))
|
||||
check('__init__ 导出 == init.CONTRACTS', pkg_exports == init_contracts,
|
||||
'missing_export=%s' % (init_contracts - pkg_exports))
|
||||
check('load_path.PATHS 契约名 == init.CONTRACTS', rbac_names == init_contracts,
|
||||
'missing_rbac=%s extra_rbac=%s' % (init_contracts - rbac_names,
|
||||
rbac_names - init_contracts))
|
||||
# QC #2 点名的 8 个此前漏接线契约必须在三处都出现
|
||||
eight = ['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_diff']
|
||||
for name in eight:
|
||||
ok = (name in pkg_exports and name in init_contracts and name in rbac_names
|
||||
and callable(getattr(pkg, name, None)))
|
||||
check('QC#2 契约 %s 三处齐全且可调用' % name, ok)
|
||||
# verify_determinism 是 US-11 验收入口，必须可从包直接取到
|
||||
check('US-11 入口 pbl_compiler_verify_determinism 可达',
|
||||
callable(getattr(pkg, 'pbl_compiler_verify_determinism', None)))
|
||||
|
||||
|
||||
def group6_no_hardcoded_db():
|
||||
print('组6 无硬编码 DB（QC #4）')
|
||||
src = io.open(os.path.join(MOD_ROOT, 'pbl_compiler', 'api.py'),
|
||||
encoding='utf-8').read()
|
||||
check("api.py 无 `DB = '...'` 常量", re.search(r"(?m)^DB\s*=\s*['\"]", src) is None)
|
||||
check('api.py 无模块级 DB 属性', not hasattr(api, 'DB'))
|
||||
check('_dbname() 走 get_module_dbname 解析', callable(api._dbname)
|
||||
and isinstance(api._dbname(), str) and api._dbname())
|
||||
# AST 扫描：模块级赋值不得出现 DB = 字面量
|
||||
tree = ast.parse(src)
|
||||
bad = []
|
||||
for node in tree.body:
|
||||
if isinstance(node, ast.Assign):
|
||||
for t in node.targets:
|
||||
if isinstance(t, ast.Name) and t.id == 'DB' \
|
||||
and isinstance(node.value, ast.Constant) \
|
||||
and isinstance(node.value.value, str):
|
||||
bad.append(t.id)
|
||||
check('AST：无模块级 DB = <字符串> 硬编码', not bad, str(bad))
|
||||
# set_dbname 注入生效（宿主/测试可覆盖，证明非写死）
|
||||
old = api._dbname()
|
||||
api.set_dbname('pbl_injected')
|
||||
injected = api._dbname()
|
||||
api.set_dbname(None)
|
||||
check('set_dbname 注入可覆盖库名', injected == 'pbl_injected', injected)
|
||||
check('set_dbname(None) 回落解析值', api._dbname() == old, api._dbname())
|
||||
|
||||
|
||||
def group7_float_and_sort():
|
||||
print('组7 浮点定点 + 无空白分隔符')
|
||||
c = cn.canonical_json({'x': 1.00000049, 'y': 2.5}, strip=False)
|
||||
check('canonical 无空白分隔符', ' ' not in c and '\n' not in c, c)
|
||||
check('浮点按 FLOAT_PRECISION 定点', cn.FLOAT_PRECISION == 6)
|
||||
a = cn.canonical_json({'k': 0.1 + 0.2}, strip=False)
|
||||
b = cn.canonical_json({'k': 0.3}, strip=False)
|
||||
check('0.1+0.2 与 0.3 定点后全等', a == b, '%s vs %s' % (a, b))
|
||||
|
||||
|
||||
def group8_registry_hash():
|
||||
print('组8 registry_hash 确定性')
|
||||
caps = [{'capability_key': 'b', 'version_no': 1}, {'capability_key': 'a', 'version_no': 2}]
|
||||
h1 = cn.registry_hash(caps)
|
||||
h2 = cn.registry_hash(list(reversed(caps)))
|
||||
check('同能力集（顺序无关）registry_hash 全等', h1 == h2 and bool(h1), '%s vs %s' % (h1, h2))
|
||||
caps2 = caps + [{'capability_key': 'c', 'version_no': 1}]
|
||||
check('新增能力 registry_hash 变化', cn.registry_hash(caps2) != h1)
|
||||
def _build(f_build, snapshot, ctx, registry):
|
||||
"""调用 build_game_definition 并归一化返回值为 (gd, fp)。"""
|
||||
out = f_build(snapshot, ctx, registry)
|
||||
if isinstance(out, tuple) and len(out) == 2:
|
||||
return out[0], out[1]
|
||||
if isinstance(out, dict):
|
||||
fp = (out.get('manifest') or {}).get('fingerprint')
|
||||
return out, fp
|
||||
raise AssertionError('build_game_definition 返回类型异常: %r' % type(out).__name__)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 主流程：6 组断言
|
||||
# --------------------------------------------------------------------------
|
||||
def main():
|
||||
print('=' * 64)
|
||||
print('pbl_compiler M3a 自检（QC #1/#2/#3/#4 整改验证）')
|
||||
print('=' * 64)
|
||||
for fn in (group1_canonical_determinism, group2_gd_top_keys,
|
||||
group3_volatile_excluded, group4_register_runs,
|
||||
group5_three_place_sync, group6_no_hardcoded_db,
|
||||
group7_float_and_sort, group8_registry_hash):
|
||||
print('=== pbl_compiler M3a selfcheck (6 groups) ===')
|
||||
print('module root : %s' % MOD_ROOT)
|
||||
|
||||
canonical, gd_builder = _load_pure_modules()
|
||||
if canonical is None or gd_builder is None:
|
||||
print('SELFCHECK_FAIL: canonical/gd_builder 加载失败，自检中止')
|
||||
return 2
|
||||
|
||||
f_canon = _pick(canonical, ('canonical_json',), 'G1')
|
||||
f_strip = _pick(canonical, ('strip_volatile',), 'G3')
|
||||
f_fp = _pick(canonical, ('sha256_fingerprint', 'fingerprint'), 'G1')
|
||||
f_reghash = _pick(canonical, ('registry_hash',), 'G6')
|
||||
f_build = _pick(gd_builder, ('build_game_definition',), 'G4')
|
||||
top_keys = getattr(gd_builder, 'GD_TOP_KEYS', None)
|
||||
err_cls = getattr(canonical, 'CanonicalError', ValueError)
|
||||
|
||||
if f_canon is None or f_fp is None or f_build is None:
|
||||
print('SELFCHECK_FAIL: 核心函数缺失，自检中止')
|
||||
return 2
|
||||
|
||||
# ---------------- G1 sort_keys：键序无关 + 指纹形态 ----------------
|
||||
print('-- G1 canonical 键序无关 / 指纹形态 --')
|
||||
a = {'name': 'quest-1', 'meta': {'z': 1, 'a': 2}, 'tags': ['x', 'y']}
|
||||
b = {'tags': ['x', 'y'], 'name': 'quest-1', 'meta': {'a': 2, 'z': 1}}
|
||||
ca, cb = f_canon(a), f_canon(b)
|
||||
_check('G1', ca == cb, '同内容不同键序 → canonical_json 完全相同')
|
||||
_check('G1', f_fp(a) == f_fp(b), '同内容不同键序 → sha256_fingerprint 相同')
|
||||
fp_a = f_fp(a)
|
||||
_check('G1', isinstance(fp_a, str) and len(fp_a) == 64
|
||||
and all(ch in '0123456789abcdef' for ch in fp_a),
|
||||
'指纹为 64 位小写十六进制 sha256（实测 %s…）' % fp_a[:16])
|
||||
_check('G1', ca == _ref_canonical(a),
|
||||
'canonical_json 与参考实现(sort_keys=True + separators=(",",":"))一致')
|
||||
_check('G1', f_fp(a) != f_fp({'name': 'quest-2', 'meta': {'z': 1, 'a': 2},
|
||||
'tags': ['x', 'y']}),
|
||||
'内容变更 → 指纹必变')
|
||||
|
||||
# ---------------- G2 零空白 / 中文不转义 / 浮点定点 / 拒 NaN ----------------
|
||||
print('-- G2 零空白 / ensure_ascii=False / 浮点定点 / NaN 拒绝 --')
|
||||
nested = {'规则': {'概率': 0.5, 'n': 3, 'list': [{'b': 1.0, 'a': '中文值'}]}}
|
||||
cn = f_canon(nested)
|
||||
_check('G2', not any(ch in cn for ch in ' \n\t\r'),
|
||||
'规范化串不含任何空白字符（零空白分隔符）')
|
||||
_check('G2', '\\u' not in cn, 'ensure_ascii=False：中文原样保留不被转义')
|
||||
_check('G2', f_canon(json.loads(cn)) == cn,
|
||||
'规范化幂等：canonical(json.loads(canonical(x))) == canonical(x)')
|
||||
_check('G2', f_fp({'v': 0.1 + 0.2}) == f_fp({'v': 0.3}),
|
||||
'浮点定点化：0.1+0.2 与 0.3 指纹相同（FLOAT_PRECISION 量化）')
|
||||
_check('G2', f_fp({'v': 1.0}) == f_fp({'v': 1}),
|
||||
'整值浮点与整数指纹相同（定点化去尾零）')
|
||||
try:
|
||||
fn()
|
||||
f_canon({'v': float('nan')})
|
||||
_check('G2', False, 'NaN 应被拒绝（allow_nan=False）')
|
||||
except err_cls:
|
||||
_check('G2', True, 'NaN 触发 CanonicalError（非有限浮点不可规范化）')
|
||||
except Exception as exc: # noqa: BLE001
|
||||
import traceback
|
||||
FAIL.append('%s(异常)' % fn.__name__)
|
||||
print(' [ERROR] %s: %s' % (fn.__name__, exc))
|
||||
traceback.print_exc()
|
||||
print('-' * 64)
|
||||
print('PASS=%d FAIL=%d' % (len(PASS), len(FAIL)))
|
||||
_check('G2', False, 'NaN 抛出非预期异常 %s: %s' % (type(exc).__name__, exc))
|
||||
|
||||
# ---------------- G3 strip_volatile：易变字段不参与指纹 ----------------
|
||||
print('-- G3 strip_volatile 易变字段不参与指纹 --')
|
||||
base = {'blueprint_id': 1001, 'gd_version': 1,
|
||||
'content': {'entities': [{'code': 'e1', 'hp': 100}]}}
|
||||
v1 = dict(base, created_at='2026-09-15 09:40:04', id=11, trace_id='t-1',
|
||||
duration_ms=5, task_no='CT-1')
|
||||
v2 = dict(base, created_at='2030-01-01 00:00:00', id=99, trace_id='t-9',
|
||||
duration_ms=888, task_no='CT-9')
|
||||
_check('G3', f_fp(v1) == f_fp(v2),
|
||||
'created_at/id/trace_id/duration_ms/task_no 变化 → 指纹不变')
|
||||
_check('G3', f_fp(v1) == f_fp(base), '易变字段整体存在与否不影响指纹')
|
||||
changed = copy.deepcopy(base)
|
||||
changed['content']['entities'][0]['hp'] = 101
|
||||
_check('G3', f_fp(changed) != f_fp(base), '业务字段(hp)变化 → 指纹必变')
|
||||
if f_strip is not None:
|
||||
stripped = f_strip(v1)
|
||||
_check('G3', not any(k in stripped for k in
|
||||
('created_at', 'id', 'trace_id', 'duration_ms', 'task_no')),
|
||||
'strip_volatile 确实剥离全部易变键')
|
||||
_check('G3', stripped.get('content') == base['content'],
|
||||
'strip_volatile 只剥不改：业务子树原样保留')
|
||||
_check('G3', f_strip({'a': [{'created_at': 'x', 'k': 1}]}) == {'a': [{'k': 1}]},
|
||||
'strip_volatile 递归进入 list[dict]')
|
||||
_check('G3', f_fp(base) == _ref_fingerprint(f_strip(v1)) if f_strip else False,
|
||||
'模块指纹 == sha256(参考规范化(strip_volatile(x)))（算法口径一致）')
|
||||
|
||||
# ---------------- G4 GD 恰好 10 个顶层键 ----------------
|
||||
print('-- G4 Game Definition 10 顶层键 --')
|
||||
EXPECTED = ('manifest', 'pbl', 'world', 'scenes', 'entities', 'events',
|
||||
'states', 'capabilities', 'assessment', 'assets')
|
||||
_check('G4', tuple(top_keys or ()) == EXPECTED,
|
||||
'gd_builder.GD_TOP_KEYS == 10 键契约（实测 %s）' % (top_keys,))
|
||||
snap = _snapshot()
|
||||
try:
|
||||
gd, fp = _build(f_build, snap, _ctx(), _registry())
|
||||
except Exception as exc: # noqa: BLE001
|
||||
gd, fp = None, None
|
||||
_bad('G4', 'build_game_definition 抛异常 %s: %s' % (type(exc).__name__, exc))
|
||||
if isinstance(gd, dict):
|
||||
keys = tuple(gd.keys())
|
||||
_check('G4', len(keys) == 10, 'GD 顶层键数量 == 10（实测 %d）' % len(keys))
|
||||
_check('G4', set(keys) == set(EXPECTED),
|
||||
'GD 顶层键集合一致；缺=%s 多=%s'
|
||||
% (sorted(set(EXPECTED) - set(keys)), sorted(set(keys) - set(EXPECTED))))
|
||||
_check('G4', isinstance(fp, str) and len(fp) == 64,
|
||||
'build_game_definition 同时返回 64 位指纹')
|
||||
_check('G4', gd['manifest'].get('fingerprint') == fp,
|
||||
'manifest.fingerprint 与返回指纹一致')
|
||||
_check('G4', gd['manifest'].get('schema') == 'pbl.game_definition.v1'
|
||||
and gd['manifest'].get('deterministic') is True
|
||||
and gd['manifest'].get('llmUsed') is False,
|
||||
'manifest 声明 schema/deterministic=True/llmUsed=False（零 LLM）')
|
||||
_check('G4', gd['manifest'].get('registryHash') == f_reghash(_registry()),
|
||||
'manifest.registryHash == canonical.registry_hash(能力注册表)')
|
||||
|
||||
# ---------------- G5 同输入重编译 N 次指纹全等（US-11 / F-CP-03） ----------------
|
||||
print('-- G5 确定性：同输入重编译指纹全等 --')
|
||||
fps = []
|
||||
for i in range(3):
|
||||
try:
|
||||
_gd, _fp = _build(f_build, copy.deepcopy(snap), _ctx(), _registry())
|
||||
fps.append(_fp)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
_bad('G5', '第 %d 次编译抛异常 %s: %s' % (i + 1, type(exc).__name__, exc))
|
||||
_check('G5', len(fps) == 3 and len(set(fps)) == 1,
|
||||
'连续 3 次编译指纹全等（%s）' % (fps[0][:16] if fps else 'N/A'))
|
||||
try:
|
||||
_gd_v, fp_v = _build(f_build, copy.deepcopy(snap),
|
||||
_ctx(created_at='2031-12-31 23:59:59', duration_ms=9999,
|
||||
task_no='CT-OTHER-999'), _registry())
|
||||
_check('G5', fps and fp_v == fps[0],
|
||||
'ctx 易变字段(created_at/duration_ms/task_no)变化 → 指纹不变')
|
||||
except Exception as exc: # noqa: BLE001
|
||||
_bad('G5', '易变 ctx 编译抛异常 %s: %s' % (type(exc).__name__, exc))
|
||||
snap2 = copy.deepcopy(snap)
|
||||
subs = snap2['content']['subobjects']
|
||||
subs['scenes'] = list(reversed(subs['scenes']))
|
||||
try:
|
||||
_gd_o, fp_o = _build(f_build, snap2, _ctx(), _registry())
|
||||
_check('G5', fps and fp_o == fps[0],
|
||||
'子对象入库顺序颠倒 → 指纹不变（集合稳定化排序生效）')
|
||||
except Exception as exc: # noqa: BLE001
|
||||
_bad('G5', '乱序快照编译抛异常 %s: %s' % (type(exc).__name__, exc))
|
||||
snap3 = copy.deepcopy(snap)
|
||||
snap3['content']['subobjects']['roles'][0]['name'] = '工程师'
|
||||
try:
|
||||
_gd_c, fp_c = _build(f_build, snap3, _ctx(), _registry())
|
||||
_check('G5', fps and fp_c != fps[0], '业务内容变更 → 指纹必变')
|
||||
except Exception as exc: # noqa: BLE001
|
||||
_bad('G5', '变更快照编译抛异常 %s: %s' % (type(exc).__name__, exc))
|
||||
|
||||
# ---------------- G6 registry_hash 语义 ----------------
|
||||
print('-- G6 registry_hash 能力注册表指纹 --')
|
||||
if f_reghash is not None:
|
||||
reg = _registry()
|
||||
h1 = f_reghash(reg)
|
||||
_check('G6', isinstance(h1, str) and len(h1) == 64,
|
||||
'registry_hash 为 64 位 sha256（%s…）' % h1[:16])
|
||||
_check('G6', f_reghash(list(reversed(reg))) == h1,
|
||||
'registry_hash 与能力行顺序无关')
|
||||
reg_more = reg + [{'capability_key': 'pbl.event.broadcast', 'category': 'event',
|
||||
'version_no': 1, 'args_schema_json': '{}',
|
||||
'permission_required': '', 'is_enabled': True}]
|
||||
_check('G6', f_reghash(reg_more) != h1, '新增一条能力 → registry_hash 变化')
|
||||
reg_bump = [dict(r) for r in reg]
|
||||
reg_bump[0]['version_no'] = 2
|
||||
_check('G6', f_reghash(reg_bump) != h1, '能力 version_no 升级 → registry_hash 变化')
|
||||
reg_schema = [dict(r) for r in reg]
|
||||
reg_schema[0]['args_schema_json'] = '{"level":"str"}'
|
||||
_check('G6', f_reghash(reg_schema) != h1, 'args_schema 变更 → registry_hash 变化')
|
||||
reg_ts = [dict(r, created_at='2030-01-01 00:00:00', id=777) for r in reg]
|
||||
_check('G6', f_reghash(reg_ts) == h1,
|
||||
'仅改 created_at/id 等易变字段 → registry_hash 不变')
|
||||
_check('G6', f_reghash([]) != h1 and len(f_reghash([])) == 64,
|
||||
'空注册表有确定哈希且与非空不同')
|
||||
else:
|
||||
_bad('G6', 'canonical.registry_hash 不可用')
|
||||
|
||||
print('--- summary: pass=%d fail=%d ---' % (len(PASS), len(FAIL)))
|
||||
if FAIL:
|
||||
print('FAILED: %s' % FAIL)
|
||||
for item in FAIL:
|
||||
print('SELFCHECK_FAIL %s' % item)
|
||||
return 1
|
||||
print('ALL GREEN')
|
||||
print('SELFCHECK_OK 6 groups all passed (G1..G6)')
|
||||
return 0
|
||||
|
||||
|
||||
# ---- pytest 兼容 ----
|
||||
def test_m3a_selfcheck():
|
||||
assert main() == 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user