#!/usr/bin/env python3 # -*- coding: utf-8 -*- """pbls 离线契约冒烟测试（无 DB / 无网络，用内存 FakeDB 驱动真实 api.py 代码路径）。 覆盖硬门禁与关键红线： A 租户强制 + CRUD 白名单丢弃越权字段 B 14 维校验 + 5 级质量状态阶梯（含 blocking 不放行） C 确定性编译（draft 门禁拒 → 同输入两次指纹一致 → 改版本指纹变 → 幂等命中） D 证据幂等（同 (source_event_id, evidence_type) 二次采集 deduped，不产生第二行） E fail-closed 8 步裁决：Critic 写 / 禁用工具(pbl.publish) / 未注册 / 参数白名单 / 审批缺失 F Rubric 权重合计≠100% 拒绝；=100% 通过且 total_score 派生 G KDB 零写入 + 匿名聚合 k=5 抑制 + 个体维度拒绝 H 运行时事件 seq/state_version 单调 + 乐观锁 + 8 人上限 + 时延预算实测 I appcodes 幂等注入（二次执行 inserted=0） J 入口挂载静态核验（ServerEnv 先于 load_rbac + 契约三处同步 + 12 loader 可导入） K G6 十一条禁止实现零命中 L 前端覆盖层 12 页面 + 渲染时延预算（可执行验证方式） """ import asyncio import os import re import sys import types HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.abspath(os.path.join(HERE, *(['..'] * 4))) for _m in sorted(os.listdir(os.path.join(ROOT, 'modules'))): if _m.startswith('pbl_'): sys.path.insert(0, os.path.join(ROOT, 'modules', _m)) sys.path.insert(0, os.path.join(ROOT, 'apps', 'pbls', 'app')) # ── FakeDB：支持 INSERT/UPDATE/DELETE/SELECT(等值+字面量+LIKE+>)/COUNT/MAX/GROUP BY ── class FakeSor: def __init__(self, db): self.db = db async def sqlExe(self, sql, params=None): p = dict(params or {}) s = ' '.join(sql.split()) self.db.log.append(s[:160]) up = s.upper() if up.startswith('INSERT'): return self._insert(s, p) if up.startswith('UPDATE'): return self._update(s, p) if up.startswith('DELETE'): return self._delete(s, p) return self._select(s, p, up) # ---- INSERT ---- def _insert(self, s, p): tbl = re.search(r'INSERT INTO `(\w+)`', s).group(1) colpart = s[s.index('(') + 1:s.index(')')] cols = re.findall(r'`(\w+)`', colpart) vals = s[s.upper().index('VALUES') + 6:] ph = re.findall(r'\$\{(\w+)\}\$', vals) self.db.next_id += 1 row = {} for i, c in enumerate(cols): row[c] = p.get(ph[i]) if i < len(ph) else None row['id'] = self.db.next_id self.db.rows.setdefault(tbl, []).append(row) return [{'id': self.db.next_id}] # ---- UPDATE ---- def _update(self, s, p): tbl = re.search(r'UPDATE `(\w+)`', s).group(1) where = s[s.upper().index(' WHERE ') + 7:] if ' WHERE ' in s.upper() else '' setpart = s[s.upper().index(' SET ') + 5: s.upper().index(' WHERE ')] if ' WHERE ' in s.upper() \ else s[s.upper().index(' SET ') + 5:] sets = re.findall(r'`(\w+)`\s*=\s*\$\{(\w+)\}\$', setpart) n = 0 for row in self.db.rows.get(tbl, []): if not self._match(row, where, p): continue for col, key in sets: row[col] = p.get(key) if '`use_count` = `use_count` + 1' in setpart: row['use_count'] = int(row.get('use_count') or 0) + 1 n += 1 return [{'affected': n}] # ---- DELETE ---- def _delete(self, s, p): tbl = re.search(r'DELETE FROM `(\w+)`', s).group(1) where = s[s.upper().index(' WHERE ') + 7:] if ' WHERE ' in s.upper() else '' keep, n = [], 0 for row in self.db.rows.get(tbl, []): if self._match(row, where, p): n += 1 else: keep.append(row) self.db.rows[tbl] = keep return [{'affected': n}] # ---- SELECT ---- def _select(self, s, p, up): if 'LAST_INSERT_ID' in up: return [{'pk': self.db.next_id}] m = re.search(r'FROM `(\w+)`', s) tbl = m.group(1) if m else '?' rows = self.db.rows.get(tbl, []) where = '' if ' WHERE ' in up: w = s[s.upper().index(' WHERE ') + 7:] where = re.split(r' ORDER BY | GROUP BY | LIMIT ', w)[0] matched = [r for r in rows if self._match(r, where, p)] gm = re.search(r'GROUP BY `(\w+)`', s) if gm: col = gm.group(1) groups = {} for r in matched: groups.setdefault(r.get(col), []).append(r) out = [] for g, items in sorted(groups.items(), key=lambda x: str(x[0])): scores = [float(i.get('total_score') or 0) for i in items] out.append({'g': g, 'n': len(items), 'avg_score': sum(scores) / len(scores) if scores else 0, 'min_score': min(scores) if scores else 0, 'max_score': max(scores) if scores else 0}) return out if 'COUNT(*)' in up: return [{'c': len(matched)}] m = re.search(r'COALESCE\((MAX|COUNT)\(`(\w+)`\),0\) AS (\w+)', s) if m: fn, col, alias = m.groups() vals = [r.get(col) for r in matched if r.get(col) is not None] if fn == 'COUNT': return [{alias: len(matched)}] try: return [{alias: max(vals) if vals else 0}] except TypeError: return [{alias: 0}] m = re.search(r'MAX\(`(\w+)`\) AS (\w+)', s) if m: col, alias = m.groups() vals = [r.get(col) for r in matched if r.get(col) is not None] try: return [{alias: max(vals) if vals else 0}] except TypeError: return [{alias: 0}] if 'DISTINCT' in up: col = re.search(r'DISTINCT `(\w+)`', s).group(1) seen = [] for r in matched: if r.get(col) is not None and r[col] not in seen: seen.append(r[col]) return [{col: v} for v in seen] lim = re.search(r'LIMIT (\d+)', s) if lim: matched = matched[:int(lim.group(1))] return [dict(r) for r in matched] def _match(self, row, where, p): if not where: return True for cond in re.split(r' AND ', where): cond = cond.strip().strip('()') m = re.match(r'`(\w+)`\s*=\s*\$\{(\w+)\}\$$?', cond) if m: col, key = m.groups() want = p.get(key) if want in (None, ''): continue if str(row.get(col)) != str(want): return False continue m = re.match(r'`(\w+)`\s*=\s*\'([^\']*)\'$', cond) if m: col, want = m.groups() if str(row.get(col)) != want: return False continue m = re.match(r'`(\w+)`\s*>\s*\$\{(\w+)\}\$$?', cond) if m: col, key = m.groups() try: if float(row.get(col) or 0) <= float(p.get(key) or 0): return False except (TypeError, ValueError): return False continue m = re.match(r'`(\w+)` IN \(([^)]*)\)$', cond) if m: col, keys = m.group(1), re.findall(r'\$\{(\w+)\}', m.group(2)) vals = [p.get(k) for k in keys if p.get(k) not in (None, '')] if vals and str(row.get(col)) not in [str(v) for v in vals]: return False continue if 'LIKE' in cond.upper(): m = re.match(r'`(\w+)` LIKE \$\{(\w+)\}\$$?', cond) if m: col, key = m.groups() pat = str(p.get(key) or '').replace('%', '') if pat and pat not in str(row.get(col) or ''): return False continue return True class FakeCtx: def __init__(self, db, name): self.db, self.name = db, name async def __aenter__(self): return FakeSor(self.db) async def __aexit__(self, *a): return False class FakeDBMod: def __init__(self): self.rows, self.log, self.next_id = {}, [], 0 def sqlorContext(self, name): return FakeCtx(self, name) fdb = FakeDBMod() _ap = types.ModuleType('apppublic') _ap.db = fdb sys.modules['apppublic'] = _ap ah = types.ModuleType('ahserver') se = types.ModuleType('ahserver.serverenv') class ServerEnv: _inst = None def __new__(cls): if cls._inst is None: cls._inst = object.__new__(cls) return cls._inst def __init__(self): self.pbl_default_tenant = 'T-DEMO' se.ServerEnv = ServerEnv ah.serverenv = se sys.modules['ahserver'] = ah sys.modules['ahserver.serverenv'] = se _ge = types.ModuleType('ahserver.globalEnv') _ge.password_encode = lambda s: s sys.modules['ahserver.globalEnv'] = _ge _wa = types.ModuleType('ahserver.webapp') _wa.webapp = lambda init: None sys.modules['ahserver.webapp'] = _wa RESULTS = [] def check(name, cond, detail=''): RESULTS.append((name, bool(cond), detail)) print('%-4s %s %s' % ('PASS' if cond else 'FAIL', name, detail)) async def main(): from pbl_common import api as C from pbl_blueprint import api as BP from pbl_validation import api as VA from pbl_compiler import api as CO from pbl_agent_runtime import api as AR from pbl_evidence import api as EV from pbl_assessment import api as AS from pbl_kdb_ext import api as KD from pbl_runtime_ext import api as RT from pbl_appcodes import api as AC from pbl_scense_ext import api as SX from pbl_domain_ext import api as DX # ── A 租户 + CRUD 白名单 ──────────────────────────────────────────── r = await BP.pbl_blueprint_create(title='湿地调查', subject='science', grade='g5', evil_column='x') check('A1 蓝图创建返回 id', r.get('ok') and r.get('id'), str(r.get('id'))) check('A2 越权字段被丢弃不入库', r.get('dropped_fields') == ['evil_column'], str(r.get('dropped_fields'))) check('A3 租户上下文注入', r.get('tenant_id') == 'T-DEMO', str(r.get('tenant_id'))) rd = await BP.pbl_blueprint_read(id=r['id']) check('A4 聚合读含 7 类子对象槽', set(rd['subobjects']) == { 'learning_goal', 'role', 'mission', 'task', 'artifact_spec', 'evidence_spec', 'reflection_spec'}, str(sorted(rd['subobjects']))) # ── B 14 维校验 + 5 级质量状态 ───────────────────────────────────── for i, (code, title) in enumerate([('LG1', '理解生态链'), ('LG2', '数据建模'), ('LG3', '公众表达')]): await BP.pbl_blueprint_subobject_save(kind='learning_goal', blueprint_id=r['id'], code=code, title=title, bloom_level='analyze', measurable=1, order_no=i + 1) await BP.pbl_blueprint_update(id=r['id'], content_json={ 'schema': 'pbl.blueprint.v1', 'driving_question': '我们如何拯救社区湿地？', 'scenario': '本地湿地富营养化，社区需要一份可执行方案', 'learning_goals': [{'uid': 'LG1', 'code': 'LG1'}, {'uid': 'LG2', 'code': 'LG2'}, {'uid': 'LG3', 'code': 'LG3'}], 'rubric_criteria': [{'code': 'C1', 'learning_goal_id': 'LG1', 'weight': 40}, {'code': 'C2', 'learning_goal_id': 'LG2', 'weight': 35}, {'code': 'C3', 'learning_goal_id': 'LG3', 'weight': 25}]}) await BP.pbl_blueprint_subobject_save(kind='mission', blueprint_id=r['id'], code='MS1', title='拯救湿地', scenario_txt='富营养化现场', order_no=1) await BP.pbl_blueprint_subobject_save(kind='role', blueprint_id=r['id'], code='R1', title='生态调研员', headcount=3, order_no=1) await BP.pbl_blueprint_subobject_save(kind='role', blueprint_id=r['id'], code='R2', title='数据分析师', headcount=3, order_no=2) await BP.pbl_blueprint_subobject_save(kind='artifact_spec', blueprint_id=r['id'], code='A1', title='治理方案', artifact_type='document', order_no=1) await BP.pbl_blueprint_subobject_save(kind='evidence_spec', blueprint_id=r['id'], code='E1', evidence_type='artifact', min_count=2, rubric_criterion_uid='C1', order_no=1) await BP.pbl_blueprint_subobject_save(kind='evidence_spec', blueprint_id=r['id'], code='E2', evidence_type='reflection', min_count=1, rubric_criterion_uid='C2', order_no=2) await BP.pbl_blueprint_subobject_save(kind='reflection_spec', blueprint_id=r['id'], code='RF1', trigger='mission.completed', prompt_txt='你学到了什么？', min_words=50, evidence_type='reflection', order_no=1) v = await VA.pbl_validation_run(blueprint_id=r['id'], apply_state=1) check('B1 校验 14 维全覆盖', v['dimension_count'] == 14, str(v['dimension_count'])) check('B2 逐维可解释结论齐备', len(v['findings']) == 14 and all( 'detail' in f for f in v['findings']), str(len(v['findings']))) check('B3 质量状态在 5 级阶梯内', v['quality_state'] in ['draft', 'needs_work', 'pbl_ready', 'playtest_ready', 'publish_ready'], '%s score=%s' % (v['quality_state'], v['score'])) check('B4 校验时延预算 ≤2000ms 实测', v['within_budget'], '%sms' % v['latency_ms']) check('B5 blocking 未清零不得跳级', not (v['blocking'] > 0 and v['quality_state'] == 'publish_ready'), 'blocking=%s' % v['blocking']) vfull = await VA.pbl_validation_run(blueprint_id=r['id']) check('B6 重复校验产生新记录（历史不覆盖）', vfull['ok'], 'idempotency=n/a') # ── C 确定性编译 ──────────────────────────────────────────────────── await CO.pbl_capability_register(capability_key='pbl.mission.show', category='mission', is_enabled=1) await CO.pbl_capability_register(capability_key='pbl.artifact.submit', category='artifact', is_enabled=1) try: await CO.pbl_compiler_compile(blueprint_id=r['id']) check('C0 draft/未达门禁编译被拒', False, '未抛错') except C.PblError as e: check('C0 质量门禁拒编译', e.code == 'PBL_COMPILE_GATE_FAILED', e.code) await BP.pbl_blueprint_version_create(blueprint_id=r['id']) c1 = await CO.pbl_compiler_compile(blueprint_id=r['id'], min_quality_state='draft') c2 = await CO.pbl_compiler_compile(blueprint_id=r['id'], min_quality_state='draft') check('C1 编译产出 SHA-256 指纹', len(str(c1.get('content_fingerprint'))) == 64, str(c1.get('content_fingerprint'))[:16]) check('C2 同输入+同编译版本 → 同指纹（29.6）', c1['content_fingerprint'] == c2['content_fingerprint'], '%s ==' % c1['content_fingerprint'][:12]) check('C2b 同指纹命中幂等（不重复建行）', c2.get('idempotent') is True, str(c2.get('note'))[:20]) pv = await CO.pbl_compiler_preview(blueprint_id=r['id']) check('C3 干跑不落库且指纹一致', pv['content_fingerprint'] == c1['content_fingerprint'] and pv['preview'], 'entities=%s' % len(pv['definition']['entities'])) check('C3b 事件映射为 script_type=1 规则 JSON（不建规则引擎）', all(b['script_type'] == 1 for b in pv['definition']['script_bindings']) and len(pv['definition']['script_bindings']) == len(pv['definition']['events']), 'bindings=%s' % len(pv['definition']['script_bindings'])) await BP.pbl_blueprint_update(id=r['id'], content_json={'schema': 'pbl.blueprint.v1', 'missions': [{'obj_uid': 'M9'}]}) await BP.pbl_blueprint_version_create(blueprint_id=r['id']) c3 = await CO.pbl_compiler_compile(blueprint_id=r['id'], blueprint_version_no=2, min_quality_state='draft', force_new=1) check('C4 改输入 → 指纹变（非缓存假阳性）', c3['content_fingerprint'] != c1['content_fingerprint'], 'v2 fp 不同') cmp_ = await CO.pbl_compiler_compare(a=c1['id'], b=c3['id']) check('C5 两版定义结构化 diff 可用', cmp_['ok'] and cmp_['identical'] is False, 'added=%s removed=%s' % (len(cmp_['diff']['added']), len(cmp_['diff']['removed']))) check('C6 编译时延预算 ≤5000ms 实测', c3['latency_ms'] <= c3['budget_ms'], '%sms' % c3['latency_ms']) mat = await DX.pbl_domain_materialize_game_definition(id=c3['id']) check('C7 物化走基座契约/降级如实回报', mat['ok'] and 'degraded' in mat and mat.get('base_tables_altered') is False, 'degraded=%s' % mat.get('degraded')) # ── E fail-closed 8 步裁决 ────────────────────────────────────────── await AR.pbl_tool_registry_save(tool_key='pbl.get', title='读蓝图', enabled=1, agent_scope='both') await AR.pbl_tool_registry_save(tool_key='pbl.update', title='改蓝图', enabled=1, agent_scope='designer', params_schema_json={'properties': {'blueprint_id': {}, 'content_json': {}}}) await AR.pbl_tool_registry_save(tool_key='pbl.publish', title='发布', enabled=0, agent_scope='none', disabled_reason='G6_no_publish_tool') from pbl_common.api import sql_exec as _x for code, mode in (('designer', 'write'), ('critic', 'read')): await _x('INSERT INTO `pbl_agent_def` (`tenant_id`,`code`,`agent_type`,`name`,' '`model_route`,`permission_mode`,`is_enabled`,`version_no`,`created_at`)' ' VALUES (${t}$,${c}$,${a}$,${n}$,${r}$,${m}$,${en}$,${vn}$,${ts}$)', {'t': 'T-DEMO', 'c': code, 'a': code, 'n': code, 'r': 'pipeline-llm', 'm': mode, 'en': 1, 'vn': 1, 'ts': '2026-09-15 10:00:00'}, 'pbl') d1 = await AR.pbl_tool_adjudicate(agent_code='critic', tool_key='pbl.update', params_json={'blueprint_id': 1}) check('E1 Critic 写工具 DENIED（零写权限）', d1['allowed'] is False and d1['error_code'] == 'AD04_PERMISSION_MODE_DENIED', d1['error_code']) d1b = await AR.pbl_tool_adjudicate(agent_code='critic', tool_key='pbl.get', params_json={}) check('E1b Critic 读工具放行', d1b['allowed'] is True, str(d1b['steps'][-1])) d2 = await AR.pbl_tool_adjudicate(agent_code='designer', tool_key='pbl.publish', params_json={}) check('E2 pbl.publish 恒拒（G6 无绕过）', d2['allowed'] is False and d2['error_code'] == 'AD03_TOOL_DISABLED', d2['error_code']) d3 = await AR.pbl_tool_adjudicate(agent_code='designer', tool_key='evil.tool', params_json={}) check('E3 未注册工具 DENIED', d3['error_code'] == 'AD02_TOOL_UNREGISTERED', d3['error_code']) d4 = await AR.pbl_tool_adjudicate(agent_code='designer', tool_key='pbl.update', params_json={'evil': 1}) check('E4 参数不在服务端白名单 DENIED', d4['error_code'] == 'AD05_PARAM_NOT_WHITELISTED', d4['error_code']) d5 = await AR.pbl_tool_adjudicate(agent_code='designer', tool_key='pbl.update', params_json={'blueprint_id': 1}) check('E5 需人工审批且无 approved 单 → DENIED', d5['error_code'] == 'AD06_APPROVAL_REQUIRED', d5['error_code']) ap = await AR.pbl_approval_create(approval_type='modify_learning_goal', target_type='pbl_blueprint', target_id=r['id']) check('E5b 审批单创建为 pending', ap.get('ok') and ap.get('approval_uid'), str(ap.get('approval_uid'))[:12]) dec = await AR.pbl_approval_decide(id=ap['id'], status='approved', comment_txt='同意') check('E5c 人工决策后放行（第6步通过）', dec.get('ok') is True, dec.get('status')) try: await AR.pbl_approval_create(approval_type='auto_publish', target_id=1) check('E6 非法审批类型被拒', False, '未抛错') except C.PblError as e: check('E6 非法审批类型被拒', e.code == 'PBL_APPROVAL_TYPE_UNKNOWN', e.code) try: await AR.pbl_tool_registry_save(tool_key='sim.configure', enabled=1) check('E7 G6 禁用工具不得启用', False, '未抛错') except C.PblError as e: check('E7 G6 禁用工具不得启用', e.code == 'PBL_TOOL_FORBIDDEN', e.code) tl = await AR.pbl_tool_registry_list() check('E8 13 启用 / 9 禁用清单齐备', len(tl['enabled_keys']) == 13 and len(tl['disabled_keys']) == 9 and 'pbl.publish' in tl['disabled_keys'], str(tl['counts'])) cr = await AR.pbl_agent_critic_run(blueprint_id=r['id']) check('E9 Critic 输出四要素且 writable=False', cr['writable'] is False and all(k in cr for k in ('recommendation', 'reason', 'evidence', 'confidence')), 'conf=%s' % cr['confidence']) dr = await AR.pbl_agent_designer_run(input_txt='给我做一个湿地 PBL', subject='science', grade='g5', template_code='TPL-NOT-EXIST') check('E10 LLM 不可达 → 兜底路径闭环不中断', dr['ok'] and dr['generation_source'] == 'template_fallback', 'err=%s %sms' % (dr['llm_error'], dr['latency_ms'])) tr = await AR.pbl_agent_trace_list() check('E11 轨迹留痕（7 要素）可读', tr['ok'] and tr['total'] >= 2, 'traces=%s' % tr['total']) # ── D 证据幂等 ────────────────────────────────────────────────────── a1 = await EV.pbl_artifact_create(title='湿地报告', session_id=1, blueprint_id=r['id'], content_json={'a': 1}) e1 = await EV.pbl_evidence_collect(evidence_type='artifact', source_event_id='EV-1', session_id=1, learner_id='U1', artifact_id=a1['id']) e2 = await EV.pbl_evidence_collect(evidence_type='artifact', source_event_id='EV-1', session_id=1, learner_id='U1', artifact_id=a1['id']) check('D1 首次采集落库', e1.get('deduped') is False, str(e1.get('id'))) check('D2 重复采集幂等（唯一索引）', e2.get('deduped') is True and e2.get('id') == e1.get('id'), '%s==%s' % (e1.get('id'), e2.get('id'))) check('D2b 幂等键为 (tenant,source_event_id,evidence_type)', 'uk_ev_dedup' in e1.get('unique_index', ''), e1.get('unique_index')) try: await EV.pbl_evidence_collect(evidence_type='artifact') check('D3 缺 source_event_id 拒绝', False, '未抛错') except C.PblError as e: check('D3 缺 source_event_id 拒绝', e.code == 'PBL_PARAM_MISSING', e.code) try: await EV.pbl_evidence_collect(evidence_type='ghost', source_event_id='X') check('D3b 非法证据类型拒绝', False, '未抛错') except C.PblError as e: check('D3b 非法证据类型拒绝', e.code == 'PBL_EVIDENCE_TYPE_INVALID', e.code) au = await EV.pbl_artifact_update(id=a1['id'], content_json={'a': 2}) check('D4 产出物更新版本递增', au.get('ok'), 'v=%s' % au.get('version_no')) el = await EV.pbl_evidence_list(session_id=1, learner_id='U1') check('D5 证据按类型聚合可读', el['ok'] and 'by_type' in el, str(el['by_type'])) # ── F Rubric ──────────────────────────────────────────────────────── try: await AS.pbl_rubric_save(title='湿地评分', blueprint_id=r['id'], criteria_json=[ {'code': 'C1', 'weight': 60, 'learning_goal_id': 'LG1', 'max_points': 5}]) check('F1 权重≠100% 拒绝', False, '未抛错') except C.PblError as e: check('F1 权重≠100% 拒绝', e.code == 'PBL_RUBRIC_WEIGHT_INVALID', e.code) try: await AS.pbl_rubric_save(title='x', blueprint_id=r['id'], criteria_json=[ {'code': 'C1', 'weight': 100}]) check('F1b 准则缺 learning_goal_id 拒绝', False, '未抛错') except C.PblError as e: check('F1b 准则缺 learning_goal_id 拒绝', e.code == 'PBL_RUBRIC_ALIGNMENT_MISSING', e.code) ru = await AS.pbl_rubric_save(title='湿地评分', blueprint_id=r['id'], criteria_json=[ {'code': 'C1', 'weight': 60, 'learning_goal_id': 'LG1', 'max_points': 5}, {'code': 'C2', 'weight': 40, 'learning_goal_id': 'LG2', 'max_points': 5}]) check('F2 权重=100% 通过', ru.get('ok') and ru.get('weight_total_ok'), str(ru.get('id'))) sc = await AS.pbl_assessment_score(rubric_id=ru['id'], learner_id='U1', scores_json={'C1': 4, 'C2': 3}) check('F3 加权总分派生（4/5×60 + 3/5×40 = 72）', abs(sc['total_score'] - 72.0) < 0.01, '%s band=%s' % (sc['total_score'], sc['band'])) check('F3b 客户端不可直传 total_score', 'total_score' not in [k for k in sc.keys() if k == 'total_score'] or sc.get('derived'), str(sc.get('derived'))[:24]) rep = await AS.pbl_assessment_report(learner_id='U1') check('F4 教师报告可读', rep.get('ok') and 'learners' in rep, str(rep.get('record_count'))) # ── G KDB 只读桩 ──────────────────────────────────────────────────── ks = await KD.pbl_kdb_search(keyword='湿地') check('G1 KDB 检索为只读桩', ks.get('writable') is False and 'stub' in ks, ks.get('stub')) check('G2 pbl_kdb_item 零 INSERT', not any('INSERT INTO `pbl_kdb_item`' in x for x in fdb.log), 'log clean') ag = await KD.pbl_research_aggregate(group_by='band') check('G3 匿名聚合只读 + k_min=5', ag.get('anonymous') and ag.get('written') is False and ag.get('k_min') == 5, 'groups=%s suppressed=%s' % (len(ag['groups']), ag['suppressed_groups'])) try: await KD.pbl_research_aggregate(group_by='learner_id') check('G4 个体维度聚合被拒', False, '未抛错') except C.PblError as e: check('G4 个体维度聚合被拒', e.code == 'PBL_AGG_DIM_INVALID', e.code) check('G5 无 kdb.write/research.write/sim.* 实现入口', not any(hasattr(KD, n) for n in ('kdb_write', 'pbl_kdb_write', 'pbl_research_write', 'pbl_sim_configure', 'pbl_kdb_add_candidate')), '零实现入口') # ── H 运行时（服务端权威）─────────────────────────────────────────── m1 = await RT.pbl_session_member_add(session_id=1, user_id='U1') m2 = await RT.pbl_session_member_add(session_id=1, user_id='U1') check('H1 成员重复加入幂等', m2.get('deduped') is True, str(m2.get('members'))) st = await RT.pbl_entity_state_apply(session_id=1, entity_id='mission:M1', state_json={'completed': True}, event_type='mission.completed', state_version=999) check('H2 客户端 state_version 被忽略', st.get('client_version_ignored') is True and st['state_version'] == 1, 'v=%s' % st['state_version']) st2 = await RT.pbl_entity_state_apply(session_id=1, entity_id='mission:M1', state_json={'x': 1}) check('H3 state_version 单调递增', st2['state_version'] == 2, str(st2['state_version'])) check('H4 单事务 + 裁决/端到端时延实测在预算内', st2['within_budget'] is True and st2['transaction'] == 'single', json.dumps(st2['measured'])) check('H4b 广播时延预算 ≤300ms 实测', st2['broadcast']['within_budget'], '%sms ch=%s' % (st2['broadcast']['latency_ms'], st2['broadcast']['channel'])) pl = await RT.pbl_runtime_event_poll(session_id=1, since_seq=0) check('H5 3s 轮询兜底可拉增量事件', pl['count'] >= 1 and pl['last_seq'] >= 1, 'events=%s interval=%ss' % (pl['count'], pl['poll_interval_seconds'])) try: await RT.pbl_entity_state_apply(session_id=1, entity_id='mission:M1', state_json={'y': 2}, if_state_version=1) check('H6 乐观锁冲突被拒', False, '未抛错') except C.PblError as e: check('H6 乐观锁冲突被拒', e.code == 'PBL_STATE_CONFLICT', e.code) for i in range(2, 9): await RT.pbl_session_member_add(session_id=1, user_id='U%d' % i) try: await RT.pbl_session_member_add(session_id=1, user_id='U99') check('H7 超 8 人上限被拒', False, '未抛错') except C.PblError as e: check('H7 超 8 人上限被拒', e.code == 'PBL_SESSION_FULL', e.code) check('H8 事件表无 UPDATE/DELETE 出口（append-only）', not any(x.startswith('UPDATE `pbl_runtime_event`') or x.startswith('DELETE FROM `pbl_runtime_event`') for x in fdb.log), 'clean') # ── I appcodes 幂等 ───────────────────────────────────────────────── i1 = await AC.pbl_appcodes_inject() i2 = await AC.pbl_appcodes_inject() check('I1 6+2 组枚举注入', i1['groups'] == 8 and i1['inserted'] > 0, 'inserted=%s' % i1['inserted']) check('I2 二次执行零新增（幂等）', i2['inserted'] == 0 and i2['skipped'] == i1['inserted'], 'skipped=%s' % i2['skipped']) cl = await AC.pbl_appcodes_list() check('I3 枚举组可读回', len(cl['groups']) >= 1, str(sorted(cl['groups'])[:3])) # ── J 入口挂载静态核验 ────────────────────────────────────────────── src = open(os.path.join(ROOT, 'apps', 'pbls', 'app', 'pbls.py'), encoding='utf-8').read() i_env, i_rbac = src.index('env = ServerEnv()'), src.index('load_rbac()') check('J1 ServerEnv() 先于 load_rbac()', 0 < i_env < i_rbac, 'env@%d < rbac@%d' % (i_env, i_rbac)) check('J2 基础模块显式 loader 调用', all(x in src for x in ('load_appbase()', 'load_rbac()', 'load_pybricks()')), 'ok') check('J2b 无 importlib 只 import 即标 loaded 的反模式', 'importlib' not in src or 'getattr(pkg' in src, 'ok') bad = [] mods = ('pbl_common', 'pbl_appcodes', 'pbl_blueprint', 'pbl_validation', 'pbl_compiler', 'pbl_agent_runtime', 'pbl_evidence', 'pbl_assessment', 'pbl_kdb_ext', 'pbl_domain_ext', 'pbl_scense_ext', 'pbl_runtime_ext') for mod in mods: pkg = os.path.join(ROOT, 'modules', mod, mod) init_src = open(os.path.join(pkg, 'init.py'), encoding='utf-8').read() exp_src = open(os.path.join(pkg, '__init__.py'), encoding='utf-8').read() api_src = open(os.path.join(pkg, 'api.py'), encoding='utf-8').read() for n in re.findall(r'async def (pbl_\w+)', api_src): if ('env.%s = %s' % (n, n)) not in init_src or n not in exp_src: bad.append('%s:%s' % (mod, n)) check('J3 契约三处同步（定义+导出+env注册）', not bad, '缺=%s' % (bad[:5] or '无')) check('J4 12 模块 load_* 均可导入', all(hasattr(__import__('importlib').import_module('%s.init' % m), 'load_%s' % m) for m in mods), 'ok') check('J5 模块嵌套包结构 modules/{m}/{m}/', all(os.path.isfile(os.path.join(ROOT, 'modules', m, m, 'api.py')) for m in mods), 'ok') check('J6 模块无 app.py / 无独立端口（模块非部署单元）', not any(os.path.exists(os.path.join(ROOT, 'modules', m, 'app.py')) or os.path.exists(os.path.join(ROOT, 'modules', m, 'app', '%s.py' % m)) for m in mods), 'ok') # ── K G6 十一条禁止实现 ───────────────────────────────────────────── g6 = ['pbl_publish', 'marketplace', 'showplace_publish', 'sim_configure', 'kdb_add_candidate', 'kdb_propose_pattern', 'research_write', 'crdt', 'frame_sync', 'unity', 'sso_billing', 'autonomous_agent'] hits = {} for term in g6: n = 0 for mod in mods: pkg = os.path.join(ROOT, 'modules', mod, mod) for fn in os.listdir(pkg): if not fn.endswith('.py'): continue t = open(os.path.join(pkg, fn), encoding='utf-8').read() n += len(re.findall(r'(?:async def|def)\s+\w*%s\w*\s*\(' % term, t, re.I)) hits[term] = n check('K1 G6 十一条禁止实现零命中（无函数定义）', all(v == 0 for v in hits.values()), str({k: v for k, v in hits.items() if v}) or 'all 0') # ── L 前端覆盖层 ──────────────────────────────────────────────────── pr = await SX.pbl_page_registry_list() check('L1 12 页面注册齐备', pr['page_count'] == 12, str(pr['page_count'])) world = [p for p in pr['data'] if p['widget_type'] == 'world_overlay' or p['page_key'] in ('pbl_play_preview', 'pbl_mission_board', 'pbl_role_panel', 'pbl_artifact_studio', 'pbl_evidence_wall')] check('L2 世界内页面预算 500ms', all(int(p['budget_ms']) == 500 for p in world), '%s pages' % len(world)) check('L2b 首屏预算 1500ms', pr['budgets']['first_screen_ms'] == 1500, json.dumps(pr['budgets'])) bs = await SX.pbl_overlay_bootstrap(session_id=1) check('L3 覆盖层一次请求拿齐上下文 + 时延实测', bs['ok'] and len(bs['pages']) == 12 and bs['within_budget'], '%sms budget=%sms' % (bs['latency_ms'], bs['budget_ms'])) fb = await SX.pbl_playtest_feedback_save(session_id=1, blueprint_id=r['id'], feedback_type='reflection', content_txt='我学会了用数据说服社区', rating=4) check('L4 反思写入并联动证据链', fb.get('ok'), str(fb.get('id'))) try: await SX.pbl_playtest_feedback_save(feedback_type='ghost') check('L4b 非法反馈类型被拒', False, '未抛错') except C.PblError as e: check('L4b 非法反馈类型被拒', e.code == 'PBL_FEEDBACK_TYPE_INVALID', e.code) ctx = await SX.pbl_session_context_get(session_id=1) check('L5 会话上下文只读（authority=server）', ctx['authority'] == 'server' and ctx['max_members'] == 8, 'members=%s' % ctx['count']) fails = [x for x in RESULTS if not x[1]] print('\n==== 结果：%d 项，通过 %d，失败 %d ====' % (len(RESULTS), len(RESULTS) - len(fails), len(fails))) for name, _ok, detail in fails: print(' FAIL', name, detail) return 1 if fails else 0 if __name__ == '__main__': sys.exit(asyncio.get_event_loop().run_until_complete(main()))