diff --git a/scripts/m11b2_selftest.py b/scripts/m11b2_selftest.py
index fa63d12..70643f5 100644
--- a/scripts/m11b2_selftest.py
+++ b/scripts/m11b2_selftest.py
@@ -15,6 +15,42 @@
任一断言不符即 RESULT: FAIL 且退出码 1。
用法: python3 modules/world_sync/scripts/m11b2_selftest.py
+
+────────────────────────────────────────────────────────────────────────────
+[cross-check-sql] 提取器的列名解析规则(M11b-2d-B 固化,下次结构变更照此扩展)
+────────────────────────────────────────────────────────────────────────────
+pbl_runtime_sql.py 里 ``EVENT_COLUMNS`` / ``STATE_COLUMNS`` 有两种合法真源形态,
+提取器必须**同时**兼容,否则自检会在代码演进时假阳(M11b-2c 实跑 4 FAIL 即此因):
+
+ 形态 A(旧·两段式手抄字面量)
+ EVENT_COLUMNS = (
+ "id", "event_id", ...
+ )
+ → 正则 ``NAME\\s*=\\s*\\((.*?)\\)`` + 引号内标识符,直接取字面量列名。
+
+ 形态 B(M11b-2b 起·四段式 models/*.json 机械派生,当前真态)
+ STATE_TABLE = "pbl_entity_state"
+ STATE_COLUMNS = columns_of(STATE_TABLE)
+ 列名不再手抄,由 :mod:`world_sync.pbl_runtime_authority` 按
+ ``models/
.json`` 的 ``fields[].name``(spec 四段式第三段之前的
+ fields 段)有序导出,fail-closed(模型缺失/非四段式/fields 空 →
+ AuthorityResolutionError,不退回旧常量)。
+ 解析优先级:
+ B1 运行时取真值:sys.path 注入 ``modules/world_sync``,import
+ ``world_sync.pbl_runtime_sql``,读模块属性(= 生产代码实际使用的列契约,
+ 与「手抄正则」相比更强:常量表达式怎么变都能拿到最终值)。
+ B2 静态回落:import 失败(依赖/环境差异)时,按形态 B 正则解析出
+ ``columns_of()`` → `` = ""`` → 读唯一真源
+ ``modules/world_sync/models/.json`` 的 ``fields[].name``。
+ A/B 都解析不出 → 记 FAIL(不静默跳过),保证门禁不失效。
+
+覆盖强度说明(未放宽任何断言):形态 A 原本只断「代码列 ⊆ fields 列」单向;
+形态 B 下列集由 fields 派生,单向断言退化为恒真,故补两条等价替代断言把强度
+恢复到「代码契约 == 表契约」双向:
+ - [cross-check-sql] 代码列 ⊆ fields 列(原有,语义不变)
+ - [cross-check-sql-nocov] fields 列 ⊆ 代码列(真源不得声明代码不认的幽灵列)
+ - [cross-check-sql-order] 代码列顺序 == fields 声明顺序(columns_of 有序导出契约)
+ - [cross-check-sql-count] 列数与代码注释宣称的列数一致(防静默丢列)
"""
from __future__ import annotations
@@ -55,8 +91,14 @@ COLS_IN_CODE = {
"pbl_runtime_event": "EVENT_COLUMNS",
"pbl_entity_state": "STATE_COLUMNS",
}
+# 代码注释里宣称的列数(防静默丢列;改模型时同步更新此表,不一致即 FAIL)
+DECLARED_COL_COUNT = {
+ "pbl_runtime_event": 25,
+ "pbl_entity_state": 10,
+}
results = [] # (ok: bool, line: str)
+_CODE_COLS_CACHE = {} # (tname, const) -> (cols|None, how)
def check(ok: bool, line: str) -> bool:
@@ -77,6 +119,87 @@ def load(path: str):
return json.load(fh)
+# --------------------------------------------------------------------------
+# 代码侧列名提取(解析规则见文件头「[cross-check-sql] 提取器的列名解析规则」)
+# --------------------------------------------------------------------------
+def _cols_from_literal(src: str, const: str):
+ """形态 A:NAME = ("a", "b", ...) 手抄字面量元组。"""
+ m = re.search(const + r"\s*=\s*\((.*?)\)", src, re.S)
+ if not m:
+ return None
+ cols = re.findall(r'"([a-zA-Z_][a-zA-Z0-9_]*)"', m.group(1))
+ return cols or None
+
+
+def _table_from_columns_of(src: str, const: str):
+ """形态 B 静态解析:NAME = columns_of() → = ""。"""
+ m = re.search(const + r"\s*=\s*columns_of\(\s*([A-Za-z_][A-Za-z0-9_]*)\s*\)", src)
+ if not m:
+ return None
+ tconst = m.group(1)
+ m2 = re.search(r"\b" + re.escape(tconst) + r"\s*=\s*[\"']([a-zA-Z_][a-zA-Z0-9_]*)[\"']", src)
+ return m2.group(1) if m2 else None
+
+
+def _cols_from_runtime(const: str):
+ """形态 B1:import 生产模块,读运行时常量真值(最贴近实际 SQL 契约)。"""
+ try:
+ if MODULE_DIR not in sys.path:
+ sys.path.insert(0, MODULE_DIR)
+ mod = __import__("world_sync.pbl_runtime_sql", fromlist=["pbl_runtime_sql"])
+ cols = getattr(mod, const, None)
+ if isinstance(cols, (list, tuple)) and cols:
+ return [str(c) for c in cols]
+ except Exception: # noqa: BLE001 - 环境不具备时回落静态解析,不因此放宽断言
+ return None
+ return None
+
+
+def _cols_from_authority_model(table: str):
+ """形态 B2:读唯一真源 models/.json 的 fields[].name(有序)。"""
+ p = os.path.join(WS_ROOT, SRC_DIR, table + ".json")
+ if not os.path.isfile(p):
+ return None
+ try:
+ doc = load(p)
+ except Exception: # noqa: BLE001
+ return None
+ fields = doc.get("fields") or []
+ cols = [str(f.get("name")) for f in fields if f.get("name")]
+ return cols or None
+
+
+def extract_code_columns(tname: str, const: str):
+ """返回 (cols|None, how)。cols=None 表示解析失败(调用方记 FAIL,不静默跳过)。"""
+ key = (tname, const)
+ if key in _CODE_COLS_CACHE:
+ return _CODE_COLS_CACHE[key]
+
+ out = (None, "unresolved")
+ sql_path = os.path.join(WS_ROOT, SQL_MODULE)
+ if os.path.isfile(sql_path):
+ src = open(sql_path, "r", encoding="utf-8").read()
+ # A) 旧两段式字面量
+ cols = _cols_from_literal(src, const)
+ if cols:
+ out = (cols, "literal-tuple(形态A)")
+ else:
+ # B1) 运行时真值
+ cols = _cols_from_runtime(const)
+ if cols:
+ out = (cols, "runtime-import columns_of(形态B1)")
+ else:
+ # B2) 静态解析表名 → 唯一真源 models/*.json
+ table = _table_from_columns_of(src, const)
+ if table:
+ cols = _cols_from_authority_model(table)
+ if cols:
+ out = (cols, "static columns_of(%s) -> models/%s.json fields[]"
+ % (table, table) + "(形态B2)")
+ _CODE_COLS_CACHE[key] = out
+ return out
+
+
def assert_spec(rel: str) -> None:
"""对单个落点文件跑全套 spec 断言。"""
path = os.path.join(WS_ROOT, rel)
@@ -169,18 +292,34 @@ def assert_spec(rel: str) -> None:
ixfs = ix.get("idxfields") or []
check(bool(ixfs) and ixfs[0] == "tenant_id",
f"[tenant-first] {rel}#{ix.get('name')} 唯一索引 tenant_id 打头 -> {ixfs}")
- # 交叉核对:代码里的列名全部在册
- sql_path = os.path.join(WS_ROOT, SQL_MODULE)
- if os.path.isfile(sql_path) and tname in COLS_IN_CODE:
- src = open(sql_path, "r", encoding="utf-8").read()
- m = re.search(COLS_IN_CODE[tname] + r"\s*=\s*\((.*?)\)", src, re.S)
- if m:
- code_cols = re.findall(r'"([a-zA-Z_][a-zA-Z0-9_]*)"', m.group(1))
- missing = [c for c in code_cols if c not in names]
- check(not missing, f"[cross-check-sql] {rel} 覆盖 {COLS_IN_CODE[tname]} {len(code_cols)} 列"
- + ("" if not missing else f" 缺列={missing}"))
+
+ # --- 交叉核对:代码里的列名全部在册(兼容形态 A/B,见文件头解析规则)---
+ const = COLS_IN_CODE.get(tname)
+ if const:
+ sql_path = os.path.join(WS_ROOT, SQL_MODULE)
+ if not os.path.isfile(sql_path):
+ check(False, f"[cross-check-sql] {rel} 代码侧真源缺失 {SQL_MODULE}")
else:
- check(False, f"[cross-check-sql] {rel} 未能解析 {COLS_IN_CODE[tname]}")
+ code_cols, how = extract_code_columns(tname, const)
+ if not code_cols:
+ check(False, f"[cross-check-sql] {rel} 未能解析 {const}"
+ f"(形态A字面量/形态B运行时/形态B静态 三级解析均失败)")
+ else:
+ missing = [c for c in code_cols if c not in names]
+ check(not missing, f"[cross-check-sql] {rel} 覆盖 {const} {len(code_cols)} 列"
+ f" via {how}" + ("" if not missing else f" 缺列={missing}"))
+ # 等价替代断言(四段式下把单向覆盖升级为双向契约一致,强度不降反升)
+ nocov = [c for c in names if c not in set(code_cols)]
+ check(not nocov, f"[cross-check-sql-nocov] {rel} fields 无代码未声明列"
+ f"(fields⊆{const})" + ("" if not nocov else f" 幽灵列={nocov}"))
+ check(list(code_cols) == list(names),
+ f"[cross-check-sql-order] {rel} {const} 顺序 == fields 声明顺序"
+ f"(columns_of 有序导出契约)")
+ declared = DECLARED_COL_COUNT.get(tname)
+ if declared is not None:
+ check(len(code_cols) == declared and len(names) == declared,
+ f"[cross-check-sql-count] {rel} 列数==宣称 {declared}"
+ f" -> code={len(code_cols)} fields={len(names)}")
# --- codes ---
codes = doc.get("codes") or []
@@ -234,6 +373,12 @@ def main() -> int:
check(False, f"[mirror-sha256-equal] {t} 文件缺失,无法比对")
print()
+ print("== 代码侧列名解析方式(形态 A/B 兼容诊断)==")
+ for t in TABLES:
+ cols, how = extract_code_columns(t, COLS_IN_CODE[t])
+ print(f" {COLS_IN_CODE[t]}: n={len(cols) if cols else 0} via {how}")
+ print()
+
passed = sum(1 for ok, _ in results if ok)
failed = [(line) for ok, line in results if not ok]
for ok, line in results: