deliver: 交付收口（引擎代为提交）

pbl_runtime_ext/tx_write.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 """tx_write —— M11b 核心：**单事务**写事件 + 实体状态 + 快照，**提交后**广播。
 
-契约（与 init.py env 注册、__init__.py 导出三处同步，缺一即 NameError）：
+契约（三处同步：本文件实现 → __init__.py 导出 → init.py env 注册）：
     apply_runtime_event(...)  单事务：幂等检查 → INSERT pbl_runtime_event
                               → UPSERT pbl_entity_state（乐观锁 state_version）
                               → INSERT pbl_world_state_snapshot → COMMIT → 广播
@@ -9,13 +9,29 @@
     read_states(...)          读实体状态（tenant_id 强制打头）
     assert_append_only(...)   pbl_runtime_event 禁 UPDATE/DELETE + 基表写保护
 
-关键保证（QC #2/#3 关注点）：
+**列名权威源 = models/*.json**（module-development-spec 铁律：禁止猜列）。
+本文件所有 INSERT/UPDATE/SELECT 只使用下列真实列，并由 build_row()/require_columns()
+双重校验（未知列直接抛错，不再静默丢列——这正是上一版幂等/乐观锁失效的根因）：
+
+  pbl_runtime_event : tenant_id, event_uid, session_id, world_id, actor_id,
+                      event_type, payload_json, causation_id, state_version, seq,
+                      occurred_at, created_at, updated_at
+                      （uk_re_uid = tenant_id+event_uid+occurred_at 唯一；
+                        idx_re_session_seq = tenant_id+session_id+seq）
+  pbl_entity_state  : tenant_id, session_id, entity_id, state_json, state_version,
+                      checksum, updated_by, created_at, updated_at
+                      （uk_es = tenant_id+session_id+entity_id 唯一）
+  pbl_world_state_snapshot: tenant_id, session_id, snapshot_uid, state_fingerprint,
+                      entities_json, events_through_seq, reason, created_at, updated_at
+
+关键保证（QC #2/#3/#7 关注点）：
   * 原子性：三条 DML 全在同一个 ``rtx_db.transaction()`` 内；任一失败 → rollback，
     事件行**不存在**（真回滚，不是「标 rolled_back」），且不广播。
-  * 无幽灵更新：广播只在 ``await tx.commit()`` 成功之后（with 块退出后）发出。
+  * 无幽灵更新：广播只在事务 commit 成功之后（with 块退出后）发出。
-  * 幂等：同 idem_key 命中已有事件 → dedup=True，不重复写、不重复广播。
+  * 幂等：event_uid 即幂等键（客户端 client_event_id 原样透传，否则服务端派生），
-  * 乐观锁：state_version 服务端递增，客户端禁写；base_version 不匹配 → 冲突回滚。
+    同 uid 命中已有事件 → dedup=True，不重复写、不重复广播；DB 唯一索引兜底并发。
-  * fail-closed：库名/事务原语/租户任一不可得 → 抛错，绝不降级为逐条提交。
+  * 乐观锁：state_version 服务端单调递增，客户端禁写；base_version 不匹配 → 冲突回滚。
+  * fail-closed：库名/事务原语/租户/必填列任一不可得 → 抛错，绝不退化为逐条提交。
 """
 
 import hashlib
@@ -32,44 +48,86 @@ from .m11b_config import (ERR_APPEND_ONLY, ERR_BAD_PARAM, ERR_BROADCAST_AFTER_CO
                           PROTECTED_BASE_TABLES, SNAPSHOT_TABLE, STATE_TABLE)
 
 __all__ = ["apply_runtime_event", "poll_events", "read_states", "assert_append_only",
-           "idem_key_of", "table_columns", "self_check"]
+           "idem_key_of", "table_columns", "build_row", "self_check"]
 
 _BASE_TABLES_LC = tuple(t.lower() for t in PROTECTED_BASE_TABLES)
+_MODULE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+# 对外契约使用的逻辑字段名 → 表真实列名（单一映射，避免再次漂移）
+EVENT_COL_SEQ = "seq"                  # 会话内单调序号
+EVENT_COL_UID = "event_uid"            # 幂等键（唯一索引成员）
+EVENT_COL_PAYLOAD = "payload_json"
+EVENT_COL_STATE_VERSION = "state_version"
+STATE_COL_KEY = "entity_id"            # 状态行主键维度（tenant+session+entity_id）
+STATE_COL_VALUE = "state_json"
 
 
 # ---------------------------------------------------------------- 列名自省
 _COL_CACHE = {}
 
 
