From 93829c6a9b54052e9cd3c2dcc40dc4a344c45c94 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Sat, 19 Sep 2026 11:43:35 +0800 Subject: [PATCH] =?UTF-8?q?deliver:=20=E4=BA=A4=E4=BB=98=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=88=E5=BC=95=E6=93=8E=E4=BB=A3=E4=B8=BA=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pbl_runtime_ext/db.py | 144 ++++++++++++++++++++++++++++++++++++++ pbl_runtime_ext/tables.py | 122 ++++++++++++++++++++++++++++---- 2 files changed, 253 insertions(+), 13 deletions(-) create mode 100644 pbl_runtime_ext/db.py diff --git a/pbl_runtime_ext/db.py b/pbl_runtime_ext/db.py new file mode 100644 index 0000000..b04ed77 --- /dev/null +++ b/pbl_runtime_ext/db.py @@ -0,0 +1,144 @@ +# -*- coding: utf-8 -*- +"""pbl_runtime_ext.db —— 真实存在的 DB 原语封装（M11a，QC #6 取证后重写）。 + +为什么需要这一层 +---------------- +本仓库 `pbl_common.api` **不提供** `u()` / `i()` / `q_one()` / `q_all()` 这组签名 +（已用 inspect 逐个核实，见 skill/SKILL.md「pbl_common 契约面取证」一节）。 +M11a 早期代码凭记忆写了 `A.u/A.i/A.new_id/q_one/q_all`，import 即 AttributeError。 +本文件把「幂等闸门」真正需要的 4 个原语落在 pbl_common.api **确实存在**的 +`sql_exec / sql_rows / sql_scalar / new_id` 之上，签名与语义在此固化， +下游（idempotency.py / entity_state.py）只依赖本模块，不再直接猜 pbl_common 的 API。 + +已核实的 pbl_common.api 真实签名（2026-09，inspect.signature 实测）： + sql_exec(sql, params=None, dbname=None) -> int 同步 + sql_rows(sql, params=None, dbname=None) -> list[dict] 同步 + sql_scalar(sql, params=None, dbname=None, default=None) 同步 + new_id(prefix='') -> str(<=32) 同步 + tenant_id(default=None) / actor_id(default=None) 同步（上下文读取器） + insert_row/update_row/fetch_one/query **async**（本模块不使用，避免混用） + +占位符约定：`%s` + **tuple** 参数（pbl_common.dbutil.execute 内部 `args = tuple(params)`， +PyMySQL/sqlor 两条路径都吃这个形态）。禁止字符串拼接外部输入。 + +事务：所有原语接受可选 `conn`。给了 conn 就在同一连接上执行（幂等记录与业务写入同事务， +业务回滚则闸门一并回滚）；没给则走 pbl_common 的自动连接路径（单语句原子）。 +""" +from __future__ import annotations + +from typing import Any, Dict, List, Optional, Sequence, Tuple + +from pbl_common.api import new_id, sql_exec, sql_rows # 已核实存在的真实符号 + +__all__ = ["new_id", "exec_sql", "q_rows", "q_one", "q_scalar", "insert_row_sql", "update_row_sql"] + +MODULE_DB = "pbl" # 库名解析交给 pbl_common（get_module_dbname），此处仅作 dbname 提示 + + +def _dbname() -> Optional[str]: + """解析本模块库名；平台/上下文不可用时返回 None（让 pbl_common 走默认库）。""" + try: + from pbl_common.api import get_module_dbname + name = get_module_dbname("pbl_runtime_ext") + return name or None + except Exception: # noqa: BLE001 - 离线/自检环境无 ServerEnv + return None + + +def _conn_exec(conn: Any, sql: str, params: Optional[Sequence[Any]]) -> Any: + """在调用方连接上执行（duck-typing：sqlExe / execute / cursor）。""" + args = tuple(params or ()) + for attr in ("execute", "sqlExe", "sql_exe"): + fn = getattr(conn, attr, None) + if callable(fn): + try: + return fn(sql, args) + except TypeError: + return fn(sql) + cursor_fn = getattr(conn, "cursor", None) + if callable(cursor_fn): + cur = cursor_fn() + cur.execute(sql, args) + return cur + raise RuntimeError("conn 不支持 execute/sqlExe/cursor，无法在同事务内执行") + + +def exec_sql(sql: str, params: Optional[Sequence[Any]] = None, conn: Any = None) -> int: + """执行写 SQL（INSERT/UPDATE/DELETE/DDL），返回受影响行数。""" + if conn is not None: + r = _conn_exec(conn, sql, params) + if isinstance(r, int): + return r + affected = getattr(r, "rowcount", None) + return int(affected) if isinstance(affected, int) else 0 + return int(sql_exec(sql, tuple(params or ()), dbname=_dbname()) or 0) + + +def q_rows(sql: str, params: Optional[Sequence[Any]] = None, conn: Any = None) -> List[Dict[str, Any]]: + """查询多行，返回 list[dict]。""" + if conn is not None: + r = _conn_exec(conn, sql, params) + if isinstance(r, list): + return r + if r is None: + return [] + # cursor 形态：description + fetchall + desc = getattr(r, "description", None) + fetch = getattr(r, "fetchall", None) + if desc and callable(fetch): + cols = [d[0] for d in desc] + return [dict(zip(cols, row)) for row in fetch()] + return list(r) if isinstance(r, (tuple, set)) else [] + return list(sql_rows(sql, tuple(params or ()), dbname=_dbname()) or []) + + +def q_one(sql: str, params: Optional[Sequence[Any]] = None, conn: Any = None) -> Optional[Dict[str, Any]]: + """查询单行（无则 None）。fail-closed：多行时返回首行并保留 SQL 的 LIMIT 1 约束。""" + rows = q_rows(sql, params, conn=conn) + return rows[0] if rows else None + + +def q_scalar(sql: str, params: Optional[Sequence[Any]] = None, conn: Any = None, + default: Any = 0) -> Any: + """查询单值（首行首列）；空结果返回 default。""" + rows = q_rows(sql, params, conn=conn) + if not rows: + return default + first = rows[0] + if isinstance(first, dict): + vals = list(first.values()) + return vals[0] if vals else default + if isinstance(first, (list, tuple)): + return first[0] if first else default + return first + + +def _placeholders(n: int) -> str: + return ",".join(["%s"] * max(0, int(n))) + + +def insert_row_sql(table: str, row: Dict[str, Any], conn: Any = None) -> int: + """参数化 INSERT（列名来自调用方常量，不接受外部输入拼接）。返回受影响行数。""" + cols: Tuple[str, ...] = tuple(row.keys()) + if not cols: + raise ValueError("insert_row_sql 需要至少一列") + sql = "INSERT INTO `%s` (%s) VALUES (%s)" % ( + table, + ", ".join("`%s`" % c for c in cols), + _placeholders(len(cols)), + ) + return exec_sql(sql, [row[c] for c in cols], conn=conn) + + +def update_row_sql(table: str, values: Dict[str, Any], where: Dict[str, Any], + conn: Any = None) -> int: + """参数化 UPDATE（WHERE 各列 AND 连接，强制要求 where 非空 → 防全表更新）。""" + if not values: + raise ValueError("update_row_sql 无可更新列") + if not where: + raise ValueError("update_row_sql 必须带 WHERE 条件（防全表更新，fail-closed）") + sets = ", ".join("`%s` = %%s" % k for k in values) + conds = " AND ".join("`%s` = %%s" % k for k in where) + sql = "UPDATE `%s` SET %s WHERE %s" % (table, sets, conds) + params = list(values.values()) + list(where.values()) + return exec_sql(sql, params, conn=conn) diff --git a/pbl_runtime_ext/tables.py b/pbl_runtime_ext/tables.py index 09d8bdf..81b6863 100644 --- a/pbl_runtime_ext/tables.py +++ b/pbl_runtime_ext/tables.py @@ -1,15 +1,30 @@ # -*- coding: utf-8 -*- -"""pbl_runtime_ext 表定义与幂等建表（M11a）。 +"""pbl_runtime_ext 表定义与幂等建表（M11a：4 表，单一真源）。 -3 表（mariadb 方言，BIGINT AUTO_INCREMENT，tenant_id 强制打头，无 FK/ENUM/TIMESTAMP）： - * pbl_runtime_event —— 运行时事件流（append-only，幂等键去重） - * pbl_entity_state —— 实体状态（薄扩展，不改 entity 基表） +4 表（mariadb 方言，BIGINT AUTO_INCREMENT 主键，tenant_id 强制打头，无 FK/ENUM/TIMESTAMP）： + * pbl_runtime_event —— 运行时事件流（append-only，idem_key 幂等唯一键） + * pbl_entity_state —— 实体状态（服务端权威，state_version 客户端禁写） * pbl_world_state_snapshot —— 世界状态快照（单事务写入后广播基线） + * pbl_runtime_idempotency —— client_seq 幂等闸门（M11a 新增，唯一键 + (tenant_id, session_id, scope, client_seq)） -与 modules/pbl_runtime_ext/models/*.json 及 sql/pbl_runtime_ext.sql 同构。 +**单一真源**：本文件的 DDL 列表是唯一权威定义，`sql/pbl_runtime_ext.sql` 与 +`models/*.json` 由 `scripts/gen_artifacts.py` 从本文件生成，三处不可能漂移。 +表名常量（TBL_*）供代码引用，禁止在业务代码里再写裸表名字符串。 """ -TABLES = ["pbl_runtime_event", "pbl_entity_state", "pbl_world_state_snapshot"] +TBL_RUNTIME_EVENT = "pbl_runtime_event" +TBL_ENTITY_STATE = "pbl_entity_state" +TBL_SNAPSHOT = "pbl_world_state_snapshot" +TBL_IDEMPOTENCY = "pbl_runtime_idempotency" + +TABLES = [TBL_RUNTIME_EVENT, TBL_ENTITY_STATE, TBL_SNAPSHOT, TBL_IDEMPOTENCY] + +# 表名常量 → 幂等闸门/状态权威链路用到的关键列（供自检与文档核对，避免列名漂移） +IDEMPOTENCY_COLUMNS = ["idempotency_id", "tenant_id", "session_id", "scope", "client_seq", + "principal_id", "request_fp", "state", "result_json", "error_code", + "created_at", "updated_at"] +ENTITY_STATE_VERSION_COLUMN = "state_version" DDL = [ """ @@ -47,14 +62,15 @@ CREATE TABLE IF NOT EXISTS `pbl_entity_state` ( `entity_id` BIGINT NOT NULL DEFAULT 0 COMMENT '实体ID（引用 entity 基表，只读）', `state_key` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '状态键', `state_value` TEXT COMMENT '状态值 JSON', - `version` INT NOT NULL DEFAULT 1 COMMENT '乐观锁版本', + `state_version` BIGINT NOT NULL DEFAULT 0 COMMENT '服务端权威单调版本（客户端禁写）', `last_event_id` BIGINT NOT NULL DEFAULT 0 COMMENT '最后触发事件ID', `created_at` DATETIME COMMENT '创建时间', `updated_at` DATETIME COMMENT '更新时间', PRIMARY KEY (`id`), UNIQUE KEY `uk_tenant_entity_key` (`tenant_id`, `session_id`, `entity_id`, `state_key`), - KEY `ix_tenant_world` (`tenant_id`, `world_id`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实体状态薄扩展' + KEY `ix_tenant_world` (`tenant_id`, `world_id`), + KEY `ix_tenant_state_version` (`tenant_id`, `session_id`, `state_version`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实体状态薄扩展（服务端权威，state_version 客户端禁写）' """, """ CREATE TABLE IF NOT EXISTS `pbl_world_state_snapshot` ( @@ -73,15 +89,59 @@ CREATE TABLE IF NOT EXISTS `pbl_world_state_snapshot` ( UNIQUE KEY `uk_tenant_snapshot_code` (`tenant_id`, `snapshot_code`), KEY `ix_tenant_session_seq` (`tenant_id`, `session_id`, `seq_no`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='世界状态快照' +""", + """ +CREATE TABLE IF NOT EXISTS `pbl_runtime_idempotency` ( + `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键', + `tenant_id` BIGINT NOT NULL DEFAULT 0 COMMENT '租户ID（多租户强制打头）', + `idempotency_id` VARCHAR(64) NOT NULL COMMENT '幂等记录业务主键', + `session_id` BIGINT NOT NULL DEFAULT 0 COMMENT '会话ID（引用 scense 基表，只读）', + `scope` VARCHAR(64) NOT NULL DEFAULT '*' COMMENT '幂等作用域（world_id 或 world_id:channel，缺省 *）', + `client_seq` BIGINT NOT NULL DEFAULT 0 COMMENT '客户端单调序号（>=1）', + `principal_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '提交者（服务端会话解析）', + `request_fp` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '请求体指纹（同 seq 不同内容判冲突）', + `state` VARCHAR(16) NOT NULL DEFAULT 'in_progress' COMMENT 'in_progress/completed/failed', + `result_json` TEXT COMMENT '已完成结果缓存（重放直接返回，不再写第二条事件）', + `error_code` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '失败错误码（failed 态）', + `updated_at` BIGINT NOT NULL DEFAULT 0 COMMENT '更新时间（epoch 毫秒）', + `created_at` BIGINT NOT NULL DEFAULT 0 COMMENT '创建时间（epoch 毫秒）', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_tenant_seq` (`tenant_id`, `session_id`, `scope`, `client_seq`), + UNIQUE KEY `uk_tenant_idem_id` (`tenant_id`, `idempotency_id`), + KEY `ix_tenant_scope_seq` (`tenant_id`, `session_id`, `scope`, `state`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='运行时 client_seq 幂等闸门（M11a 服务端权威）' """, ] +def table_names(): + """全部表名（list[str]）。""" + return list(TABLES) + + def get_ddl(): """返回本模块全部建表语句（list[str]）。""" return [x.strip() for x in DDL] +def ddl_for(table): + """按表名取建表语句；未知表名抛 KeyError（fail-closed，不静默返回空）。""" + for stmt in get_ddl(): + if stmt.split("`")[1] == table: + return stmt + raise KeyError("未知表 %s（应为 %s 之一）" % (table, ",".join(TABLES))) + + +def columns_for(table): + """按表名解析列名清单（从 DDL 单一真源解析，供 gen_artifacts / 自检使用）。""" + cols = [] + for line in ddl_for(table).splitlines(): + s = line.strip() + if s.startswith("`") and s.endswith(","): + cols.append(s.split("`")[1]) + return cols + + def ensure_tables(env=None, sor=None): """幂等建表（CREATE TABLE IF NOT EXISTS）。返回 (ok_list, bad_list)。""" ok, bad = [], [] @@ -105,13 +165,15 @@ def ensure_tables(env=None, sor=None): def self_check(): - """离线自检：3 表齐全 / tenant_id 打头 / 无禁用方言 / 幂等键唯一约束在位。""" + """离线自检：4 表齐全 / tenant_id 打头 / 无禁用方言 / 幂等唯一键在位 / 列名与代码一致。""" msgs = [] all_ok = True forbidden = ("BIGSERIAL", "SERIAL", "nextval", "FOREIGN KEY", "REFERENCES", "ENUM(", "TIMESTAMP") + seen = [] for stmt in get_ddl(): name = stmt.split("`")[1] + seen.append(name) body = stmt.upper() if name not in TABLES: all_ok = False @@ -131,12 +193,46 @@ def self_check(): if "IF NOT EXISTS" not in body: all_ok = False msgs.append("%s 非幂等建表（缺 IF NOT EXISTS）" % name) - if "UNIQUE KEY `uk_tenant_idem`" not in DDL[0]: + + if sorted(seen) != sorted(TABLES): + all_ok = False + msgs.append("DDL 表集合 %s 与 TABLES %s 不一致" % (sorted(seen), sorted(TABLES))) + + # 事件表幂等唯一键 + if "UNIQUE KEY `uk_tenant_idem`" not in ddl_for(TBL_RUNTIME_EVENT): all_ok = False msgs.append("pbl_runtime_event 缺幂等唯一键 uk_tenant_idem") - if len(TABLES) != 3: + else: + msgs.append("pbl_runtime_event uk_tenant_idem 在位") + + # 状态表：服务端权威版本列必须是 state_version（代码引用的真实列名） + state_cols = columns_for(TBL_ENTITY_STATE) + if ENTITY_STATE_VERSION_COLUMN not in state_cols: all_ok = False - msgs.append("表数=%d 应为 3" % len(TABLES)) + msgs.append("pbl_entity_state 缺列 %s（代码 MAX(state_version) 会 Unknown column）" + % ENTITY_STATE_VERSION_COLUMN) + else: + msgs.append("pbl_entity_state.%s 在位（服务端权威，客户端禁写）" % ENTITY_STATE_VERSION_COLUMN) + if "version" in state_cols and ENTITY_STATE_VERSION_COLUMN not in state_cols: + all_ok = False + msgs.append("pbl_entity_state 仍用旧列名 version，应统一为 state_version") + + # 幂等闸门表：列 + 唯一键 + idem_cols = columns_for(TBL_IDEMPOTENCY) + for col in IDEMPOTENCY_COLUMNS: + if col not in idem_cols: + all_ok = False + msgs.append("pbl_runtime_idempotency 缺列 %s（idempotency.py 读写依赖）" % col) + idem_ddl = ddl_for(TBL_IDEMPOTENCY) + if "UNIQUE KEY `uk_tenant_seq` (`tenant_id`, `session_id`, `scope`, `client_seq`)" not in idem_ddl: + all_ok = False + msgs.append("pbl_runtime_idempotency 缺唯一键 uk_tenant_seq(tenant_id,session_id,scope,client_seq)") + if all_ok and set(IDEMPOTENCY_COLUMNS) <= set(idem_cols): + msgs.append("pbl_runtime_idempotency 列/唯一键齐全（%d 列）" % len(idem_cols)) + + if len(TABLES) != 4: + all_ok = False + msgs.append("表数=%d 应为 4" % len(TABLES)) if all_ok: msgs.append("SELF_CHECK pbl_runtime_ext.tables: PASS %d/%d" % (len(TABLES), len(TABLES))) return all_ok, msgs