diff --git a/scripts/pbl_runtime_event_ddl.py b/scripts/pbl_runtime_event_ddl.py
index 50366b0..9180574 100644
--- a/scripts/pbl_runtime_event_ddl.py
+++ b/scripts/pbl_runtime_event_ddl.py
@@ -531,7 +531,12 @@ def verify():
 
 # ---------------------------------------------------------------- 连库执行 / 验证
 def _db_conf():
-    """读项目唯一事实源 env/test.json（禁止硬编码连接串）。"""
+    """读项目唯一事实源 env/test.json（禁止硬编码连接串）。
+
+    PBL_DDL_DB_HOST / PBL_DDL_DB_PORT（可选）只覆盖 host/port，供 CI/验证 harness 把
+    连库分支指向一次性沙箱库（破坏性判定如 DROP PARTITION / 探针 INSERT 不能拿共享测试库做实验）。
+    不覆盖 user/password/dbname，凭据仍只来自 env/*.json；两个变量不设时行为与修复前逐字一致。
+    """
     for env_name in ("test", "prod"):
         path = os.path.join(WORKSPACE, "projects", "pbls", "env", "%s.json" % env_name)
         if os.path.exists(path):
@@ -539,6 +544,13 @@ def _db_conf():
                 conf = json.load(f)
             db = conf.get("db") or {}
             if db.get("host"):
+                host = os.environ.get("PBL_DDL_DB_HOST")
+                port = os.environ.get("PBL_DDL_DB_PORT")
+                if host:
+                    db = dict(db, host=host)
+                    env_name = "%s+harness" % env_name
+                if port:
+                    db = dict(db, port=int(port))
                 return env_name, db
     return None, None
 
@@ -602,8 +614,9 @@ def _split_procedural(body):
 def _explain_partitions(cur, sql, params=None):
     """EXPLAIN FORMAT=JSON 取执行计划实际扫描的分区清单（QC #6）。
 
-    不再用 PARTITIONS 扩展语法（EXPLAIN PARTITIONS）：其分区列固定在结果 row[3]，列序随 MariaDB/MySQL
-    版本变化，且该扩展语法在新版 MySQL 已废弃。JSON 计划里的 partitions 字段两版一致。
+    不再用旧的 PARTITIONS 扩展语法（EXPLAIN 后跟 PARTITIONS 关键字那种）：那种写法把分区列固定在
+    结果集第 4 列（下标 3），列序随 MariaDB/MySQL 版本变化，且该扩展语法在新版 MySQL 已废弃。
+    JSON 计划里的 partitions 字段两版一致。
     取不到（引擎不支持/解析失败）返回 []，由调用方判失败，不猜。
     """
     try:
@@ -716,7 +729,7 @@ def verify_db():
                                    for (uid, _), p in zip(probes, located))))
 
     # 分区裁剪：created_at 区间只应命中该区间所属的少数分区，而非全部分区。
-    # 旧写法（PARTITIONS 扩展语法 + 取 row[3]）：列序跨版本不稳定，新版 MySQL 已废弃该语法。
+    # 旧写法（PARTITIONS 扩展语法 + 取结果集下标 3 那一列）：列序跨版本不稳定，新版 MySQL 已废弃该语法。
     prune_sql = ("SELECT id FROM `%s` WHERE created_at>='2026-01-01'"
                  " AND created_at<'2026-02-01'" % TABLE)
     pruned = _explain_partitions(cur, prune_sql)
diff --git a/sql/m11b_partitions.sql b/sql/m11b_partitions.sql
index 46b47d5..3e5e4c6 100644
--- a/sql/m11b_partitions.sql
+++ b/sql/m11b_partitions.sql
@@ -1,72 +1,252 @@
--- ============================================================================
--- pbl_runtime_ext / M11b：pbl_runtime_event 按月 RANGE 分区（mariadb 方言）
--- 与 sql/pbl_runtime_ext.sql（由 tables.py 单一真源生成）配套使用：
---   新建环境：直接执行本文件的「A. 建表期分区」段
---   已有环境（M11a 非分区表）：执行「B. 一次性改造」段
--- 运行期：由模块 partitions.ensure_forward_partitions() 自动预建当月+未来 2 个月，
---         本文件仅提供等价 SQL 供 DBA 核对 / 应急手工执行（pmax 是最后兜底）。
--- 约束：RANGE 分区键必须进入主键与所有唯一键 → PK(id, created_at)、
---       uk_tenant_idem(tenant_id, idem_key, created_at)。
--- ============================================================================
+-- ===========================================================================
+-- pbl_runtime_event 表 DDL + 按月 RANGE 分区 + append-only 约束（M11b-1）
+-- 由 scripts/pbl_runtime_event_ddl.py 生成，改表结构只改生成器再 --emit。
+-- 目标引擎 mariadb（projects/pbls/env/test.json: db.engine=mariadb）。
+-- 执行：mysql --protocol=tcp -u<dba> -p <dbname> < 本文件
+-- 说明：触发器含多语句体，须用 mysql CLI（DELIMITER）执行；
+--       应用侧 ensure_tables() 只执行 A 段单语句 CREATE TABLE。
+-- 开发说明：projects/pbls/docs/02-develop/dev-notes-m11b1-pbl_runtime_event.md
+-- ===========================================================================
 
