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

.m11b2d_probe.py

@@ -0,0 +1,66 @@
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tests"))
+import _m11b2_compat as H  # noqa: E402
+
+e = H.RuntimeEngine()
+H.install(e)
+try:
+    r1 = H.apply(tenant_id=7, world_id=1, session_id=11, event_type="unit.move",
+                 payload={"dx": 1}, state_updates=[{"entity_id": "e1", "state": {"hp": 2}}],
+                 client_event_id="ev-1", actor_id="u1")
+    print("R1", sorted(r1.keys()))
+    print("R1 vals", r1["ok"], r1["dedup"], r1["event_id"], r1["seq"], r1["state_rows"], r1["stmt_count"], r1["channel"])
+    print("states", r1["states"])
+    print("trace", H.trace_kinds(e))
+    print("counts", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE))
+    print("state row", H.read(tenant_id=7, session_id=11))
+    # dedup
+    r2 = H.apply(tenant_id=7, world_id=1, session_id=11, event_type="unit.move",
+                 payload={"dx": 1}, state_updates=[{"entity_id": "e1", "state": {"hp": 9}}],
+                 client_event_id="ev-1")
+    print("R2", r2["dedup"], r2["event_id"], r2["seq"], r2["state_rows"], sorted(r2.keys()))
+    print("counts after dedup", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE))
+    # conflict
+    try:
+        H.apply(tenant_id=7, world_id=1, session_id=11, event_type="x",
+                payload={}, state_updates=[{"entity_id": "e1", "state": {"hp": 0}}],
+                client_event_id="ev-3", base_version=999)
+        print("NO CONFLICT !!")
+    except H.RtxError as exc:
+        print("conflict code", exc.code, "trace", H.trace_kinds(e)[-3:])
+        print("counts after conflict", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE))
+    # snapshot
+    r3 = H.apply(tenant_id=7, world_id=1, session_id=11, event_type="snap",
+                 payload={}, state_updates=[{"entity_id": "e1", "state": {"hp": 5}}],
+                 client_event_id="ev-4", snapshot={"reason": "join", "hp": 5})
+    print("snapshot rows", e.count(H.SNAPSHOT_TABLE), "broadcasts", len(e.broadcasts))
+    print("broadcast payload keys", sorted(r3.keys()))
+    print("UPDATE stmts", H.update_statements(e))
+    print("errors:")
+    for kw in ({"tenant_id": None}, {"world_id": None}, {"session_id": None}, {"event_type": None}):
+        base = dict(tenant_id=7, world_id=1, session_id=11, event_type="x", payload={})
+        base.update(kw)
+        try:
+            H.apply(**base)
+            print("  no raise for", kw)
+        except H.RtxError as exc:
+            print("  ", kw, "->", exc.code)
+    # bad state_updates
+    for su in ([{"state": {"a": 1}}], [{"entity_id": "e1"}], "notalist"):
+        try:
+            H.apply(tenant_id=7, world_id=1, session_id=11, event_type="x", payload={},
+                    state_updates=su, client_event_id="ev-b%d" % id(su) % 100)
+            print("  no raise for", su)
+        except H.RtxError as exc:
+            print("   su", su, "->", exc.code)
+    # missing tenant
+    try:
+        H.apply(tenant_id=None, world_id=1, session_id=11, event_type="x", payload={})
+    except H.RtxError as exc:
+        print("tenant->", exc.code)
+    print("final counts", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE), e.count(H.SNAPSHOT_TABLE))
+    print("trace tail", H.trace_kinds(e)[-6:])
+finally:
+    H.uninstall()

tests/_m11b2_compat.py

@@ -1,165 +1,449 @@
 # -*- coding: utf-8 -*-
-"""M11b-2c 测试夹具层兼容垫片（**仅测试使用，不属于模块运行时 API**）。
+"""M11b-2d-A 测试夹具层：**函数式 API 直连真源**的运行时事务引擎（仅测试使用）。
 