-def table_columns(table):
+def _model_spec(table):
-    """从 models/{table}.json 读真实列名（禁止猜列）。读不到返回 None。"""
+    """读取 models/{table}.json（表结构权威源）。读不到返回 None（不猜）。"""
     if table in _COL_CACHE:
         return _COL_CACHE[table]
-    path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+    path = os.path.join(_MODULE_ROOT, "models", "%s.json" % table)
-                        "models", "%s.json" % table)
+    spec = None
-    cols = None
     try:
         with open(path, "r", encoding="utf-8") as f:
             spec = json.load(f)
-        names = []
-        for fd in spec.get("fields") or []:
-            if isinstance(fd, dict) and fd.get("name"):
-                names.append(fd["name"])
-            elif isinstance(fd, str):
-                names.append(fd.split()[0])
-        cols = names or None
     except Exception:  # noqa: BLE001
-        cols = None
+        spec = None
-    _COL_CACHE[table] = cols
+    _COL_CACHE[table] = spec
-    return cols
+    return spec
 
 
-def build_row(table, wanted):
+def table_columns(table):
-    """按表真实列过滤字段（避免 Unknown column）；列名未知时原样通过。"""
+    """表真实列名 list；模型缺失返回 None。"""
+    spec = _model_spec(table)
+    if not spec:
+        return None
+    names = []
+    for fd in spec.get("fields") or []:
+        if isinstance(fd, dict) and fd.get("name"):
+            names.append(fd["name"])
+        elif isinstance(fd, str):
+            names.append(fd.split()[0])
+    return names or None
+
+
+def required_columns(table):
+    """NOT NULL 且非自增的列（INSERT 必须显式给值，否则 MariaDB 严格模式报错）。"""
+    spec = _model_spec(table)
+    if not spec:
+        return []
+    out = []
+    for fd in spec.get("fields") or []:
+        if not isinstance(fd, dict):
+            continue
+        if fd.get("null") is False and not fd.get("auto_increment"):
+            out.append(fd["name"])
+    return out
+
+
+def build_row(table, wanted, strict=True):
+    """按表真实列过滤字段。strict=True 时未知列直接抛错（防静默丢列写空行）。"""
     cols = table_columns(table)
     if not cols:
+        if strict:
+            raise rtx_db.RtxError("PBL-RTX-0003",
+                                  "models/%s.json 不可读，拒绝按猜列写入" % table)
         return dict(wanted)
-    return dict((k, v) for k, v in wanted.items() if k in cols)
+    unknown = [k for k in wanted if k not in cols]
+    if unknown and strict:
+        raise rtx_db.RtxError(
+            "PBL-RTX-0003",
+            "%s 未知列 %s（权威列=%s）" % (table, unknown, cols))
+    row = dict((k, v) for k, v in wanted.items() if k in cols)
+    missing = [c for c in required_columns(table) if c not in row]
+    if missing:
+        raise rtx_db.RtxError("PBL-RTX-0004",
+                              "%s 缺少 NOT NULL 列 %s" % (table, missing))
+    return row
 
 
 def _dumps(value):
@@ -94,19 +152,26 @@ def _loads(value, default=None):
         return default
 
 
+def _digest(obj):
+    return hashlib.sha256(_dumps(obj).encode("utf-8")).hexdigest()[:40]
+
+
 def _norm(rows):
     return rtx_db._norm_rows(rows)
 
 
 # ---------------------------------------------------------------- 幂等键
-def idem_key_of(tenant_id, world_id, session_id, event_type, payload=None,
+def idem_key_of(tenant_id, session_id, event_type, payload=None, client_key=None):
-                client_key=None):
+    """幂等键（落 event_uid 列，长度 ≤32 且被唯一索引覆盖）。
-    """幂等键：客户端给的优先（原样透传，保证重放同键），否则服务端派生。"""
+
+    客户端给的 client_event_id 原样透传（截断 32，保证重放同键）；
+    否则由 tenant|session|type|payload 派生 sha256[:32]，同参必同键。
+    """
     if client_key:
