diff --git a/models/pbl_runtime_event.json b/models/pbl_runtime_event.json index bfd9d8b..d4a32f9 100644 --- a/models/pbl_runtime_event.json +++ b/models/pbl_runtime_event.json @@ -5,7 +5,7 @@ "title": "运行时事件流(append-only)", "primary": ["id", "created_at"], "catelog": "entity", - "comment": "【本表定位 · world_sync 侧内联副本】运行时事件流（append-only：禁 UPDATE/DELETE，由权威模块 pbl_runtime_ext 的存储过程 pbl_assert_append_only + BEFORE UPDATE/BEFORE DELETE 触发器强制；按 created_at 做按月 RANGE COLUMNS(created_at) 分区，分区保留 14 个月）。物理主键为复合主键 (id, created_at) —— MySQL/MariaDB 要求分区表的所有唯一键（含主键）必须包含分区列，否则建表报错 1503。【结构权威源 · PM 裁决】PM 裁决（任务 u0EO4x3ajntR41acaHj5H 问答）：选 B —— 以 pbl_runtime_ext 为唯一真源；结构权威定义位于 modules/pbl_runtime_ext/models/pbl_runtime_event.json（M11b-1 已批准产物，本任务禁止反向修改），DDL 派生源为该模块 scripts/pbl_runtime_event_ddl.py 与 scripts/migrations/*.sql，建表/分区/触发器归 pbl_runtime_ext 所有，本文件不复制那套逻辑。【为什么是内联而不是引用】database-table-definition-spec 只承认 summary/fields/indexes/codes 四段式；scripts/validate_models_json.py（M11b-2a 子任务 A 修复版）对 key=schema 引用式写法直接判 FAIL（QC #5），故本文件按真源逐字段内联权威列集，作为 world_sync 侧的消费方契约视图；两份定义的一致性由 scripts/m11b2b_column_diff.py 与 scripts/m11b2b_selftest.py 机械核对，真源列集变更时必须同步本副本。【id 类型偏离处理 · 本子任务的选择与理由】真源 id 为 type=long + auto_increment（BIGINT AUTO_INCREMENT，真源侧已在其 summary[0].comment 自行登记该偏离）；子任务 A 的最终登记表 COMPLIANCE_DEVIATIONS 为空 {}（A 按 QC #5 选 (a)：不放行任何历史失效项），且本任务范围明确禁止修改校验器脚本，因此 world_sync 侧不再走「登记放行 id=long」路线，而是内联为 spec 合规写法 type=str / length=32、不带 auto_increment 键，由应用层生成 32 位 uuid 去横线十六进制串作为主键值（见 world_sync/pbl_runtime_tx.py::new_event_uid，正是 uuid4().hex 32 位）。与真源的 id 类型差异由 world_sync 侧消费方代码适配：写路径 INSERT 不显式提交 id 时交由真源 DDL 的 AUTO_INCREMENT 生成（BIGINT），本地/测试库（scripts 渲染的 sqlite DDL）则使用应用层生成的 str(32)；对外幂等与跨系统追溯一律使用 event_uid（str 64，uk_event_uid 唯一键），不依赖 id 的物理类型。此差异为有意选择、非疏漏，评审请勿据此判为结构缺陷。【消费方清单】modules/world_sync/world_sync/pbl_runtime_sql.py（EVENT_TABLE/EVENT_COLUMNS）、modules/world_sync/world_sync/pbl_runtime_tx.py::write_event_with_state（单事务事件+状态写入）、modules/world_sync/world_sync/pbl_runtime_authority.py（列契约解析层）；上述代码引用的列名必须全部出现在本文件 fields[] 中。【打包镜像说明】apps/scense/pkgs/world_sync/models/pbl_runtime_event.json 是本文件的打包镜像，只由 modules/world_sync/scripts/sync_models_mirror.py 生成，禁止手工双写；一致性由 modules/world_sync/scripts/m11b2b_selftest.py 的 sha256 断言守护。【私有键剥离说明】真源使用的 spec 未定义顶层扩展键 module/engine/charset/collate/tenant_scoped/append_only/write_paths/guards/partition/ddl_source，以及 fields[].ddl_type/comment/auto_increment、indexes[].comment、codes[].comment，在本副本中一律不保留（本文件根键仅 summary/fields/indexes/codes，summary[0] 仅 name/title/primary/catelog/comment，fields[] 仅 name/title/type/length/dec/nullable/default，indexes[] 仅 name/idxtype/idxfields，codes[] 仅 field/table/valuefield/textfield/cond）；这些物理实现细节以真源与其 DDL 派生源为准。" + "comment": "【本表定位】world_sync 侧消费方契约副本：运行时事件流（append-only，禁 UPDATE/DELETE；生产库由权威模块 pbl_runtime_ext 的存储过程 pbl_assert_append_only + BEFORE UPDATE/BEFORE DELETE 触发器强制，本副本只声明列契约、不复制那套物理实现）。【结构权威源 · PM 裁决】PM 裁决（任务 u0EO4x3ajntR41acaHj5H）选 B：以 modules/pbl_runtime_ext/models/pbl_runtime_event.json 为唯一真源（M11b-1 已批准产物，本任务禁止反向修改）；建表 DDL 派生源为该模块 scripts/pbl_runtime_event_ddl.py 与 scripts/migrations/*.sql。本文件是按真源逐字段内联的 world_sync 侧副本（25 列全量抄录），不是引用式 schema —— database-table-definition-spec 只承认 summary/fields/indexes/codes 四段式，scripts/validate_models_json.py 对 key=schema 引用式写法直接判 FAIL（QC #5），故 QC 要求内联。两份定义的一致性由 scripts/m11b2b_column_diff.py（代码列名 vs 权威列集双向 diff）与 scripts/m11b2b_selftest.py（四段式/抽象类型/主键/索引/编码 6 项断言 + 真源镜像 sha256）机械核对，真源列集变更时必须同步本副本。【id 生成方 · 单一口径，QC #4】本副本声明 id 由应用层生成：world_sync/pbl_runtime_tx.py::new_event_uid()（uuid4().hex，32 位小写十六进制），写路径 INSERT 列集（= 本文件 fields[] 全列，见 pbl_runtime_sql.py::EVENT_COLUMNS 由 fields[] 机械导出）显式提交 id，不依赖任何数据库 AUTO_INCREMENT。真源侧 id 为 type=long + auto_increment（BIGINT AUTO_INCREMENT），属其私有物理实现，本副本按 spec 归一为 str(32) 且不保留 auto_increment 键；两者差异已在交付件正文登记为 DEVIATION(WS-2) 并冒泡 agent.pm 裁决生产库列型收敛方案（改 VARCHAR(32) 或统一走 event_uid 定位），本副本不再同时声称两种生成方。对外幂等与跨系统追溯一律使用 event_uid（str 64，uk_event_uid），不依赖 id 的物理类型。【复合主键偏离 · 已登记，QC #4】summary[0].primary=[\"id\",\"created_at\"] 是对 spec「单键表恒为 [\\\"id\\\"]」的有意偏离：本表按 created_at 做按月 RANGE COLUMNS 分区，MySQL/MariaDB 要求分区表所有唯一键（含主键）必须包含分区列，否则建表报错 1503。真源依据行号：modules/pbl_runtime_ext/models/pbl_runtime_event.json:12（primary=[\"id\",\"created_at\"]）、同文件:6（comment 中 1503 约束说明与分区保留 14 个月）、同文件:40（fields[0].id type=long + auto_increment）。本副本不写 partition/guards/engine 等私有键，但主键必须与真源一致，否则 uk_event_uid/uk_tenant_idem（含 created_at）无法在本副本声明的 PK 下成立。该偏离已在 M11b-2b-B 交付件正文登记（DEVIATION(WS-1)），未写进校验器 COMPLIANCE_DEVIATIONS（子任务 A 最终登记表为空 {}，且本任务禁止修改校验器）。【私有键剥离】真源使用的 spec 未定义顶层扩展键 module/engine/charset/collate/tenant_scoped/append_only/write_paths/guards/partition/ddl_source，以及 fields[].ddl_type/comment/auto_increment、indexes[].comment、codes[].comment，本副本一律不保留：根键仅 summary/fields/indexes/codes，summary[0] 仅 name/title/primary/catelog/comment，fields[] 仅 name/title/type/length/dec/nullable/default，indexes[] 仅 name/idxtype/idxfields（主键不写进 indexes，由 summary[0].primary 表达），codes[] 仅 field/table/valuefield/textfield/cond。物理细节（分区、触发器、字符集、存储过程）以真源与其 DDL 派生源为准。【消费方清单】modules/world_sync/world_sync/pbl_runtime_sql.py（EVENT_TABLE/EVENT_COLUMNS，由本文件 fields[] 机械导出）、modules/world_sync/world_sync/pbl_runtime_tx.py::write_event_with_state（单事务事件+状态写入）、modules/world_sync/world_sync/pbl_runtime_authority.py（列契约解析层，已适配本内联四段式副本）、modules/world_sync/world_sync/init.py（事件写入门面）。上述代码引用的列名必须全部出现在本文件 fields[] 中，核对证据见 M11b-2b-B 交付件「字段交叉核对清单」。【打包镜像】apps/scense/pkgs/world_sync/models/pbl_runtime_event.json 是本文件的打包镜像，只由 modules/world_sync/scripts/sync_models_mirror.py 生成，禁止手工双写；一致性由 scripts/m11b2b_selftest.py 的 sha256 断言守护。" } ], "fields": [ diff --git a/world_sync/pbl_runtime_authority.py b/world_sync/pbl_runtime_authority.py index e4781d4..e86deab 100644 --- a/world_sync/pbl_runtime_authority.py +++ b/world_sync/pbl_runtime_authority.py @@ -1,25 +1,29 @@ # -*- coding: utf-8 -*- -"""pbl_runtime_authority —— M11b-2b：``key=schema`` 引用的解析层（列名唯一权威源）。 +"""pbl_runtime_authority —— M11b-2b：表定义 JSON 的列契约解析层（列名唯一入口）。 -**为什么存在** -PM 裁决（任务 ``u0EO4x3ajntR41acaHj5H``，方案 B）：``pbl_runtime_event`` / -``pbl_entity_state`` 两张表的**唯一真源**是 ``modules/pbl_runtime_ext/models/*.json`` -（M11b-1 已批准产物）。world_sync 侧 ``models/*.json`` 只保留 ``key=schema`` 引用， -**不再自带** summary/fields/indexes/codes 结构体。 +**结构现状（M11b-2b-B 起，QC #1/#2/#3 整改）** +PM 裁决 B：``pbl_runtime_event`` / ``pbl_entity_state`` 的**结构权威源**是 +``modules/pbl_runtime_ext/models/*.json``（M11b-1 已批准产物，禁止反向修改）。 +但 QC 明确要求 world_sync 侧 ``models/*.json`` 必须是 **database-table-definition-spec +四段式（summary/fields/indexes/codes）内联副本**，而不是 ``key=schema`` 引用式 +——校验器 ``scripts/validate_models_json.py`` 对引用式写法直接判 FAIL。 +因此本模块现在解析的是「按真源逐字段内联、并按 spec 归一（抽象类型、去私有键）的 +world_sync 侧副本」，它同时就是 world_sync 消费方的列契约（列名与真源一致）。 -于是「消费方列契约」必须从权威定义**机械导出**，而不是人工抄一遍——人工双写正是 -QC #1/#3 判定的「多份互相矛盾定义」的根因。本模块就是那个导出层： +本模块是列契约的**机械导出层**，不是人工抄写层： -* :func:`load_reference` —— 读本模块 ``models/.json``（key=schema 引用）； -* :func:`authority_model` —— 顺着引用解析到 pbl_runtime_ext 的真源 JSON； +* :func:`load_model` —— 读本模块 ``models/
.json``（四段式）并校验结构； * :func:`columns` / :func:`insert_columns` / :func:`primary` / :func:`indexes` - —— 暴露权威列集（``insert_columns`` 自动剔除 auto_increment 列）； -* :func:`ddl` —— 由权威列集渲染本地建表语句（仅供测试/本地起库；生产 DDL、 - 分区与 append-only 触发器归 pbl_runtime_ext 的 ``scripts/pbl_runtime_event_ddl.py`` - 与 ``scripts/migrations/*.sql`` 所有，本模块不复制那套逻辑）。 + —— 暴露列契约（``insert_columns`` 自动剔除自增列，兼容真源私有 ``auto_increment`` 键）； +* :func:`authority_path` —— 反查结构权威源（pbl_runtime_ext）在磁盘上的真实路径， + 供审计与 QC 取证（写入结果的 ``authority`` 字段）； +* :func:`ddl` —— 由列契约渲染本地建表语句（仅供测试/本地起库；生产 DDL、分区与 + append-only 触发器归 pbl_runtime_ext 的 ``scripts/pbl_runtime_event_ddl.py`` + 与 ``scripts/migrations/*.sql`` 所有，本模块不复制那套逻辑）； +* :func:`column_diff` —— 消费方用到的列 vs 列契约的双向 diff（fail-closed 的判据）。 -**fail-closed**：引用缺失、真源找不到、真源不是四段式、列集为空 —— 一律抛 -:class:`AuthorityResolutionError`，绝不退化成「猜列名」。 +**fail-closed**：文件缺失、JSON 不可解析、不是四段式、fields 为空 —— 一律抛 +:class:`AuthorityResolutionError`，绝不退化成「猜列名」或回落到旧的手写常量。 """ import json @@ -27,10 +31,13 @@ import os __all__ = [ "AuthorityResolutionError", - "REFERENCE_ROOT_KEYS", - "REFERENCE_SCHEMA_KEYS", "SPEC_ROOT_KEYS", - "load_reference", + "SPEC_SUMMARY_KEYS", + "SPEC_FIELD_KEYS", + "SPEC_INDEX_KEYS", + "SPEC_CODE_KEYS", + "AUTHORITY_MODULE", + "load_model", "authority_path", "authority_model", "columns", @@ -44,15 +51,16 @@ __all__ = [ "column_diff", ] -# models/
.json 是 key=schema 引用：根键只允许这两个 -REFERENCE_ROOT_KEYS = ("key", "schema") -# 引用体内允许的元信息键（白名单，自测脚本按此做机械断言） -REFERENCE_SCHEMA_KEYS = ( - "table", "authority_module", "authority_path", "owner", "milestone", - "decision_ref", "decision_basis", "structure_note", "consumers", "mirror_note", -) -# 真源（表定义）必须是四段式 +# world_sync 侧 models/
.json 必须是 spec 四段式（database-table-definition-spec） SPEC_ROOT_KEYS = ("summary", "fields", "indexes", "codes") +# 各段允许的键（白名单，自测脚本按此做机械断言；comment 为 spec 允许的可选说明键） +SPEC_SUMMARY_KEYS = ("name", "title", "primary", "catelog", "comment") +SPEC_FIELD_KEYS = ("name", "title", "type", "length", "dec", "nullable", "default") +SPEC_INDEX_KEYS = ("name", "idxtype", "idxfields") +SPEC_CODE_KEYS = ("field", "table", "valuefield", "textfield", "cond") + +# 结构权威源所在模块（PM 裁决 B；本模块只反查路径，不消费其私有键） +AUTHORITY_MODULE = "pbl_runtime_ext" MODULE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) MODELS_DIR = os.path.join(MODULE_ROOT, "models") @@ -77,7 +85,7 @@ _MYSQL_TYPES = { class AuthorityResolutionError(Exception): - """引用/真源解析失败（fail-closed，调用方不得吞掉）。""" + """模型缺失/不可解析/非四段式（fail-closed，调用方不得吞掉）。""" def __init__(self, message, **details): Exception.__init__(self, message) @@ -91,61 +99,83 @@ class AuthorityResolutionError(Exception): return "%s (%s)" % (self.message, parts) -def _reference_dir(): +def _models_dir(): return MODELS_DIR -def load_reference(table): - """读取本模块 ``models/
.json``，校验它是合法的 key=schema 引用。""" - if not table or not isinstance(table, str): - raise AuthorityResolutionError("table name is required", table=table) - path = os.path.join(_reference_dir(), table + ".json") +def _read_json(path): if not os.path.isfile(path): - raise AuthorityResolutionError("schema reference missing", path=path) + raise AuthorityResolutionError("model file missing", path=path) try: with open(path, "r", encoding="utf-8") as fh: - ref = json.load(fh) + data = json.load(fh) except ValueError as exc: - raise AuthorityResolutionError("schema reference not valid JSON: %s" % exc, - path=path) - if not isinstance(ref, dict) or ref.get("key") != "schema": + raise AuthorityResolutionError("model not valid JSON: %s" % exc, path=path) + if not isinstance(data, dict): + raise AuthorityResolutionError("model must be a JSON object", path=path) + return data + + +_MODEL_CACHE = {} + + +def load_model(table): + """读取并校验本模块 ``models/
.json``（spec 四段式内联副本）。 + + 校验项（fail-closed）：根键含 summary/fields/indexes/codes；summary 恰好 1 条； + fields 非空且每列有 name。列集与真源的一致性由 scripts/m11b2b_column_diff.py + 与 scripts/m11b2b_selftest.py 机械核对，不在运行期重复解析真源私有键。 + """ + if not table or not isinstance(table, str): + raise AuthorityResolutionError("table name is required", table=table) + if table in _MODEL_CACHE: + return _MODEL_CACHE[table] + path = os.path.join(_models_dir(), table + ".json") + model = _read_json(path) + missing = [k for k in SPEC_ROOT_KEYS if k not in model] + if missing: raise AuthorityResolutionError( - "models/%s.json must be a key=schema reference (no local structure)" % table, - path=path, root_keys=sorted(ref) if isinstance(ref, dict) else type(ref).__name__) - extra = set(ref) - set(REFERENCE_ROOT_KEYS) - if extra: - raise AuthorityResolutionError("reference carries non-whitelist root keys", - path=path, extra=sorted(extra)) - schema = ref.get("schema") - if not isinstance(schema, dict): - raise AuthorityResolutionError("reference.schema must be an object", path=path) - if schema.get("table") != table: - raise AuthorityResolutionError("reference.schema.table mismatches filename", - path=path, declared=schema.get("table")) - bad = set(schema) - set(REFERENCE_SCHEMA_KEYS) - if bad: - raise AuthorityResolutionError("reference.schema carries non-whitelist keys", - path=path, extra=sorted(bad)) - return ref + "model is not database-table-definition-spec four-section", + path=path, missing=missing, + hint="world_sync 侧必须是内联四段式副本（PM 裁决 B + QC #1/#5），不得用 key=schema 引用式") + summary = model.get("summary") + if not isinstance(summary, list) or len(summary) != 1 or not isinstance(summary[0], dict): + raise AuthorityResolutionError("summary must be an array with exactly 1 record", + path=path, got=type(summary).__name__) + if not str(summary[0].get("name") or "").strip(): + raise AuthorityResolutionError("summary[0].name is required", path=path) + fields = model.get("fields") + if not isinstance(fields, list) or not fields: + raise AuthorityResolutionError("model has no fields", path=path) + for f in fields: + if not isinstance(f, dict) or not str(f.get("name") or "").strip(): + raise AuthorityResolutionError("field without name", path=path, field=f) + _MODEL_CACHE[table] = model + return model + + +# 兼容旧调用名：本模块过去暴露 authority_model（解析引用指向的真源），现在列契约 +# 就是内联副本本身，语义等价。 +def authority_model(table): + """列契约模型（= world_sync 侧内联四段式副本，逐字段抄自 pbl_runtime_ext 真源）。""" + return load_model(table) def authority_path(table): - """把引用解析成磁盘上真实存在的真源路径（多候选，全部落空即报错）。""" - ref = load_reference(table) - schema = ref["schema"] - mod = schema.get("authority_module") - candidates = [] - declared = schema.get("authority_path") - if declared: - candidates.append(os.path.normpath(os.path.join(WORKSPACE_ROOT, declared))) - candidates.append(os.path.normpath(os.path.join(MODULE_ROOT, declared))) - if mod: - candidates.append(os.path.join(WORKSPACE_ROOT, "modules", mod, "models", - table + ".json")) - candidates.append(os.path.join(WORKSPACE_ROOT, "apps", "scense", "pkgs", mod, - "models", table + ".json")) - candidates.append(os.path.normpath(os.path.join(MODULE_ROOT, os.pardir, mod, - "models", table + ".json"))) + """反查结构权威源（pbl_runtime_ext）在磁盘上的真实路径，供审计取证。 + + 只用于写入结果里的 ``authority`` 字段与 QC 核对，**不参与列集解析**（列集来自 + 本模块内联副本）。找不到真源不抛错，返回声明式相对路径（真源缺失属部署期问题， + 不应让运行时写入失败）。 + """ + declared = os.path.join("modules", AUTHORITY_MODULE, "models", table + ".json") + candidates = [ + os.path.normpath(os.path.join(WORKSPACE_ROOT, declared)), + os.path.normpath(os.path.join(MODULE_ROOT, os.pardir, AUTHORITY_MODULE, + "models", table + ".json")), + os.path.join(WORKSPACE_ROOT, "apps", "scense", "pkgs", AUTHORITY_MODULE, + "models", table + ".json"), + ] tried = [] for cand in candidates: if cand in tried: @@ -153,55 +183,28 @@ def authority_path(table): tried.append(cand) if os.path.isfile(cand): return cand - raise AuthorityResolutionError("authority model not found for table", - table=table, tried=tried) - - -_MODEL_CACHE = {} - - -def authority_model(table): - """读取并校验真源 JSON（四段式 + fields 非空），带缓存。""" - if table in _MODEL_CACHE: - return _MODEL_CACHE[table] - path = authority_path(table) - try: - with open(path, "r", encoding="utf-8") as fh: - model = json.load(fh) - except ValueError as exc: - raise AuthorityResolutionError("authority model not valid JSON: %s" % exc, - path=path) - if not isinstance(model, dict): - raise AuthorityResolutionError("authority model must be an object", path=path) - missing = set(SPEC_ROOT_KEYS) - set(model) - if missing: - raise AuthorityResolutionError("authority model is not four-section", - path=path, missing=sorted(missing)) - fields = model.get("fields") - if not isinstance(fields, list) or not fields: - raise AuthorityResolutionError("authority model has no fields", path=path) - _MODEL_CACHE[table] = model - return model + return declared def field_map(table): - """``{列名: 列定义 dict}``（保持真源顺序）。""" + """``{列名: 列定义 dict}``（保持模型顺序）。""" out = {} - for f in authority_model(table)["fields"]: - name = f.get("name") - if not name: - raise AuthorityResolutionError("authority field without name", table=table) - out[name] = f + for f in load_model(table)["fields"]: + out[f["name"]] = f return out def columns(table): - """权威全列集（有序元组）。""" + """权威全列集（有序元组）—— 消费方 SQL 里允许出现的列名以此为唯一白名单。""" return tuple(field_map(table).keys()) def auto_increment_columns(table): - """由数据库自增生成的列（INSERT 时不得出现）。""" + """由数据库自增生成的列（INSERT 时不得出现）。 + + 内联副本按 spec 剥离了 ``auto_increment`` 私有键，故通常为空集；保留该逻辑是为了 + 当列契约指向真源（其 id 带 auto_increment）时仍能正确剔除该列。 + """ return tuple(n for n, f in field_map(table).items() if f.get("auto_increment") or str(f.get("default", "")).upper() in ("AUTO_INCREMENT", "IDENTITY")) @@ -214,29 +217,38 @@ def insert_columns(table): def primary(table): - summary = authority_model(table).get("summary") or [] - if not summary or not isinstance(summary[0], dict): - raise AuthorityResolutionError("authority model has no summary[0]", table=table) + """summary[0].primary（非空数组）。主键不写进 indexes（spec 约定）。""" + summary = load_model(table).get("summary") or [{}] pk = summary[0].get("primary") or summary[0].get("key") if isinstance(pk, str): pk = [p.strip() for p in pk.split(",") if p.strip()] - return tuple(pk or ()) + if not isinstance(pk, list) or not pk: + raise AuthorityResolutionError("model summary[0].primary must be a non-empty array", + table=table) + cols = set(columns(table)) + unknown = [p for p in pk if p not in cols] + if unknown: + raise AuthorityResolutionError("primary references unknown column", + table=table, unknown=unknown) + return tuple(pk) def indexes(table): - """规范化索引：``[{'name','idxtype','idxfields'}]``（兼容 unique/fields 写法）。""" + """二级索引/唯一键列表：``[{name, idxtype, idxfields}]``（兼容真源 unique/fields 写法）。""" out = [] - for idx in authority_model(table).get("indexes") or []: + for idx in (load_model(table).get("indexes") or []): + if not isinstance(idx, dict): + raise AuthorityResolutionError("index must be an object", table=table, index=idx) fields = idx.get("idxfields") or idx.get("fields") or [] if isinstance(fields, str): fields = [f.strip() for f in fields.split(",") if f.strip()] - kind = idx.get("idxtype") or ("unique" if idx.get("unique") else "index") - out.append({"name": idx.get("name"), "idxtype": kind, "idxfields": tuple(fields)}) + idxtype = idx.get("idxtype") or ("unique" if idx.get("unique") else "index") + out.append({"name": idx.get("name"), "idxtype": idxtype, "idxfields": list(fields)}) return out def table_title(table): - summary = authority_model(table).get("summary") or [{}] + summary = load_model(table).get("summary") or [{}] return summary[0].get("title") or summary[0].get("name") or table @@ -264,21 +276,19 @@ def _nullable(f): def ddl(table, dialect="sqlite"): - """由权威列集渲染 CREATE TABLE（测试/本地起库用）。 + """由列契约渲染 CREATE TABLE（测试/本地起库用）。 生产库的 DDL（分区、append-only 触发器、存储过程）归 pbl_runtime_ext 所有， 本函数**只保证列集与主键/唯一键一致**，不复制分区与触发器逻辑。 """ if dialect not in ("sqlite", "mysql"): raise AuthorityResolutionError("unsupported ddl dialect", dialect=dialect) - model = authority_model(table) + model = load_model(table) + gen = set(auto_increment_columns(table)) lines = [] for f in model["fields"]: - col = "`%s`" % f["name"] if dialect == "mysql" else f["name"] + col = ("`%s`" % f["name"]) if dialect == "mysql" else f["name"] piece = [" %s %s" % (col, _render_type(f, dialect))] - gen = bool(f.get("auto_increment")) - if dialect == "mysql" and gen: - piece.append(" AUTO_INCREMENT") if not _nullable(f) and not gen: piece.append(" NOT NULL") default = f.get("default") diff --git a/world_sync/pbl_runtime_sql.py b/world_sync/pbl_runtime_sql.py index 6bf192e..dafc571 100644 --- a/world_sync/pbl_runtime_sql.py +++ b/world_sync/pbl_runtime_sql.py @@ -2,8 +2,15 @@ """M11b-2 运行时单事务写入 —— SQL 方言层（纯 SQL 文本 + 占位符，不绑定任何 ORM/驱动）。 职责边界：本模块只负责「把 SQL 写对」——方言识别、占位符风格、READ COMMITTED 设置语句、 -悲观锁子句、以及两张表的建表 DDL（与 M11b-1 的 pbl_runtime_event schema 对齐）。 -真正的事务编排在 :mod:`world_sync.pbl_runtime_tx`。 +悲观锁子句、以及两张表的建表 DDL。真正的事务编排在 :mod:`world_sync.pbl_runtime_tx`。 + +**列集单一真源（M11b-2b-B 起，QC #3 整改）** +``EVENT_COLUMNS`` / ``STATE_COLUMNS`` 不再手写常量，而是从 +``modules/world_sync/models/pbl_runtime_event.json`` / ``pbl_entity_state.json`` +（spec 四段式内联副本，结构权威源 = pbl_runtime_ext，PM 裁决 B）的 ``fields[]`` +**机械导出**。旧版在本文件里手抄的 ``event_id`` / ``world_id``(state 表) / ``state`` +/ ``updated_by_event`` 等列名在真源中不存在，构成「代码一套、JSON 另一套」的双写分叉， +是 QC #3 判定的消费方契约断裂根因；现在这类漂移在结构上不可能再发生。 设计要点： 1. 用 DB-API 2.0 规范接口（``paramstyle`` / ``execute`` / ``rowcount`` / ``commit`` / @@ -12,11 +19,15 @@ ``pyformat`` → ``%(name)s``），避免在 MySQL 上写出 sqlite 的 ``?``； 3. ``SELECT ... FOR UPDATE`` 只有 MySQL/PG 支持，sqlite 侧返回空串（sqlite 写锁由事务本身保证）， 并发正确性在两条路径上都由「乐观锁 state_version」兜底； -4. 不引入任何广播 / 轮询 / 时延优化逻辑（明确排除在 M11b-2 范围之外）。 +4. 不引入任何广播 / 轮询 / 时延优化逻辑（明确排除在 M11b-2 范围之外）； +5. 建表 DDL 由 :func:`world_sync.pbl_runtime_authority.ddl` 按同一份模型渲染， + 本模块不再自带第二套列定义（生产库的分区/触发器 DDL 归 pbl_runtime_ext 所有）。 """ import sqlite3 +from . import pbl_runtime_authority as auth + __all__ = [ "EVENT_TABLE", "STATE_TABLE", @@ -34,6 +45,7 @@ __all__ = [ "ddl_for", "EVENT_COLUMNS", "STATE_COLUMNS", + "columns_of", "percentile", ] @@ -44,30 +56,24 @@ DIALECT_SQLITE = "sqlite" DIALECT_MYSQL = "mysql" DIALECT_POSTGRES = "postgres" -# 与 M11b-1 dev-notes-m11b1-pbl_runtime_event.md 的表定义对齐（append-only，无 updated_at）。 -EVENT_COLUMNS = ( - "event_id", # 事件主键（uuid/ULID，由调用方或本模块生成） - "tenant_id", # 租户，强制打头，缺失即拒绝写入 - "world_id", # 世界隔离维度，强制 - "session_id", # 运行时会话（可空） - "entity_id", # 关联实体 - "event_type", # 事件类型（如 state.updated） - "payload", # 事件体（JSON 文本） - "causation_id", # 因果链上游事件（可空） - "source", # 写入来源标识（模块名/引擎版本） - "state_version", # 本事件落库后实体的状态版本（与 pbl_entity_state 一致，便于回放） - "created_at", # 服务端时间戳（TEXT ISO8601 UTC） -) -STATE_COLUMNS = ( - "tenant_id", - "world_id", - "entity_id", - "state", # 实体状态（JSON 文本） - "state_version", # 乐观锁版本号，每次成功写入 +1 - "updated_at", - "updated_by_event", -) +def columns_of(table): + """取某表在模型 JSON（spec 四段式内联副本）里声明的全列集（有序）。 + + fail-closed：模型缺失 / 不是四段式 / fields 为空，一律由 + :mod:`world_sync.pbl_runtime_authority` 抛 + :class:`~world_sync.pbl_runtime_authority.AuthorityResolutionError`，不退回旧常量。 + """ + return auth.columns(table) + + +# append-only 事件流列集（= models/pbl_runtime_event.json fields[] 顺序，25 列，含应用层 +# 生成的 id；id 生成方见 world_sync/pbl_runtime_tx.py::new_event_uid 与该文件 summary[0].comment） +EVENT_COLUMNS = columns_of(EVENT_TABLE) + +# 实体状态列集（= models/pbl_entity_state.json fields[] 顺序，10 列；定位键是 +# tenant_id + session_id + entity_id，对应唯一键 uk_es，真源无 world_id / state 列） +STATE_COLUMNS = columns_of(STATE_TABLE) def detect_dialect(conn_or_module): @@ -162,80 +168,18 @@ def supports_transaction_control(dialect): def _ddl_sqlite(): - """返回 sqlite 建表语句列表（单测 / 本地环境用）。""" - return [ - """ - CREATE TABLE IF NOT EXISTS %s ( - event_id TEXT PRIMARY KEY, - tenant_id TEXT NOT NULL, - world_id TEXT NOT NULL, - session_id TEXT, - entity_id TEXT NOT NULL, - event_type TEXT NOT NULL, - payload TEXT NOT NULL, - causation_id TEXT, - source TEXT, - state_version INTEGER NOT NULL, - created_at TEXT NOT NULL - ) - """ % EVENT_TABLE, - """ - CREATE INDEX IF NOT EXISTS idx_pbl_runtime_event_world_time - ON %s (tenant_id, world_id, created_at) - """ % EVENT_TABLE, - """ - CREATE INDEX IF NOT EXISTS idx_pbl_runtime_event_entity - ON %s (tenant_id, world_id, entity_id, state_version) - """ % EVENT_TABLE, - """ - CREATE TABLE IF NOT EXISTS %s ( - tenant_id TEXT NOT NULL, - world_id TEXT NOT NULL, - entity_id TEXT NOT NULL, - state TEXT NOT NULL, - state_version INTEGER NOT NULL DEFAULT 0, - updated_at TEXT NOT NULL, - updated_by_event TEXT, - PRIMARY KEY (tenant_id, world_id, entity_id) - ) - """ % STATE_TABLE, - ] + """sqlite 建表语句列表（单测 / 本地环境用），由模型 JSON 渲染，无第二套列定义。""" + return [auth.ddl(EVENT_TABLE, DIALECT_SQLITE), auth.ddl(STATE_TABLE, DIALECT_SQLITE)] def _ddl_mysql(): - """返回 MySQL 建表语句列表（生产环境用，InnoDB + utf8mb4）。""" - return [ - """ - CREATE TABLE IF NOT EXISTS `%s` ( - `event_id` VARCHAR(64) NOT NULL COMMENT '事件主键', - `tenant_id` VARCHAR(64) NOT NULL COMMENT '租户ID(强制)', - `world_id` VARCHAR(64) NOT NULL COMMENT '世界ID(隔离维度)', - `session_id` VARCHAR(64) DEFAULT NULL COMMENT '运行时会话ID', - `entity_id` VARCHAR(64) NOT NULL COMMENT '关联实体ID', - `event_type` VARCHAR(64) NOT NULL COMMENT '事件类型', - `payload` JSON NOT NULL COMMENT '事件体', - `causation_id` VARCHAR(64) DEFAULT NULL COMMENT '因果链上游事件', - `source` VARCHAR(64) DEFAULT NULL COMMENT '写入来源', - `state_version` BIGINT NOT NULL COMMENT '落库后实体状态版本', - `created_at` DATETIME(3) NOT NULL COMMENT '服务端时间戳', - PRIMARY KEY (`event_id`), - KEY `idx_pbl_runtime_event_world_time` (`tenant_id`, `world_id`, `created_at`), - KEY `idx_pbl_runtime_event_entity` (`tenant_id`, `world_id`, `entity_id`, `state_version`) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL运行时事件(append-only,M11b-1)' - """ % EVENT_TABLE, - """ - CREATE TABLE IF NOT EXISTS `%s` ( - `tenant_id` VARCHAR(64) NOT NULL COMMENT '租户ID(强制)', - `world_id` VARCHAR(64) NOT NULL COMMENT '世界ID(隔离维度)', - `entity_id` VARCHAR(64) NOT NULL COMMENT '实体ID', - `state` JSON NOT NULL COMMENT '实体状态', - `state_version` BIGINT NOT NULL DEFAULT 0 COMMENT '乐观锁版本号', - `updated_at` DATETIME(3) NOT NULL COMMENT '更新时间', - `updated_by_event` VARCHAR(64) DEFAULT NULL COMMENT '最近生效事件', - PRIMARY KEY (`tenant_id`, `world_id`, `entity_id`) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL实体状态(单事务写入,M11b-2)' - """ % STATE_TABLE, - ] + """MySQL 建表语句列表（本地/预发用，InnoDB + utf8mb4），同样由模型 JSON 渲染。 + + 生产库真正的 ``pbl_runtime_event`` DDL（按月 RANGE COLUMNS 分区、append-only 存储过程 + 与触发器）归 pbl_runtime_ext 的 ``scripts/pbl_runtime_event_ddl.py`` 与 + ``scripts/migrations/*.sql`` 所有，本函数不复制那套物理细节。 + """ + return [auth.ddl(EVENT_TABLE, DIALECT_MYSQL), auth.ddl(STATE_TABLE, DIALECT_MYSQL)] #: 各方言建表语句集合，供宿主初始化 / 单测建库使用。