-背景（QC #3 允许路径：只改夹具/断言，不改业务实现、不改表定义 JSON）
+本文件的历史与本轮收敛（响应 QC #2 硬门禁）
-=====================================================================
+==========================================================
-``tests/test_m11b2_single_tx.py`` 是在 M11b-2a「世界同步真源迁移」之前写成的，它
+M11b-2a「世界同步真源迁移」把写入实现迁到 **函数式 API**
-绑定的是当时的类式写入器 API：
+``pbl_runtime_ext.tx_write.apply_runtime_event`` / ``read_states``
+（见 ``modules/pbl_runtime_ext/pbl_runtime_ext/tx_write.py:344`` / ``:466``）。
 
-    tx.RuntimeStateEventWriter / tx.build_plan / tx.percentile / tx.configure_writer
+本文件在 M11b-2c 时的形态是「类式垫片」：定义 ``RuntimeStateEventWriter`` 类，
+持有 ``conn_factory`` / ``dialect`` / ``close_connection``，再把方法调用转发到
+``world_sync/pbl_runtime_tx.write_event_with_state``。那是**第二条夹具路径**——
+它把测试钉在「旧类式调用形态 + world_sync 本地同步实现」上，既没验证现行真源
+``apply_runtime_event``，又与真源返回键（``action`` / ``states[]`` / ``stmt_count``
+/ ``latency_ms``）不一致，垫片只能靠 ``_augment`` 补假键（``state_inserted`` 等）
+维持旧断言，实测 22 failed（``Cannot operate on a closed database.`` /
+``KeyError: 'expected_version'``）。
 
-M11b-2a 把真源实现改为**函数式 API**（``pbl_runtime_tx.write_event_with_state`` /
+**本轮已删除该类别垫片**，本文件只保留一条夹具路径：
-``read_entity_state``，见 ``docs/02-develop/dev-notes-m11b2a-world-sync-migration.md``
-与 ``world_sync/init.py`` 第 17 行的导出说明），类式写入器与模块级全局 writer 单例
-被一并移除——``git log -S RuntimeStateEventWriter -- world_sync/pbl_runtime_tx.py``
-为空即证据：该类从未存在于当前真源。
 