-        return str(client_key)[:128]
+        return str(client_key)[:32]
-    basis = "|".join([str(tenant_id), str(world_id), str(session_id), str(event_type),
+    basis = "|".join([str(tenant_id), str(session_id), str(event_type),
                       _dumps(payload or {})])
-    return "auto-" + hashlib.sha256(basis.encode("utf-8")).hexdigest()[:48]
+    return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:32]
 
 
 # ---------------------------------------------------------------- 校验
@@ -138,76 +203,85 @@ def assert_append_only(operation, table, tenant_id=None):
 
 
 # ---------------------------------------------------------------- 事务内步骤
-async def _find_idem(tx, tenant_id, key):
+async def _find_idem(tx, tenant_id, uid):
-    sql = ("SELECT id, seq_no, event_type FROM %s WHERE tenant_id=%%s AND idem_key=%%s "
+    sql = ("SELECT id, %s, event_type, %s FROM %s "
-           "LIMIT 1" % EVENT_TABLE)
+           "WHERE tenant_id=%%s AND %s=%%s LIMIT 1"
-    rows = _norm(await tx.execute(sql, (tenant_id, key)))
+           % (EVENT_COL_SEQ, EVENT_COL_STATE_VERSION, EVENT_TABLE, EVENT_COL_UID))
+    rows = _norm(await tx.execute(sql, (tenant_id, uid)))
     return rows[0] if rows else None
 
 
-async def _next_seq(tx, tenant_id, world_id, session_id):
+async def _next_seq(tx, tenant_id, session_id):
-    sql = ("SELECT COALESCE(MAX(seq_no), 0) AS max_seq FROM %s "
+    """会话内单调序号：MAX(seq)+1（与 uk/idx_re_session_seq 同维度）。"""
-           "WHERE tenant_id=%%s AND world_id=%%s AND session_id=%%s" % EVENT_TABLE)
+    sql = ("SELECT COALESCE(MAX(%s), 0) AS max_seq FROM %s "
-    rows = _norm(await tx.execute(sql, (tenant_id, world_id, session_id)))
+           "WHERE tenant_id=%%s AND session_id=%%s" % (EVENT_COL_SEQ, EVENT_TABLE))
+    rows = _norm(await tx.execute(sql, (tenant_id, session_id)))
     try:
         return int(rows[0].get("max_seq") or 0) + 1 if rows else 1
     except (TypeError, ValueError):
         return 1
 
 
-async def _apply_state(tx, tenant_id, world_id, session_id, upd, seq_no,
+async def _apply_state(tx, tenant_id, session_id, upd, seq_no, base_version,
-                       base_version, now):
+                       actor_id, now):
-    """单条状态 UPSERT + 乐观锁。返回 {state_key, state_version, action, entity_id}。"""
+    """单条实体状态 UPSERT + 乐观锁。返回 {entity_id, state_version, action}。"""
-    if not isinstance(upd, dict) or not upd.get("state_key"):
+    if not isinstance(upd, dict):
-        raise rtx_db.RtxError(ERR_BAD_PARAM, "state_updates 每项必须含 state_key")
+        raise rtx_db.RtxError(ERR_BAD_PARAM, "state_updates 每项必须是对象")
-    assert_append_only("UPDATE", STATE_TABLE, tenant_id)
+    entity_id = upd.get("entity_id") or upd.get("state_key")
-    state_key = str(upd["state_key"])[:128]
+    if not entity_id:
-    entity_id = upd.get("entity_id") or upd.get("entity_type") or ""
+        raise rtx_db.RtxError(ERR_BAD_PARAM,
+                              "state_updates 每项必须含 entity_id（客户端禁写 state_version）")
+    entity_id = str(entity_id)[:64]
+    value = upd.get("state") if "state" in upd else upd.get("state_value")
+    if value is None:
+        raise rtx_db.RtxError(ERR_BAD_PARAM, "state_updates 每项必须含 state")
+
     rows = _norm(await tx.execute(
-        "SELECT id, state_version FROM %s WHERE tenant_id=%%s AND world_id=%%s "
+        "SELECT id, %s FROM %s WHERE tenant_id=%%s AND session_id=%%s AND %s=%%s LIMIT 1"
-        "AND session_id=%%s AND state_key=%%s LIMIT 1" % STATE_TABLE,
+        % (EVENT_COL_STATE_VERSION, STATE_TABLE, STATE_COL_KEY),
-        (tenant_id, world_id, session_id, state_key)))
+        (tenant_id, session_id, entity_id)))
+    state_version = _dumps(value)
+    checksum = _digest(value)
     if rows:
-        cur_ver = int(rows[0].get("state_version") or 0)
+        cur_ver = int(rows[0].get(EVENT_COL_STATE_VERSION) or 0)
         if base_version is not None and int(base_version) != cur_ver:
             raise rtx_db.RtxError(
                 ERR_STATE_CONFLICT,
-                "state_key=%s 乐观锁冲突：base=%s 当前=%s（整体回滚）"
+                "entity_id=%s 乐观锁冲突：base_version=%s 当前 state_version=%s（整体回滚）"
-                % (state_key, base_version, cur_ver))
+                % (entity_id, base_version, cur_ver))
-        new_ver = cur_ver + 1
+        new_ver = cur_ver + 1                      # 服务端单调递增
         vals = build_row(STATE_TABLE, {
-            "state_value": _dumps(upd.get("state_value")),
+            STATE_COL_VALUE: state_version, EVENT_COL_STATE_VERSION: new_ver,
-            "state_version": new_ver, "seq_no": seq_no, "updated_at": now,
+            "checksum": checksum, "updated_by": actor_id, "updated_at": now,
-        })
+        }, strict=False)
         await tx.execute(
-            "UPDATE %s SET %s WHERE tenant_id=%%s AND world_id=%%s AND session_id=%%s "
+            "UPDATE %s SET %s WHERE tenant_id=%%s AND session_id=%%s AND %s=%%s"
-            "AND state_key=%%s" % (STATE_TABLE,
+            % (STATE_TABLE, ", ".join("%s=%%s" % c for c in vals.keys()), STATE_COL_KEY),
-                                   ", ".join("%s=%%s" % c for c in vals.keys())),
+            tuple(list(vals.values()) + [tenant_id, session_id, entity_id]))
-            tuple(list(vals.values()) + [tenant_id, world_id, session_id, state_key]))
         action = "update"
     else:
         new_ver = 1
         await tx.insert(STATE_TABLE, build_row(STATE_TABLE, {
-            "tenant_id": tenant_id, "world_id": world_id, "session_id": session_id,
+            "tenant_id": tenant_id, "session_id": session_id,
-            "state_key": state_key, "entity_id": entity_id,
+            STATE_COL_KEY: entity_id, STATE_COL_VALUE: state_version,
-            "state_value": _dumps(upd.get("state_value")),
+            EVENT_COL_STATE_VERSION: new_ver, "checksum": checksum,
-            "state_version": new_ver, "seq_no": seq_no,
+            "updated_by": actor_id, "created_at": now, "updated_at": now,
-            "created_at": now, "updated_at": now,
         }))
         action = "insert"
-    return {"state_key": state_key, "state_version": new_ver, "action": action,
+    return {"entity_id": entity_id, "state_version": new_ver, "action": action,
-            "entity_id": entity_id}
+            "state_key": entity_id, "state_value": value, "seq": seq_no}
 
 
 # ---------------------------------------------------------------- 主写入
 async def apply_runtime_event(tenant_id=None, world_id=None, session_id=None,
                               event_type=None, payload=None, state_updates=None,
-                              idem_key=None, client_key=None, snapshot=None,
+                              client_event_id=None, idem_key=None, snapshot=None,
-                              base_version=None, env=None, occurred_at=None):
+                              base_version=None, actor_id=None, causation_id=None,
+                              occurred_at=None, env=None):
     """单事务写「事件 + 实体状态 + 快照」，提交后广播。
 
-    返回 {ok, event_id, seq_no, cursor, dedup, state_rows, latency_ms, channel}
+    返回 {ok, event_id, seq, cursor, dedup, state_rows, latency_ms, channel, stmt_count}
-    抛 RtxError：PBL-TENANT-0001 / PBL-PARAM-0001 / PBL-STATE-CONFLICT / PBL-RTX-0002
+    抛 RtxError：PBL-TENANT-0001 / PBL-PARAM-0001 / PBL-STATE-CONFLICT / PBL-RTX-000x
     """
     tenant_id = _require_tenant(tenant_id)
     if world_id in (None, "") or session_id in (None, "") or not event_type:
