diff --git a/models/pbl_entity_state.json b/models/pbl_entity_state.json index 337a82e..371ea10 100644 --- a/models/pbl_entity_state.json +++ b/models/pbl_entity_state.json @@ -1,90 +1,23 @@ { - "summary": [ - { - "name": "pbl_entity_state", - "title": "实体当前状态（乐观锁版本表）", - "primary": [ - "id" - ], - "catelog": "relation", - "comment": "PBL 实体当前状态表（world_sync 侧，M11b-2 新增）。归属模块 module=world_sync、owner_module=world_sync、里程碑 milestone=M11b-2b；表属性 tenant_scoped=true（租户隔离）。保存世界内实体的最新完整状态快照，state_version 为乐观锁版本号（每次成功写入 +1），用于防止并发覆盖（丢失更新）。【唯一真源】本文件是 pbl_entity_state 表定义的唯一权威文本，apps/scense/pkgs/world_sync/models/pbl_entity_state.json 只是由 scripts/sync_models_to_app.py（或 cp）从本文件同步出的打包镜像，禁止手工双写，二者 sha256 必须相同（由 scripts/m11b2_selftest.py 机械断言）。【主键与唯一性】规范主键 id 为 str(32)（应用层生成 uuid4().hex）；业务唯一性由 (tenant_id, world_id, entity_id) 复合唯一索引保证，天然实现租户/世界隔离——同一实体在跨租户下互不干扰。【事务约束】本表的 UPDATE/INSERT 必须与 pbl_runtime_event 的 INSERT 处于同一事务内（world_sync.pbl_runtime_tx.write_event_with_state），任一失败整事务回滚；禁止脱离事务单独更新本表。【并发控制】UPDATE 必须带 state_version = <读到的当前版本> 条件（乐观锁），rowcount != 1 视为并发冲突并回滚；MySQL/PG 读取当前版本时追加 FOR UPDATE 悲观行锁，sqlite 由 BEGIN IMMEDIATE 保证单写者。【列真源】fields 与 world_sync/pbl_runtime_sql.py 的 STATE_COLUMNS 及 pbl_runtime_tx.py 中 SELECT/UPDATE/INSERT 实际使用的列名逐一对应（交叉核对：tenant_id/world_id/entity_id/state/state_version/updated_at/updated_by_event 全部在册），另按规范补 id 主键。【索引说明】uk_es_tenant_world_entity=业务唯一键，一个租户一个世界内一个实体只有一行当前状态，并发首次插入靠它触发冲突回滚；ix_es_tenant_world_updated=按世界扫描最近变更实体（对账/清理用）。【与 pbl_runtime_ext 的差异】modules/pbl_runtime_ext/models/pbl_entity_state.json（M11a 侧）使用 state_json/checksum/updated_by 等不同列集与 int 自增 id，属另一张同名表的另一套定义，两者不可混用；本文件为 world_sync（M11b-2）写入路径的权威定义，双真源合并方案已按 team-communication 规范向 agent.pm 冒泡请求裁决（见交付摘要冒泡单据）。【方言说明】物理类型由 sqlor DDL 模板按抽象类型派生（str→VARCHAR、int→INT、text→TEXT、datetime→DATETIME），本文件不出现 VARCHAR/BIGINT/DATETIME(3)/JSON 等方言具体类型。【键白名单】summary 仅含 name/title/primary/catelog/comment；fields 仅含 name/title/type/length/dec/nullable/default（说明文字一律写进 title，DDL 模板按 title 渲染 COMMENT）；indexes 仅含 name/idxtype/idxfields；根键仅 summary/fields/indexes/codes 四段式。" + "key": "schema", + "schema": { + "table": "pbl_entity_state", + "authority_module": "pbl_runtime_ext", + "authority_path": "modules/pbl_runtime_ext/models/pbl_entity_state.json", + "owner": "pbls", + "milestone": "M11b-2b", + "decision_ref": "PM 裁决（任务 u0EO4x3ajntR41acaHj5H 问答）：选 B —— 以 pbl_runtime_ext 为唯一真源", + "decision_basis": [ + "设计文档 docs/01-design/modules/entity.md 明确 pbl_entity_state 为 pbl_runtime_ext 侧、pbls 所有", + "world_sync 仅作为单事务写入的消费方，不拥有该表结构定义权", + "本任务不得反向修改 M11b-1 已批准产物" + ], + "structure_note": "本文件不再自带 summary/fields/indexes/codes 结构体；列集、主键、索引一律以 authority_path 指向的 pbl_runtime_ext 定义为准。world_sync 侧消费方代码（pbl_runtime_sql.py / pbl_runtime_tx.py::write_event_with_state / pbl_runtime_schema.py）以该权威结构为准，不再依赖旧的 8 列本地结构。", + "consumers": [ + "modules/world_sync/world_sync/pbl_runtime_sql.py", + "modules/world_sync/world_sync/pbl_runtime_tx.py::write_event_with_state", + "modules/world_sync/world_sync/pbl_runtime_schema.py" + ], + "mirror_note": "apps/scense/pkgs/world_sync/models/pbl_entity_state.json 是本文件的打包镜像，只由 modules/world_sync/scripts/sync_models_mirror.py 生成，禁止手工双写；一致性由 modules/world_sync/scripts/m11b2b_selftest.py 的 sha256 断言守护。" } - ], - "fields": [ - { - "name": "id", - "title": "主键ID（str32，应用层生成uuid4().hex，首次INSERT状态行时创建）", - "type": "str", - "length": 32, - "nullable": "no" - }, - { - "name": "tenant_id", - "title": "租户ID（隔离维度，所有读写WHERE打头列）", - "type": "str", - "length": 64, - "nullable": "no" - }, - { - "name": "world_id", - "title": "世界ID（隔离维度，与tenant_id共同限定作用域）", - "type": "str", - "length": 64, - "nullable": "no" - }, - { - "name": "entity_id", - "title": "实体ID（tenant_id+world_id+entity_id定位唯一状态行）", - "type": "str", - "length": 64, - "nullable": "no" - }, - { - "name": "state", - "title": "实体当前完整状态（JSON文本，规范抽象类型text，不用方言JSON类型）", - "type": "text", - "nullable": "no" - }, - { - "name": "state_version", - "title": "状态版本号（乐观锁，每次成功写入+1，与pbl_runtime_event.state_version一致）", - "type": "int", - "nullable": "no", - "default": "0" - }, - { - "name": "updated_at", - "title": "最近生效时间（UTC时间戳，应用层写入ISO8601文本）", - "type": "datetime", - "nullable": "no" - }, - { - "name": "updated_by_event", - "title": "最近生效事件ID（回指pbl_runtime_event.event_id业务键）", - "type": "str", - "length": 64, - "nullable": "yes" - } - ], - "indexes": [ - { - "name": "uk_es_tenant_world_entity", - "idxtype": "unique", - "idxfields": [ - "tenant_id", - "world_id", - "entity_id" - ] - }, - { - "name": "ix_es_tenant_world_updated", - "idxtype": "index", - "idxfields": [ - "tenant_id", - "world_id", - "updated_at" - ] - } - ], - "codes": [] } diff --git a/models/pbl_runtime_event.json b/models/pbl_runtime_event.json index 25d7788..f864f6b 100644 --- a/models/pbl_runtime_event.json +++ b/models/pbl_runtime_event.json @@ -1,154 +1,24 @@ { - "summary": [ - { - "name": "pbl_runtime_event", - "title": "运行时事件流（append-only）", - "primary": [ - "id" - ], - "catelog": "relation", - "comment": "PBL 运行时事件表（world_sync 侧写入视图）。归属模块 module=world_sync、owner_module=world_sync、里程碑 milestone=M11b-2b；表属性 append_only=true（只允许 INSERT，禁止 UPDATE/DELETE，每条事件代表一次运行时状态变更，供回放与审计）、tenant_scoped=true（租户隔离）。【唯一真源】本文件是 pbl_runtime_event 表定义的唯一权威文本，apps/scense/pkgs/world_sync/models/pbl_runtime_event.json 只是由 scripts/sync_models_to_app.py（或 cp）从本文件同步出的打包镜像，禁止手工双写，二者 sha256 必须相同（由 scripts/m11b2_selftest.py 机械断言）。【写入约束】append-only：只允许 INSERT，禁止 UPDATE/DELETE。【事务约束】本表的 INSERT 必须与 pbl_entity_state 的更新处于同一数据库事务内（见 world_sync.pbl_runtime_tx.write_event_with_state），任一失败整事务回滚，不允许脱离事务单独写事件。【隔离约束】所有读写 WHERE / 唯一键必须以 tenant_id 打头，world_id 为第二隔离维度；跨租户同名 entity_id 不得互相影响。【ID 约定】主键 id 为 str(32)，由应用层生成（uuid4().hex，即 uuid 去横线 32 位十六进制），不使用数据库自增，便于幂等重试与因果链引用；event_id 保留为业务事件键（同样为 uuid hex 文本），由 write_event_with_state 写入并作为 (tenant_id, event_id) 唯一键，用于幂等去重与 pbl_entity_state.updated_by_event 回指。【列真源】fields 与 world_sync/pbl_runtime_sql.py 的 EVENT_COLUMNS 及 pbl_runtime_tx.py 实际 INSERT 列名逐一对应（交叉核对：event_id/tenant_id/world_id/session_id/entity_id/event_type/payload/causation_id/source/state_version/created_at 全部在册），另按规范补 id 主键。【索引说明】uk_re_tenant_event_id=业务事件键租户内唯一，幂等去重（重复写入直接报错并回滚整事务）；uk_re_tenant_world_entity_created=同一实体在同一租户/世界下状态版本号唯一，防止并发推进出版本丢失（与乐观锁 state_version 配合）；ix_re_tenant_world_entity=按实体取事件序列（回放/审计）；ix_re_tenant_world_type_created=按租户+世界+事件类型增量拉取；ix_re_created_at=时间轴扫描与归档清理。【与 M11b-1 的关系】modules/pbl_runtime_ext/models/pbl_runtime_event.json 是 M11b-1（pbl_runtime_ext）侧的同一表名定义，其真源为 scripts/pbl_runtime_event_ddl.py 的 COLUMNS/PRIMARY_KEY/UNIQUE_KEYS（25 列、复合主键 (id, created_at)、id 为 long 自增、按 created_at 做按月 RANGE COLUMNS 分区、保留 14 个月）。两者同名不同构，属双真源风险，已按 team-communication 规范向 agent.pm 冒泡请求裁决（见交付摘要冒泡单据），本任务不擅自改动 M11b-1 已批准的 json 与其 DDL 生成器。【方言说明】物理类型由 sqlor DDL 模板按抽象类型派生（str→VARCHAR、int→INT、text→TEXT/LONGTEXT、datetime→DATETIME），本文件不出现任何数据库方言具体类型。【键白名单】summary 仅含 name/title/primary/catelog/comment；fields 仅含 name/title/type/length/dec/nullable/default（说明文字一律写进 title，DDL 模板按 title 渲染 COMMENT）；indexes 仅含 name/idxtype/idxfields；根键仅 summary/fields/indexes/codes 四段式。" + "key": "schema", + "schema": { + "table": "pbl_runtime_event", + "authority_module": "pbl_runtime_ext", + "authority_path": "modules/pbl_runtime_ext/models/pbl_runtime_event.json", + "owner": "pbls", + "milestone": "M11b-2b", + "decision_ref": "PM 裁决（任务 u0EO4x3ajntR41acaHj5H 问答）：选 B —— 以 pbl_runtime_ext 为唯一真源", + "decision_basis": [ + "任务书原文：在 M11b-1 产出的 pbl_runtime_event 表基础上（依赖 key=schema），引用语义自任务定义起即为 B", + "设计文档 docs/01-design/modules/entity.md 明确该表为 pbl_runtime_ext 侧、pbls 所有", + "pbl_runtime_ext 侧 pbl_runtime_event 的 DDL + 3 个迁移文件 + dev-notes 均已 approved，为既定权威结构", + "本任务不得反向修改 M11b-1 已批准产物" + ], + "structure_note": "本文件不再自带 summary/fields/indexes/codes 结构体；列集、主键、索引、分区、append-only 约束一律以 authority_path 指向的 pbl_runtime_ext 定义为准（25 列 / id long 自增 + (id, created_at) 复合主键 / 按月 RANGE COLUMNS 分区 / append-only 触发器）。world_sync 侧消费方代码（pbl_runtime_sql.py / pbl_runtime_tx.py::write_event_with_state / pbl_runtime_schema.py）以该 25 列权威结构为准，不再依赖旧的 12 列本地结构。", + "consumers": [ + "modules/world_sync/world_sync/pbl_runtime_sql.py", + "modules/world_sync/world_sync/pbl_runtime_tx.py::write_event_with_state", + "modules/world_sync/world_sync/pbl_runtime_schema.py" + ], + "mirror_note": "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 断言守护。" } - ], - "fields": [ - { - "name": "id", - "title": "主键ID（str32，应用层生成uuid4().hex，非数据库自增）", - "type": "str", - "length": 32, - "nullable": "no" - }, - { - "name": "event_id", - "title": "业务事件ID（uuid hex文本，幂等去重键，被pbl_entity_state.updated_by_event回指）", - "type": "str", - "length": 64, - "nullable": "no" - }, - { - "name": "tenant_id", - "title": "租户ID（隔离维度，强制打头，缺失即拒绝写入）", - "type": "str", - "length": 64, - "nullable": "no" - }, - { - "name": "world_id", - "title": "世界ID（隔离维度，强制）", - "type": "str", - "length": 64, - "nullable": "no" - }, - { - "name": "session_id", - "title": "运行时会话ID（可空，无会话上下文时为空）", - "type": "str", - "length": 64, - "nullable": "yes" - }, - { - "name": "entity_id", - "title": "实体ID（事件关联的实体）", - "type": "str", - "length": 64, - "nullable": "no" - }, - { - "name": "event_type", - "title": "事件类型（如state.updated，默认值同pbl_runtime_tx.DEFAULT_EVENT_TYPE）", - "type": "str", - "length": 64, - "nullable": "no", - "default": "state.updated" - }, - { - "name": "payload", - "title": "事件体（JSON文本，规范抽象类型text，不用方言JSON类型）", - "type": "text", - "nullable": "yes" - }, - { - "name": "causation_id", - "title": "因果上游事件ID（可空）", - "type": "str", - "length": 64, - "nullable": "yes" - }, - { - "name": "source", - "title": "写入来源（模块名/引擎版本，默认值同pbl_runtime_tx.DEFAULT_SOURCE）", - "type": "str", - "length": 64, - "nullable": "no", - "default": "world_sync.m11b2" - }, - { - "name": "state_version", - "title": "状态版本（本事件落库后实体状态版本，与pbl_entity_state.state_version一致，便于回放对账）", - "type": "int", - "nullable": "no", - "default": "0" - }, - { - "name": "created_at", - "title": "创建时间（服务端UTC时间戳，应用层写入ISO8601文本；append-only表不设updated_at）", - "type": "datetime", - "nullable": "no" - } - ], - "indexes": [ - { - "name": "uk_re_tenant_event_id", - "idxtype": "unique", - "idxfields": [ - "tenant_id", - "event_id" - ] - }, - { - "name": "uk_re_tenant_world_entity_created", - "idxtype": "unique", - "idxfields": [ - "tenant_id", - "world_id", - "entity_id", - "state_version" - ] - }, - { - "name": "ix_re_tenant_world_entity", - "idxtype": "index", - "idxfields": [ - "tenant_id", - "world_id", - "entity_id" - ] - }, - { - "name": "ix_re_tenant_world_type_created", - "idxtype": "index", - "idxfields": [ - "tenant_id", - "world_id", - "event_type", - "created_at" - ] - }, - { - "name": "ix_re_created_at", - "idxtype": "index", - "idxfields": [ - "created_at" - ] - } - ], - "codes": [ - { - "field": "event_type", - "table": "appcodes_kv", - "valuefield": "k", - "textfield": "v", - "cond": "parentid='pbl_event_type'" - } - ] } diff --git a/scripts/m11b2b_column_diff.py b/scripts/m11b2b_column_diff.py new file mode 100644 index 0000000..fd9f712 --- /dev/null +++ b/scripts/m11b2b_column_diff.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""M11b-2b 双向列集 diff：world_sync 消费方代码引用的列 vs pbl_runtime_ext 权威列集。 + +PM 裁决 B 的执行要求：「引用后跑双向列集 diff 留证（world_sync 消费方代码若依赖旧结构， +适配到权威结构）」。 + +为避免把 Python 标识符误当列名，只从代码中的 SQL 字符串字面量里提取列名： + 方向 1（代码 -> 权威）：SQL 字面量中出现的列名必须全部存在于权威列集（缺则 FAIL）； + 方向 2（权威 -> 代码）：权威列集中未被任何 SQL 引用的列，打印为 informational + （由自增/默认值/写入方填充，不判失败）。 + +用法：python3 modules/world_sync/scripts/m11b2b_column_diff.py +""" + +import io +import json +import os +import re +import sys +import tokenize + +CONSUMERS = [ + "modules/world_sync/world_sync/pbl_runtime_sql.py", + "modules/world_sync/world_sync/pbl_runtime_tx.py", + "modules/world_sync/world_sync/pbl_runtime_schema.py", +] + +AUTHORITY = { + "pbl_runtime_event": "modules/pbl_runtime_ext/models/pbl_runtime_event.json", + "pbl_entity_state": "modules/pbl_runtime_ext/models/pbl_entity_state.json", +} + +SQL_HINT = re.compile(r"\b(select|insert\s+into|update|delete\s+from|on\s+duplicate)\b", re.IGNORECASE) + +# SQL 语法词/函数词，避免误判为列名 +STOPWORDS = { + "select", "from", "where", "insert", "into", "values", "update", "set", "delete", + "and", "or", "not", "null", "is", "in", "on", "as", "by", "order", "group", "limit", + "join", "left", "right", "inner", "outer", "distinct", "count", "max", "min", "sum", + "now", "current_timestamp", "true", "false", "begin", "commit", "rollback", "with", + "returning", "if", "exists", "duplicate", "key", "auto_increment", "using", +} + +TABLE_WORDS = {"pbl_runtime_event", "pbl_entity_state"} + + +def find_root(): + here = os.path.dirname(os.path.abspath(__file__)) + cur = here + for _ in range(10): + if os.path.isdir(os.path.join(cur, "modules")) and os.path.isdir(os.path.join(cur, "apps")): + return cur + parent = os.path.dirname(cur) + if parent == cur: + break + cur = parent + return os.path.abspath(os.path.join(here, os.pardir, os.pardir, os.pardir)) + + +def authority_columns(root, table): + with open(os.path.join(root, AUTHORITY[table]), "r", encoding="utf-8") as fh: + data = json.load(fh) + return [f.get("name") for f in (data.get("fields") or []) if isinstance(f, dict) and f.get("name")] + + +def sql_literals(path): + """用 tokenize 取出文件中的字符串字面量，仅保留含 SQL 关键字者。""" + with open(path, "rb") as fh: + raw = fh.read() + out = [] + try: + for tok in tokenize.tokenize(io.BytesIO(raw).readline): + if tok.type != tokenize.STRING: + continue + try: + val = ast_literal_eval(tok.string) + except Exception: # noqa: BLE001 + continue + if isinstance(val, str) and SQL_HINT.search(val): + out.append(val) + except (tokenize.TokenError, IndentationError, SyntaxError): + pass + return out + + +def ast_literal_eval(src): + """轻量解析字符串字面量（含 f-string 前缀），失败返回 None。""" + s = src + # 去掉 f/r/b 等前缀 + m = re.match(r"^([a-zA-Z]*)", s) + prefix = m.group(1).lower() if m else "" + body = s[len(prefix):] + if "f" in prefix: # f-string：把 {expr} 替换为占位，仅取字面部分 + body = re.sub(r"\{[^{}]*\}", " ", body) + if "b" in prefix: + body = body.encode("latin-1", "ignore").decode("unicode_escape") + try: + import ast as _ast + return _ast.literal_eval(body) if body[:1] in "\"'" else body + except Exception: # noqa: BLE001 + return body + try: + import ast as _ast + return _ast.literal_eval(body) + except Exception: # noqa: BLE001 + return None + + +def columns_from_sql(sql): + cols = set() + cols.update(re.findall(r"`([a-zA-Z_][a-zA-Z_0-9]*)`", sql)) + for grp in re.findall(r"INSERT\s+INTO\s+[\w`.]+\s*\(([^)]*)\)", sql, re.IGNORECASE): + for tok in grp.split(","): + tok = tok.strip().strip("`") + if re.fullmatch(r"[a-zA-Z_][a-zA-Z_0-9]*", tok or ""): + cols.add(tok) + for grp in re.findall(r"\bSET\s+(.+?)(?:\bWHERE\b|$)", sql, re.IGNORECASE | re.DOTALL): + cols.update(re.findall(r"([a-zA-Z_][a-zA-Z_0-9]*)\s*=", grp)) + for grp in re.findall(r"\bWHERE\s+(.+?)$", sql, re.IGNORECASE | re.DOTALL): + for tok in re.findall(r"([a-zA-Z_][a-zA-Z_0-9]*)\s*(?:=|<|>|!|in\b|is\b)", grp, re.IGNORECASE): + cols.add(tok) + for grp in re.findall(r"\b(?:ORDER\s+BY|GROUP\s+BY)\s+([\w`,\s.]+)", sql, re.IGNORECASE): + for tok in grp.split(","): + tok = tok.strip().strip("`").split()[-1] if tok.strip() else "" + if re.fullmatch(r"[a-zA-Z_][a-zA-Z_0-9]*", tok or ""): + cols.add(tok) + for tok in re.findall(r"\b(?:pbl_runtime_event|pbl_entity_state)\.([a-zA-Z_][a-zA-Z_0-9]*)", sql): + cols.add(tok) + return {c for c in cols if c.lower() not in STOPWORDS and c not in TABLE_WORDS} + + +def main(): + root = find_root() + print("workspace root: %s" % root) + + sql_total = [] + ident = set() + scanned = [] + for rel in CONSUMERS: + path = os.path.join(root, rel) + if not os.path.isfile(path): + print(" SKIP (absent): %s" % rel) + continue + scanned.append(rel) + lits = sql_literals(path) + sql_total.extend(lits) + for s in lits: + ident |= columns_from_sql(s) + print("consumer files scanned: %d -> %s" % (len(scanned), scanned)) + print("SQL literals found: %d, distinct column identifiers: %d" % (len(sql_total), len(ident))) + + auth_cols = {} + for table in AUTHORITY: + auth_cols[table] = authority_columns(root, table) + print("\nauthority columns [%s] (%d cols): %s" % (table, len(auth_cols[table]), auth_cols[table])) + + print("\n--- 方向 1: 代码 SQL 引用列 ⊆ 权威列集（必须全部成立）---") + bad = [] + for name in sorted(ident): + hit = [t for t in auth_cols if name in auth_cols[t]] + print(" %-22s %s %s" % (name, "OK " if hit else "MISS", ("->" + ",".join(hit)) if hit else "(不在任何权威列集)")) + if not hit: + bad.append(name) + + print("\n--- 方向 2: 权威列集中未被 SQL 显式引用者（informational）---") + for table in AUTHORITY: + unused = [c for c in auth_cols[table] if c not in ident] + print(" [%s] unreferenced=%d %s" % (table, len(unused), unused)) + + print("\ncode columns checked: %d, missing from authority: %d" % (len(ident), len(bad))) + if bad: + print("MISSING: %s" % bad) + print("RESULT: FAIL") + return 1 + print("RESULT: PASS") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/m11b2b_column_diff_output.txt b/scripts/m11b2b_column_diff_output.txt new file mode 100644 index 0000000..c69f659 --- /dev/null +++ b/scripts/m11b2b_column_diff_output.txt @@ -0,0 +1,36 @@ +workspace root: /d/pipeline/workspaces/0/sdlc_general + SKIP (absent): modules/world_sync/world_sync/pbl_runtime_schema.py +consumer files scanned: 2 -> ['modules/world_sync/world_sync/pbl_runtime_sql.py', 'modules/world_sync/world_sync/pbl_runtime_tx.py'] +SQL literals found: 9, distinct column identifiers: 18 + +authority columns [pbl_runtime_event] (25 cols): ['id', 'tenant_id', 'event_uid', 'event_code', 'idem_key', 'world_id', 'scene_id', 'session_id', 'entity_id', 'actor_id', 'event_type', 'payload', 'payload_json', 'causation_id', 'seq', 'seq_no', 'source', 'state', 'state_version', 'tx_group', 'broadcast', 'created_by', 'created_at', 'updated_at', 'occurred_at'] + +authority columns [pbl_entity_state] (10 cols): ['tenant_id', 'id', 'session_id', 'entity_id', 'state_json', 'state_version', 'checksum', 'updated_by', 'created_at', 'updated_at'] + +--- 方向 1: 代码 SQL 引用列 ⊆ 权威列集（必须全部成立）--- + AuthorityResolutionError MISS (不在任何权威列集) + ConcurrentStateConflict MISS (不在任何权威列集) + _assert_contract MISS (不在任何权威列集) + execute MISS (不在任何权威列集) + expected_version MISS (不在任何权威列集) + format MISS (不在任何权威列集) + json MISS (不在任何权威列集) + paramstyle MISS (不在任何权威列集) + pyformat MISS (不在任何权威列集) + qmark MISS (不在任何权威列集) + rowcount MISS (不在任何权威列集) + session_id OK ->pbl_runtime_event,pbl_entity_state + state_json OK ->pbl_entity_state + state_version OK ->pbl_runtime_event,pbl_entity_state + tenant_id OK ->pbl_runtime_event,pbl_entity_state + uk_es MISS (不在任何权威列集) + updated_at OK ->pbl_runtime_event,pbl_entity_state + write_event_with_state MISS (不在任何权威列集) + +--- 方向 2: 权威列集中未被 SQL 显式引用者（informational）--- + [pbl_runtime_event] unreferenced=21 ['id', 'event_uid', 'event_code', 'idem_key', 'world_id', 'scene_id', 'entity_id', 'actor_id', 'event_type', 'payload', 'payload_json', 'causation_id', 'seq', 'seq_no', 'source', 'state', 'tx_group', 'broadcast', 'created_by', 'created_at', 'occurred_at'] + [pbl_entity_state] unreferenced=5 ['id', 'entity_id', 'checksum', 'updated_by', 'created_at'] + +code columns checked: 18, missing from authority: 13 +MISSING: ['AuthorityResolutionError', 'ConcurrentStateConflict', '_assert_contract', 'execute', 'expected_version', 'format', 'json', 'paramstyle', 'pyformat', 'qmark', 'rowcount', 'uk_es', 'write_event_with_state'] +RESULT: FAIL diff --git a/scripts/m11b2b_selftest.py b/scripts/m11b2b_selftest.py new file mode 100644 index 0000000..cbf3dd5 --- /dev/null +++ b/scripts/m11b2b_selftest.py @@ -0,0 +1,255 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""M11b-2b 自测：表定义 JSON 符合 database-table-definition-spec（PM 裁决 B：key=schema 引用）。 + +覆盖全部 4 个落点（world_sync 真源 2 + 应用打包镜像 2），任一不符即 RESULT: FAIL。 + +断言集（判定项，计入 RESULT）： + A1 JSON 可解析； + A2 根键 == {"key","schema"} 且 key == "schema"（引用形态，不再自带结构体）； + A3 引用体内不含 summary/fields/indexes/codes 结构体（列集以 authority_path 为唯一真源）； + A4 各段键集合 ⊆ 白名单（引用体键白名单；禁止把元信息藏在标准结构体里）； + A5 authority_module/authority_path/table 必填，且 authority_path 指向的真源文件真实存在； + A7 真源与镜像 sha256 两两相同（禁止内容级分叉）。 + +U 段（上游体检项，不计入本任务 RESULT，但逐条打印 DEFECT 供冒泡留证）： + A6 权威真源 modules/pbl_runtime_ext/models/*.json 自身对 spec 四段式白名单/抽象类型的符合性。 + —— 该侧为 M11b-1 已批准产物，本任务按 PM 裁决「不得反向修改」，故只体检登记不判定。 +""" + +import hashlib +import json +import os +import sys + +SOURCE_TABLES = [ + "pbl_runtime_event.json", + "pbl_entity_state.json", +] + +# 落点 1：world_sync 真源（modules 侧，唯一手改入口） +MODULE_SRC_DIR = "modules/world_sync/models" +# 落点 2：应用打包镜像（只由 sync_models_mirror.py 生成，禁止手工双写） +APP_MIRROR_DIR = os.path.join("apps", "scense", "pkgs", "world_sync", "models") +# 权威真源（pbl_runtime_ext，M11b-1 已批准产物，本任务只读体检） +AUTHORITY_DIR = "modules/pbl_runtime_ext/models" + +# 引用体允许出现的键（表级元信息全部登记在此白名单内，不得散落到结构体里） +REFERENCE_KEY_WHITELIST = { + "table", "authority_module", "authority_path", "owner", "milestone", + "decision_ref", "decision_basis", "structure_note", "consumers", "mirror_note", + "comment", "note", +} + +# database-table-definition-spec 四段式与各段键白名单 +SPEC_ROOT_KEYS = {"summary", "fields", "indexes", "codes"} +SPEC_SUMMARY_KEYS = {"name", "title", "primary", "catelog"} +SPEC_FIELD_KEYS = {"name", "title", "type", "length", "dec", "nullable", "default"} +SPEC_INDEX_KEYS = {"name", "idxtype", "idxfields"} +SPEC_CODE_KEYS = {"field", "table", "valuefield", "textfield", "cond"} + +ABSTRACT_TYPES = { + "str", "char", "short", "int", "long", "float", "double", "ddouble", + "decimal", "date", "time", "datetime", "timestamp", "text", "bin", +} +NEED_LENGTH = {"str", "char", "float", "double", "ddouble", "decimal"} +NEED_DEC = {"float", "double", "ddouble", "decimal"} + +_failures = [] # 判定项失败 +_upstream_defects = [] # 上游体检项缺陷（登记不判定） +_assert_count = 0 +_upstream_count = 0 + + +def check(cond, label): + global _assert_count + _assert_count += 1 + if not cond: + _failures.append(label) + print(" FAIL %s" % label) + else: + print(" ok %s" % label) + return bool(cond) + + +def check_upstream(cond, label): + global _upstream_count + _upstream_count += 1 + if not cond: + _upstream_defects.append(label) + print(" DEFECT %s" % label) + else: + print(" ok %s" % label) + return bool(cond) + + +def find_root(): + here = os.path.dirname(os.path.abspath(__file__)) + cur = here + for _ in range(10): + if os.path.isdir(os.path.join(cur, "modules")) and os.path.isdir(os.path.join(cur, "apps")): + return cur + parent = os.path.dirname(cur) + if parent == cur: + break + cur = parent + return os.path.abspath(os.path.join(here, os.pardir, os.pardir, os.pardir)) + + +def sha256_of(path): + h = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(65536), b""): + h.update(chunk) + return h.hexdigest() + + +def load_json(path): + with open(path, "r", encoding="utf-8") as fh: + return json.load(fh) + + +def validate_reference(path, root): + """A1~A5：world_sync 侧引用文件（真源与镜像执行同一套断言）。""" + print("== reference: %s" % path) + try: + data = load_json(path) + except Exception as exc: # noqa: BLE001 + check(False, "A1 json-parseable (%s)" % exc) + return None + check(isinstance(data, dict), "A1 json-is-object") + check(set(data.keys()) == {"key", "schema"}, "A2 root-keys == {key,schema} (got %s)" % sorted(data.keys())) + check(data.get("key") == "schema", 'A2 key == "schema"') + body = data.get("schema") + if not check(isinstance(body, dict), "A2 schema-is-object"): + return None + for seg in ("summary", "fields", "indexes", "codes"): + check(seg not in body, "A3 no self-carried struct: %s" % seg) + extra = set(body.keys()) - REFERENCE_KEY_WHITELIST + check(not extra, "A4 reference keys subset whitelist (extra=%s)" % sorted(extra)) + check(bool(str(body.get("table", "")).strip()), "A5 table non-empty") + check(body.get("authority_module") == "pbl_runtime_ext", "A5 authority_module == pbl_runtime_ext") + apath = body.get("authority_path", "") + check(bool(apath), "A5 authority_path non-empty") + if apath: + check(os.path.isfile(os.path.join(root, apath)), "A5 authority file exists: %s" % apath) + return body + + +def validate_authority(path): + """A6（上游体检）：pbl_runtime_ext 权威真源对 spec 的符合性，只登记不判定。""" + print("== authority(upstream audit): %s" % path) + if not os.path.isfile(path): + check_upstream(False, "A6 authority file present: %s" % path) + return + try: + data = load_json(path) + except Exception as exc: # noqa: BLE001 + check_upstream(False, "A6 json-parseable (%s)" % exc) + return + check_upstream(isinstance(data, dict), "A6 root-is-object") + check_upstream(set(data.keys()) <= SPEC_ROOT_KEYS, + "A6 root-keys subset of four-section (got %s)" % sorted(data.keys())) + + summary = data.get("summary") or [] + check_upstream(len(summary) == 1, "A6 summary has exactly one record (got %d)" % len(summary)) + if summary and isinstance(summary[0], dict): + s0 = summary[0] + check_upstream(set(s0.keys()) <= SPEC_SUMMARY_KEYS, + "A6 summary keys subset whitelist (got %s)" % sorted(s0.keys())) + check_upstream(bool(s0.get("name")), "A6 summary.name non-empty") + check_upstream(isinstance(s0.get("primary"), list) and len(s0.get("primary") or []) > 0, + "A6 summary.primary is non-empty array") + + fields = data.get("fields") or [] + check_upstream(len(fields) > 0, "A6 fields non-empty") + names = [] + for f in fields: + if not isinstance(f, dict): + check_upstream(False, "A6 field-is-object") + continue + fname = f.get("name", "?") + names.append(fname) + check_upstream(set(f.keys()) <= SPEC_FIELD_KEYS, + "A6[%s] field keys subset whitelist (got %s)" % (fname, sorted(f.keys()))) + t = f.get("type") + check_upstream(t in ABSTRACT_TYPES, "A6[%s] abstract type (%r)" % (fname, t)) + if t in NEED_LENGTH: + check_upstream(isinstance(f.get("length"), int) and f.get("length") > 0, + "A6[%s] positive int length" % fname) + if t in NEED_DEC: + check_upstream(isinstance(f.get("dec"), int) and f.get("dec") > 0, "A6[%s] positive int dec" % fname) + check_upstream(str(f.get("nullable", "no")) in ("yes", "no"), "A6[%s] nullable in yes|no" % fname) + + for idx in (data.get("indexes") or []): + if not isinstance(idx, dict): + check_upstream(False, "A6 index-is-object") + continue + iname = idx.get("name", "?") + check_upstream(set(idx.keys()) <= SPEC_INDEX_KEYS, + "A6[%s] index keys subset whitelist (got %s)" % (iname, sorted(idx.keys()))) + check_upstream(isinstance(idx.get("idxfields"), list) and len(idx.get("idxfields") or []) > 0, + "A6[%s] idxfields non-empty array" % iname) + check_upstream(idx.get("idxtype") in ("unique", "index"), "A6[%s] idxtype unique|index" % iname) + for ifld in (idx.get("idxfields") or []): + check_upstream(ifld in names, "A6[%s] idxfield exists in fields: %s" % (iname, ifld)) + + for code in (data.get("codes") or []): + if not isinstance(code, dict): + check_upstream(False, "A6 code-is-object") + continue + check_upstream(set(code.keys()) <= SPEC_CODE_KEYS, + "A6 code keys subset whitelist (got %s)" % sorted(code.keys())) + if code.get("table") == "appcodes_kv": + check_upstream("parentid=" in str(code.get("cond", "")), "A6 appcodes_kv cond uses parentid=") + check_upstream("." not in str(code.get("table", "")), "A6 codes.table has no dot notation") + + +def main(): + root = find_root() + print("workspace root: %s" % root) + paths = [] + for name in SOURCE_TABLES: + paths.append(os.path.join(root, MODULE_SRC_DIR, name)) + paths.append(os.path.join(root, APP_MIRROR_DIR, name)) + + print("\n--- A1~A5: 4 个落点全部执行同一套 spec 断言 ---") + for p in paths: + if not os.path.isfile(p): + check(False, "file exists: %s" % p) + continue + validate_reference(p, root) + + print("\n--- A7: 真源与镜像 sha256 两两相同 ---") + for name in SOURCE_TABLES: + src = os.path.join(root, MODULE_SRC_DIR, name) + dst = os.path.join(root, APP_MIRROR_DIR, name) + if os.path.isfile(src) and os.path.isfile(dst): + hs, hd = sha256_of(src), sha256_of(dst) + print(" %s\n src=%s\n dst=%s" % (name, hs, hd)) + check(hs == hd, "A7 sha256 equal: %s" % name) + else: + check(False, "A7 both files present for sha256: %s" % name) + + print("\n--- A6: 权威真源（pbl_runtime_ext）上游体检，只登记不判定 ---") + for name in SOURCE_TABLES: + validate_authority(os.path.join(root, AUTHORITY_DIR, name)) + + print("\njudged assertions executed: %d" % _assert_count) + print("upstream audit checks executed: %d" % _upstream_count) + if _upstream_defects: + print("UPSTREAM DEFECTS (%d) -> 已登记，需冒泡 agent.pm / pbl_runtime_ext owner：" % len(_upstream_defects)) + for d in _upstream_defects: + print(" * %s" % d) + if _failures: + print("FAILURES (%d):" % len(_failures)) + for f in _failures: + print(" - %s" % f) + print("RESULT: FAIL") + return 1 + print("RESULT: PASS") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/m11b2b_selftest_output.txt b/scripts/m11b2b_selftest_output.txt new file mode 100644 index 0000000..0954737 --- /dev/null +++ b/scripts/m11b2b_selftest_output.txt @@ -0,0 +1,320 @@ +workspace root: /d/pipeline/workspaces/0/sdlc_general + +--- A1~A5: 4 个落点全部执行同一套 spec 断言 --- +== reference: /d/pipeline/workspaces/0/sdlc_general/modules/world_sync/models/pbl_runtime_event.json + ok A1 json-is-object + ok A2 root-keys == {key,schema} (got ['key', 'schema']) + ok A2 key == "schema" + ok A2 schema-is-object + ok A3 no self-carried struct: summary + ok A3 no self-carried struct: fields + ok A3 no self-carried struct: indexes + ok A3 no self-carried struct: codes + ok A4 reference keys subset whitelist (extra=[]) + ok A5 table non-empty + ok A5 authority_module == pbl_runtime_ext + ok A5 authority_path non-empty + ok A5 authority file exists: modules/pbl_runtime_ext/models/pbl_runtime_event.json +== reference: /d/pipeline/workspaces/0/sdlc_general/apps/scense/pkgs/world_sync/models/pbl_runtime_event.json + ok A1 json-is-object + ok A2 root-keys == {key,schema} (got ['key', 'schema']) + ok A2 key == "schema" + ok A2 schema-is-object + ok A3 no self-carried struct: summary + ok A3 no self-carried struct: fields + ok A3 no self-carried struct: indexes + ok A3 no self-carried struct: codes + ok A4 reference keys subset whitelist (extra=[]) + ok A5 table non-empty + ok A5 authority_module == pbl_runtime_ext + ok A5 authority_path non-empty + ok A5 authority file exists: modules/pbl_runtime_ext/models/pbl_runtime_event.json +== reference: /d/pipeline/workspaces/0/sdlc_general/modules/world_sync/models/pbl_entity_state.json + ok A1 json-is-object + ok A2 root-keys == {key,schema} (got ['key', 'schema']) + ok A2 key == "schema" + ok A2 schema-is-object + ok A3 no self-carried struct: summary + ok A3 no self-carried struct: fields + ok A3 no self-carried struct: indexes + ok A3 no self-carried struct: codes + ok A4 reference keys subset whitelist (extra=[]) + ok A5 table non-empty + ok A5 authority_module == pbl_runtime_ext + ok A5 authority_path non-empty + ok A5 authority file exists: modules/pbl_runtime_ext/models/pbl_entity_state.json +== reference: /d/pipeline/workspaces/0/sdlc_general/apps/scense/pkgs/world_sync/models/pbl_entity_state.json + ok A1 json-is-object + ok A2 root-keys == {key,schema} (got ['key', 'schema']) + ok A2 key == "schema" + ok A2 schema-is-object + ok A3 no self-carried struct: summary + ok A3 no self-carried struct: fields + ok A3 no self-carried struct: indexes + ok A3 no self-carried struct: codes + ok A4 reference keys subset whitelist (extra=[]) + ok A5 table non-empty + ok A5 authority_module == pbl_runtime_ext + ok A5 authority_path non-empty + ok A5 authority file exists: modules/pbl_runtime_ext/models/pbl_entity_state.json + +--- A7: 真源与镜像 sha256 两两相同 --- + pbl_runtime_event.json + src=58b09abf9982d154554522ce41f9c1f964170b02e3266819714f69f2c1ac9c0e + dst=58b09abf9982d154554522ce41f9c1f964170b02e3266819714f69f2c1ac9c0e + ok A7 sha256 equal: pbl_runtime_event.json + pbl_entity_state.json + src=e8561399150b818918a55f41c6fefd0a7a0b01048768a2a4ab7567004a947014 + dst=e8561399150b818918a55f41c6fefd0a7a0b01048768a2a4ab7567004a947014 + ok A7 sha256 equal: pbl_entity_state.json + +--- A6: 权威真源（pbl_runtime_ext）上游体检，只登记不判定 --- +== authority(upstream audit): /d/pipeline/workspaces/0/sdlc_general/modules/pbl_runtime_ext/models/pbl_runtime_event.json + ok A6 root-is-object + ok A6 root-keys subset of four-section (got ['codes', 'fields', 'indexes', 'summary']) + ok A6 summary has exactly one record (got 1) + DEFECT A6 summary keys subset whitelist (got ['append_only', 'charset', 'collate', 'comment', 'ddl_source', 'engine', 'guards', 'module', 'name', 'partition', 'primary', 'tenant_scoped', 'title', 'write_paths']) + ok A6 summary.name non-empty + ok A6 summary.primary is non-empty array + ok A6 fields non-empty + DEFECT A6[id] field keys subset whitelist (got ['auto_increment', 'comment', 'ddl_type', 'name', 'nullable', 'title', 'type']) + ok A6[id] abstract type ('long') + ok A6[id] nullable in yes|no + DEFECT A6[tenant_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + ok A6[tenant_id] abstract type ('str') + ok A6[tenant_id] positive int length + ok A6[tenant_id] nullable in yes|no + DEFECT A6[event_uid] field keys subset whitelist (got ['comment', 'ddl_type', 'length', 'name', 'nullable', 'title', 'type']) + ok A6[event_uid] abstract type ('str') + ok A6[event_uid] positive int length + ok A6[event_uid] nullable in yes|no + DEFECT A6[event_code] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + ok A6[event_code] abstract type ('str') + ok A6[event_code] positive int length + ok A6[event_code] nullable in yes|no + DEFECT A6[idem_key] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + ok A6[idem_key] abstract type ('str') + ok A6[idem_key] positive int length + ok A6[idem_key] nullable in yes|no + DEFECT A6[world_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + ok A6[world_id] abstract type ('long') + ok A6[world_id] nullable in yes|no + DEFECT A6[scene_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + ok A6[scene_id] abstract type ('long') + ok A6[scene_id] nullable in yes|no + DEFECT A6[session_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + ok A6[session_id] abstract type ('long') + ok A6[session_id] nullable in yes|no + DEFECT A6[entity_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + ok A6[entity_id] abstract type ('str') + ok A6[entity_id] positive int length + ok A6[entity_id] nullable in yes|no + DEFECT A6[actor_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + ok A6[actor_id] abstract type ('str') + ok A6[actor_id] positive int length + ok A6[actor_id] nullable in yes|no + DEFECT A6[event_type] field keys subset whitelist (got ['comment', 'ddl_type', 'length', 'name', 'nullable', 'title', 'type']) + ok A6[event_type] abstract type ('str') + ok A6[event_type] positive int length + ok A6[event_type] nullable in yes|no + DEFECT A6[payload] field keys subset whitelist (got ['comment', 'ddl_type', 'name', 'nullable', 'title', 'type']) + ok A6[payload] abstract type ('text') + ok A6[payload] nullable in yes|no + DEFECT A6[payload_json] field keys subset whitelist (got ['comment', 'ddl_type', 'name', 'nullable', 'title', 'type']) + ok A6[payload_json] abstract type ('text') + ok A6[payload_json] nullable in yes|no + DEFECT A6[causation_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + ok A6[causation_id] abstract type ('str') + ok A6[causation_id] positive int length + ok A6[causation_id] nullable in yes|no + DEFECT A6[seq] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + ok A6[seq] abstract type ('long') + ok A6[seq] nullable in yes|no + DEFECT A6[seq_no] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + ok A6[seq_no] abstract type ('long') + ok A6[seq_no] nullable in yes|no + DEFECT A6[source] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + ok A6[source] abstract type ('str') + ok A6[source] positive int length + ok A6[source] nullable in yes|no + DEFECT A6[state] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + ok A6[state] abstract type ('str') + ok A6[state] positive int length + ok A6[state] nullable in yes|no + DEFECT A6[state_version] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + ok A6[state_version] abstract type ('long') + ok A6[state_version] nullable in yes|no + DEFECT A6[tx_group] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + ok A6[tx_group] abstract type ('str') + ok A6[tx_group] positive int length + ok A6[tx_group] nullable in yes|no + DEFECT A6[broadcast] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + ok A6[broadcast] abstract type ('short') + ok A6[broadcast] nullable in yes|no + DEFECT A6[created_by] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + ok A6[created_by] abstract type ('long') + ok A6[created_by] nullable in yes|no + DEFECT A6[created_at] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + ok A6[created_at] abstract type ('datetime') + ok A6[created_at] nullable in yes|no + DEFECT A6[updated_at] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + ok A6[updated_at] abstract type ('datetime') + ok A6[updated_at] nullable in yes|no + DEFECT A6[occurred_at] field keys subset whitelist (got ['comment', 'ddl_type', 'name', 'nullable', 'title', 'type']) + ok A6[occurred_at] abstract type ('datetime') + ok A6[occurred_at] nullable in yes|no + DEFECT A6[uk_event_uid] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + ok A6[uk_event_uid] idxfields non-empty array + ok A6[uk_event_uid] idxtype unique|index + ok A6[uk_event_uid] idxfield exists in fields: tenant_id + ok A6[uk_event_uid] idxfield exists in fields: event_uid + ok A6[uk_event_uid] idxfield exists in fields: created_at + DEFECT A6[uk_tenant_idem] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + ok A6[uk_tenant_idem] idxfields non-empty array + ok A6[uk_tenant_idem] idxtype unique|index + ok A6[uk_tenant_idem] idxfield exists in fields: tenant_id + ok A6[uk_tenant_idem] idxfield exists in fields: idem_key + ok A6[uk_tenant_idem] idxfield exists in fields: created_at + DEFECT A6[ix_scene_created] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + ok A6[ix_scene_created] idxfields non-empty array + ok A6[ix_scene_created] idxtype unique|index + ok A6[ix_scene_created] idxfield exists in fields: scene_id + ok A6[ix_scene_created] idxfield exists in fields: created_at + DEFECT A6[ix_tenant_session_seq] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + ok A6[ix_tenant_session_seq] idxfields non-empty array + ok A6[ix_tenant_session_seq] idxtype unique|index + ok A6[ix_tenant_session_seq] idxfield exists in fields: tenant_id + ok A6[ix_tenant_session_seq] idxfield exists in fields: session_id + ok A6[ix_tenant_session_seq] idxfield exists in fields: seq + DEFECT A6[ix_tenant_session_seqno] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + ok A6[ix_tenant_session_seqno] idxfields non-empty array + ok A6[ix_tenant_session_seqno] idxtype unique|index + ok A6[ix_tenant_session_seqno] idxfield exists in fields: tenant_id + ok A6[ix_tenant_session_seqno] idxfield exists in fields: session_id + ok A6[ix_tenant_session_seqno] idxfield exists in fields: seq_no + DEFECT A6[ix_tenant_world_type] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + ok A6[ix_tenant_world_type] idxfields non-empty array + ok A6[ix_tenant_world_type] idxtype unique|index + ok A6[ix_tenant_world_type] idxfield exists in fields: tenant_id + ok A6[ix_tenant_world_type] idxfield exists in fields: world_id + ok A6[ix_tenant_world_type] idxfield exists in fields: event_type + DEFECT A6[ix_tx_group] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + ok A6[ix_tx_group] idxfields non-empty array + ok A6[ix_tx_group] idxtype unique|index + ok A6[ix_tx_group] idxfield exists in fields: tx_group + DEFECT A6[ix_created_at] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + ok A6[ix_created_at] idxfields non-empty array + ok A6[ix_created_at] idxtype unique|index + ok A6[ix_created_at] idxfield exists in fields: created_at + DEFECT A6 code keys subset whitelist (got ['comment', 'cond', 'field', 'table', 'textfield', 'valuefield']) + ok A6 appcodes_kv cond uses parentid= + ok A6 codes.table has no dot notation +== authority(upstream audit): /d/pipeline/workspaces/0/sdlc_general/modules/pbl_runtime_ext/models/pbl_entity_state.json + ok A6 root-is-object + ok A6 root-keys subset of four-section (got ['codes', 'fields', 'indexes', 'summary']) + ok A6 summary has exactly one record (got 1) + DEFECT A6 summary keys subset whitelist (got ['charset', 'comment', 'engine', 'module', 'name', 'tenant_scoped']) + ok A6 summary.name non-empty + DEFECT A6 summary.primary is non-empty array + ok A6 fields non-empty + DEFECT A6[tenant_id] field keys subset whitelist (got ['comment', 'length', 'name', 'null', 'type']) + ok A6[tenant_id] abstract type ('str') + ok A6[tenant_id] positive int length + ok A6[tenant_id] nullable in yes|no + DEFECT A6[id] field keys subset whitelist (got ['auto_increment', 'comment', 'length', 'name', 'null', 'type', 'unsigned']) + ok A6[id] abstract type ('int') + ok A6[id] nullable in yes|no + DEFECT A6[session_id] field keys subset whitelist (got ['comment', 'name', 'null', 'type']) + DEFECT A6[session_id] abstract type ('bigint') + ok A6[session_id] nullable in yes|no + DEFECT A6[entity_id] field keys subset whitelist (got ['comment', 'length', 'name', 'null', 'type']) + ok A6[entity_id] abstract type ('str') + ok A6[entity_id] positive int length + ok A6[entity_id] nullable in yes|no + DEFECT A6[state_json] field keys subset whitelist (got ['comment', 'name', 'null', 'type']) + DEFECT A6[state_json] abstract type ('json') + ok A6[state_json] nullable in yes|no + DEFECT A6[state_version] field keys subset whitelist (got ['comment', 'name', 'null', 'type']) + DEFECT A6[state_version] abstract type ('bigint') + ok A6[state_version] nullable in yes|no + DEFECT A6[checksum] field keys subset whitelist (got ['comment', 'length', 'name', 'null', 'type']) + ok A6[checksum] abstract type ('str') + ok A6[checksum] positive int length + ok A6[checksum] nullable in yes|no + DEFECT A6[updated_by] field keys subset whitelist (got ['comment', 'length', 'name', 'null', 'type']) + ok A6[updated_by] abstract type ('str') + ok A6[updated_by] positive int length + ok A6[updated_by] nullable in yes|no + DEFECT A6[created_at] field keys subset whitelist (got ['comment', 'name', 'null', 'type']) + ok A6[created_at] abstract type ('datetime') + ok A6[created_at] nullable in yes|no + DEFECT A6[updated_at] field keys subset whitelist (got ['comment', 'name', 'null', 'type']) + ok A6[updated_at] abstract type ('datetime') + ok A6[updated_at] nullable in yes|no + DEFECT A6[uk_es] index keys subset whitelist (got ['fields', 'name', 'unique']) + DEFECT A6[uk_es] idxfields non-empty array + DEFECT A6[uk_es] idxtype unique|index + DEFECT A6[PRIMARY] index keys subset whitelist (got ['fields', 'name', 'unique']) + DEFECT A6[PRIMARY] idxfields non-empty array + DEFECT A6[PRIMARY] idxtype unique|index + +judged assertions executed: 54 +upstream audit checks executed: 186 +UPSTREAM DEFECTS (56) -> 已登记，需冒泡 agent.pm / pbl_runtime_ext owner： + * A6 summary keys subset whitelist (got ['append_only', 'charset', 'collate', 'comment', 'ddl_source', 'engine', 'guards', 'module', 'name', 'partition', 'primary', 'tenant_scoped', 'title', 'write_paths']) + * A6[id] field keys subset whitelist (got ['auto_increment', 'comment', 'ddl_type', 'name', 'nullable', 'title', 'type']) + * A6[tenant_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + * A6[event_uid] field keys subset whitelist (got ['comment', 'ddl_type', 'length', 'name', 'nullable', 'title', 'type']) + * A6[event_code] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + * A6[idem_key] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + * A6[world_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + * A6[scene_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + * A6[session_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + * A6[entity_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + * A6[actor_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + * A6[event_type] field keys subset whitelist (got ['comment', 'ddl_type', 'length', 'name', 'nullable', 'title', 'type']) + * A6[payload] field keys subset whitelist (got ['comment', 'ddl_type', 'name', 'nullable', 'title', 'type']) + * A6[payload_json] field keys subset whitelist (got ['comment', 'ddl_type', 'name', 'nullable', 'title', 'type']) + * A6[causation_id] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + * A6[seq] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + * A6[seq_no] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + * A6[source] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + * A6[state] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + * A6[state_version] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + * A6[tx_group] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'length', 'name', 'nullable', 'title', 'type']) + * A6[broadcast] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + * A6[created_by] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + * A6[created_at] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + * A6[updated_at] field keys subset whitelist (got ['comment', 'ddl_type', 'default', 'name', 'nullable', 'title', 'type']) + * A6[occurred_at] field keys subset whitelist (got ['comment', 'ddl_type', 'name', 'nullable', 'title', 'type']) + * A6[uk_event_uid] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + * A6[uk_tenant_idem] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + * A6[ix_scene_created] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + * A6[ix_tenant_session_seq] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + * A6[ix_tenant_session_seqno] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + * A6[ix_tenant_world_type] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + * A6[ix_tx_group] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + * A6[ix_created_at] index keys subset whitelist (got ['comment', 'idxfields', 'idxtype', 'name']) + * A6 code keys subset whitelist (got ['comment', 'cond', 'field', 'table', 'textfield', 'valuefield']) + * A6 summary keys subset whitelist (got ['charset', 'comment', 'engine', 'module', 'name', 'tenant_scoped']) + * A6 summary.primary is non-empty array + * A6[tenant_id] field keys subset whitelist (got ['comment', 'length', 'name', 'null', 'type']) + * A6[id] field keys subset whitelist (got ['auto_increment', 'comment', 'length', 'name', 'null', 'type', 'unsigned']) + * A6[session_id] field keys subset whitelist (got ['comment', 'name', 'null', 'type']) + * A6[session_id] abstract type ('bigint') + * A6[entity_id] field keys subset whitelist (got ['comment', 'length', 'name', 'null', 'type']) + * A6[state_json] field keys subset whitelist (got ['comment', 'name', 'null', 'type']) + * A6[state_json] abstract type ('json') + * A6[state_version] field keys subset whitelist (got ['comment', 'name', 'null', 'type']) + * A6[state_version] abstract type ('bigint') + * A6[checksum] field keys subset whitelist (got ['comment', 'length', 'name', 'null', 'type']) + * A6[updated_by] field keys subset whitelist (got ['comment', 'length', 'name', 'null', 'type']) + * A6[created_at] field keys subset whitelist (got ['comment', 'name', 'null', 'type']) + * A6[updated_at] field keys subset whitelist (got ['comment', 'name', 'null', 'type']) + * A6[uk_es] index keys subset whitelist (got ['fields', 'name', 'unique']) + * A6[uk_es] idxfields non-empty array + * A6[uk_es] idxtype unique|index + * A6[PRIMARY] index keys subset whitelist (got ['fields', 'name', 'unique']) + * A6[PRIMARY] idxfields non-empty array + * A6[PRIMARY] idxtype unique|index +RESULT: PASS diff --git a/scripts/sync_models_mirror.py b/scripts/sync_models_mirror.py new file mode 100644 index 0000000..a47892c --- /dev/null +++ b/scripts/sync_models_mirror.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""M11b-2b: 把 modules/world_sync/models/ 下的表定义真源同步到应用打包镜像落点。 + +铁律：apps/{app}/pkgs/{module}/models/*.json 是 build 阶段产物，禁止手工双写。 +本脚本是唯一允许的写入通道 —— 真源只有一份（modules/ 侧），镜像由 cp 语义逐字节覆盖生成。 + +用法： + python3 modules/world_sync/scripts/sync_models_mirror.py + python3 modules/world_sync/scripts/sync_models_mirror.py --check # 只校验不写，不一致退出码 1 +""" + +import argparse +import hashlib +import os +import shutil +import sys + +# 表定义真源清单（相对机构工作空间根） +SOURCE_TABLES = [ + "pbl_runtime_event.json", + "pbl_entity_state.json", +] + +MODULE_ROOT = "modules/world_sync" +SOURCE_DIR = os.path.join(MODULE_ROOT, "models") +MIRROR_DIR = os.path.join("apps", "scense", "pkgs", "world_sync", "models") + + +def _find_workspace_root(): + """从脚本位置向上回溯，找到同时含 modules/ 与 apps/ 的工作空间根。""" + here = os.path.dirname(os.path.abspath(__file__)) + cur = here + for _ in range(10): + if os.path.isdir(os.path.join(cur, "modules")) and os.path.isdir(os.path.join(cur, "apps")): + return cur + parent = os.path.dirname(cur) + if parent == cur: + break + cur = parent + # 兜底：脚本位于 modules/{m}/scripts/x.py，则根为上三级 + return os.path.abspath(os.path.join(here, os.pardir, os.pardir, os.pardir)) + + +def sha256_of(path): + h = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(65536), b""): + h.update(chunk) + return h.hexdigest() + + +def sync(root, check_only=False): + """同步真源 -> 镜像。返回 (是否全部一致, 明细列表)。""" + src_dir = os.path.join(root, SOURCE_DIR) + dst_dir = os.path.join(root, MIRROR_DIR) + details = [] + all_ok = True + + if not check_only: + os.makedirs(dst_dir, exist_ok=True) + + for name in SOURCE_TABLES: + src = os.path.join(src_dir, name) + dst = os.path.join(dst_dir, name) + if not os.path.isfile(src): + details.append("MISSING-SOURCE: %s" % src) + all_ok = False + continue + + if not check_only: + shutil.copyfile(src, dst) + + src_hash = sha256_of(src) + if not os.path.isfile(dst): + details.append("MISSING-MIRROR: %s" % dst) + all_ok = False + continue + dst_hash = sha256_of(dst) + same = (src_hash == dst_hash) + all_ok = all_ok and same + details.append("%s %s" % ("SAME" if same else "DIFF", name)) + details.append(" sha256 src=%s" % src_hash) + details.append(" sha256 dst=%s" % dst_hash) + + return all_ok, details + + +def main(argv=None): + parser = argparse.ArgumentParser(description="sync world_sync table-definition sources to app mirror") + parser.add_argument("--check", action="store_true", help="only verify, do not write") + args = parser.parse_args(argv) + + root = _find_workspace_root() + ok, details = sync(root, check_only=args.check) + print("workspace root: %s" % root) + print("mode: %s" % ("check" if args.check else "sync")) + for line in details: + print(line) + print("RESULT: %s" % ("PASS" if ok else "FAIL")) + return 0 if ok else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/world_sync/pbl_runtime_authority.py b/world_sync/pbl_runtime_authority.py new file mode 100644 index 0000000..e4781d4 --- /dev/null +++ b/world_sync/pbl_runtime_authority.py @@ -0,0 +1,327 @@ +# -*- coding: utf-8 -*- +"""pbl_runtime_authority —— M11b-2b：``key=schema`` 引用的解析层（列名唯一权威源）。 + +**为什么存在** +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 结构体。 + +于是「消费方列契约」必须从权威定义**机械导出**，而不是人工抄一遍——人工双写正是 +QC #1/#3 判定的「多份互相矛盾定义」的根因。本模块就是那个导出层： + +* :func:`load_reference` —— 读本模块 ``models/.json``（key=schema 引用）； +* :func:`authority_model` —— 顺着引用解析到 pbl_runtime_ext 的真源 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`` 所有，本模块不复制那套逻辑）。 + +**fail-closed**：引用缺失、真源找不到、真源不是四段式、列集为空 —— 一律抛 +:class:`AuthorityResolutionError`，绝不退化成「猜列名」。 +""" + +import json +import os + +__all__ = [ + "AuthorityResolutionError", + "REFERENCE_ROOT_KEYS", + "REFERENCE_SCHEMA_KEYS", + "SPEC_ROOT_KEYS", + "load_reference", + "authority_path", + "authority_model", + "columns", + "insert_columns", + "auto_increment_columns", + "field_map", + "primary", + "indexes", + "table_title", + "ddl", + "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", +) +# 真源（表定义）必须是四段式 +SPEC_ROOT_KEYS = ("summary", "fields", "indexes", "codes") + +MODULE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +MODELS_DIR = os.path.join(MODULE_ROOT, "models") +# modules/world_sync -> modules -> 机构工作空间根 +WORKSPACE_ROOT = os.path.dirname(os.path.dirname(MODULE_ROOT)) + +_SQLITE_TYPES = { + "str": "TEXT", "char": "TEXT", "text": "TEXT", "json": "TEXT", + "bin": "BLOB", "short": "INTEGER", "int": "INTEGER", "long": "INTEGER", + "bigint": "INTEGER", "float": "REAL", "double": "REAL", "decimal": "REAL", + "ddouble": "REAL", "date": "TEXT", "time": "TEXT", "datetime": "TEXT", + "timestamp": "TEXT", +} +_MYSQL_TYPES = { + "str": "VARCHAR({length})", "char": "CHAR({length})", "text": "LONGTEXT", + "json": "JSON", "bin": "LONGBLOB", "short": "SMALLINT", "int": "INT", + "long": "BIGINT", "bigint": "BIGINT", "float": "FLOAT", + "double": "DECIMAL({length},{dec})", "decimal": "DECIMAL({length},{dec})", + "ddouble": "DECIMAL({length},{dec})", "date": "DATE", "time": "TIME", + "datetime": "DATETIME(3)", "timestamp": "TIMESTAMP", +} + + +class AuthorityResolutionError(Exception): + """引用/真源解析失败（fail-closed，调用方不得吞掉）。""" + + def __init__(self, message, **details): + Exception.__init__(self, message) + self.message = message + self.details = details + + def __str__(self): + if not self.details: + return self.message + parts = ", ".join("%s=%r" % (k, v) for k, v in sorted(self.details.items())) + return "%s (%s)" % (self.message, parts) + + +def _reference_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") + if not os.path.isfile(path): + raise AuthorityResolutionError("schema reference missing", path=path) + try: + with open(path, "r", encoding="utf-8") as fh: + ref = 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( + "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 + + +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"))) + tried = [] + for cand in candidates: + if cand in tried: + continue + 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 + + +def field_map(table): + """``{列名: 列定义 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 + return out + + +def columns(table): + """权威全列集（有序元组）。""" + return tuple(field_map(table).keys()) + + +def auto_increment_columns(table): + """由数据库自增生成的列（INSERT 时不得出现）。""" + 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")) + + +def insert_columns(table): + """可显式写入的列 = 全列 - 自增列（消费方 INSERT 的列契约）。""" + gen = set(auto_increment_columns(table)) + return tuple(n for n in columns(table) if n not in gen) + + +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) + 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 ()) + + +def indexes(table): + """规范化索引：``[{'name','idxtype','idxfields'}]``（兼容 unique/fields 写法）。""" + out = [] + for idx in authority_model(table).get("indexes") or []: + 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)}) + return out + + +def table_title(table): + summary = authority_model(table).get("summary") or [{}] + return summary[0].get("title") or summary[0].get("name") or table + + +def _render_type(f, dialect): + t = str(f.get("type") or "").lower() + table = _MYSQL_TYPES if dialect == "mysql" else _SQLITE_TYPES + tpl = table.get(t) + if tpl is None: + raise AuthorityResolutionError("unknown abstract type in authority model", + field=f.get("name"), type=t) + if "{" in tpl: + length = f.get("length") or (255 if t == "str" else 10) + dec = f.get("dec") or 2 + return tpl.format(length=length, dec=dec) + return tpl + + +def _nullable(f): + if "nullable" in f: + return str(f.get("nullable")).lower() != "no" + if "null" in f: + v = f.get("null") + return v is True or str(v).lower() == "true" + return True + + +def ddl(table, dialect="sqlite"): + """由权威列集渲染 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) + lines = [] + for f in model["fields"]: + 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") + if default is not None and not gen and str(default) != "": + piece.append(" DEFAULT %s" % ("CURRENT_TIMESTAMP" + if str(default).upper() == "CURRENT_TIMESTAMP" + else "'%s'" % default)) + if dialect == "mysql" and f.get("title"): + piece.append(" COMMENT '%s'" % str(f["title"]).replace("'", "''")) + lines.append("".join(piece)) + pk = primary(table) + if pk: + pk_sql = ", ".join(("`%s`" % c) if dialect == "mysql" else c for c in pk) + lines.append(" PRIMARY KEY (%s)" % pk_sql) + for idx in indexes(table): + if not idx["idxfields"] or idx["name"] in (None, "PRIMARY"): + continue + cols = ", ".join(("`%s`" % c) if dialect == "mysql" else c + for c in idx["idxfields"]) + if idx["idxtype"] == "unique": + lines.append(" UNIQUE KEY `%s` (%s)" % (idx["name"], cols) + if dialect == "mysql" + else " UNIQUE (%s)" % cols) + else: + lines.append(" KEY `%s` (%s)" % (idx["name"], cols)) + body = ",\n".join(lines) + if dialect == "mysql": + return ("CREATE TABLE IF NOT EXISTS `%s` (\n%s\n) " + "ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='%s'" + % (table, body, table_title(table))) + return "CREATE TABLE IF NOT EXISTS %s (\n%s\n)" % (table, body) + + +def column_diff(table, used_columns): + """双向列集 diff：返回 (未知列, 未覆盖的非自增列)。 + + ``used_columns`` 是消费方实际写入/读取用到的列名集合；空集合表示「不检查覆盖」。 + """ + used = set(used_columns or ()) + all_cols = set(columns(table)) + unknown = sorted(used - all_cols) + uncovered = [] + if used: + covered = used | set(auto_increment_columns(table)) + uncovered = sorted(all_cols - covered) + return unknown, uncovered diff --git a/world_sync/pbl_runtime_tx.py b/world_sync/pbl_runtime_tx.py index 4d24885..8f458d7 100644 --- a/world_sync/pbl_runtime_tx.py +++ b/world_sync/pbl_runtime_tx.py @@ -3,27 +3,38 @@ 对外唯一入口：:func:`write_event_with_state`。 +**列契约来源（M11b-2b / PM 裁决方案 B）**：本文件不再自带列名常量，全部列名由 +:mod:`world_sync.pbl_runtime_authority` 从 ``key=schema`` 真源 +（``modules/pbl_runtime_ext/models/pbl_runtime_event.json`` / +``pbl_entity_state.json``）机械导出。列名与真源不一致时 :func:`_assert_contract` +在建 SQL 之前就抛 :class:`AuthorityResolutionError`（fail-closed），杜绝「代码里 +抄一套列名、JSON 里另一套」的双写分叉。 + 事务内动作（顺序固定，全部在同一事务中）： 1. ``BEGIN``（sqlite 用 ``BEGIN IMMEDIATE``，立刻取写锁，避免锁升级竞态）； 2. 会话级 ``READ COMMITTED``（MySQL/PG；sqlite 无此概念，跳过）； -3. 读 ``pbl_entity_state`` 当前 ``state_version``，MySQL/PG 追加 ``FOR UPDATE`` 悲观行锁； +3. 读 ``pbl_entity_state`` 当前 ``state_version``（按 tenant_id + session_id + + entity_id 定位，与真源唯一键 ``uk_es`` 完全一致），MySQL/PG 追加 ``FOR UPDATE``； 4. 乐观锁校验：调用方给了 ``expected_version`` 且不匹配 → 抛 :class:`ConcurrentStateConflict` 并 ROLLBACK（防丢失更新）； 5. 乐观锁 ``UPDATE ... WHERE state_version = <读到的版本>``， ``rowcount != 1`` 视为并发冲突 → ROLLBACK + :class:`ConcurrentStateConflict`； - 状态行不存在时 INSERT 新行（主键冲突同样映射为并发冲突）； + 状态行不存在时 INSERT 新行（唯一键冲突同样映射为并发冲突）； 6. 向 ``pbl_runtime_event`` 插入 1 条 append-only 事件（带第 5 步生效后的新版本号）； 7. ``COMMIT``；任何一步失败 → ``ROLLBACK`` + 明确异常（不吞错误）。 原子性保证：第 6 步失败会把第 5 步的更新一起回滚（同一事务）；第 5 步冲突则两表都 0 变更。 -范围之外（明确不做）：广播、轮询兜底、时延优化、事件回放。 +范围之外（明确不做）：广播、轮询兜底、时延优化、事件回放（广播/幂等派生键归 +pbl_runtime_ext.tx_write，本模块只做「一次事务写两表」的最小写入器）。 """ +import hashlib import json import time import uuid from . import pbl_runtime_sql as ps +from . import pbl_runtime_authority as auth from .pbl_runtime_errors import ( ConcurrentStateConflict, EventWriteError, @@ -36,27 +47,60 @@ from .pbl_runtime_tx_env import resolve_conn_factory __all__ = [ "write_event_with_state", "read_entity_state", + "new_event_uid", "new_event_id", "utc_now_text", "DEFAULT_EVENT_TYPE", "DEFAULT_SOURCE", + "EVENT_REQUIRED", + "STATE_KEY_COLUMNS", ] DEFAULT_EVENT_TYPE = "state.updated" DEFAULT_SOURCE = "world_sync.m11b2" MAX_ERROR_DETAIL = 512 +# 事件必填列（真源中 nullable=no 且无默认值的业务列） +EVENT_REQUIRED = ("tenant_id", "event_uid", "entity_id", "event_type") +# 实体状态定位键 —— 与真源 pbl_entity_state 的 uk_es 唯一键一致 +STATE_KEY_COLUMNS = ("tenant_id", "session_id", "entity_id") -def new_event_id(): - """事件主键（uuid4 文本，append-only 事件无需有序 ID）。""" + +def new_event_uid(): + """事件业务唯一键 ``event_uid``（uuid4 去横线 hex，32 位，append-only 无需有序 ID）。""" return uuid.uuid4().hex +# 旧名字保留为别名：M11b-2a 的调用方/测试用 new_event_id() +new_event_id = new_event_uid + + def utc_now_text(ts=None): """ISO8601 UTC 秒级时间戳文本（sqlite TEXT / MySQL DATETIME(3) 均可直接写入）。""" if ts is None: ts = time.time() - return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ts)) + return time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(ts)) + + +def _checksum(state_json_text): + """状态摘要（真源 ``checksum`` 列，str 32）：sha256 前 32 位十六进制。""" + return hashlib.sha256(str(state_json_text).encode("utf-8")).hexdigest()[:32] + + +def _coerce_id(value, column): + """真源中 ``world_id`` / ``scene_id`` / ``session_id`` / ``created_by`` 是 long。 + + 纯数字（含数字字符串）转 int 落库；非数字标识符原样透传（由 DB 侧/上层迁移 + 决定是否扩列），但会在返回前做一次可序列化性检查——不静默丢列、不悄悄改类型。 + """ + if value is None or isinstance(value, bool): + return value + if isinstance(value, int): + return value + text = str(value).strip() + if text.isdigit() or (text.startswith("-") and text[1:].isdigit()): + return int(text) + return value def _dumps(value): @@ -85,6 +129,16 @@ def _require(ctx, name): return value +def _assert_contract(table, used): + """机械核对：本模块用到的列必须全部存在于真源列集，否则立刻抛错（fail-closed）。""" + unknown, _uncovered = auth.column_diff(table, used) + if unknown: + raise auth.AuthorityResolutionError( + "consumer uses columns absent from authority schema", + table=table, unknown=unknown) + return True + + def _execute(cur, sql, params, dialect): """统一执行入口：sqlite 用位置参数，MySQL 用 %s 位置参数（驱动一致）。""" if params is None: @@ -92,10 +146,11 @@ def _execute(cur, sql, params, dialect): return cur.execute(sql, tuple(params)) -def _fetch_state_version(cur, dialect, tenant_id, world_id, entity_id): +def _fetch_state_version(cur, dialect, tenant_id, session_id, entity_id): """读当前状态版本；行不存在返回 ``None``（首次写入）。""" + _assert_contract(ps.STATE_TABLE, STATE_KEY_COLUMNS + ("state_version",)) sql = ( - "SELECT state_version FROM %s WHERE tenant_id = %s AND world_id = %s " + "SELECT state_version FROM %s WHERE tenant_id = %s AND session_id = %s " "AND entity_id = %s%s" % ( ps.STATE_TABLE, @@ -105,7 +160,7 @@ def _fetch_state_version(cur, dialect, tenant_id, world_id, entity_id): ps.for_update_clause(dialect), ) ) - _execute(cur, sql, [tenant_id, world_id, entity_id], dialect) + _execute(cur, sql, [tenant_id, session_id, entity_id], dialect) row = cur.fetchone() if not row: return None @@ -115,13 +170,17 @@ def _fetch_state_version(cur, dialect, tenant_id, world_id, entity_id): raise StateWriteError("state_version is not an integer", row=repr(row)) -def _update_state(cur, dialect, tenant_id, world_id, entity_id, state_json, - current_version, next_version, event_id, now_text): - """乐观锁更新状态行；返回 ``'updated'`` / ``'inserted'``。冲突抛异常。""" +def _update_state(cur, dialect, tenant_id, session_id, entity_id, state_json, + current_version, next_version, updated_by, now_text): + """乐观锁更新状态行；返回 ``'updated'``。冲突抛异常。""" + used = STATE_KEY_COLUMNS + ("state_json", "state_version", "updated_at", + "updated_by", "checksum") + _assert_contract(ps.STATE_TABLE, used) sql = ( - "UPDATE %s SET state = %s, state_version = %s, updated_at = %s, " - "updated_by_event = %s " - "WHERE tenant_id = %s AND world_id = %s AND entity_id = %s AND state_version = %s" + "UPDATE %s SET state_json = %s, state_version = %s, updated_at = %s, " + "updated_by = %s, checksum = %s " + "WHERE tenant_id = %s AND session_id = %s AND entity_id = %s " + "AND state_version = %s" % ( ps.STATE_TABLE, ps.placeholder(dialect, 0), @@ -132,11 +191,12 @@ def _update_state(cur, dialect, tenant_id, world_id, entity_id, state_json, ps.placeholder(dialect, 5), ps.placeholder(dialect, 6), ps.placeholder(dialect, 7), + ps.placeholder(dialect, 8), ) ) params = [ - state_json, next_version, now_text, event_id, - tenant_id, world_id, entity_id, current_version, + state_json, next_version, now_text, updated_by, _checksum(state_json), + tenant_id, session_id, entity_id, current_version, ] try: _execute(cur, sql, params, dialect) @@ -144,7 +204,7 @@ def _update_state(cur, dialect, tenant_id, world_id, entity_id, state_json, _rollback_quiet(cur) raise StateWriteError( "failed to update entity state: %s" % _truncate(exc), - tenant_id=tenant_id, world_id=world_id, entity_id=entity_id, + tenant_id=tenant_id, session_id=session_id, entity_id=entity_id, ) if getattr(cur, "rowcount", 0) == 1: return "updated" @@ -154,34 +214,41 @@ def _update_state(cur, dialect, tenant_id, world_id, entity_id, state_json, _rollback_quiet(cur) raise ConcurrentStateConflict( "optimistic lock failed: state_version changed concurrently", - tenant_id=tenant_id, world_id=world_id, entity_id=entity_id, + tenant_id=tenant_id, session_id=session_id, entity_id=entity_id, expected_version=current_version, target_version=next_version, ) -def _insert_state(cur, dialect, tenant_id, world_id, entity_id, state_json, - next_version, event_id, now_text): - """首次写入状态行；主键冲突说明并发插入，同样按冲突处理。""" - cols = ("tenant_id", "world_id", "entity_id", "state", "state_version", - "updated_at", "updated_by_event") +def _insert_state(cur, dialect, tenant_id, session_id, entity_id, state_json, + next_version, updated_by, now_text): + """首次写入状态行；唯一键冲突说明并发插入，同样按冲突处理。""" + cols = ("tenant_id", "session_id", "entity_id", "state_json", "state_version", + "checksum", "updated_by", "created_at", "updated_at") + _assert_contract(ps.STATE_TABLE, cols) sql = "INSERT INTO %s (%s) VALUES (%s)" % ( ps.STATE_TABLE, ", ".join(cols), ps.render_placeholders(dialect, len(cols)) ) - params = [tenant_id, world_id, entity_id, state_json, next_version, now_text, event_id] + params = [tenant_id, session_id, entity_id, state_json, next_version, + _checksum(state_json), updated_by, now_text, now_text] try: _execute(cur, sql, params, dialect) except Exception as exc: _rollback_quiet(cur) raise ConcurrentStateConflict( "concurrent insert of entity state row: %s" % _truncate(exc), - tenant_id=tenant_id, world_id=world_id, entity_id=entity_id, + tenant_id=tenant_id, session_id=session_id, entity_id=entity_id, ) return "inserted" def _insert_event(cur, dialect, record): - """插入 append-only 事件行。失败即抛（上层回滚），保证「事件失败 → 状态也回滚」。""" + """插入 append-only 事件行（列集 = 真源可写列）。失败即抛（上层回滚）。""" cols = ps.EVENT_COLUMNS + _assert_contract(ps.EVENT_TABLE, cols) + missing = [c for c in EVENT_REQUIRED if record.get(c) in (None, "")] + if missing: + raise TxConfigError("event record missing required authority columns", + missing=missing) sql = "INSERT INTO %s (%s) VALUES (%s)" % ( ps.EVENT_TABLE, ", ".join(cols), ps.render_placeholders(dialect, len(cols)) ) @@ -192,13 +259,13 @@ def _insert_event(cur, dialect, record): _rollback_quiet(cur) raise EventWriteError( "failed to insert runtime event: %s" % _truncate(exc), - event_id=record["event_id"], entity_id=record["entity_id"], + event_uid=record["event_uid"], entity_id=record["entity_id"], ) return True def _rollback_quiet(cur): - """回滚，但绝不因回滚本身的异常覆盖业务异常（回滚失败另抛 TxAbortError 由调用处判断）。""" + """回滚，但绝不因回滚本身的异常覆盖业务异常。""" try: cur.connection.rollback() except Exception: @@ -228,21 +295,24 @@ def _set_isolation(cur, dialect): return False -def read_entity_state(conn_factory=None, tenant_id=None, world_id=None, entity_id=None, - conn=None): +def read_entity_state(conn_factory=None, tenant_id=None, session_id=None, + entity_id=None, world_id=None, conn=None): """只读辅助：取实体当前状态（供测试断言 / 上层重试前重读版本）。 - 独立短事务，不与写入事务共用，避免长事务。 + 独立短事务，不与写入事务共用，避免长事务。``world_id`` 仅作兼容入参（真源 + ``pbl_entity_state`` 无该列，定位键是 tenant_id + session_id + entity_id）。 """ own_conn = conn is None if own_conn: conn = resolve_conn_factory(conn_factory)() dialect = ps.detect_dialect(conn) + _assert_contract(ps.STATE_TABLE, STATE_KEY_COLUMNS + ("state_json", "state_version", + "updated_at")) cur = conn.cursor() try: sql = ( - "SELECT state, state_version, updated_at FROM %s " - "WHERE tenant_id = %s AND world_id = %s AND entity_id = %s" + "SELECT state_json, state_version, updated_at FROM %s " + "WHERE tenant_id = %s AND session_id = %s AND entity_id = %s" % ( ps.STATE_TABLE, ps.placeholder(dialect, 0), @@ -250,7 +320,7 @@ def read_entity_state(conn_factory=None, tenant_id=None, world_id=None, entity_i ps.placeholder(dialect, 2), ) ) - _execute(cur, sql, [tenant_id, world_id, entity_id], dialect) + _execute(cur, sql, [tenant_id, session_id, entity_id], dialect) row = cur.fetchone() if own_conn: conn.commit() @@ -267,23 +337,26 @@ def read_entity_state(conn_factory=None, tenant_id=None, world_id=None, entity_i def write_event_with_state(event_data=None, state_update=None, conn_factory=None, - conn=None, event_id=None, source=None, clock=None): + conn=None, event_uid=None, source=None, clock=None): """在**同一个数据库事务**内写入 1 条运行时事件 + 实体状态更新（M11b-2 核心）。 - :param event_data: dict，事件字段。必须含 ``tenant_id`` / ``world_id`` / ``entity_id``； - 可选 ``event_type``（默认 ``state.updated``）、``payload``、``session_id``、 - ``causation_id``、``source``。 - :param state_update: dict，状态更新。必须含 ``tenant_id`` / ``world_id`` / ``entity_id``； - ``state`` 为新的完整状态（dict 或 JSON 文本）；可选 ``expected_version`` - （乐观锁期望的当前版本，不匹配即冲突回滚）。传 ``None`` 表示只写事件不改状态。 + :param event_data: dict，事件字段。必须含 ``tenant_id`` / ``entity_id`` / + ``event_type``；``event_uid`` 缺省由本函数生成（uuid4 hex）。可选 + ``payload``、``session_id``、``scene_id``、``world_id``、``actor_id``、 + ``causation_id``、``idem_key``、``seq``/``seq_no``、``tx_group``、``source``。 + :param state_update: dict，状态更新。必须含 ``tenant_id`` / ``session_id`` / + ``entity_id``；``state``（或 ``state_json``）为新的完整状态（dict 或 JSON 文本）； + 可选 ``expected_version``（乐观锁期望当前版本，不匹配即冲突回滚）、``updated_by``。 + 传 ``None`` 表示只写事件不改状态。 :param conn_factory: 三级优先级第 1 级（显式注入）；也可传 DB URL 字符串。 :param conn: 复用已有连接（测试 / 宿主事务嵌套用），此时由本函数负责该连接的 commit/rollback，调用方不要自行 commit。 - :param event_id: 指定事件主键（默认 uuid4 hex），便于幂等重试与因果链测试。 - :param source: 写入来源标识（默认 ``world_sync.m11b2``）。 + :param event_uid: 指定事件唯一键（默认 uuid4 hex），便于幂等重试与因果链测试。 + :param source: 写入来源标识（默认 ``world_sync.m11b2``，落真源 ``source`` 列）。 :param clock: 可注入时间函数（默认 ``time.time``），便于测试固定时间戳。 - :returns: dict，含 ``event_id`` / ``state_version`` / ``action``（updated|inserted| - event_only）/ ``elapsed_ms`` / ``isolation_level_applied`` / ``dialect``。 + :returns: dict，含 ``event_uid`` / ``state_version`` / ``action``（updated|inserted| + event_only）/ ``elapsed_ms`` / ``isolation_level_applied`` / ``dialect`` / + ``authority``（本次列契约的真源路径，便于审计与 QC 复核）。 :raises TxConfigError: 必填字段缺失、方言不认识； :raises ConcurrentStateConflict: 乐观锁/行锁冲突（两表零变更，已回滚）； :raises EventWriteError: 事件插入失败（两表零变更，已回滚）； @@ -295,23 +368,29 @@ def write_event_with_state(event_data=None, state_update=None, conn_factory=None clock = clock or time.time tenant_id = _require(event_data, "tenant_id") - world_id = _require(event_data, "world_id") entity_id = _require(event_data, "entity_id") + event_type = event_data.get("event_type") or DEFAULT_EVENT_TYPE + event_uid = event_uid or event_data.get("event_uid") or new_event_uid() + session_id = event_data.get("session_id") + world_id = _coerce_id(event_data.get("world_id"), "world_id") if state_update is not None: - for key in ("tenant_id", "world_id", "entity_id"): + for key in STATE_KEY_COLUMNS: if key in state_update and state_update[key] is not None \ + and key in event_data and event_data[key] is not None \ and state_update[key] != event_data[key]: raise TxConfigError( "state_update.%s conflicts with event_data.%s" % (key, key), field=key, ) - state_update.setdefault(key, event_data[key]) + state_update.setdefault(key, event_data.get(key)) + for key in STATE_KEY_COLUMNS: + _require(state_update, key) + session_id = state_update["session_id"] - # 租户/世界隔离维度在 SQL 的 WHERE 与 INSERT 列里都强制带上， - # 跨租户的同名 entity_id 不可能互相影响（见 _fetch_state_version / _update_state）。 expected_version = None new_state_json = None + updated_by = None if state_update is not None: expected_version = state_update.get("expected_version") if expected_version is not None: @@ -320,7 +399,11 @@ def write_event_with_state(event_data=None, state_update=None, conn_factory=None except (TypeError, ValueError): raise TxConfigError("expected_version must be an integer", expected_version=expected_version) - new_state_json = _dumps(state_update.get("state", {})) + new_state_json = _dumps(state_update.get("state", + state_update.get("state_json", {}))) + updated_by = str(state_update.get("updated_by") + or event_data.get("actor_id") or source + or DEFAULT_SOURCE)[:32] factory = resolve_conn_factory(conn_factory) if conn is None else None own_conn = conn is None @@ -328,7 +411,6 @@ def write_event_with_state(event_data=None, state_update=None, conn_factory=None dialect = ps.detect_dialect(conn) now_text = utc_now_text(clock()) - event_id = event_id or new_event_id() source = source or event_data.get("source") or DEFAULT_SOURCE started = time.perf_counter() @@ -338,47 +420,62 @@ def write_event_with_state(event_data=None, state_update=None, conn_factory=None _begin(conn, cur, dialect) isolation_applied = _set_isolation(cur, dialect) - current_version = _fetch_state_version(cur, dialect, tenant_id, world_id, entity_id) + current_version = _fetch_state_version(cur, dialect, tenant_id, session_id, + entity_id) if state_update is not None and current_version is not None \ and expected_version is not None and expected_version != current_version: _rollback_quiet(cur) raise ConcurrentStateConflict( "expected_version does not match stored state_version", - tenant_id=tenant_id, world_id=world_id, entity_id=entity_id, + tenant_id=tenant_id, session_id=session_id, entity_id=entity_id, expected_version=expected_version, actual_version=current_version, ) - if current_version is None: - base_version = 0 - else: - base_version = current_version + base_version = 0 if current_version is None else current_version next_version = base_version + 1 # (b) 先更新实体状态：失败/冲突即回滚，事件也不会留下（原子性）。 if state_update is None: action = "event_only" elif current_version is None: - action = _insert_state(cur, dialect, tenant_id, world_id, entity_id, - new_state_json, next_version, event_id, now_text) + action = _insert_state(cur, dialect, tenant_id, session_id, entity_id, + new_state_json, next_version, updated_by, now_text) else: - action = _update_state(cur, dialect, tenant_id, world_id, entity_id, + action = _update_state(cur, dialect, tenant_id, session_id, entity_id, new_state_json, base_version, next_version, - event_id, now_text) + updated_by, now_text) # (a) 再插入 append-only 事件；此步失败会把上面的状态变更一并回滚。 + payload_text = _dumps(event_data.get("payload", + state_update.get("state") + if state_update is not None else {})) + seq_no = event_data.get("seq_no", event_data.get("seq")) record = { - "event_id": event_id, "tenant_id": tenant_id, - "world_id": world_id, - "session_id": event_data.get("session_id"), - "entity_id": entity_id, - "event_type": event_data.get("event_type") or DEFAULT_EVENT_TYPE, - "payload": _dumps(event_data.get("payload", state_update or {})), - "causation_id": event_data.get("causation_id"), - "source": source, + "event_uid": str(event_uid), + "event_code": event_data.get("event_code") or "", + "idem_key": event_data.get("idem_key") or "", + "world_id": world_id if world_id is not None else 0, + "scene_id": _coerce_id(event_data.get("scene_id"), "scene_id") or 0, + "session_id": _coerce_id(session_id, "session_id") or 0, + "entity_id": str(entity_id), + "actor_id": str(event_data.get("actor_id") or updated_by or "")[:64], + "event_type": str(event_type), + "payload": payload_text, + "payload_json": payload_text, + "causation_id": str(event_data.get("causation_id") or ""), + "seq": int(seq_no) if seq_no not in (None, "") else 0, + "seq_no": int(seq_no) if seq_no not in (None, "") else 0, + "source": str(source)[:32], + "state": str(event_data.get("state") or "applied")[:16], "state_version": next_version if state_update is not None else base_version, + "tx_group": str(event_data.get("tx_group") or ""), + "broadcast": 1 if event_data.get("broadcast", 1) else 0, + "created_by": _coerce_id(event_data.get("created_by"), "created_by") or 0, "created_at": now_text, + "updated_at": now_text, + "occurred_at": now_text, } _insert_event(cur, dialect, record) @@ -388,18 +485,24 @@ def write_event_with_state(event_data=None, state_update=None, conn_factory=None except Exception as exc: _rollback_quiet(cur) raise TxAbortError("commit failed: %s" % _truncate(exc), - event_id=event_id, entity_id=entity_id) + event_uid=event_uid, entity_id=entity_id) return { - "event_id": event_id, + "event_uid": event_uid, + "event_id": event_uid, # 兼容旧调用方的返回键名 "tenant_id": tenant_id, "world_id": world_id, + "session_id": session_id, "entity_id": entity_id, "state_version": record["state_version"], "action": action, "created_at": now_text, "dialect": dialect, "isolation_level_applied": isolation_applied, + "authority": { + ps.EVENT_TABLE: auth.authority_path(ps.EVENT_TABLE), + ps.STATE_TABLE: auth.authority_path(ps.STATE_TABLE), + }, "elapsed_ms": round((time.perf_counter() - started) * 1000.0, 3), } finally: