#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M5a 自检脚本（pbl_evidence 模块交付门禁）。 用法 ---- cd modules/pbl_evidence && python3 scripts/selfcheck_m5a.py # 全部通过 → 逐条打印 [PASS]，以 `ALL PASS (N/N checks)` 收尾，exit 0 # 任一失败 → 打印 [FAIL] + 原因，exit 1 幂等契约（本脚本自身） -------------------- 本脚本**只读**：不连库、不写文件、不打印时间戳/随机数/绝对路径，所有明细一律 `sorted()` 后输出。因此重复执行输出逐字节一致（`diff <(run) <(run)` 为空即证）。 校验项（与任务书一一对应） ------------------------ 1. 三处同步：包内定义符号 ⊇ __init__.py __all__ ⊇ init.py env 注册符号，且 env 注册数 ≥ 10 2. RBAC：wwwroot/api/*.dspy 文件名集合 ⊇ wwwroot/index.ui 声明的 api 路径（≥9），且均已在 scripts/load_path.py 显式登记 3. 模型四段式：models/*.json 每文件含 summary(primary 为非空数组)/fields/indexes/codes，且至少 1 个 unique 约束 4. 幂等键语义：uk 列组合 == (tenant_id, source_event_id, evidence_type)；dedup_key 在 editexclouded；新增走采集端点 5. pyproject.toml dependencies 不含 apppublic/ahserver/appbase/rbac（基础包由宿主 build.sh 安装） 6. init/data.json 存在且 json.load 通过（禁占位词） 7. C-3 fail-closed：resolve_event_table() 返回空时 collector 必须 raise CollectError 且消息含 'M11b' 8. README 引用的 scripts/*.py 路径全部在磁盘命中（含本脚本自身） 9. harness 静态门禁：main() 外层有 try/except 收敛未捕获异常；raw()/q_all()/q_exec() 不再出现 `args or {}` / `dict(params or {})` 反模式；种子 INSERT 用具名参数 `%(event_id)s` 而非位置 `%s` 10. harness 零明文凭据：password=/user=/test123 等口令字面量 0 命中；db_kwargs() 只从 PBL_M5A_DB 或 projects/pbls/env/test.json 取凭据，取不到即 raise（禁 fallback 到硬编码口令） 11. 运行证据日志：projects/pbls/docs/02-develop/ 下 m5a-live-db-harness.log / m5a-selfcheck.log / m5a-pytest-offline.log 三份均存在且含 `# RC=0` 收尾标记；harness 日志额外含 `LIVE DB HARNESS ALL PASS` 与 C-3.1~C-3.4、IDM.1~IDM.9 全 [PASS] 行； .qc_tmp/ 下的崩溃 traceback 日志不计为证据 12. harness DDL 与模型一致：DDL_PBL_EVIDENCE 解析出的列集合 ⊆ models/pbl_evidence.json fields， 且 UNIQUE KEY uk_ev_dedup 列组合 == (tenant_id, source_event_id, evidence_type) 关于「四段式」口径：database-table-definition-spec 规定的四段是 summary/fields/indexes/codes（任务书写的 primary/columns/indexes/constraints 是其 语义等价映射：primary→summary[0].primary、columns→fields、indexes→indexes、 constraints→indexes 中 idxtype='unique' 的唯一约束）。本脚本按规范口径校验， 并逐条断言映射后的四项语义成立。 """ import ast import glob import json import os import re import sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) PKG_DIR = os.path.join(ROOT, 'pbl_evidence') WWWROOT = os.path.join(ROOT, 'wwwroot') MODELS_DIR = os.path.join(ROOT, 'models') JSON_DIR = os.path.join(ROOT, 'json') HARNESS_PATH = os.path.join(ROOT, 'tests', 'm5a_live_db_harness.py') # 期望的幂等键三元组（C-3 硬约束） UK_EXPECTED = ['tenant_id', 'source_event_id', 'evidence_type'] # 禁止出现在 pyproject dependencies 的基础包（非 PyPI，由宿主 build.sh git 安装） FORBIDDEN_DEPS = ('apppublic', 'appPublic', 'ahserver', 'appbase', 'rbac', 'accounting') # 三处同步的 10 个对外契约函数（api.py 定义 → __init__ 导出 → init.py 挂载） CONTRACT_APIS = ( 'pbl_artifact_create', 'pbl_artifact_read', 'pbl_artifact_update', 'pbl_artifact_delete', 'pbl_artifact_list', 'pbl_evidence_collect', 'pbl_evidence_collect_from_events', 'pbl_evidence_list', 'pbl_evidence_stats', 'pbl_evidence_watermark', ) MIN_ENV_REGISTRATIONS = 10 MIN_UI_API_REFS = 9 # ── 第 9~12 项门禁用的常量 ──────────────────────────────────────────── # 项目过程文档落点（相对机构工作空间，见 project-directory-spec 第五章） DOCS_REL = os.path.join('projects', 'pbls', 'docs', '02-develop') # 必须存在的三份运行证据日志（只认 docs/02-develop 顶层，.qc_tmp/ 下的一律不算） EVIDENCE_LOGS = ( 'm5a-live-db-harness.log', 'm5a-selfcheck.log', 'm5a-pytest-offline.log', ) RC_MARKER = '# RC=0' HARNESS_ALL_PASS_TOKEN = 'LIVE DB HARNESS ALL PASS' # harness 必须逐条 [PASS] 的断言编号（C-3 fail-closed 4 条 + 幂等 9 条） HARNESS_CHECK_TOKENS = ( 'C-3.1', 'C-3.2', 'C-3.3', 'C-3.4', 'IDM.1', 'IDM.2', 'IDM.3', 'IDM.4', 'IDM.5', 'IDM.6', 'IDM.7', 'IDM.8', 'IDM.9', ) # 明文口令字面量黑名单（harness 源码里 0 命中才算过） CREDENTIAL_PATTERNS = ( r"password\s*=\s*['\"]", r'user\s*=\s*[\'"]test[\'"]', r"['\"]test123['\"]", r"passwd\s*=\s*['\"]", ) # 数据访问层「静默吞参」反模式（会把 None 变成 {} 掩盖调用方漏传参的缺陷） PARAM_SWALLOW_PATTERNS = ('args or {}', 'params or {}', 'dict(params or {})', 'or {}') # harness 里需要收敛为断言的 phase 调用（main 外层 try 必须包住它们） HARNESS_PHASE_CALLS = ('install_stub', 'phase_fail_closed', 'phase_idempotent', 'phase_third_run') # 崩溃 traceback 特征（出现即说明该日志是崩溃现场，不是通过证据） TRACEBACK_TOKEN = 'Traceback (most recent call last)' def _read(path): with open(path, 'r', encoding='utf-8') as fh: return fh.read() def _rel(path): return os.path.relpath(path, ROOT).replace(os.sep, '/') def _pkg_py_files(): return sorted(glob.glob(os.path.join(PKG_DIR, '*.py'))) def _top_level_symbols(path): """一个 .py 文件里顶层定义的函数/类/模块级常量名集合。""" tree = ast.parse(_read(path), filename=path) out = set() for node in tree.body: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): out.add(node.name) elif isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Name): out.add(tgt.id) return out def _dunder_all(path): """解析 __init__.py 里的 __all__ 列表字面量。""" tree = ast.parse(_read(path), filename=path) for node in tree.body: if isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Name) and tgt.id == '__all__': if isinstance(node.value, (ast.List, ast.Tuple)): return [e.value for e in node.value.elts if isinstance(e, ast.Constant) and isinstance(e.value, str)] return [] def _env_registrations(path, known_symbols=None): """解析 init.py 中 `env. = ` 注册，返回 {attr: symbol_or_None}。 右侧不止 `= name` 一种合法形态：`env.pbl_evidence_types = list(EVIDENCE_TYPES)` 这种「把包内常量浅拷贝一份再挂载」的写法同样可追溯——取表达式里引用的、且在包内 确有定义的符号名（内建 list/dict 等不算）。只有完全追溯不到包内符号（属性链、 字面量、lambda）才返回 None，由调用方按可疑处理。 """ tree = ast.parse(_read(path), filename=path) out = {} known = set(known_symbols or ()) for node in ast.walk(tree): if not isinstance(node, ast.Assign): continue for tgt in node.targets: if (isinstance(tgt, ast.Attribute) and isinstance(tgt.value, ast.Name) and tgt.value.id == 'env'): val = node.value if isinstance(val, ast.Name): sym = val.id else: hits = [n.id for n in ast.walk(val) if isinstance(n, ast.Name) and n.id in known] sym = hits[0] if hits else None out[tgt.attr] = sym return out def _funcs_by_name(src, path, names): """返回 {name: 源码片段}，收集文件内所有指定名字的函数/方法定义（含类内方法）。""" tree = ast.parse(src, filename=path) want = set(names) out = {} for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in want: seg = ast.get_source_segment(src, node) or '' # 同名函数（如类方法 vs 顶层）都保留，拼接后一起判定 out[node.name] = (out.get(node.name, '') + '\n' + seg) return out def _find_docs_dir(): """从模块目录向上定位机构工作空间根，返回 docs/02-develop 绝对路径（可能不存在）。""" cur = ROOT for _ in range(6): parent = os.path.dirname(cur) if parent == cur: break cand = os.path.join(parent, DOCS_REL) if os.path.isdir(cand): return cand cur = parent return os.path.join(ROOT, os.pardir, os.pardir, DOCS_REL) # ── 校验项 ───────────────────────────────────────────────────────────── def check_three_place_sync(): """① 包内定义 ⊇ ② __init__.__all__ ⊇ ③ init.py env 注册（≥10）。""" defined = set() api_defined = set() for f in _pkg_py_files(): syms = _top_level_symbols(f) defined |= syms if os.path.basename(f) == 'api.py': api_defined |= syms init_py = os.path.join(PKG_DIR, '__init__.py') load_py = os.path.join(PKG_DIR, 'init.py') if not os.path.isfile(init_py): return False, '缺少 %s' % _rel(init_py) if not os.path.isfile(load_py): return False, '缺少 %s' % _rel(load_py) all_names = _dunder_all(init_py) env_map = _env_registrations(load_py, defined) problems = [] if not all_names: problems.append('__init__.py 未声明 __all__') not_defined = sorted(set(all_names) - defined) if not_defined: problems.append('__all__ 中符号在包内无定义: %s' % ', '.join(not_defined)) missing_api = sorted(set(CONTRACT_APIS) - api_defined) if missing_api: problems.append('api.py 缺契约函数: %s' % ', '.join(missing_api)) missing_in_all = sorted(set(CONTRACT_APIS) - set(all_names)) if missing_in_all: problems.append('__all__ 未导出契约函数: %s' % ', '.join(missing_in_all)) env_syms = sorted({s for s in env_map.values() if s}) env_not_in_all = sorted(set(env_syms) - set(all_names)) if env_not_in_all: problems.append('init.py 注册的符号未在 __all__ 导出: %s' % ', '.join(env_not_in_all)) env_missing_attr = sorted(a for a, s in env_map.items() if not s) if env_missing_attr: problems.append('env 注册右侧追溯不到包内符号（可疑）: %s' % ', '.join(env_missing_attr)) if len(env_map) < MIN_ENV_REGISTRATIONS: problems.append('init.py env 注册数 %d < %d' % (len(env_map), MIN_ENV_REGISTRATIONS)) if problems: return False, '; '.join(problems) return True, ('包内定义 %d 符号 ⊇ __all__ %d 项 ⊇ env 注册 %d 项；契约函数 %d 个三处齐全；' 'env 注册样例: %s' % (len(defined), len(all_names), len(env_map), len(CONTRACT_APIS), ', '.join(sorted(env_map)[:5]) + ' ...')) def check_rbac_dspy_vs_ui(): """wwwroot/api/*.dspy ⊇ index.ui 声明的 api 路径；且 load_path.py 已显式登记。""" api_dir = os.path.join(WWWROOT, 'api') if not os.path.isdir(api_dir): return False, '缺少目录 %s' % _rel(api_dir) have = sorted(os.path.basename(p) for p in glob.glob(os.path.join(api_dir, '*.dspy'))) index_ui = os.path.join(WWWROOT, 'index.ui') if not os.path.isfile(index_ui): return False, '缺少 %s' % _rel(index_ui) ui_text = _read(index_ui) refs = sorted(set(re.findall(r'api/([A-Za-z0-9_]+\.dspy)', ui_text))) if not refs: return False, 'index.ui 中未解析到任何 api/*.dspy 引用' problems = [] if len(refs) < MIN_UI_API_REFS: problems.append('index.ui 声明的 api 路径数 %d < %d' % (len(refs), MIN_UI_API_REFS)) missing = sorted(set(refs) - set(have)) if missing: problems.append('index.ui 引用但磁盘缺失: %s' % ', '.join(missing)) lp = os.path.join(ROOT, 'scripts', 'load_path.py') if os.path.isfile(lp): lp_text = _read(lp) unregistered = sorted(r for r in refs if r not in lp_text) if unregistered: problems.append('未在 scripts/load_path.py 显式登记（RBAC 403 风险）: %s' % ', '.join(unregistered)) else: problems.append('缺少 %s（RBAC 无登记入口）' % _rel(lp)) if problems: return False, '; '.join(problems) return True, ('磁盘 %d 个 .dspy ⊇ index.ui 声明 %d 个（%s）；均已在 load_path.py 逐条登记' % (len(have), len(refs), ', '.join(refs))) def check_models_four_sections(): """models/*.json 四段式（summary/fields/indexes/codes + unique 约束）。""" files = sorted(glob.glob(os.path.join(MODELS_DIR, '*.json'))) if len(files) < 2: return False, 'models/*.json 数量 %d < 2' % len(files) problems = [] detail = [] for f in files: rel = _rel(f) try: data = json.loads(_read(f)) except Exception as exc: problems.append('%s 不是合法 JSON: %s' % (rel, exc)) continue for seg in ('summary', 'fields', 'indexes'): if not isinstance(data.get(seg), list) or not data.get(seg): problems.append('%s 缺非空数组段 %s' % (rel, seg)) if not isinstance(data.get('codes'), list): problems.append('%s 缺 codes 段（数组，可为空）' % rel) summ = (data.get('summary') or [{}])[0] primary = summ.get('primary') if not isinstance(primary, list) or not primary: problems.append('%s summary[0].primary 必须是非空数组' % rel) names = [fd.get('name') for fd in (data.get('fields') or [])] if 'id' not in names: problems.append('%s fields 缺主键列 id' % rel) for fd in (data.get('fields') or []): if fd.get('type') in ('str', 'char') and not isinstance(fd.get('length'), int): problems.append('%s 字段 %s: str/char 缺整数 length' % (rel, fd.get('name'))) if not fd.get('title'): problems.append('%s 字段 %s 缺 title（DDL COMMENT）' % (rel, fd.get('name'))) idx_names = [i.get('name') for i in (data.get('indexes') or [])] if len(idx_names) != len(set(idx_names)): problems.append('%s 索引名重复' % rel) for idx in (data.get('indexes') or []): if not isinstance(idx.get('idxfields'), list) or not idx.get('idxfields'): problems.append('%s 索引 %s 的 idxfields 必须是非空数组' % (rel, idx.get('name'))) unknown = sorted(c for c in (idx.get('idxfields') or []) if c not in names) if unknown: problems.append('%s 索引 %s 引用未知列 %s' % (rel, idx.get('name'), ','.join(unknown))) uniques = sorted(i.get('name') for i in (data.get('indexes') or []) if i.get('idxtype') == 'unique') if not uniques: problems.append('%s 无 unique 约束（constraints 语义缺失）' % rel) detail.append('%s: primary=%s unique=[%s] fields=%d' % (os.path.basename(f), primary, ','.join(uniques), len(names))) if problems: return False, '; '.join(problems) return True, ' | '.join(detail) def check_idempotency_key(): """uk 列组合 == (tenant_id, source_event_id, evidence_type) 且全链路语义闭环。""" problems = [] mpath = os.path.join(MODELS_DIR, 'pbl_evidence.json') if not os.path.isfile(mpath): return False, '缺少 %s' % _rel(mpath) model = json.loads(_read(mpath)) uk_indexes = [i for i in (model.get('indexes') or []) if i.get('idxtype') == 'unique' and str(i.get('name', '')).startswith('uk')] if len(uk_indexes) != 1: problems.append('pbl_evidence 唯一索引数 %d != 1' % len(uk_indexes)) uk_cols = list(uk_indexes[0].get('idxfields') or []) if uk_indexes else [] if uk_cols != UK_EXPECTED: problems.append('uk 列组合 %s != %s' % (uk_cols, UK_EXPECTED)) jpath = os.path.join(JSON_DIR, 'pbl_evidence.json') if not os.path.isfile(jpath): problems.append('缺少 %s' % _rel(jpath)) return False, '; '.join(problems) crud = json.loads(_read(jpath)) params = crud.get('params') or {} edit_ex = params.get('editexclouded') or [] if 'dedup_key' not in edit_ex: problems.append('params.editexclouded 未含 dedup_key（去重键可被表单手改）') # editexclouded 的语义是「把列从表单中排除」。幂等键列必须被排除才安全： # 旧断言方向写反（要求 tenant_id 出现在表单里），既与 crud_api.py # 「唯一索引三元组中的 tenant_id 永不开放」矛盾，也会把跨租户漂移的真实 # 缺陷判成 PASS。故正确断言是 tenant_id 必须被屏蔽。 if 'tenant_id' not in edit_ex: problems.append('tenant_id 未列入 editexclouded（表单可改租户 → 幂等键跨租户漂移）') # source_event_id / evidence_type 允许人工修订（见 crud_api.EVIDENCE_EDITABLE_COLS）， # 但后端必须在三元组变化时重算 dedup_key，否则与 uk_ev_dedup 漂移。 crud_path = os.path.join(PKG_DIR, 'crud_api.py') if not os.path.isfile(crud_path): problems.append('缺少 %s（无法核对 dedup_key 重算守卫）' % _rel(crud_path)) elif 'build_dedup_key(' not in _read(crud_path): problems.append('crud_api.py 修订三元组后未重算 dedup_key（与 uk_ev_dedup 漂移）') new_url = str(params.get('new_data_url') or '') if 'pbl_evidence_collect.dspy' not in new_url: problems.append('params.new_data_url 未指向采集端点: %s' % new_url) for key in ('new_data_url', 'update_data_url', 'delete_data_url'): if key not in params: problems.append('params 缺 %s（editable 三 URL 须在 params 顶层）' % key) if str(params.get('logined_userorgid') or '') != 'tenant_id': problems.append('params.logined_userorgid != tenant_id（列表未按租户隔离）') # 代码侧闭环：dedup_key 必须由同一三元组派生，且写入 record 含三列 col_src = _read(os.path.join(PKG_DIR, 'collector.py')) if not re.search(r"build_dedup_key\(\s*tid\s*,\s*src_id\s*,\s*evidence_type\s*\)", col_src): problems.append('collector.py 未按 (tenant_id, source_event_id, evidence_type) 构造 dedup_key') for col in UK_EXPECTED: if re.search(r"'%s'\s*:" % col, col_src) is None: problems.append('collector.py 写入 record 缺列 %s' % col) if 'ON DUPLICATE KEY UPDATE' not in col_src: problems.append('collector.py 缺 ON DUPLICATE KEY UPDATE（并发防重第二层保险）') if problems: return False, '; '.join(problems) return True, ('uk_ev_dedup 列组合 == %s；tenant_id/dedup_key 已在 editexclouded 屏蔽，' 'source_event_id/evidence_type 可人工修订但 crud_api 重算 dedup_key；' '新增走 %s；写入含 ON DUPLICATE KEY UPDATE 双保险' % (tuple(UK_EXPECTED), new_url.replace('{{entire_url(', '').replace(')}}', ''))) def check_pyproject_deps(): """dependencies 不得声明基础包（由宿主 build.sh git 安装，非 PyPI）。""" pp = os.path.join(ROOT, 'pyproject.toml') if not os.path.isfile(pp): return False, '缺少 pyproject.toml' text = _read(pp) block = re.search(r'^dependencies\s*=\s*\[([^\]]*)\]', text, re.S | re.M) if not block: return False, 'pyproject.toml 未找到 dependencies 数组' deps = [d.strip().strip('"\'') for d in block.group(1).split(',') if d.strip()] bad = sorted({d for d in deps for f in FORBIDDEN_DEPS if d.split('[')[0].split('>=')[0].strip().lower() == f.lower()}) if bad: return False, 'dependencies 含基础包（应交由宿主 build.sh 安装）: %s' % ', '.join(bad) if 'sqlor' not in deps: return False, 'dependencies 缺直接依赖 sqlor' return True, 'dependencies = [%s]，不含 apppublic/ahserver/appbase/rbac' % ', '.join(deps) def check_init_data(): """init/data.json 存在且是合法 JSON 的真实种子数据（禁占位符文本）。""" p = os.path.join(ROOT, 'init', 'data.json') if not os.path.isfile(p): return False, '缺少 %s' % _rel(p) try: data = json.loads(_read(p)) except Exception as exc: return False, 'init/data.json 解析失败（部署时 json.load 会中断）: %s' % exc if not isinstance(data, (dict, list)) or not data: return False, 'init/data.json 内容为空或不是对象/数组（禁止占位符文本）' text = json.dumps(data, ensure_ascii=False) for placeholder in ('待明确', '待确认', 'TBD', 'placeholder'): if placeholder in text: return False, 'init/data.json 含占位词 %s' % placeholder groups = data.get('appcodes') if isinstance(data, dict) else None if isinstance(groups, list): problems = [] n_items = 0 for g in groups: pid = str(g.get('parentid') or '') items = g.get('items') or [] n_items += len(items) if not pid or not items: problems.append('组 %s 缺 parentid 或 items' % (pid or '?')) continue longest_k = max(len(str(i.get('k') or '')) for i in items) if len(pid) + 1 + longest_k > 32: problems.append('parentid=%s 过长：id 生成 %s_%%s 超 VARCHAR(32)' % (pid, pid)) for i in items: if not i.get('k') or not i.get('v'): problems.append('组 %s 存在缺 k/v 的条目' % pid) if problems: return False, '; '.join(problems) return True, ('init/data.json 合法：appcodes %d 组 / kv %d 项（parentid 长度合规）: %s' % (len(groups), n_items, ', '.join(sorted(str(g.get('parentid')) for g in groups)))) return True, 'init/data.json 合法（顶层键 %s）' % ', '.join(sorted(data.keys())[:6]) def _find_fail_closed_guard(src, path): """在 collect_evidence_from_events 中定位「事件表缺失 → raise CollectError(M11b)」守卫。""" tree = ast.parse(src, filename=path) fn = None for node in tree.body: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == 'collect_evidence_from_events': fn = node break if fn is None: return False, '未定义 collect_evidence_from_events' # 找出 `X = await resolve_event_table()` 中被赋值的变量名 resolved_names = set() for node in ast.walk(fn): if isinstance(node, (ast.Assign, ast.AnnAssign)): call = node.value if isinstance(call, ast.Await): call = call.value if isinstance(call, ast.Call): fname = getattr(call.func, 'id', None) or getattr(call.func, 'attr', None) if fname == 'resolve_event_table': tgts = [node.target] if isinstance(node, ast.AnnAssign) else node.targets for t in tgts: if isinstance(t, ast.Name): resolved_names.add(t.id) if not resolved_names: return False, 'collect_evidence_from_events 未调用 resolve_event_table()' for node in ast.walk(fn): if not isinstance(node, ast.If): continue test = node.test if (isinstance(test, ast.UnaryOp) and isinstance(test.op, ast.Not) and isinstance(test.operand, ast.Name) and test.operand.id in resolved_names): seg = ast.get_source_segment(src, node) or '' has_raise = 'raise CollectError' in seg has_m11b = 'M11b' in seg if has_raise and has_m11b: return True, '守卫 `if not %s: raise CollectError(...M11b...)` 成立' % test.operand.id return False, ('变量 %s 由 resolve_event_table() 赋值，但其空值分支未 raise 含 M11b 的 ' 'CollectError（可能空成功返回）' % ', '.join(sorted(resolved_names))) def check_fail_closed_source(): """C-3：事件表缺失时 collector 必须 raise CollectError 且消息含 'M11b'。""" p = os.path.join(PKG_DIR, 'collector.py') if not os.path.isfile(p): return False, '缺少 collector.py' src = _read(p) problems = [] if 'class CollectError' not in src: problems.append('未定义 CollectError') if "'pbl_runtime_event'" not in src: problems.append("EVENT_TABLE_CANDIDATES 未包含 'pbl_runtime_event'") ok, detail = _find_fail_closed_guard(src, p) if not ok: problems.append(detail) if problems: return False, '; '.join(problems) return True, detail def check_readme_reference(): """README 引用的 scripts/*.py 必须在磁盘命中（含本脚本自身）。""" p = os.path.join(ROOT, 'README.md') if not os.path.isfile(p): return False, '缺少 README.md' text = _read(p) # README 里同时存在两类 scripts/*.py 引用：①本模块自己的脚本（必须磁盘命中）； # ②宿主/中央应用（apps/pbls 等）的同名目录脚本，按设计不在本模块仓库内。 # 旧逻辑不区分两类，把宿主侧路径也拿来查本模块磁盘，必然误报。 local, external = set(), set() for m in re.finditer(r'((?:[A-Za-z0-9_./-]*/)?)scripts/[A-Za-z0-9_./-]+\.py', text): token = m.group(0) before = text[max(0, m.start() - 24):m.start()] if 'apps/' in token or token.count('/') > 1 or re.search(r'宿主|中央', before): external.add(token) else: local.add(token) refs = sorted(local) missing = [r for r in refs if not os.path.isfile(os.path.join(ROOT, r))] if missing: return False, 'README 引用但磁盘缺失: %s' % ', '.join(missing) if 'scripts/selfcheck_m5a.py' not in refs: return False, 'README 未引用 scripts/selfcheck_m5a.py（自检入口缺失）' return True, ('README 引用 %d 个本模块脚本路径全部命中（%d 个宿主侧路径按外部引用豁免）: %s' % (len(refs), len(external), ', '.join(refs))) # ── 第 9~12 项：harness 与运行证据门禁（QC #4 退回项）───────────────── def check_harness_static(): """⑨ harness 源码静态门禁：异常收敛 + 无吞参反模式 + 种子 INSERT 具名参数。 三项都只读源码做 AST/正则判定，不执行 harness（本脚本不连库、不跑子进程）。 """ if not os.path.isfile(HARNESS_PATH): return False, '缺少 %s' % _rel(HARNESS_PATH) src = _read(HARNESS_PATH) try: tree = ast.parse(src, filename=HARNESS_PATH) except Exception as exc: return False, '%s 语法错误，无法静态校验: %s' % (_rel(HARNESS_PATH), exc) problems = [] detail = [] # (a) main() 外层 try/except 收敛未捕获异常：崩溃必须变成 FAIL 断言而非 traceback 逃逸 main_fn = None for node in tree.body: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == 'main': main_fn = node break if main_fn is None: problems.append('harness 未定义 main()') else: converged = None for node in ast.walk(main_fn): if not isinstance(node, ast.Try): continue body_src = ast.get_source_segment(src, node) or '' if not any(c in body_src for c in HARNESS_PHASE_CALLS): continue # 只认包住 phase 调用的「外层」try for handler in node.handlers: htype = getattr(handler.type, 'id', None) or getattr(handler.type, 'attr', None) if htype not in ('Exception', 'BaseException'): continue hsrc = ast.get_source_segment(src, handler) or '' # 收敛 = 记失败/给出非零退出，而不是原样 raise 或静默 pass if re.search(r'\braise\b', hsrc) and 'check(' not in hsrc: continue if 'check(' in hsrc or 'return 1' in hsrc or 'sys.exit(1)' in hsrc \ or 'FAILED' in hsrc or 'format_exc' in hsrc: converged = handler break if converged is not None: break if converged is None: problems.append('main() 缺少包住 phase 调用的外层 try/except Exception 收敛' '（未捕获异常会以 traceback 崩溃逃逸，日志变 .qc_tmp 不算证据）') else: detail.append('main() 外层 try/except 收敛未捕获异常') # (b) raw()/q_all()/q_exec() 不再出现 `args or {}` / `dict(params or {})` 吞参反模式 funcs = _funcs_by_name(src, HARNESS_PATH, ('raw', 'q_all', 'q_exec')) missing_fn = sorted(set(('raw', 'q_all', 'q_exec')) - set(funcs)) if missing_fn: problems.append('harness 缺函数 %s' % ', '.join(missing_fn)) else: bad = [] for name in sorted(funcs): for pat in PARAM_SWALLOW_PATTERNS: if pat in funcs[name]: bad.append('%s() 含 %s' % (name, pat)) if bad: problems.append('数据访问层静默吞参反模式（None→{} 掩盖漏传参）: %s' % '; '.join(sorted(bad))) else: detail.append('raw()/q_all()/q_exec() 无 args or {} / dict(params or {}) 反模式') # (c) 种子 INSERT 用具名参数 %(event_id)s，而非位置 %s（位置参数与列序耦合，改列即错位） seed_hits = list(re.finditer(r'INSERT\s+INTO\s+`?pbl_runtime_event`?', src, re.I)) if not seed_hits: problems.append('harness 中未找到 pbl_runtime_event 种子 INSERT 语句') else: named_ok = 0 positional = [] for m in seed_hits: window = src[m.start():m.start() + 600] flat = re.sub(r'\s+', ' ', window) if '%(event_id)s' in window: named_ok += 1 if re.search(r'VALUES\s*\(\s*%s\s*(,|\))', flat): positional.append(flat[:60]) if named_ok == 0: problems.append('种子 INSERT 未使用具名参数 %%(event_id)s（%d 处均为位置参数）' % len(seed_hits)) elif named_ok < len(seed_hits): problems.append('种子 INSERT 具名/位置混用：具名 %d 处 / 共 %d 处' % (named_ok, len(seed_hits))) else: detail.append('种子 INSERT %d 处均用具名参数 %%(event_id)s' % named_ok) if positional: problems.append('种子 INSERT 仍含位置参数 VALUES (%s,...)：%d 处' % ('%s', len(positional))) if problems: return False, '; '.join(problems) return True, ' | '.join(detail) def check_no_hardcoded_credentials(): """⑩ harness 零明文凭据：口令字面量 0 命中；db_kwargs() 只从 env 变量/env/test.json 取。 环境信息唯一事实源是 projects/pbls/env/test.json（project-directory-spec 第八章）， 应用/模块仓库内不得内嵌部署凭据——写死 fallback 口令既泄密又造成多份矛盾环境。 """ if not os.path.isfile(HARNESS_PATH): return False, '缺少 %s' % _rel(HARNESS_PATH) src = _read(HARNESS_PATH) problems = [] detail = [] hits = [] for lineno, line in enumerate(src.splitlines(), 1): for pat in CREDENTIAL_PATTERNS: if re.search(pat, line, re.I): hits.append('%s@L%d' % (re.sub(r'\s+', '', pat), lineno)) if hits: problems.append('harness 源码命中明文口令字面量 %d 处: %s' % (len(hits), ', '.join(sorted(set(hits))))) else: detail.append('password=/user=/test123 等明文凭据字面量 0 命中') try: tree = ast.parse(src, filename=HARNESS_PATH) except Exception as exc: problems.append('harness 语法错误，无法解析 db_kwargs(): %s' % exc) return False, '; '.join(problems) db_kwargs_src = '' for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == 'db_kwargs': db_kwargs_src += (ast.get_source_segment(src, node) or '') if not db_kwargs_src: problems.append('harness 未定义 db_kwargs()（凭据来源无从校验）') else: has_env_src = ('PBL_M5A_DB' in db_kwargs_src) or ('test.json' in db_kwargs_src) if not has_env_src: problems.append("db_kwargs() 未从 PBL_M5A_DB 或 env/test.json 读取凭据" "（唯一事实源缺失，只能靠硬编码）") else: srcs = [] if 'PBL_M5A_DB' in db_kwargs_src: srcs.append('PBL_M5A_DB') if 'test.json' in db_kwargs_src: srcs.append('env/test.json') detail.append('db_kwargs() 凭据来源: %s' % ' / '.join(srcs)) # 取不到必须 raise：静默回落到硬编码口令 = 用错库还自称 PASS if 'raise' not in db_kwargs_src: problems.append('db_kwargs() 取不到凭据时未 raise（会静默回落到硬编码口令连错库）') else: detail.append('db_kwargs() 缺凭据即 raise，无静默回落') if problems: return False, '; '.join(problems) return True, ' | '.join(detail) def check_evidence_logs(): """⑪ 运行证据日志：三份日志齐备且含 `# RC=0`；harness 日志含 ALL PASS 与全 [PASS] 行。 只认 projects/pbls/docs/02-develop/ 顶层文件：.qc_tmp/ 下的是崩溃 traceback 现场， 属于「跑挂了」的记录而非「跑过了」的证据，计入会把失败伪装成交付。 """ docs = _find_docs_dir() if not os.path.isdir(docs): return False, '缺少证据目录 %s（相对工作空间 projects/pbls/docs/02-develop）' % DOCS_REL.replace(os.sep, '/') top = sorted(f for f in os.listdir(docs) if os.path.isfile(os.path.join(docs, f))) problems = [] detail = [] # .qc_tmp/ 下的同名文件不算证据（显式统计出来，便于定位「日志放错地方」） qc_dir = os.path.join(docs, '.qc_tmp') qc_names = sorted(f for f in (os.listdir(qc_dir) if os.path.isdir(qc_dir) else [])) qc_only = sorted(set(EVIDENCE_LOGS) & set(qc_names) - set(top)) contents = {} for name in EVIDENCE_LOGS: if name not in top: where = '（仅存在于 .qc_tmp/，不计为证据）' if name in qc_only else '' problems.append('缺少证据日志 %s%s' % (name, where)) continue text = _read(os.path.join(docs, name)) contents[name] = text if RC_MARKER not in text: problems.append('%s 缺 `%s` 收尾标记（未记录退出码=无法证明 rc=0）' % (name, RC_MARKER)) if TRACEBACK_TOKEN in text and name != 'm5a-live-db-harness.log': problems.append('%s 含崩溃 traceback，非通过证据' % name) if problems: return False, '; '.join(problems) detail.append('三份证据日志齐备且均含 `%s`: %s' % (RC_MARKER, ', '.join(sorted(EVIDENCE_LOGS)))) hlog = contents['m5a-live-db-harness.log'] if HARNESS_ALL_PASS_TOKEN not in hlog: problems.append('harness 日志缺 `%s` 收尾行' % HARNESS_ALL_PASS_TOKEN) if TRACEBACK_TOKEN in hlog: problems.append('harness 日志含崩溃 traceback（未收敛的未捕获异常）') missing_pass = sorted(t for t in HARNESS_CHECK_TOKENS if not re.search(r'\[PASS\]\s*' + re.escape(t) + r'\b', hlog)) if missing_pass: problems.append('harness 日志缺 [PASS] 断言行: %s' % ', '.join(missing_pass)) else: detail.append('harness 日志含 %s 与 C-3.1~C-3.4 / IDM.1~IDM.9 共 %d 条 [PASS] 行' % (HARNESS_ALL_PASS_TOKEN, len(HARNESS_CHECK_TOKENS))) if problems: return False, '; '.join(problems) return True, ' | '.join(detail) def _parse_ddl_columns(ddl): """从 CREATE TABLE 正文里取「列定义」名（排除 PRIMARY/UNIQUE/KEY/CONSTRAINT 等索引行）。""" body = re.search(r'\((.*)\)\s*(?:ENGINE|COMMENT|=|$)', ddl, re.S) seg = body.group(1) if body else ddl cols = [] for line in seg.splitlines(): s = line.strip().rstrip(',') m = re.match(r'`(\w+)`\s+[A-Za-z]', s) if not m: continue if re.match(r'(PRIMARY|UNIQUE|KEY|CONSTRAINT|INDEX)\b', s, re.I): continue cols.append(m.group(1)) return cols def _parse_ddl_unique(ddl, uk_name): """取 `UNIQUE KEY \\`uk_ev_dedup\\` (\\`a\\`, \\`b\\`)` 的列组合（保序）。""" m = re.search(r'UNIQUE\s+KEY\s+`?%s`?\s*\(([^)]*)\)' % re.escape(uk_name), ddl, re.I) if not m: return None return [c.strip().strip('`').strip() for c in m.group(1).split(',') if c.strip()] def check_harness_matches_model(): """⑫ harness 沙箱 DDL 与 models/pbl_evidence.json 对齐（列 ⊆ 模型列、uk 三元组一致）。 harness 自建表跑出来的幂等结论，只有与真源（models/*.json）同构才有效；否则「沙箱 里幂等、生产上漂移」会被日志证据掩盖。 """ problems = [] if not os.path.isfile(HARNESS_PATH): return False, '缺少 %s' % _rel(HARNESS_PATH) src = _read(HARNESS_PATH) try: tree = ast.parse(src, filename=HARNESS_PATH) except Exception as exc: return False, 'harness 语法错误，无法解析 DDL_PBL_EVIDENCE: %s' % exc ddl = None for node in tree.body: if isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Name) and tgt.id == 'DDL_PBL_EVIDENCE': v = node.value if isinstance(v, ast.Constant) and isinstance(v.value, str): ddl = v.value if ddl is None: return False, 'harness 未找到模块级常量 DDL_PBL_EVIDENCE（字符串字面量）' mpath = os.path.join(MODELS_DIR, 'pbl_evidence.json') if not os.path.isfile(mpath): return False, '缺少 %s' % _rel(mpath) model = json.loads(_read(mpath)) model_cols = [fd.get('name') for fd in (model.get('fields') or []) if fd.get('name')] uk_idx = [i for i in (model.get('indexes') or []) if i.get('name') == 'uk_ev_dedup'] if len(uk_idx) != 1: problems.append('models/pbl_evidence.json 中 uk_ev_dedup 索引数 %d != 1' % len(uk_idx)) model_uk = list(uk_idx[0].get('idxfields') or []) if uk_idx else [] ddl_cols = _parse_ddl_columns(ddl) ddl_uk = _parse_ddl_unique(ddl, 'uk_ev_dedup') if not ddl_cols: problems.append('DDL_PBL_EVIDENCE 未解析出任何列定义') unknown = sorted(set(ddl_cols) - set(model_cols)) if unknown: problems.append('DDL 列不在模型 fields 中: %s' % ', '.join(unknown)) if sorted(set(ddl_uk or [])) != sorted(UK_EXPECTED): problems.append('DDL uk_ev_dedup 列组合 %s != %s' % (ddl_uk, UK_EXPECTED)) if model_uk and ddl_uk and list(ddl_uk) != list(model_uk): problems.append('DDL uk_ev_dedup 列序 %s != 模型 idxfields %s' % (ddl_uk, model_uk)) if problems: return False, '; '.join(problems) return True, ('DDL_PBL_EVIDENCE %d 列 ⊆ models/pbl_evidence.json %d 列；' 'uk_ev_dedup == %s（与模型 idxfields 逐列一致）' % (len(ddl_cols), len(model_cols), tuple(ddl_uk))) CHECKS = [ ('三处同步：包内定义 ⊇ __init__.__all__ ⊇ init.py env 注册（≥10）', check_three_place_sync), ('RBAC：wwwroot/api/*.dspy ⊇ index.ui 声明路径（≥9）且逐条登记 load_path', check_rbac_dspy_vs_ui), ('模型四段式：models/*.json summary(primary)/fields/indexes/codes + unique', check_models_four_sections), ('幂等键语义：uk == (tenant_id, source_event_id, evidence_type) 全链路闭环', check_idempotency_key), ('pyproject.toml dependencies 不含 apppublic/ahserver/appbase/rbac', check_pyproject_deps), ('init/data.json 存在且 json.load 通过（无占位词、parentid 不超长）', check_init_data), ('C-3 fail-closed：事件表缺失时抛 CollectError 且消息含 M11b', check_fail_closed_source), ('README 引用的 scripts/*.py 路径全部磁盘命中（含 selfcheck_m5a.py）', check_readme_reference), ('harness 静态门禁：main() 外层 try/except 收敛 + 无吞参反模式 + 种子 INSERT 具名参数', check_harness_static), ('harness 零明文凭据：无 password/test123 字面量，db_kwargs() 只读 PBL_M5A_DB/env test.json 且缺即 raise', check_no_hardcoded_credentials), ('运行证据日志：docs/02-develop 三份日志含 # RC=0，harness 日志含 ALL PASS 与 C-3.x/IDM.x 全 [PASS]', check_evidence_logs), ('harness DDL 与模型一致：DDL 列 ⊆ models fields 且 uk_ev_dedup == (tenant_id, source_event_id, evidence_type)', check_harness_matches_model), ] def main(): print('selfcheck_m5a: target = modules/pbl_evidence') failed = [] for i, (name, fn) in enumerate(CHECKS, 1): try: ok, detail = fn() except SystemExit: raise except Exception as exc: ok, detail = False, '校验执行异常 %s: %s' % (type(exc).__name__, exc) tag = 'PASS' if ok else 'FAIL' print('[%s] %d. %s' % (tag, i, name)) print(' %s' % detail) if not ok: failed.append(name) if failed: print('SELF CHECK FAILED (%d/%d)' % (len(failed), len(CHECKS))) for n in failed: print(' - %s' % n) return 1 print('ALL PASS (%d/%d checks)' % (len(CHECKS), len(CHECKS))) return 0 if __name__ == '__main__': sys.exit(main())