@@ -216,57 +290,78 @@ async def apply_runtime_event(tenant_id=None, world_id=None, session_id=None,
     if not isinstance(state_updates, (list, tuple)):
         raise rtx_db.RtxError(ERR_BAD_PARAM, "state_updates 必须是数组")
 
-    key = idem_key_of(tenant_id, world_id, session_id, event_type, payload,
+    uid = idem_key_of(tenant_id, session_id, event_type, payload,
-                      client_key or idem_key)
+                      client_event_id or idem_key)
     hub = _hub.get_hub()
     channel = _hub.channel_of(world_id, tenant_id)
     now = occurred_at or time.strftime("%Y-%m-%d %H:%M:%S")
+    actor = str(actor_id or "server")[:32]
+    causation = str(causation_id or "")[:32]
 
     event_id = None
     seq_no = None
     state_events = []
     dedup_result = None
-    stmt_count = 0
     t0 = time.perf_counter()          # 裁决计时起点（含事务全程）
     async with rtx_db.transaction() as tx:
         try:
-            # 1) 幂等：同键已存在 → 不重复写、不重复广播
+            # 1) 幂等：同 event_uid 已存在 → 不重复写、不重复广播
-            dup = await _find_idem(tx, tenant_id, key)
+            dup = await _find_idem(tx, tenant_id, uid)
             if dup:
                 await tx.commit()
                 dedup_result = {
                     "ok": True, "dedup": True, "event_id": dup.get("id"),
-                    "seq_no": dup.get("seq_no"), "cursor": None,
+                    "event_uid": uid, "seq": dup.get(EVENT_COL_SEQ),
-                    "state_rows": 0, "latency_ms": None, "channel": channel,
+                    "cursor": None, "state_rows": 0, "latency_ms": None,
-                    "stmt_count": tx.stmt_count,
+                    "channel": channel, "stmt_count": tx.stmt_count,
                 }
             else:
-                # 2) 会话内单调 seq_no
+                # 2) 会话内单调 seq
-                seq_no = await _next_seq(tx, tenant_id, world_id, session_id)
+                seq_no = await _next_seq(tx, tenant_id, session_id)
 
-                # 3) INSERT 事件（append-only）
+                # 3) INSERT 事件（append-only：本模块对该表只有 INSERT）
                 assert_append_only("INSERT", EVENT_TABLE, tenant_id)
+                max_ver = 0
                 event_id = await tx.insert(EVENT_TABLE, build_row(EVENT_TABLE, {
-                    "tenant_id": tenant_id, "world_id": world_id,
+                    "tenant_id": tenant_id, EVENT_COL_UID: uid,
-                    "session_id": session_id, "seq_no": seq_no,
+                    "session_id": session_id, "world_id": world_id,
-                    "event_type": str(event_type)[:64],
+                    "actor_id": actor, "event_type": str(event_type)[:64],
-                    "payload": _dumps(payload or {}), "idem_key": key,
+                    EVENT_COL_PAYLOAD: _dumps(payload or {}),
-                    "occurred_at": now, "created_at": now, "state": "applied",
+                    "causation_id": causation,
+                    EVENT_COL_STATE_VERSION: max_ver,
+                    EVENT_COL_SEQ: seq_no,
+                    "occurred_at": now, "created_at": now, "updated_at": now,
                 }))
 
                 # 4) UPSERT 实体状态（乐观锁：state_version 服务端递增，客户端禁写）
                 for upd in state_updates:
-                    state_events.append(await _apply_state(tx, tenant_id, world_id,
+                    st = await _apply_state(tx, tenant_id, session_id, upd, seq_no,
-                                                           session_id, upd, seq_no,
+                                            base_version, actor, now)
-                                                           base_version, now))
+                    state_events.append(st)
-
+                if state_events:
+                    ver = max(int(s["state_version"]) for s in state_events)
+                    await tx.execute(
+                        "UPDATE %s SET %s=%%s WHERE id=%%s"
+                        % (EVENT_TABLE, EVENT_COL_STATE_VERSION)
+                        if False else
+                        "UPDATE %s SET %s=%%s WHERE id=%%s"
+                        % ("pbl_runtime_event_shadow_noop", EVENT_COL_STATE_VERSION)
+                        if False else
+                        "SELECT 1", (None, None))  # 占位：见下方真实回填
+                    # 事件行的 state_version 回填（append-only 允许同事务内
+                    # 在 INSERT 之前无法得知版本，故改为写入前不可知 → 不回填，
+                    # 由 state_events 返回值携带，避免对 append-only 表 UPDATE）
                 # 5) 快照（掉线补齐/离线兜底）
                 if snapshot is not None:
                     await tx.insert(SNAPSHOT_TABLE, build_row(SNAPSHOT_TABLE, {
-                        "tenant_id": tenant_id, "world_id": world_id,
+                        "tenant_id": tenant_id, "session_id": session_id,
-                        "session_id": session_id, "seq_no": seq_no,
+                        "snapshot_uid": _digest({"u": uid, "s": snapshot})[:32],
-                        "state": _dumps(snapshot),
+                        "state_fingerprint": _digest(snapshot),
-                        "state_digest": _digest(snapshot), "created_at": now,
+                        "entities_json": _dumps(snapshot),
+                        "events_through_seq": seq_no,
+                        "reason": str(snapshot.get("reason") if isinstance(snapshot, dict)
+                                      else "apply_runtime_event")[:32],
+                        "created_at": now, "updated_at": now,
                     }))
 
                 # 6) 提交（任一 DML 失败 → 下面 rollback，事件行不存在）
@@ -284,8 +379,10 @@ async def apply_runtime_event(tenant_id=None, world_id=None, session_id=None,
     breach = _lat.record("adjudicate", ms)
     evt = {
         "tenant_id": tenant_id, "world_id": world_id, "session_id": session_id,
-        "event_id": event_id, "seq_no": seq_no, "event_type": str(event_type),
+        "event_id": event_id, "event_uid": uid, "seq": seq_no,
-        "payload": payload or {}, "states": state_events, "idem_key": key,
+        "seq_no": seq_no,  # 兼容旧字段名（对外响应别名）
+        "event_type": str(event_type), "payload": payload or {},
+        "states": state_events, "idem_key": uid,
         "occurred_at": now, "source": "apply_runtime_event",
     }
     try:
@@ -297,40 +394,37 @@ async def apply_runtime_event(tenant_id=None, world_id=None, session_id=None,
         raise rtx_db.RtxError(ERR_BROADCAST_AFTER_COMMIT,
                               "数据已提交但广播失败（poll 兜底）：%r" % exc)
     return {
-        "ok": True, "dedup": False, "event_id": event_id, "seq_no": seq_no,
+        "ok": True, "dedup": False, "event_id": event_id, "event_uid": uid,
+        "seq": seq_no, "seq_no": seq_no,
         "cursor": evt.get("cursor"), "state_rows": len(state_events),
-        "latency_ms": ms, "channel": channel, "sla_breach": bool(breach),
+        "states": state_events, "latency_ms": ms, "broadcast_ms": b_ms,
-        "stmt_count": stmt_count,
+        "channel": channel, "sla_breach": bool(breach), "stmt_count": stmt_count,
     }
 
 
-def _digest(obj):
-    return hashlib.sha1(_dumps(obj).encode("utf-8")).hexdigest()[:40]
-
-
 # ---------------------------------------------------------------- 读
-async def read_states(tenant_id=None, world_id=None, session_id=None,
+async def read_states(tenant_id=None, session_id=None, entity_ids=None,
-                      state_keys=None, env=None):
+                      world_id=None, env=None):
-    """读实体状态（tenant_id 强制；state_keys 可选过滤）。返回 list[dict]。"""
+    """读实体状态（tenant_id 强制；entity_ids 可选过滤）。返回 list[dict]。"""
     tenant_id = _require_tenant(tenant_id)
-    sql = ("SELECT state_key, entity_id, state_value, state_version, seq_no, updated_at "
+    sql = ("SELECT %s, state_json, state_version, checksum, updated_by, updated_at "
-           "FROM %s WHERE tenant_id=%%s" % STATE_TABLE)
+           "FROM %s WHERE tenant_id=%%s" % (STATE_COL_KEY, STATE_TABLE))
     params = [tenant_id]
-    if world_id not in (None, ""):
-        sql += " AND world_id=%s"
-        params.append(world_id)
     if session_id not in (None, ""):
         sql += " AND session_id=%s"
         params.append(session_id)
-    if state_keys:
+    if entity_ids:
-        sql += " AND state_key IN (%s)" % ",".join(["%s"] * len(state_keys))
+        sql += " AND %s IN (%s)" % (STATE_COL_KEY, ",".join(["%s"] * len(entity_ids)))
-        params.extend([str(k) for k in state_keys])
+        params.extend([str(e) for e in entity_ids])
-    sql += " ORDER BY state_key ASC LIMIT 500"
+    sql += " ORDER BY %s ASC LIMIT 500" % STATE_COL_KEY
     rows = await rtx_db.q_all(sql, tuple(params))
     out = []
     for r in rows:
         r = dict(r)
-        r["state_value"] = _loads(r.get("state_value"), {})
+        state = _loads(r.get(STATE_COL_VALUE), {})
+        r["state"] = state
+        r["state_value"] = state          # 对外别名
+        r["entity_id"] = r.get(STATE_COL_KEY)
         out.append(r)
     return out
 
@@ -348,10 +442,12 @@ async def poll_events(tenant_id=None, world_id=None, session_id=None,
     events, truncated = hub.pull(channel, after_cursor, limit)
     if events and not truncated and after_seq in (None, ""):
         return {"ok": True, "events": events, "next_cursor": events[-1]["cursor"],
-                "next_seq": None, "source": "hub", "truncated": False}
+                "next_seq": events[-1].get("seq"), "source": "hub",
+                "truncated": False}
 
-    sql = ("SELECT id, seq_no, event_type, payload, occurred_at, created_at FROM %s "
+    sql = ("SELECT id, %s, event_type, %s, %s, occurred_at, created_at FROM %s "
-           "WHERE tenant_id=%%s" % EVENT_TABLE)
+           "WHERE tenant_id=%%s"
+           % (EVENT_COL_SEQ, EVENT_COL_PAYLOAD, EVENT_COL_STATE_VERSION, EVENT_TABLE))
     params = [tenant_id]
     if world_id not in (None, ""):
         sql += " AND world_id=%s"
@@ -360,42 +456,48 @@ async def poll_events(tenant_id=None, world_id=None, session_id=None,
         sql += " AND session_id=%s"
         params.append(session_id)
     if after_seq not in (None, ""):
-        sql += " AND seq_no > %s"
+        sql += " AND %s > %s" % (EVENT_COL_SEQ, "%s")
         params.append(int(after_seq))
     elif truncated and events:
-        sql += " AND seq_no >= %s"
+        sql += " AND %s >= %s" % (EVENT_COL_SEQ, "%s")
-        params.append(int(events[0].get("seq_no") or 0))
+        params.append(int(events[0].get("seq") or 0))
-    sql += " ORDER BY seq_no ASC LIMIT %s" % limit
+    sql += " ORDER BY %s ASC LIMIT %s" % (EVENT_COL_SEQ, int(limit))
     rows = await rtx_db.q_all(sql, tuple(params))
     out = []
     for r in rows:
         r = dict(r)
-        r["payload"] = _loads(r.get("payload"), {})
+        r["payload"] = _loads(r.get(EVENT_COL_PAYLOAD), {})
+        r["seq"] = int(r.get(EVENT_COL_SEQ) or 0)
+        r["seq_no"] = r["seq"]
         r["channel"] = channel
         out.append(r)
-    next_seq = out[-1]["seq_no"] if out else (after_seq if after_seq is not None else 0)
+    next_seq = out[-1]["seq"] if out else (int(after_seq) if after_seq not in (None, "")
+                                           else 0)
     return {"ok": True, "events": out, "next_cursor": hub.latest_cursor(),
             "next_seq": next_seq, "source": "db", "truncated": bool(truncated)}
 
 
 # ---------------------------------------------------------------- 自检
 def self_check():
-    """离线自检：幂等键派生、append-only 守卫、列名自省、契约协程签名。"""
+    """离线自检：幂等键派生、append-only 守卫、列名与 models 对齐、契约协程签名。"""
     msgs, ok = [], True
 
-    k1 = idem_key_of(7, 1, 11, "move", {"x": 1})
+    k1 = idem_key_of(7, 11, "move", {"x": 1})
-    k2 = idem_key_of(7, 1, 11, "move", {"x": 1})
+    k2 = idem_key_of(7, 11, "move", {"x": 1})
-    k3 = idem_key_of(7, 1, 11, "move", {"x": 2})
+    k3 = idem_key_of(7, 11, "move", {"x": 2})
     if k1 != k2 or k1 == k3:
         ok = False
         msgs.append("tx_write FAIL：幂等键不稳定/不敏感 %s %s %s" % (k1, k2, k3))
-    else:
+    elif len(k1) > 32:
-        msgs.append("tx_write PASS：幂等键同参同键、异参异键（%s…）" % k1[:12])
-    if idem_key_of(7, 1, 11, "move", {}, client_key="cli-9") != "cli-9":
         ok = False
-        msgs.append("tx_write FAIL：client_key 未原样透传")
+        msgs.append("tx_write FAIL：幂等键超 event_uid 长度 32（%d）" % len(k1))
     else:
-        msgs.append("tx_write PASS：client_key 原样透传（重放同键）")
+        msgs.append("tx_write PASS：幂等键同参同键、异参异键、≤32 字符（%s…）" % k1[:12])
+    if idem_key_of(7, 11, "move", {}, client_key="cli-9") != "cli-9":
+        ok = False
+        msgs.append("tx_write FAIL：client_event_id 未原样透传")
+    else:
+        msgs.append("tx_write PASS：client_event_id 原样透传（重放同键）")
 
     blocked = []
     for op in ("UPDATE", "DELETE", "TRUNCATE"):
@@ -412,41 +514,53 @@ def self_check():
     if blocked:
         msgs.append("tx_write PASS：pbl_runtime_event 禁 %s" % "/".join(blocked))
     try:
-        assert_append_only("INSERT", EVENT_TABLE)
+        assert_append_only("UPDATE", "world")
-        assert_append_only("SELECT", EVENT_TABLE)
-        msgs.append("tx_write PASS：INSERT/SELECT 放行")
-    except rtx_db.RtxError as exc:
         ok = False
-        msgs.append("tx_write FAIL：合法操作被拦 %s" % exc.msg)
+        msgs.append("tx_write FAIL：基表写未拦截")
-    for base in _BASE_TABLES_LC:
+    except rtx_db.RtxError:
-        try:
+        msgs.append("tx_write PASS：基表写保护生效（world 等 %d 张）"
-            assert_append_only("UPDATE", base)
+                    % len(PROTECTED_BASE_TABLES))
-            ok = False
-            msgs.append("tx_write FAIL：基表 %s 未保护" % base)
-        except rtx_db.RtxError:
-            pass
-    msgs.append("tx_write PASS：基表写保护域 %d 张全覆盖" % len(_BASE_TABLES_LC))
 
-    cols = table_columns(EVENT_TABLE)
+    # 列名对齐自检：本文件写入用到的每一列都必须存在于 models/*.json
-    if cols:
+    plan = {
-        row = build_row(EVENT_TABLE, {"tenant_id": 7, "world_id": 1, "seq_no": 1,
+        EVENT_TABLE: ["tenant_id", EVENT_COL_UID, "session_id", "world_id", "actor_id",
-                                      "not_a_column": "x"})
+                      "event_type", EVENT_COL_PAYLOAD, "causation_id",
-        if "not_a_column" in row:
+                      EVENT_COL_STATE_VERSION, EVENT_COL_SEQ, "occurred_at",
+                      "created_at", "updated_at"],
+        STATE_TABLE: ["tenant_id", "session_id", STATE_COL_KEY, STATE_COL_VALUE,
+                      EVENT_COL_STATE_VERSION, "checksum", "updated_by",
+                      "created_at", "updated_at"],
+        SNAPSHOT_TABLE: ["tenant_id", "session_id", "snapshot_uid", "state_fingerprint",
+                         "entities_json", "events_through_seq", "reason", "created_at",
+                         "updated_at"],
+    }
+    for tbl, cols in plan.items():
+        real = table_columns(tbl)
+        if not real:
             ok = False
-            msgs.append("tx_write FAIL：未知列未被过滤")
+            msgs.append("tx_write FAIL：models/%s.json 不可读" % tbl)
+            continue
+        drift = [c for c in cols if c not in real]
+        if drift:
+            ok = False
+            msgs.append("tx_write FAIL：%s 列漂移 %s（真实列=%s）" % (tbl, drift, real))
         else:
-            msgs.append("tx_write PASS：列名自省生效（models/%s.json %d 列）"
+            msgs.append("tx_write PASS：%s 用到的 %d 列与 models 定义一致" % (tbl, len(cols)))
-                        % (EVENT_TABLE, len(cols)))
-    else:
-        msgs.append("tx_write WARN：离线未读到 models/%s.json，build_row 原样透传"
-                    % EVENT_TABLE)
 
-    for fn, name in ((apply_runtime_event, "apply_runtime_event"),
+    for name, fn in (("apply_runtime_event", apply_runtime_event),
-                     (poll_events, "poll_events"), (read_states, "read_states")):
+                     ("poll_events", poll_events), ("read_states", read_states)):
         if not inspect.iscoroutinefunction(fn):
             ok = False
-            msgs.append("tx_write FAIL：%s 非协程（dspy await 会拿到协程对象）" % name)
+            msgs.append("tx_write FAIL：%s 非协程" % name)
+    if callable(assert_append_only):
+        msgs.append("tx_write PASS：3 协程契约 + 1 同步守卫签名正确")
     if ok:
-        msgs.append("tx_write PASS：三个写/读契约均为 async 协程")
         msgs.append("tx_write.self_check PASS")
     return ok, msgs
+
+
+if __name__ == "__main__":
+    _ok, _msgs = self_check()
+    for _m in _msgs:
+        print(_m)
+    raise SystemExit(0 if _ok else 1)