-因此本垫片**不新增业务能力**，只做三件事：
+  1. ``RuntimeEngine``：以 **真实 sqlite3** 为存储的事务引擎。表结构（列名/类型/
-1. ``RuntimeStateEventWriter``：把「持有 conn_factory + dialect + close_connection」
+     NOT NULL/DEFAULT/唯一索引）**从 ``pbl_runtime_ext/models/*.json`` 读取生成**
-   的旧调用形态适配到当前 ``write_event_with_state`` 函数（同一条代码路径）。
+     ——不另立第二份 schema 定义，避免列名再次漂移（这正是 28 failed 的根因之一：
-2. ``build_plan``：转发到 ``world_sync.init.build_runtime_write_plan``（已存在的
+     旧夹具里手写 DDL 用了 ``event_id`` / ``state`` 等真源不存在的列）。
-   不落库预演函数），保持旧名字可用。
+  2. ``install(engine)`` / ``uninstall()``：把 ``rtx_db.transaction`` /
-3. ``percentile``：转发到 ``pbl_runtime_sql.percentile``（真源唯一实现，避免第二份
+     ``rtx_db.dbname`` / ``rtx_db.q_all`` 换成 sqlite 引擎实现，使
-   分位数算法口径漂移）。
+     ``tx_write.apply_runtime_event`` / ``read_states`` **走自身真实代码路径**
-4. ``configure_writer``：旧「模块级默认 writer」语义在真源已不存在；这里只在**测试
+     （幂等预检 → INSERT 事件 → UPSERT 状态 → 快照 → COMMIT → 提交后广播）。
-   进程内**保存一个默认连接工厂，供 ``write_event_with_state`` 的兼容包装取用，
+     与 ``modules/pbl_runtime_ext/scripts/m11b_selftest.py::install_fake`` 同一
-   不写 ServerEnv、不影响生产路径。
+     思路，差别只在存储换成真 sqlite（可验证真实 ROLLBACK 与 DB 唯一索引兜底）。
+  3. ``apply(**kw)`` / ``read(**kw)``：函数式 API 的同步入口（``asyncio.run``）。
+     **不做任何返回键增补**——测试直接断言真源返回键。
+  4. ``percentile``：转发到 ``world_sync.pbl_runtime_sql.percentile``（真源唯一实现）。
 
-返回键差异也在垫片里收敛：真源返回 ``action``（``inserted``/``updated``/
+边界（QC #3 允许路径）：只改测试夹具层，**不改业务实现、不改 models/*.json 表定义、
-``event_only``），旧断言读 ``state_inserted``（bool）。垫片补 ``state_inserted``
+不改任何 load_path.py RBAC 注册**。本文件不含任何生产能力：只有被 pytest 夹具调用时
-= ``action == "inserted"``，不改动真源返回值。
+才安装引擎补丁，``uninstall()`` 逐样还原。
 """
 
-import importlib.util
+import asyncio
+import json
 import os
+import sqlite3
 import sys
-import types
+import threading
+import time
 
 HERE = os.path.dirname(os.path.abspath(__file__))
-PKG_DIR = os.path.dirname(HERE)
+WS_ROOT = os.path.dirname(HERE)                       # modules/world_sync
-_PKG = "_ws_m11b2_compat"
+RTX_ROOT = os.path.join(os.path.dirname(WS_ROOT), "pbl_runtime_ext")   # modules/pbl_runtime_ext
 
-# 与 test_m11b2_single_tx.py 使用同一个合成包名，确保加载的是同一批模块实例
+for _p in (WS_ROOT, RTX_ROOT):
-# （否则 configure_writer 保存的默认工厂与被测函数看到的不是同一份状态）。
+    if _p not in sys.path:
-_SHARED_PKG = "_ws_m11b2"
+        sys.path.insert(0, _p)
 
+# --- 真源导入（被测对象本体，不是副本）---------------------------------
+from pbl_runtime_ext import broadcast as _bcast        # noqa: E402
+from pbl_runtime_ext import m11b_config as _cfg        # noqa: E402
+from pbl_runtime_ext import rtx_db as _rtx_db          # noqa: E402
+from pbl_runtime_ext import tx_write as _tx_write      # noqa: E402
+from world_sync import pbl_runtime_sql as ps           # noqa: E402
 
-def _load(name, pkg=None):
+# 旧夹具里以名字引用的常量：统一取真源，避免第二份字面量
-    pkg = pkg or _SHARED_PKG
-    if pkg not in sys.modules:
-        mod = types.ModuleType(pkg)
-        mod.__path__ = [PKG_DIR]
-        sys.modules[pkg] = mod
-    full = "%s.%s" % (pkg, name)
-    if full in sys.modules:
-        return sys.modules[full]
-    spec = importlib.util.spec_from_file_location(
-        full, os.path.join(PKG_DIR, name + ".py"))
-    mod = importlib.util.module_from_spec(spec)
-    sys.modules[full] = mod
-    spec.loader.exec_module(mod)
-    return mod
-
-
-tx = _load("pbl_runtime_tx")
-ps = _load("pbl_runtime_sql")
-
-# --- 旧名字 -> 真源实现 -------------------------------------------------
 SQLITE = ps.DIALECT_SQLITE
 MYSQL = ps.DIALECT_MYSQL
-percentile = ps.percentile
+percentile = ps.percentile                       # 真源唯一实现（world_sync.pbl_runtime_sql）
+apply_runtime_event = _tx_write.apply_runtime_event   # 现行函数式真源 API
+read_states = _tx_write.read_states
+build_row = _tx_write.build_row
+assert_append_only = _tx_write.assert_append_only
+RtxError = _rtx_db.RtxError
 
-_write_event_with_state = tx.write_event_with_state
+EVENT_TABLE = _cfg.EVENT_TABLE
-_read_entity_state = tx.read_entity_state
+STATE_TABLE = _cfg.STATE_TABLE
+SNAPSHOT_TABLE = _cfg.SNAPSHOT_TABLE
+PROTECTED_BASE_TABLES = _cfg.PROTECTED_BASE_TABLES
+ERR_TENANT_MISSING = _cfg.ERR_TENANT_MISSING
+ERR_BAD_PARAM = _cfg.ERR_BAD_PARAM
+ERR_STATE_CONFLICT = _cfg.ERR_STATE_CONFLICT
+ERR_APPEND_ONLY = _cfg.ERR_APPEND_ONLY
+ERR_BROADCAST_AFTER_COMMIT = _cfg.ERR_BROADCAST_AFTER_COMMIT
+# build_row 的三道门禁码（真源 tx_write.py 内以字面量抛出）
+ERR_UNKNOWN_COLUMN = "PBL-RTX-0003"
+ERR_MISSING_NOT_NULL = "PBL-RTX-0004"
+ERR_IMMUTABLE_COLUMN = "PBL-RTX-0006"
 
-# 测试进程内的「默认连接工厂」（仅服务于旧 configure_writer 语义）
+_MODELS_DIR = os.path.join(RTX_ROOT, "models")
-_default = {"factory": None, "dialect": None, "close_connection": True}
 
 
-def configure_writer(conn_factory=None, dialect=None, close_connection=True,
+# ---------------------------------------------------------------- schema
-                     **_ignored):
+def _sql_type(t):
-    """旧模块级默认 writer 语义的测试内等价物：登记默认连接工厂。
+    """抽象类型 → sqlite 存储类（sqlite 动态类型，映射只为可读性/亲和性）。"""
-
+    t = str(t or "").lower()
-    真源已无全局 writer 单例（见 ``world_sync/init.py:356`` 的
+    if t in ("long", "int", "bigint", "smallint", "short"):
-    ``configure_runtime_writer`` —— 它登记的是 ServerEnv 契约，不是类实例）。
+        return "INTEGER"
-    本函数只在测试进程内记一个 factory，供同名兼容包装使用。
+    if t in ("float", "double", "ddouble", "decimal"):
-    """
+        return "REAL"
-    _default["factory"] = conn_factory
+    if t in ("text", "json"):
-    _default["dialect"] = dialect
+        return "TEXT"
-    _default["close_connection"] = close_connection
+    if t in ("date", "datetime", "time", "timestamp"):
-    return conn_factory
+        return "TEXT"
+    return "TEXT"
 
 
-def write_event_with_state(event_data=None, state_update=None, conn_factory=None,
+def _not_null(fd):
-                           conn=None, event_uid=None, source=None, clock=None):
+    if fd.get("null") is False:
-    """兼容包装：未显式给 conn/conn_factory 时取 ``configure_writer`` 登记的工厂。"""
+        return True
-    if conn is None and conn_factory is None:
+    return str(fd.get("nullable") or "").strip().lower() in ("no", "false", "0")
-        conn_factory = _default["factory"]
-    res = _write_event_with_state(event_data, state_update,
-                                  conn_factory=conn_factory, conn=conn,
-                                  event_uid=event_uid, source=source, clock=clock)
-    return _augment(res)
 
 
-def _augment(result):
+def _default_clause(fd):
-    """补旧断言用到的键，不改动真源已有键的语义。"""
+    dv = fd.get("default")
-    if isinstance(result, dict):
+    if dv is None:
-        action = result.get("action")
+        return ""
-        result.setdefault("state_inserted", action == "inserted")
+    s = str(dv).strip()
-        result.setdefault("state_updated", action == "updated")
+    if s.upper() in ("CURRENT_TIMESTAMP", "NULL", "NONE", ""):
-        result.setdefault("event_only", action == "event_only")
+        return ""
-    return result
+    if isinstance(dv, bool):
+        return " DEFAULT %d" % int(dv)
+    if isinstance(dv, (int, float)):
+        return " DEFAULT %s" % dv
+    return " DEFAULT '%s'" % s.replace("'", "''")
 
 
-class RuntimeStateEventWriter(object):
+def model_fields(table):
-    """旧类式写入器的测试适配：持有 conn_factory/dialect/close_connection，
+    """表列定义（权威源 = pbl_runtime_ext/models/{table}.json）。"""
-    方法体直接调用真源函数式 API（同一条代码路径，不复制业务逻辑）。
+    path = os.path.join(_MODELS_DIR, "%s.json" % table)
+    with open(path, "r", encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+def ddl_statements(table):
+    """按 models/*.json 生成 sqlite DDL（列/NOT NULL/DEFAULT/唯一索引）。"""
+    spec = model_fields(table)
+    cols = []
+    for fd in spec.get("fields") or []:
+        name = fd.get("name")
+        if not name:
+            continue
+        if fd.get("auto_increment"):
+            cols.append('"%s" INTEGER PRIMARY KEY AUTOINCREMENT' % name)
+            continue
+        d = '"%s" %s' % (name, _sql_type(fd.get("type")))
+        if _not_null(fd):
+            d += " NOT NULL"
+        d += _default_clause(fd)
+        cols.append(d)
+    out = ['CREATE TABLE IF NOT EXISTS "%s" (\n  %s\n)' % (table, ",\n  ".join(cols))]
+    for idx in spec.get("indexes") or []:
+        fields = idx.get("idxfields") or []
+        if not fields:
+            continue
+        kind = "UNIQUE INDEX" if str(idx.get("idxtype")) == "unique" else "INDEX"
+        cols_sql = ", ".join('"%s"' % f for f in fields)
+        out.append('CREATE %s IF NOT EXISTS "%s" ON "%s" (%s)'
+                   % (kind, idx.get("name") or ("%s_%d" % (table, len(out))), table, cols_sql))
+    return out
+
+
+def _to_qmark(sql):
+    """真源 SQL 用 sqlor 的 ``%s`` 占位；sqlite 用 ``?``。"""
+    return sql.replace("%s", "?")
+
+
+# ---------------------------------------------------------------- 引擎
+class RuntimeEngine(object):
+    """真实 sqlite 事务引擎（测试专用），适配 ``rtx_db.Transaction`` 接口。
+
+    记录三条可断言的轨迹：
+      * ``trace``   有序事务轨迹：``BEGIN`` / ``SQL`` / ``COMMIT`` / ``ROLLBACK`` /
+                    ``BROADCAST`` —— 用于「广播只在 commit 之后」「BEGIN/COMMIT 配平」
+      * ``sql_log`` 实际发出的 SQL（含参数），用于「SET 子句不含不可变列」「无广播/
+                    轮询/睡眠语句」这类语句级断言
+      * ``broadcasts`` 提交后广播消息（hub.publish 录制）
     """
 
-    def __init__(self, conn_factory=None, dialect=None, close_connection=True,
+    def __init__(self, path=":memory:", tables=(EVENT_TABLE, STATE_TABLE, SNAPSHOT_TABLE)):
-                 source=None, clock=None, hooks=None, **extra):
+        self.path = path
-        self.conn_factory = conn_factory
+        self.tables = tuple(tables)
-        self.dialect = dialect
+        self.trace = []
-        self.close_connection = close_connection
+        self.sql_log = []
-        self.source = source
+        self.broadcasts = []
-        self.clock = clock
+        self.txn_depth = 0
-        self.hooks = hooks or {}
+        self._lock = threading.RLock()
-        self._extra = dict(extra)
+        self._mem = None
-        self.calls = []
+        if path == ":memory:":
+            self._mem = sqlite3.connect(":memory:", isolation_level=None,
+                                        check_same_thread=False)
+            self._mem.execute("PRAGMA busy_timeout=15000")
+            self._create(self._mem)
+        else:
+            c = sqlite3.connect(path, isolation_level=None)
+            c.execute("PRAGMA busy_timeout=15000")
+            self._create(c)
+            c.close()
 
-    def write_event_with_state(self, event_data=None, state_update=None, **kw):
+    # -- 建表 -------------------------------------------------------------
-        conn = kw.pop("conn", None)
+    def _create(self, conn):
-        event_uid = kw.pop("event_uid", self._extra.get("event_uid"))
+        for t in self.tables:
-        source = kw.pop("source", self.source)
+            for stmt in ddl_statements(t):
-        clock = kw.pop("clock", self.clock)
+                conn.execute(stmt)
-        conn_factory = kw.pop("conn_factory", None) or self.conn_factory
+        conn.commit()
-        pre = self.hooks.get("before_write")
-        if callable(pre):
-            pre(event_data, state_update)
-        if conn is None and conn_factory is None:
-            conn_factory = _default["factory"]
-        post = self.hooks.get("after_write")
-        result = _write_event_with_state(event_data, state_update,
-                                         conn_factory=conn_factory, conn=conn,
-                                         event_uid=event_uid, source=source,
-                                         clock=clock)
-        self.calls.append(result.get("event_uid"))
-        if callable(post):
-            post(result)
-        return _augment(result)
 
-    # 旧代码里偶有以 read_state 名称取当前状态的用法
+    # -- 连接 -------------------------------------------------------------
-    def read_entity_state(self, tenant_id=None, session_id=None, entity_id=None,
+    def admin_conn(self):
-                          world_id=None):
+        """给测试直接查库用的连接（文件库返回新连接，调用方负责 close）。"""
-        res = _read_entity_state(conn_factory=self.conn_factory,
+        if self._mem is not None:
-                                 tenant_id=tenant_id, session_id=session_id,
+            return self._mem, False
-                                 entity_id=entity_id, world_id=world_id)
+        c = sqlite3.connect(self.path, isolation_level=None)
-        return res
+        c.execute("PRAGMA busy_timeout=15000")
+        return c, True
+
+    def query(self, sql, params=()):
+        conn, own = self.admin_conn()
+        try:
+            cur = conn.execute(_to_qmark(sql), tuple(params or ()))
+            cols = [d[0] for d in cur.description] if cur.description else []
+            return [dict(zip(cols, r)) for r in cur.fetchall()]
+        finally:
+            if own:
+                conn.close()
+
+    def scalar(self, sql, params=(), default=None):
+        rows = self.query(sql, params)
+        if not rows:
+            return default
+        vals = list(rows[0].values())
+        return vals[0] if vals else default
+
+    def count(self, table):
+        return int(self.scalar("SELECT COUNT(*) AS c FROM \"%s\"" % table, (), 0) or 0)
+
+    def seed_state(self, tenant_id, session_id, entity_id, state_json,
+                   state_version=0, checksum="", updated_by="seed"):
+        """预置状态行（走真实 INSERT，不绕过列契约）。"""
+        now = time.strftime("%Y-%m-%d %H:%M:%S")
+        conn, own = self.admin_conn()
+        try:
+            conn.execute("DELETE FROM \"%s\" WHERE tenant_id=? AND session_id=? AND entity_id=?"
+                         % STATE_TABLE, (str(tenant_id), str(session_id), str(entity_id)))
+            conn.execute(
+                "INSERT INTO \"%s\" (tenant_id, session_id, entity_id, state_json,"
+                " state_version, checksum, updated_by, created_at, updated_at)"
+                " VALUES (?,?,?,?,?,?,?,?,?)" % STATE_TABLE,
+                (str(tenant_id), str(session_id), str(entity_id), state_json,
+                 int(state_version), checksum, updated_by, now, now))
+            conn.commit()
+        finally:
+            if own:
+                conn.close()
+
+    # -- 事务轨迹 ---------------------------------------------------------
+    def _note(self, kind, detail=None):
+        with self._lock:
+            self.trace.append((kind, detail))
+
+    # -- 只读查询（供 rtx_db.q_all 补丁）----------------------------------
+    async def q_all(self, sql, params=None, db=None):
+        return _rtx_db._norm_rows(self.query(sql, params))
 
 
-def build_plan(event_data=None, state_update=None, event_uid=None, source=None,
+class _Tx(object):
-               now=None):
+    """sqlite 事务句柄：与 ``rtx_db.Transaction`` 同接口（execute/insert/update/
-    """旧 build_plan 名字 -> 真源 init.build_runtime_write_plan（不落库预演）。"""
+    select/commit/rollback/stmt_count），**stmt_count 只计 DML**（与真源一致：
-    init_mod = _load("init", pkg=_PKG)
+    真源 ``execute`` 不计数、``insert``/``update`` 计数）。
-    return init_mod.build_runtime_write_plan(
+    """
-        event_data, state_update, event_uid=event_uid, source=source, now=now)
+
+    def __init__(self, engine):
+        self.e = engine
+        self.conn = None
+        self.committed = False
+        self.rolled_back = False
+        self.stmt_count = 0
+
+    async def __aenter__(self):
+        self.conn = self.e._mem if self.e._mem is not None else sqlite3.connect(
+            self.e.path, isolation_level=None)
+        if self.e._mem is None:
+            self.conn.execute("PRAGMA busy_timeout=15000")
+        with self.e._lock:
+            self.conn.execute("BEGIN IMMEDIATE")
+            self.e.txn_depth += 1
+        self.e._note("BEGIN")
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        if exc_type is not None or not self.committed:
+            await self.rollback()
+        self._close()
+        return False
+
+    def _close(self):
+        if self.e._mem is None and self.conn is not None:
+            try:
+                self.conn.close()
+            except Exception:  # noqa: BLE001
+                pass
+        self.conn = None
+
+    def _run(self, sql, params):
+        self.e.sql_log.append((" ".join(str(sql).split()), tuple(params or ())))
+        cur = self.conn.cursor()
+        try:
+            cur.execute(_to_qmark(sql), tuple(params or ()))
+            cols = [d[0] for d in cur.description] if cur.description else []
+            rows = [dict(zip(cols, r)) for r in cur.fetchall()] if cols else []
+            return rows
+        finally:
+            try:
+                cur.close()
+            except Exception:  # noqa: BLE001
+                pass
+
+    # -- DML / 查询 -------------------------------------------------------
+    async def execute(self, sql, params=None):
+        return self._run(sql, params)
+
+    async def insert(self, table, row):
+        row = dict(row)
+        if "tenant_id" in row and row.get("tenant_id") in (None, "", 0):
+            raise RtxError(ERR_TENANT_MISSING, "%s 写入缺少 tenant_id" % table)
+        cols = list(row.keys())
+        sql = 'INSERT INTO "%s" (%s) VALUES (%s)' % (
+            table, ", ".join('"%s"' % c for c in cols), ", ".join(["?"] * len(cols)))
+        self.e.sql_log.append((" ".join(sql.split()), tuple(row[c] for c in cols)))
+        cur = self.conn.cursor()
+        try:
+            cur.execute(sql, tuple(row[c] for c in cols))
+            rid = cur.lastrowid
+        finally:
+            try:
+                cur.close()
+            except Exception:  # noqa: BLE001
+                pass
+        self.stmt_count += 1
+        return rid
+
+    async def update(self, table, values, where):
+        cols = list(dict(values).keys())
+        sql = 'UPDATE "%s" SET %s' % (table, ", ".join('"%s" = ?' % c for c in cols))
+        self.e.sql_log.append((" ".join(sql.split()), tuple(dict(values)[c] for c in cols)))
+        cur = self.conn.cursor()
+        try:
+            cur.execute(sql, tuple(dict(values)[c] for c in cols))
+            n = cur.rowcount
+        finally:
+            try:
+                cur.close()
+            except Exception:  # noqa: BLE001
+                pass
+        self.stmt_count += 1
+        return n
+
+    async def select(self, table, where="", fields="*", order="", limit=200):
+        sql = 'SELECT %s FROM "%s"' % (fields or "*", table)
+        if where:
+            sql += " WHERE %s" % where
+        if order:
+            sql += " ORDER BY %s" % order
+        sql += " LIMIT %d" % int(limit or 200)
+        return self._run(sql, ())
+
+    # -- 事务控制 ---------------------------------------------------------
+    async def commit(self):
+        if self.committed or self.rolled_back:
+            return
+        self.conn.commit()
+        self.committed = True
+        with self.e._lock:
+            self.e.txn_depth -= 1
+        self.e._note("COMMIT")
+
+    async def rollback(self):
+        if self.committed or self.rolled_back:
+            return
+        try:
+            self.conn.rollback()
+        except Exception:  # noqa: BLE001
+            pass
+        self.rolled_back = True
+        with self.e._lock:
+            self.e.txn_depth -= 1
+        self.e._note("ROLLBACK")
+
+
+# ---------------------------------------------------------------- 安装 / 还原
+_ORIG = {}
+_INSTALLED = False
+
+
+def install(engine):
+    """把 ``rtx_db`` 的事务/查询入口换成 sqlite 引擎，并录制提交后广播。"""
+    global _INSTALLED
+    if _INSTALLED:
+        uninstall()
+    _ORIG["transaction"] = _rtx_db.transaction
+    _ORIG["dbname"] = _rtx_db.dbname
+    _ORIG["q_all"] = _rtx_db.q_all
+    _rtx_db.dbname = lambda env=None: "m11b2d_test_sqlite"
+    _rtx_db.transaction = lambda db=None: _Tx(engine)
+    _rtx_db.q_all = engine.q_all
+
+    _bcast.reset_hub()
+    hub = _bcast.get_hub()
+    original_publish = hub.publish
+
+    def _recording_publish(channel, event):
+        engine.broadcasts.append({"channel": channel, "event": event})
+        engine._note("BROADCAST", channel)
+        return original_publish(channel, event)
+
+    hub.publish = _recording_publish
+    _ORIG["hub"] = (hub, original_publish)
+    _INSTALLED = True
+    return engine
+
+
+def uninstall():
+    global _INSTALLED
+    if not _INSTALLED:
+        return
+    for key in ("transaction", "dbname", "q_all"):
+        if key in _ORIG:
+            setattr(_rtx_db, key, _ORIG[key])
+    _INSTALLED = False
+    _ORIG.clear()
+
+
+def apply(**kwargs):
+    """函数式真源 API 的同步入口（不做返回键增补）。"""
+    return asyncio.run(apply_runtime_event(**kwargs))
+
+
+def read(**kwargs):
+    return asyncio.run(read_states(**kwargs))
+
+
+def trace_kinds(engine):
+    return [k for k, _d in engine.trace]
+
+
+def update_statements(engine):
+    """引擎实际发出的 UPDATE 语句（大写、压空格），用于不可变列/乐观锁断言。"""
+    return [(s, p) for s, p in engine.sql_log if s.upper().startswith("UPDATE ")]
+
+
+def insert_statements(engine, table):
+    return [(s, p) for s, p in engine.sql_log
+            if s.upper().startswith('INSERT INTO "%s"' % table.upper())]