# -*- coding: utf-8 -*- """M11b-2c-A3 一次性补丁器：修 pbl_runtime_ext/tx_write.py 中 PBL-RTX-0004 的业务缺陷。 四段精确字符串替换；任一片段命中数 != 1 即报错退出（绝不静默跳过）。 改完由 py_compile + self_check + selftest + pytest 验证。 """ import io import os import sys TARGET = os.path.join("modules", "pbl_runtime_ext", "pbl_runtime_ext", "tx_write.py") # ---------------------------------------------------------------- 1) required_columns OLD_1 = '''def required_columns(table): """NOT NULL 且非自增的列（INSERT 必须显式给值，否则 MariaDB 严格模式报错）。""" spec = _model_spec(table) if not spec: return [] out = [] for fd in spec.get("fields") or []: if not isinstance(fd, dict): continue if fd.get("null") is False and not fd.get("auto_increment"): out.append(fd["name"]) return out ''' NEW_1 = '''def _is_not_null(fd): """字段是否声明 NOT NULL —— 兼容 models/*.json 并存的两种写法。 * ``{"null": false}`` ：xlsx→json 产物写法（pbl_entity_state / pbl_world_state_snapshot） * ``{"nullable": "no"}`` ：database-table-definition-spec 规范写法（pbl_runtime_event） M11b-2c-A3 修复（PBL-RTX-0004 **漏报侧**根因）：此前只认 ``null: false``， 而 pbl_runtime_event 用的是规范写法 ``nullable: "no"``，使该表 ``required_columns()`` 恒为 ``[]`` —— build_row 的 INSERT 完整性门禁对事件表**完全失效** （该报的缺列不报，静默写出半空行）。 """ if fd.get("null") is False: return True if str(fd.get("nullable") or "").strip().lower() in ("no", "false", "0"): return True return False def _has_db_default(fd): """列是否在 DDL 中声明 DEFAULT（含 CURRENT_TIMESTAMP）。 ``NOT NULL DEFAULT x`` 的列在 INSERT 省略时由数据库补值，**不构成"缺 NOT NULL 列"**。 M11b-2c-A3 修复（PBL-RTX-0004 **误报侧**根因之二）：pbl_runtime_event 的 event_code/idem_key/scene_id/entity_id/actor_id/causation_id/seq/source/state/ tx_group/broadcast/created_by/created_at/updated_at 等列均为 NOT NULL DEFAULT， 写路径按语义不填它们；若计入必填集合，合法调用会被误报缺列。 （与 _recognize nullable 写法配套：只修识别不修默认值排除，事件表主链路会被误挡。） """ dv = fd.get("default") if dv is None: return False s = str(dv).strip() if s == "" or s.upper() in ("NULL", "NONE"): return False return True def required_columns(table): """INSERT 必须**显式给值**的列：NOT NULL + 非自增 + 无数据库 DEFAULT。 自增列由数据库生成、NOT NULL DEFAULT 列由默认值补上，两者都不需要调用方提供； 纳入必填集合只会造成 PBL-RTX-0004 误报。真正的必填列（pbl_entity_state 的 tenant_id/session_id/entity_id/state_version/checksum/updated_by/created_at/ updated_at，pbl_runtime_event 的 event_uid/event_type）缺值时数据库严格模式会 直接报错，故必须在这里 fail-closed 拦住，不得放宽。 """ spec = _model_spec(table) if not spec: return [] out = [] for fd in spec.get("fields") or []: if not isinstance(fd, dict) or not fd.get("name"): continue if _is_not_null(fd) and not fd.get("auto_increment") and not _has_db_default(fd): out.append(fd["name"]) return out ''' # ---------------------------------------------------------------- 2) build_row OLD_2 = '''def build_row(table, wanted, strict=True): """按表真实列过滤字段。strict=True 时未知列直接抛错（防静默丢列写空行）。""" cols = table_columns(table) if not cols: if strict: raise rtx_db.RtxError("PBL-RTX-0003", "models/%s.json 不可读，拒绝按猜列写入" % table) return dict(wanted) unknown = [k for k in wanted if k not in cols] if unknown and strict: raise rtx_db.RtxError( "PBL-RTX-0003", "%s 未知列 %s（权威列=%s）" % (table, unknown, cols)) row = dict((k, v) for k, v in wanted.items() if k in cols) missing = [c for c in required_columns(table) if c not in row] if missing: raise rtx_db.RtxError("PBL-RTX-0004", "%s 缺少 NOT NULL 列 %s" % (table, missing)) return row ''' NEW_2 = '''def build_row(table, wanted, strict=True, full_row=True): """按表真实列过滤字段。strict=True 时未知列直接抛错（防静默丢列写空行）。 full_row=True（默认，**INSERT 语义**）：额外校验 required_columns() 齐备， 缺列抛 PBL-RTX-0004。 full_row=False（**UPDATE SET 子句语义**）：被改动的行在库中已存在，未出现在 SET 列表里的 NOT NULL 列保持原值，**不存在"缺列"**，因此不得套用 INSERT 的完整性 校验；同时禁止 SET 不可变列（tenant_id/id/created_at），违者抛 PBL-RTX-0006。 PBL-RTX-0004 闭环（M11b-2c-A3，A1 判定为业务实现缺陷）：误报有两条独立路径—— (1) UPDATE 分支沿用 INSERT 校验（此前用 strict=False 绕，但 strict 只管未知列， 缺列校验照跑 → tenant_id/session_id/entity_id/created_at 必被误报）。 现按 full_row=False 构造：定位维度走 WHERE、created_at 保持建档原值。 (2) 必填集合把「自增列 / NOT NULL DEFAULT 列」也算进来 → required_columns() 已排除（见 _has_db_default）。 漏报侧同修：required_columns() 此前不认 ``nullable: "no"`` 写法，事件表门禁恒真。 修后语义：该报的（真缺 event_uid/event_type 等无默认 NOT NULL 列）一定报， 不该报的（UPDATE SET 子集、数据库能补值的列）一定不报。 """ cols = table_columns(table) if not cols: if strict: raise rtx_db.RtxError("PBL-RTX-0003", "models/%s.json 不可读，拒绝按猜列写入" % table) return dict(wanted) unknown = [k for k in wanted if k not in cols] if unknown and strict: raise rtx_db.RtxError( "PBL-RTX-0003", "%s 未知列 %s（权威列=%s）" % (table, unknown, cols)) row = dict((k, v) for k, v in wanted.items() if k in cols) if full_row: missing = [c for c in required_columns(table) if c not in row] if missing: raise rtx_db.RtxError("PBL-RTX-0004", "%s 缺少 NOT NULL 列 %s" % (table, missing)) else: immutable = [c for c in ("tenant_id", "id", "created_at") if c in row] if immutable: raise rtx_db.RtxError("PBL-RTX-0006", "%s UPDATE 禁止 SET 不可变列 %s" % (table, immutable)) return row ''' # ---------------------------------------------------------------- 3) UPDATE 调用点 OLD_3 = ''' "checksum": checksum, "updated_by": actor_id, "updated_at": now, }, strict=False)''' NEW_3 = ''' "checksum": checksum, "updated_by": actor_id, "updated_at": now, }, strict=True, full_row=False)''' # ---------------------------------------------------------------- 4) self_check 闭环门禁 OLD_4 = ''' for name, fn in (("apply_runtime_event", apply_runtime_event), ("poll_events", poll_events), ("read_states", read_states)):''' NEW_4 = ''' # ---- PBL-RTX-0004 闭环门禁（M11b-2c-A3）：误报侧不报 + 漏报侧必报 ---- try: upd_row = build_row(STATE_TABLE, { STATE_COL_VALUE: "{}", EVENT_COL_STATE_VERSION: 2, "checksum": "c", "updated_by": "server", "updated_at": "2026-09-21 12:00:00", }, strict=True, full_row=False) expect = [STATE_COL_VALUE, EVENT_COL_STATE_VERSION, "checksum", "updated_by", "updated_at"] if sorted(upd_row.keys()) != sorted(expect): ok = False msgs.append("tx_write FAIL：UPDATE SET 列被意外裁剪 %s（应为 %s）" % (sorted(upd_row.keys()), sorted(expect))) else: msgs.append("tx_write PASS：UPDATE SET 语义不误报缺 NOT NULL 列" "（PBL-RTX-0004 误报侧已闭环）") except rtx_db.RtxError as exc: ok = False msgs.append("tx_write FAIL：UPDATE SET 仍被 %s 拒绝（PBL-RTX-0004 未闭环）" % exc.code) try: build_row(STATE_TABLE, {"session_id": 11, STATE_COL_KEY: "pos"}) ok = False msgs.append("tx_write FAIL：INSERT 真缺 NOT NULL 列未被拦（门禁恒真 = 漏报回归）") except rtx_db.RtxError as exc: if exc.code == "PBL-RTX-0004": msgs.append("tx_write PASS：INSERT 真缺列仍被 PBL-RTX-0004 拦下（fail-closed 未弱化）") else: ok = False msgs.append("tx_write FAIL：INSERT 缺列错误码异常 %s（应为 PBL-RTX-0004）" % exc.code) # 事件表用 nullable:"no" 写法，识别修复后必填集合必须非空（否则门禁又退回恒真） for tbl in (EVENT_TABLE, STATE_TABLE, SNAPSHOT_TABLE): req = required_columns(tbl) if not req: ok = False msgs.append("tx_write FAIL：%s required_columns 为空" "（模型 NOT NULL 写法未被识别，INSERT 门禁失效）" % tbl) else: msgs.append("tx_write PASS：%s 必填列 %d 个（已排除自增/DEFAULT 列）：%s" % (tbl, len(req), ",".join(req))) # 只给「必填列真子集」必须被拦（事件表缺 event_type / 状态表缺 tenant_id） for tbl, partial in ((EVENT_TABLE, {EVENT_COL_UID: "u1"}), (STATE_TABLE, {"session_id": 11, STATE_COL_KEY: "pos", EVENT_COL_STATE_VERSION: 1, "checksum": "c", "updated_by": "a", "created_at": "2026-09-21", "updated_at": "2026-09-21"})): try: build_row(tbl, partial) ok = False msgs.append("tx_write FAIL：%s 只给部分必填列却通过校验" % tbl) except rtx_db.RtxError as exc: if exc.code != "PBL-RTX-0004": ok = False msgs.append("tx_write FAIL：%s 部分列错误码 %s" % (tbl, exc.code)) else: msgs.append("tx_write PASS：%s 部分必填列被 PBL-RTX-0004 拦下" % tbl) # 写路径实际使用的 INSERT 列集合必须通过（防修复过头把主链路挡死） try: build_row(EVENT_TABLE, { "tenant_id": "7", EVENT_COL_UID: "u", "session_id": 11, "world_id": 1, "actor_id": "server", "event_type": "move", EVENT_COL_PAYLOAD: "{}", "causation_id": "", EVENT_COL_STATE_VERSION: 0, EVENT_COL_SEQ: 1, "occurred_at": "2026-09-21 12:00:00", "created_at": "2026-09-21 12:00:00", "updated_at": "2026-09-21 12:00:00"}) build_row(SNAPSHOT_TABLE, { "tenant_id": "7", "session_id": 11, "snapshot_uid": "s", "state_fingerprint": "f", "entities_json": "{}", "events_through_seq": 1, "reason": "apply_runtime_event", "created_at": "2026-09-21", "updated_at": "2026-09-21"}) msgs.append("tx_write PASS：事件表/快照表 INSERT 写路径列集合通过校验（无回归）") except rtx_db.RtxError as exc: ok = False msgs.append("tx_write FAIL：写路径 INSERT 列集合被 %s 拒绝（修复过头）" % exc.code) # 不可变列守卫：SET tenant_id/created_at 必须被 PBL-RTX-0006 拒绝 try: build_row(STATE_TABLE, {"tenant_id": "7", STATE_COL_VALUE: "{}"}, strict=True, full_row=False) ok = False msgs.append("tx_write FAIL：UPDATE SET 不可变列未被拦") except rtx_db.RtxError as exc: if exc.code == "PBL-RTX-0006": msgs.append("tx_write PASS：UPDATE 禁止 SET 不可变列（PBL-RTX-0006）") else: ok = False msgs.append("tx_write FAIL：不可变列错误码 %s（应为 PBL-RTX-0006）" % exc.code) for name, fn in (("apply_runtime_event", apply_runtime_event), ("poll_events", poll_events), ("read_states", read_states)):''' def main(): with io.open(TARGET, "r", encoding="utf-8") as f: src = f.read() pairs = [("required_columns", OLD_1, NEW_1), ("build_row", OLD_2, NEW_2), ("_apply_state UPDATE 调用点", OLD_3, NEW_3), ("self_check 闭环门禁", OLD_4, NEW_4)] for tag, old, new in pairs: n = src.count(old) if n != 1: print("FAIL: 片段[%s] 命中 %d 次（应为 1）" % (tag, n)) return 2 src = src.replace(old, new) with io.open(TARGET, "w", encoding="utf-8") as f: f.write(src) print("OK: 已应用 %d 处补丁 -> %s" % (len(pairs), TARGET)) return 0 if __name__ == "__main__": sys.exit(main())