From 8eb1e2464a07917027940b075fc97850f28a2fe7 Mon Sep 17 00:00:00 2001
From: "agent.develop" <agent.develop@hrs6.local>
Date: Tue, 22 Sep 2026 14:19:39 +0800
Subject: [PATCH] =?UTF-8?q?deliver:=20=E4=BA=A4=E4=BB=98=E6=94=B6=E5=8F=A3?=
 =?UTF-8?q?=EF=BC=88=E5=BC=95=E6=93=8E=E4=BB=A3=E4=B8=BA=E6=8F=90=E4=BA=A4?=
 =?UTF-8?q?=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 scripts/selfcheck_m5a.py | 480 +++++++++++++++++++++++++++++++++++++++
 1 file changed, 480 insertions(+)
 create mode 100644 scripts/selfcheck_m5a.py

diff --git a/scripts/selfcheck_m5a.py b/scripts/selfcheck_m5a.py
new file mode 100644
index 0000000..3062776
--- /dev/null
+++ b/scripts/selfcheck_m5a.py
@@ -0,0 +1,480 @@
+#!/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 路径全部在磁盘命中（含本脚本自身）
+
+关于「四段式」口径：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')
+
+# 期望的幂等键三元组（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
+
+
+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):
+    """解析 init.py 中 `env.<attr> = <symbol>` 注册，返回 {attr: symbol_or_None}。"""
+    tree = ast.parse(_read(path), filename=path)
+    out = {}
+    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'):
+                sym = node.value.id if isinstance(node.value, ast.Name) else None
+                out[tgt.attr] = sym
+    return out
+
+
+# ── 校验项 ─────────────────────────────────────────────────────────────
+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)
+
+    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（幂等键可被表单手改）')
+    for col in UK_EXPECTED:
+        if col in edit_ex:
+            problems.append('幂等键列 %s 不应出现在 editexclouded（会漏进新增表单）' % col)
+    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；dedup_key=md5(三元组) 且在 editexclouded 受保护；'
+                  '新增走 %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)
+    refs = sorted(set(re.findall(r'(scripts/[A-Za-z0-9_./-]+\.py)', text)))
+    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 text:
+        return False, 'README 未引用 scripts/selfcheck_m5a.py（自检入口缺失）'
+    return True, 'README 引用 %d 个脚本路径全部命中: %s' % (len(refs), ', '.join(refs))
+
+
+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),
+]
+
+
+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())