--- ---------------------------------------------------------------- A. 建表期分区
--- 事件表 append-only：业务侧只 INSERT，本文件不提供任何 UPDATE/DELETE 语句。
+-- ---------------------------------------------------------------- A. 建表(分区)
 CREATE TABLE IF NOT EXISTS `pbl_runtime_event` (
-  `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
-  `tenant_id` BIGINT NOT NULL DEFAULT 0 COMMENT '租户ID（多租户强制打头）',
-  `event_code` VARCHAR(64) NOT NULL COMMENT '事件编码',
-  `idem_key` VARCHAR(128) NOT NULL DEFAULT '' COMMENT '幂等键（同键重复投递只落一条）',
-  `world_id` BIGINT NOT NULL DEFAULT 0 COMMENT '世界ID（引用 world 基表，只读）',
-  `session_id` BIGINT NOT NULL DEFAULT 0 COMMENT '游戏会话ID（引用 scense 基表，只读）',
-  `entity_id` BIGINT NOT NULL DEFAULT 0 COMMENT '实体ID（引用 entity 基表，只读）',
-  `event_type` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '事件类型',
-  `payload` TEXT COMMENT '事件负载 JSON',
-  `seq_no` BIGINT NOT NULL DEFAULT 0 COMMENT '会话内单调序号（轮询游标）',
-  `source` VARCHAR(32) NOT NULL DEFAULT 'runtime' COMMENT '来源：runtime/agent/script/client_intent',
+  `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '自增ID(逻辑主键;物理主键=(id,created_at))',
+  `tenant_id` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '租户ID(多租户强制打头)',
+  `event_uid` VARCHAR(64) NOT NULL COMMENT '事件UID(uuid,应用层生成)',
+  `event_code` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '事件编码',
+  `idem_key` VARCHAR(128) NOT NULL DEFAULT '' COMMENT '幂等键(同键重复投递只落一条)',
+  `world_id` BIGINT NOT NULL DEFAULT 0 COMMENT '世界ID(引用 world 基表,只读)',
+  `scene_id` BIGINT NOT NULL DEFAULT 0 COMMENT '场景/会话ID(引用 scense 基表,只读)',
+  `session_id` BIGINT NOT NULL DEFAULT 0 COMMENT '游戏会话ID(M11a 语义,与 scene_id 同域)',
+  `entity_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '实体ID(引用 entity 基表,只读)',
+  `actor_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '触发者(服务端会话解析,客户端禁填)',
+  `event_type` VARCHAR(64) NOT NULL COMMENT '事件类型',
+  `payload` TEXT NULL COMMENT '事件负载JSON(任务书字段)',
+  `payload_json` LONGTEXT NULL COMMENT '事件负载JSON(M11a 写路径列名,与 payload 同义留宽)',
+  `causation_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '因果链:引发本事件的事件UID',
+  `seq` BIGINT NOT NULL DEFAULT 0 COMMENT '会话内单调序号(轮询游标,应用层生成)',
+  `seq_no` BIGINT NOT NULL DEFAULT 0 COMMENT '会话内单调序号(M11a 写路径列名)',
+  `source` VARCHAR(32) NOT NULL DEFAULT 'runtime' COMMENT '来源:runtime/agent/script/client_intent',
   `state` VARCHAR(16) NOT NULL DEFAULT 'applied' COMMENT 'applied/rejected/rolled_back',
-  `tx_group` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '事务组（同组同事务）',
-  `broadcast` TINYINT NOT NULL DEFAULT 1 COMMENT '是否参与广播 0/1',
   `state_version` BIGINT NOT NULL DEFAULT 0 COMMENT '本事件推进到的服务端权威版本',
+  `tx_group` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '事务组(同组同事务)',
+  `broadcast` TINYINT NOT NULL DEFAULT 1 COMMENT '是否参与广播 0/1',
   `created_by` BIGINT NOT NULL DEFAULT 0 COMMENT '创建人',
-  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间（分区键）',
+  `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间(分区键,禁TIMESTAMP)',
+  `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '记录时间戳(append-only:永不 UPDATE)',
+  `occurred_at` DATETIME NULL COMMENT '兼容列(M11a 旧分区键,新写入可空)',
   PRIMARY KEY (`id`, `created_at`),
+  UNIQUE KEY `uk_event_uid` (`tenant_id`, `event_uid`, `created_at`),
   UNIQUE KEY `uk_tenant_idem` (`tenant_id`, `idem_key`, `created_at`),
-  KEY `ix_tenant_session_seq` (`tenant_id`, `session_id`, `seq_no`),
-  KEY `ix_tenant_world` (`tenant_id`, `world_id`),
-  KEY `ix_tenant_type` (`tenant_id`, `event_type`)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
-  COMMENT='运行时事件流（append-only，按月 RANGE 分区，保留 14 个月）'
-PARTITION BY RANGE (TO_DAYS(`created_at`)) (
-  PARTITION `pbl_runtime_event_202609` VALUES LESS THAN (TO_DAYS('2026-10-01')),
-  PARTITION `pbl_runtime_event_202610` VALUES LESS THAN (TO_DAYS('2026-11-01')),
-  PARTITION `pbl_runtime_event_202611` VALUES LESS THAN (TO_DAYS('2026-12-01')),
-  PARTITION `pbl_runtime_event_202612` VALUES LESS THAN (TO_DAYS('2027-01-01')),
-  PARTITION `pbl_runtime_event_pmax` VALUES LESS THAN MAXVALUE
+  KEY `ix_scene_created` (`scene_id`, `created_at`),
+  KEY `ix_tenant_session_seq` (`tenant_id`, `session_id`, `seq`),
+  KEY `ix_tenant_session_seqno` (`tenant_id`, `session_id`, `seq_no`),
+  KEY `ix_tenant_world_type` (`tenant_id`, `world_id`, `event_type`),
+  KEY `ix_tx_group` (`tx_group`),
+  KEY `ix_created_at` (`created_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci
+  COMMENT='运行时事件流(append-only,按月RANGE分区,保留14个月)' 
+PARTITION BY RANGE COLUMNS(`created_at`) (
+  PARTITION `p202609` VALUES LESS THAN (DATE '2026-10-01'),
+  PARTITION `p202610` VALUES LESS THAN (DATE '2026-11-01'),
+  PARTITION `p202611` VALUES LESS THAN (DATE '2026-12-01'),
+  PARTITION `p202612` VALUES LESS THAN (DATE '2027-01-01'),
+  PARTITION `p202701` VALUES LESS THAN (DATE '2027-02-01'),
+  PARTITION `p202702` VALUES LESS THAN (DATE '2027-03-01'),
+  PARTITION `pmax` VALUES LESS THAN (MAXVALUE)
 );
 
--- ---------------------------------------------------------------- B. 一次性改造（M11a 非分区表 → 分区表）
--- 顺序不可调换：先扩主键/唯一键，再 PARTITION BY。执行前请备份。
--- ALTER TABLE `pbl_runtime_event` DROP PRIMARY KEY, ADD PRIMARY KEY (`id`, `created_at`);
--- ALTER TABLE `pbl_runtime_event` DROP INDEX `uk_tenant_idem`,
---   ADD UNIQUE KEY `uk_tenant_idem` (`tenant_id`, `idem_key`, `created_at`);
--- ALTER TABLE `pbl_runtime_event` PARTITION BY RANGE (TO_DAYS(`created_at`)) (
---   PARTITION `pbl_runtime_event_202609` VALUES LESS THAN (TO_DAYS('2026-10-01')),
---   PARTITION `pbl_runtime_event_pmax` VALUES LESS THAN MAXVALUE
--- );
+DELIMITER $$
 
--- ---------------------------------------------------------------- C. 月度维护（每月 1 日定时任务）
--- 模块写入前会自动预建（ensure_forward_partitions，进程内 TTL 6h），下面是等价手工语句：
--- ALTER TABLE `pbl_runtime_event` ADD PARTITION (PARTITION `pbl_runtime_event_202701` VALUES LESS THAN (TO_DAYS('2027-02-01')));
--- 查询当前分区：
---   SELECT PARTITION_NAME, PARTITION_DESCRIPTION FROM information_schema.PARTITIONS
---    WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'pbl_runtime_event';
--- 过期清理（保留 14 个月；DROP PARTITION 是元数据操作，不产生 DELETE 洪峰）：
---   由 partitions.drop_expired_partitions(retain=14, dry_run=True) 先出清单，
---   人工确认后再 dry_run=False 执行，等价：
--- ALTER TABLE `pbl_runtime_event` DROP PARTITION `pbl_runtime_event_202507`;
+-- ---------------------------------------------------------------- B. append-only
+-- append-only 断言过程：非白名单操作一律 SIGNAL 拒绝
+DROP PROCEDURE IF EXISTS `pbl_assert_append_only`;
 
--- ---------------------------------------------------------------- D. 实体状态 / 快照（M11a 定义不变）
--- pbl_entity_state：服务端权威，state_version 由 apply_runtime_event 单调分配，
---   客户端提交值一律忽略；uk(tenant_id, session_id, entity_id, state_key) 支撑 UPSERT。
--- pbl_world_state_snapshot：掉线补齐/离线兜底基线，与事件同事务写入。
--- 见 sql/pbl_runtime_ext.sql（单一真源生成），本文件不重复定义，避免两处漂移。
+CREATE$$
+
+CREATE PROCEDURE `pbl_assert_append_only`(
+    IN p_table VARCHAR(64), IN p_priv VARCHAR(16), IN p_detail VARCHAR(255))
+    READS SQL DATA
+BEGIN
+    IF p_table <> 'pbl_runtime_event' THEN
+        RETURN;                       -- 只守护事件表，其他表不受影响
+    END IF;
+    IF IFNULL(@pbl_guard_ctx, '') = 'append_only_admin' THEN
+        RETURN;                       -- break-glass：一次性历史数据回填
+    END IF;
+    SIGNAL SQLSTATE '45000'
+        SET MESSAGE_TEXT = 'PBL_APPEND_ONLY_DENIED',
+            MYSQL_ERRNO = 45001;
+END;
+
+-- BEFORE UPDATE 护栏：把违规语句原文带进错误消息，便于定位调用方
+DROP TRIGGER IF EXISTS `trg_pbl_runtime_event_append_only_upd`$$
+
+CREATE$$
+
+CREATE TRIGGER `trg_pbl_runtime_event_append_only_upd`
+    BEFORE UPDATE ON `pbl_runtime_event` FOR EACH ROW
+BEGIN
+    DECLARE v_sql TEXT;
+    DECLARE v_msg VARCHAR(255);
+    SET v_sql = '';
+    BEGIN
+        DECLARE CONTINUE HANDLER FOR NOT FOUND SET v_sql = '';
+        SELECT INFO INTO v_sql FROM information_schema.processlist
+         WHERE ID = CONNECTION_ID();
+    END;
+    IF IFNULL(@pbl_guard_ctx, '') = 'append_only_admin' THEN
+        SET v_msg = CONCAT('break-glass update on ', 'pbl_runtime_event');
+    ELSE
+        SET v_msg = CONCAT('UPDATE denied: ', LEFT(IFNULL(v_sql, 'n/a'), 120));
+    END IF;
+    CALL pbl_assert_append_only('pbl_runtime_event', 'update', v_msg);
+    IF IFNULL(@pbl_guard_ctx, '') <> 'append_only_admin' THEN
+        SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'PBL_APPEND_ONLY_DENIED_UPDATE',
+            MYSQL_ERRNO = 45002;
+    END IF;
+END;
+
+-- BEFORE DELETE 护栏
+DROP TRIGGER IF EXISTS `trg_pbl_runtime_event_append_only_del`$$
+
+CREATE$$
+
+CREATE TRIGGER `trg_pbl_runtime_event_append_only_del`
+    BEFORE DELETE ON `pbl_runtime_event` FOR EACH ROW
+BEGIN
+    DECLARE v_sql TEXT;
+    DECLARE v_msg VARCHAR(255);
+    SET v_sql = '';
+    BEGIN
+        DECLARE CONTINUE HANDLER FOR NOT FOUND SET v_sql = '';
+        SELECT INFO INTO v_sql FROM information_schema.processlist
+         WHERE ID = CONNECTION_ID();
+    END;
+    IF IFNULL(@pbl_guard_ctx, '') = 'append_only_admin' THEN
+        SET v_msg = CONCAT('break-glass delete on ', 'pbl_runtime_event');
+    ELSE
+        SET v_msg = CONCAT('DELETE denied: ', LEFT(IFNULL(v_sql, 'n/a'), 120));
+    END IF;
+    CALL pbl_assert_append_only('pbl_runtime_event', 'delete', v_msg);
+    IF IFNULL(@pbl_guard_ctx, '') <> 'append_only_admin' THEN
+        SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'PBL_APPEND_ONLY_DENIED_DELETE',
+            MYSQL_ERRNO = 45003;
+    END IF;
+END;
+
+-- 权限层加固（可选，需 DBA 账号执行；app 账号只保留增/查）
+-- GRANT INSERT, SELECT, UPDATE, DELETE ON `pbls`.`pbl_runtime_event` TO NULL;  -- 占位，勿执行
+-- REVOKE UPDATE, DELETE ON `pbls`.`pbl_runtime_event` FROM 'pbls'@'%'$$
+
+-- ---------------------------------------------------------------- C. 分区维护
+-- 1) 幂等补建未来月分区：pmax 存在时用 REORGANIZE PARTITION 拆出新月份（不丢数据）
+DROP PROCEDURE IF EXISTS `pbl_runtime_event_ensure_partitions`;
+
+CREATE$$
+
+CREATE PROCEDURE `pbl_runtime_event_ensure_partitions`(IN p_months_ahead INT)
+    MODIFIES SQL DATA
+BEGIN
+    DECLARE v_done INT DEFAULT 0;
+    DECLARE v_name VARCHAR(16);
+    DECLARE v_upper DATE;
+    DECLARE v_exists INT DEFAULT 0;
+    DECLARE v_sql TEXT;
+    DECLARE cur CURSOR FOR
+        WITH RECURSIVE ms(month_start) AS (
+            SELECT DATE_FORMAT(CURDATE(), '%Y-%m-01')
+            UNION ALL
+            SELECT DATE_ADD(month_start, INTERVAL 1 MONTH) FROM ms
+             WHERE month_start < DATE_ADD(DATE_FORMAT(CURDATE(), '%Y-%m-01'),
+                                          INTERVAL p_months_ahead MONTH)
+        )
+        SELECT CONCAT('p', DATE_FORMAT(month_start, '%Y%m')),
+               DATE_ADD(month_start, INTERVAL 1 MONTH)
+          FROM ms;
+    DECLARE CONTINUE HANDLER FOR NOT FOUND SET v_done = 1;
+    OPEN cur;
+    read_loop: LOOP
+        FETCH cur INTO v_name, v_upper;
+        IF v_done = 1 THEN
+            LEAVE read_loop;
+        END IF;
+        SELECT COUNT(*) INTO v_exists FROM information_schema.PARTITIONS
+         WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'pbl_runtime_event'
+           AND PARTITION_NAME = v_name;
+        IF v_exists = 0 THEN
+            SET @pbl_part_sql = CONCAT(
+                'ALTER TABLE `pbl_runtime_event` REORGANIZE PARTITION `pmax` INTO (',
+                'PARTITION `', v_name, '` VALUES LESS THAN (DATE ''', v_upper, '''), ',
+                'PARTITION `pmax` VALUES LESS THAN (MAXVALUE))');
+            PREPARE pbl_part_st FROM @pbl_part_sql;
+            EXECUTE pbl_part_st;
+            DEALLOCATE PREPARE pbl_part_st;
+        END IF;
+    END LOOP;
+    CLOSE cur;
+END;
+
+-- 2) 到期归档：整月 DROP PARTITION（append-only 表唯一合规的清理手段）
+DROP PROCEDURE IF EXISTS `pbl_runtime_event_prune_partitions`$$
+
+CREATE$$
+
+CREATE PROCEDURE `pbl_runtime_event_prune_partitions`(IN p_keep_months INT)
+    MODIFIES SQL DATA
+BEGIN
+    DECLARE v_done INT DEFAULT 0;
+    DECLARE v_name VARCHAR(64);
+    DECLARE v_cutoff DATE;
+    DECLARE cur CURSOR FOR
+        SELECT PARTITION_NAME FROM information_schema.PARTITIONS
+         WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'pbl_runtime_event'
+           AND PARTITION_NAME IS NOT NULL
+           AND PARTITION_NAME <> 'pmax'
+           AND PARTITION_NAME < CONCAT('p', DATE_FORMAT(v_cutoff, '%Y%m'));
+    DECLARE CONTINUE HANDLER FOR NOT FOUND SET v_done = 1;
+    SET v_cutoff = DATE_ADD(DATE_FORMAT(CURDATE(), '%Y-%m-01'), INTERVAL -p_keep_months MONTH);
+    OPEN cur;
+    prune_loop: LOOP
+        FETCH cur INTO v_name;
+        IF v_done = 1 THEN
+            LEAVE prune_loop;
+        END IF;
+        SET @pbl_drop_sql = CONCAT('ALTER TABLE `pbl_runtime_event` DROP PARTITION `', v_name, '`');
+        PREPARE pbl_drop_st FROM @pbl_drop_sql;
+        EXECUTE pbl_drop_st;
+        DEALLOCATE PREPARE pbl_drop_st;
+    END LOOP;
+    CLOSE cur;
+END;
+
+-- 3) 每月自动补分区（需 event_scheduler=ON；否则用 cron，见开发说明）
+-- SET GLOBAL event_scheduler = ON$$
+
+DROP$$
+
+DROP EVENT IF EXISTS `ev_pbl_runtime_event_partition_maintain`;
+
+CREATE$$
+
+CREATE EVENT IF NOT EXISTS `ev_pbl_runtime_event_partition_maintain`
+    ON SCHEDULE EVERY 1 MONTH
+    STARTS TIMESTAMP(DATE_FORMAT(DATE_ADD(CURDATE(), INTERVAL 1 MONTH), '%Y-%m-01 03:00:00'))
+    ON COMPLETION PRESERVE
+    ENABLE
+    COMMENT '每月预建未来 6 个月分区'
+DO CALL pbl_runtime_event_ensure_partitions(6)$$
+
+DELIMITER ;
+
+-- ---------------------------------------------------------------- D. 一次性改造
+-- 一次性改造（仅当 pbl_runtime_event 已存在且未分区时手工执行；先备份）：
+--   1) CREATE TABLE pbl_runtime_event_new ... （本文件 A 段 DDL，改名 _new）
+--   2) SET @pbl_guard_ctx = 'append_only_admin';   -- break-glass，仅回填期间
+--      INSERT INTO pbl_runtime_event_new (...) SELECT ... FROM pbl_runtime_event;
+--      SET @pbl_guard_ctx = NULL;
+--   3) RENAME TABLE pbl_runtime_event TO pbl_runtime_event_m11a_bak,
+--                   pbl_runtime_event_new TO pbl_runtime_event;
+--   4) 重放 append-only 触发器段（B 段），核对 EXPLAIN 分区裁剪后删备份表。
+-- 本脚本不自动执行改造：DROP/RENAME 生产表必须人工确认（见开发说明「回滚」段）。
+SELECT 'm11b-1 migrate: 人工执行，见注释' AS note;
diff --git a/tests/run_m11b1a_localdb.sh b/tests/run_m11b1a_localdb.sh
new file mode 100644
index 0000000..19d1a38
--- /dev/null
+++ b/tests/run_m11b1a_localdb.sh
@@ -0,0 +1,107 @@
+#!/usr/bin/env bash
+# m11b-1a 实库验证：拉起一次性本地 MariaDB（独立 datadir + 独立端口），
+# 用**部署同款路径**（mariadb CLI + DELIMITER）装载生成器产出的 DDL，
+# 再跑 tests/test_m11b1a_ddl_runtime.py 的 A+B 两层，结束后彻底销毁。
+#
+# 为什么自造库而不是连 projects/pbls/env/test.json：
+#   verify_db()/prune_partitions 会真做 DROP PARTITION 与 INSERT/DELETE 探针，
+#   在共享测试库上跑等于拿别人的数据做破坏性实验；且本沙箱内 3306 端口的
+#   MariaDB 对 pbls 账号是 unix_socket 鉴权（1698 Access denied），根本连不上。
+#   连接参数（user/dbname/password）仍从 env/test.json 读取，不硬编码凭据。
+#
+# 装载方式说明：生成器产出的 B/C 段含多语句过程体，按文件头注释规定必须用
+#   mysql/mariadb CLI（DELIMITER）执行；故本 runner 用 CLI 装载，
+#   顺带证明 --emit/full_script() 产出的 SQL 在 MariaDB 10.6 上可原样执行。
+#
+# 用法：bash tests/run_m11b1a_localdb.sh   （退出码 0 = 全 PASS）
+set -u
+HERE="$(cd "$(dirname "$0")" && pwd)"
+MODULE_DIR="$(dirname "$HERE")"
+D="${TMPDIR:-/tmp}/m11b1a_localdb_$$"
+PORT="${M11B1A_DB_PORT:-13306}"
+SOCK="$D/my.sock"
+LOG="$D/server.err"
+
+cleanup() {
+  if [[ -f "$D/my.pid" ]]; then
+    kill "$(cat "$D/my.pid")" 2>/dev/null || true
+    sleep 1
+  fi
+  pkill -f "datadir=$D" 2>/dev/null || true
+  rm -rf "$D"
+}
+trap cleanup EXIT INT TERM
+
+echo "[runner] datadir=$D port=$PORT"
+mkdir -p "$D"
+
+mariadb-install-db --user="$(id -un)" --datadir="$D" \
+  --auth-root-authentication-method=normal > "$D/install.log" 2>&1 || {
+  echo "[runner] FAIL: mariadb-install-db 失败"; tail -20 "$D/install.log"; exit 1; }
+
+mariadbd --user="$(id -un)" --datadir="$D" --socket="$SOCK" \
+  --port="$PORT" --bind-address=127.0.0.1 --pid-file="$D/my.pid" \
+  --skip-name-resolve --log-error="$LOG" > /dev/null 2>&1 &
+
+for i in $(seq 1 40); do
+  if mariadb --protocol=tcp -h 127.0.0.1 -P "$PORT" -u root -e "SELECT 1" >/dev/null 2>&1; then
+    echo "[runner] mariadbd ready (${i}s)"; break
+  fi
+  sleep 1
+done
+
+if ! mariadb --protocol=tcp -h 127.0.0.1 -P "$PORT" -u root -e "SELECT VERSION()" >/dev/null 2>&1; then
+  echo "[runner] FAIL: mariadbd 未在 40s 内就绪"; tail -20 "$LOG" 2>/dev/null; exit 1
+fi
+
+# 按 env/test.json 建库建号（凭据只在进程内使用，不 echo、不落盘、不 commit）
+python3 - "$MODULE_DIR" "$PORT" <<'PY' || exit 1
+import json, os, subprocess, sys
+module_dir, port = sys.argv[1], sys.argv[2]
+ws = os.path.dirname(os.path.dirname(module_dir))
+db = json.load(open(os.path.join(ws, "projects", "pbls", "env", "test.json"), encoding="utf-8"))["db"]
+name, user, pw = db["dbname"], db["user"], db["password"]
+sql = ("CREATE DATABASE IF NOT EXISTS `%s` CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;\n"
+       "CREATE USER IF NOT EXISTS '%s'@'%%' IDENTIFIED BY '%s';\n"
+       "GRANT ALL PRIVILEGES ON *.* TO '%s'@'%%' WITH GRANT OPTION;\n"
+       "FLUSH PRIVILEGES;\n"
+       % (name, user, pw.replace("'", "''"), user))
+r = subprocess.run(["mariadb", "--protocol=tcp", "-h", "127.0.0.1", "-P", port, "-u", "root"],
+                   input=sql, capture_output=True, text=True)
+if r.returncode:
+    print("[runner] FAIL: 建库建号失败(错误信息已脱敏): %s" % r.stderr.replace(pw, "***")[:300])
+    sys.exit(1)
+print("[runner] db=%s user=%s 就绪（一次性库，与项目 test/prod 无关）" % (name, user))
+PY
+
+# 生成器产出的完整脚本（A 建表 + B append-only + C 分区维护 + D 一次性改造）落到临时目录，
+# 用 mariadb CLI（支持 DELIMITER）装载 —— 与部署执行路径一致。
+python3 - "$MODULE_DIR" "$D/full.sql" <<'PY' || exit 1
+import importlib.util, os, sys
+module_dir, out = sys.argv[1], sys.argv[2]
+script = os.path.join(module_dir, "scripts", "pbl_runtime_event_ddl.py")
+spec = importlib.util.spec_from_file_location("pbl_runtime_event_ddl", script)
+mod = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(mod)
+open(out, "w", encoding="utf-8").write(mod.full_script())
+print("[runner] full_script() 渲染 %d bytes -> %s" % (os.path.getsize(out), out))
+PY
+
+echo "[runner] 装载 DDL（mariadb CLI + DELIMITER）"
+mariadb --protocol=tcp -h 127.0.0.1 -P "$PORT" -u root pbls < "$D/full.sql" 2> "$D/load.err"
+LOAD_RC=$?
+if [[ "$LOAD_RC" != "0" ]]; then
+  echo "[runner] FAIL: DDL 装载 rc=$LOAD_RC"
+  sed -e "s/'[^']*'@/'***'@/g" "$D/load.err" | head -20
+else
+  echo "[runner] DDL 装载 rc=0（stderr 见下，应为空）"
+  head -5 "$D/load.err"
+fi
+
+export M11B1A_LOCAL_DB=1
+export M11B1A_DB_HOST=127.0.0.1
+export M11B1A_DB_PORT="$PORT"
+python3 "$HERE/test_m11b1a_ddl_runtime.py"
+rc=$?
+echo "[runner] harness rc=$rc"
+exit $rc
diff --git a/tests/test_m11b1a_ddl_runtime.py b/tests/test_m11b1a_ddl_runtime.py
new file mode 100644
index 0000000..6c68f62
--- /dev/null
+++ b/tests/test_m11b1a_ddl_runtime.py
@@ -0,0 +1,247 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""m11b-1a 运行时缺陷验证 harness（响应 QC #1 % 转义 / QC #6 verify_db 判定 / QC #5 落点）。
+
+被验对象：scripts/pbl_runtime_event_ddl.py（唯一被改文件）。
+
+两层：
+  A. 离线层（无依赖，必跑）
+     A1 maintenance_sql() 渲染不再抛 ValueError（QC #1 崩溃点）
+     A2 渲染正文里 DATE_FORMAT 参数为**单 %** 形式（'%Y-%m-01' / '%Y%m'），且无 '%%Y' 残留
+     A3 append_only_sql()/full_script() 既有转义语义未改坏：'%(dbname)s'/'%(app_user)s'
+        被 replace 掉、host 通配符输出为单 % （'pbls'@'%'），无 '@'%%'' 残留
+     A4 DDL 未被改动：create_table_sql() 的 sha256 == 修复前基线（验收标准 4）
+     A5 源码无 `EXPLAIN PARTITIONS ` 废弃语法残留、无「查询结果未使用」的空转 for 循环
+     A6 verify_db 判定改用 information_schema.PARTITIONS(按 ORDINAL 排序) +
+        EXPLAIN FORMAT=JSON + PARTITION(p) 反查（静态断言三要素齐备）
+     A7 CLI 冒烟：--verify / --print / --print-maintain / --emit 全部 rc=0
+  B. 实库层（仅当 M11B1A_LOCAL_DB=1，由 tests/run_m11b1a_localdb.sh 拉起一次性
+     MariaDB 后运行；**不会**连项目 test/prod 库，避免 DROP PARTITION 误伤真数据）
+     B1 用 mariadb CLI(DELIMITER) 装载 full_script()，核对表/分区/过程/触发器齐备
+     B2 --verify-db 全项 PASS（分区数/兜底/跨月落不同分区/分区裁剪/UPDATE·DELETE 被拒/探针清理）
+     B3 CALL pbl_runtime_event_ensure_partitions(9) 真能 REORGANIZE 出新月分区（A2 的单 % SQL 在
+        服务端可执行，不只是渲染正确）
+     B4 CALL pbl_runtime_event_prune_partitions(2) 真能 DROP 旧月分区且保留 pmax
+
+用法：
+  python3 tests/test_m11b1a_ddl_runtime.py                 # 只跑 A 层
+  bash tests/run_m11b1a_localdb.sh                          # A + B 层（自带一次性库）
+退出码：0 = 无 FAIL；1 = 有 FAIL；2 = B 层被跳过且 A 层通过时仍返回 0（B 缺失会显式打印 SKIP 行）。
+"""
+
+import hashlib
+import importlib.util
+import os
+import re
+import subprocess
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+MODULE_DIR = os.path.dirname(HERE)                                    # modules/pbl_runtime_ext
+SCRIPT = os.path.join(MODULE_DIR, "scripts", "pbl_runtime_event_ddl.py")
+WORKSPACE = os.path.dirname(os.path.dirname(MODULE_DIR))              # 机构工作空间根
+ENV_JSON = os.path.join(WORKSPACE, "projects", "pbls", "env", "test.json")
+
+# 修复前（git 049e09a）`--print` 输出的 sha256 —— DDL 未改动的机械证据（验收标准 4）
+DDL_BASELINE_SHA256 = "30a83c8a9d729db987f82de1a5bc30229dc4a7b1cc9881c7a5a45673b1a62619"
+
+RESULTS = []
+
+
+def chk(name, passed, detail=""):
+    RESULTS.append((name, bool(passed), detail))
+    print("  %-4s %-46s %s" % ("PASS" if passed else "FAIL", name, detail))
+    return bool(passed)
+
+
+def load_gen():
+    spec = importlib.util.spec_from_file_location("pbl_runtime_event_ddl", SCRIPT)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+# ------------------------------------------------------------------ A. 离线层
+def layer_a():
+    print("[A] 离线层（% 转义 + 判定逻辑静态核验）")
+    src = open(SCRIPT, encoding="utf-8").read()
+
+    # A1/A2 —— QC #1 崩溃点
+    try:
+        maint = subprocess.run([sys.executable, SCRIPT, "--print-maintain"],
+                               capture_output=True, text=True)
+        rc_maint = maint.returncode
+        body = maint.stdout
+    except Exception as exc:  # noqa: BLE001
+        rc_maint, body = 99, ""
+        chk("A1 maintenance_sql 渲染不抛异常", False, str(exc))
+    chk("A1 maintenance_sql 渲染不抛异常(QC#1)", rc_maint == 0,
+        "rc=%s, %d bytes（修复前 rc=1 ValueError: unsupported format character 'Y'）"
+        % (rc_maint, len(body)))
+    for token in ("pbl_runtime_event_ensure_partitions", "pbl_runtime_event_prune_partitions",
+                  "ev_pbl_runtime_event_partition_maintain"):
+        chk("A2 维护段含完整对象 %s" % token, token in body, "%d 处" % body.count(token))
+    chk("A2 DATE_FORMAT 参数为单 %% 形式('%Y-%m-01')",
+        "DATE_FORMAT(CURDATE(), '%Y-%m-01')" in body and "DATE_FORMAT(month_start, '%Y%m')" in body,
+        "命中 %d 处单 %% 形式" % (body.count("'%Y-%m-01'") + body.count("'%Y%m'")))
+    chk("A2 无 '%%Y' 双写转义残留", "%%Y" not in body and "%%'" not in body,
+        "grep -c '%%Y' = %d" % body.count("%%Y"))
+
+    # A3 —— append_only 既有转义语义
+    full = subprocess.run([sys.executable, SCRIPT, "--print-maintain"], capture_output=True, text=True)
+    assert full.returncode == 0
+    script_txt = subprocess.run([sys.executable, "-c",
+                                 "import sys;sys.path.insert(0,%r);"
+                                 "import importlib.util as u;"
+                                 "s=u.spec_from_file_location('m',%r);m=u.module_from_spec(s);s.loader.exec_module(m);"
+                                 "print(m.full_script(),end='')" % (MODULE_DIR, SCRIPT)],
+                                capture_output=True, text=True)
+    chk("A3 full_script() rc=0", script_txt.returncode == 0, "rc=%s" % script_txt.returncode)
+    fs = script_txt.stdout
+    chk("A3 %(dbname)s/%(app_user)s 已被替换", "%(dbname)s" not in fs and "%(app_user)s" not in fs,
+        "残留 %d 处" % (fs.count("%(dbname)s") + fs.count("%(app_user)s")))
+    chk("A3 REVOKE 主机通配符为单 %%（'pbls'@'%'）", "FROM 'pbls'@'%'" in fs and "@'%%'" not in fs,
+        "命中 'pbls'@'%%' = %d，'@%%' = %d" % (fs.count("FROM 'pbls'@'%'"), fs.count("@'%%'")))
+    chk("A3 break-glass ctx 原样输出", "append_only_admin" in fs, "%d 处" % fs.count("append_only_admin"))
+
+    # A4 —— DDL 逐字未动
+    gen = load_gen()
+    digest = hashlib.sha256((gen.create_table_sql() + "\n").encode("utf-8")).hexdigest()
+    chk("A4 create_table_sql sha256 == 修复前基线", digest == DDL_BASELINE_SHA256, digest[:16] + "…")
+
+    # A5/A6 —— 判定逻辑（QC #6）
+    chk("A5 源码无 'EXPLAIN PARTITIONS ' 废弃语法", "EXPLAIN PARTITIONS " not in src,
+        "grep 命中 %d" % src.count("EXPLAIN PARTITIONS "))
+    chk("A5 无未使用查询结果的空转 for 循环",
+        not re.search(r"for\s+k\s+in\s*\(", src) and "row[3]" not in src,
+        "row[3] 命中 %d，for k in 命中 %d" % (src.count("row[3]"), len(re.findall(r"for\s+k\s+in\s*\(", src))))
+    vsrc = src[src.index("def verify_db("):]
+    chk("A6 分区清单来自 information_schema.PARTITIONS+ORDINAL",
+        "information_schema.PARTITIONS" in vsrc and "PARTITION_ORDINAL_POSITION" in vsrc, "OK")
+    chk("A6 落分区判定用 PARTITION(p) 反查 + EXPLAIN FORMAT=JSON",
+        "_row_in_partition" in src and "PARTITION (`%s`)" in src
+        and "EXPLAIN FORMAT=JSON" in src and "_locate_row_partition" in vsrc, "OK")
+    chk("A6 探针清理 break-glass 逻辑保留",
+        "SET @pbl_guard_ctx=" in vsrc and "GUARD_ADMIN_CTX" in vsrc
+        and "SET @pbl_guard_ctx=NULL" in vsrc,
+        "guard_ctx 设置 %d 处 / 复位 %d 处"
+        % (vsrc.count("SET @pbl_guard_ctx="), vsrc.count("SET @pbl_guard_ctx=NULL")))
+
+    # A7 —— CLI 冒烟
+    for args in (["--verify"], ["--print"], ["--print-guard"], ["--print-maintain"]):
+        r = subprocess.run([sys.executable, SCRIPT] + args, capture_output=True, text=True)
+        tail = (r.stdout.strip().splitlines() or [""])[-1][:40]
+        chk("A7 CLI %s rc=0" % " ".join(args), r.returncode == 0, "末行: %s" % tail)
+    tmp_emit = os.path.join(HERE, "_emit_check")
+    r = subprocess.run([sys.executable, SCRIPT, "--emit", tmp_emit, "--dbname", "pbls"],
+                       capture_output=True, text=True)
+    files = sorted(os.listdir(tmp_emit)) if os.path.isdir(tmp_emit) else []
+    chk("A7 CLI --emit rc=0", r.returncode == 0 and len(files) >= 3,
+        "%d 文件: %s" % (len(files), ",".join(files)))
+    # 生成的维护 SQL 同样必须是单 % 形式（emit 路径回归）
+    m3 = os.path.join(tmp_emit, "20260919_m11b_03_pbl_runtime_event_partition_maintenance.sql")
+    if os.path.exists(m3):
+        t = open(m3, encoding="utf-8").read()
+        chk("A7 --emit 维护段 DATE_FORMAT 单 %", "DATE_FORMAT(CURDATE(), '%Y-%m-01')" in t and "%%Y" not in t,
+            "%d bytes" % len(t))
+    subprocess.run(["rm", "-rf", tmp_emit])
+    return gen
+
+
+# ------------------------------------------------------------------ B. 实库层
+def layer_b(gen):
+    print("\n[B] 实库层（一次性本地 MariaDB，不触碰项目 test/prod 库）")
+    if os.environ.get("M11B1A_LOCAL_DB") != "1":
+        print("  SKIP B 层：未设 M11B1A_LOCAL_DB=1（用 bash tests/run_m11b1a_localdb.sh 跑全量）")
+        return True
+    import json
+    import pymysql
+    conf = json.load(open(ENV_JSON, encoding="utf-8"))["db"]
+    db = dict(conf)
+    db["host"] = os.environ.get("M11B1A_DB_HOST", "127.0.0.1")
+    db["port"] = int(os.environ.get("M11B1A_DB_PORT", "13306"))
+    os.environ["PBL_DDL_DB_HOST"] = db["host"]
+    os.environ["PBL_DDL_DB_PORT"] = str(db["port"])
+
+    conn0 = pymysql.connect(host=db["host"], port=db["port"], user=db["user"],
+                            password=db["password"], database=db["dbname"],
+                            charset="utf8mb4", autocommit=True)
+    cur0 = conn0.cursor()
+    cur0.execute("SELECT COUNT(*) FROM information_schema.PARTITIONS "
+                 "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=%s "
+                 "AND PARTITION_NAME IS NOT NULL", (gen.TABLE,))
+    n0 = cur0.fetchone()[0]
+    cur0.execute("SELECT COUNT(*) FROM information_schema.ROUTINES "
+                 "WHERE ROUTINE_SCHEMA=DATABASE() AND ROUTINE_NAME LIKE 'pbl_runtime_event%%'")
+    n_proc = cur0.fetchone()[0]
+    cur0.execute("SELECT COUNT(*) FROM information_schema.TRIGGERS "
+                 "WHERE EVENT_OBJECT_TABLE=%s", (gen.TABLE,))
+    n_trg = cur0.fetchone()[0]
+    cur0.close(); conn0.close()
+    chk("B1 CLI(mariadb+DELIMITER) 装载 full_script() 后对象齐备",
+        n0 >= gen.MONTHS_PRECREATE + 1 and n_proc >= 2 and n_trg >= 2,
+        "分区 %d 个 / 维护过程 %d 个 / 触发器 %d 个" % (n0, n_proc, n_trg))
+    rc2 = gen.verify_db()
+    chk("B2 verify_db() rc=0 全项 PASS(QC#6)", rc2 == 0, "verify-db rc=%s" % rc2)
+
+    conn = pymysql.connect(host=db["host"], port=db["port"], user=db["user"],
+                           password=db["password"], database=db["dbname"],
+                           charset="utf8mb4", autocommit=True)
+    cur = conn.cursor()
+
+    def parts():
+        cur.execute("SELECT PARTITION_NAME FROM information_schema.PARTITIONS "
+                    "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=%s "
+                    "AND PARTITION_NAME IS NOT NULL ORDER BY PARTITION_ORDINAL_POSITION", (gen.TABLE,))
+        return [r[0] for r in cur.fetchall()]
+
+    before = parts()
+    cur.execute("CALL pbl_runtime_event_ensure_partitions(9)")
+    after = parts()
+    new = [p for p in after if p not in before]
+    chk("B3 CALL ensure_partitions(9) 补建出新月分区", len(new) >= 3 and after.count("pmax") == 1,
+        "新增 %s（分区数 %d→%d）" % (",".join(new), len(before), len(after)))
+
+    cur.execute("CALL pbl_runtime_event_prune_partitions(2)")
+    pruned = parts()
+    dropped = [p for p in after if p not in pruned]
+    chk("B4 CALL prune_partitions(2) DROP 旧月分区且保留 pmax",
+        len(dropped) >= 1 and "pmax" in pruned,
+        "DROP %s（剩余 %d 个）" % (",".join(dropped), len(pruned)))
+
+    # 分区裁剪 + 落分区判定的服务端可执行性（EXPLAIN FORMAT=JSON 解析）
+    planned = gen._explain_partitions(cur, "SELECT id FROM `%s` WHERE created_at>='2026-01-01'"
+                                      " AND created_at<'2026-02-01'" % gen.TABLE)
+    chk("B5 _explain_partitions 解析出分区清单", bool(planned) and len(planned) < len(pruned),
+        "命中 %s（表内共 %d 分区）" % (",".join(planned), len(pruned)))
+    cur.execute("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)"
+                " VALUES ('t_harness',%%s,'probe',0,%%s,%%s)" % gen.TABLE,
+                ("h_a", 1, "2026-01-05 00:00:00"))
+    located = gen._locate_row_partition(cur, "h_a", "2026-01-05 00:00:00", pruned)
+    chk("B6 _locate_row_partition 定位到 information_schema 清单内的分区",
+        located in pruned, "h_a -> %s（清单 %s）" % (located, ",".join(pruned)))
+    cur.execute("SET @pbl_guard_ctx='%s'" % gen.GUARD_ADMIN_CTX)
+    cur.execute("DELETE FROM `%s` WHERE tenant_id='t_harness'" % gen.TABLE)
+    cur.execute("SET @pbl_guard_ctx=NULL")
+    cur.close(); conn.close()
+    return True
+
+
+def main():
+    print("harness: %s" % SCRIPT)
+    gen = layer_a()
+    ok_b = layer_b(gen)
+    fails = [n for n, p, _ in RESULTS if not p]
+    print("\n[汇总] 共 %d 项，PASS %d，FAIL %d%s"
+          % (len(RESULTS), len(RESULTS) - len(fails), len(fails),
+             "" if ok_b else "（B 层 SKIP）"))
+    if fails:
+        print("FAIL 明细: %s" % " | ".join(fails))
+        print("HARNESS FAIL")
+        return 1
+    print("HARNESS PASS")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())