approve: [M11b-2c-B] 补 dev-notes-m11b2c 文档 + RBAC/路径声明 + git 收口

scripts/m11b4_selftest.py

@@ -66,18 +66,44 @@ def make_db(path):
     conn.row_factory = sqlite3.Row
     for stmt in ps.ddl_for(ps.DIALECT_SQLITE):
         conn.execute(stmt)
+    # 索引由 models/pbl_runtime_event.json 渲染（与 MySQL 侧 ix_scene_created 同口径）：
+    # auth.ddl 只出 CREATE TABLE，测试库必须自己补 CREATE INDEX，否则 EXPLAIN 必然 SCAN，
+    # AC2 的"命中索引"就无从验证。
+    for idx in model_indexes("pbl_runtime_event"):
+        conn.execute(idx)
     conn.commit()
     return conn
 
 
+def model_indexes(table):
+    """从 models/{table}.json 生成 CREATE INDEX 语句（真源是模型 JSON，不手写列序）。"""
+    import json as _json
+    path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..",
+                        "models", table + ".json")
+    spec = _json.load(open(path, encoding="utf-8"))
+    out = []
+    for idx in spec.get("indexes") or []:
+        uniq = "UNIQUE " if idx.get("idxtype") == "unique" else ""
+        cols = ", ".join(idx["idxfields"])
+        out.append("CREATE %sINDEX IF NOT EXISTS %s ON %s (%s)"
+                   % (uniq, idx["name"], table, cols))
+    return out
+
+
 def seed_event(conn, scene, uid, created_at, event_type="m11b4.probe", seq=1):
-    cols = [c for c in ps.EVENT_COLUMNS]
+    """按 M11b-1 模型 JSON 的列契约灌一条事件（NOT NULL 列全部给值，不猜列名）。"""
+    cols = list(ps.EVENT_COLUMNS)
     row = {c: None for c in cols}
     row.update({
-        "id": uid, "event_uid": uid, "tenant_id": "t_m11b4", "world_id": "w_m11b4",
+        "id": uid, "event_uid": uid, "event_code": "m11b4.probe",
-        "scene_id": scene, "entity_id": "e_m11b4", "event_type": event_type,
+        "idem_key": "idem-" + uid, "tenant_id": "t_m11b4", "world_id": "w_m11b4",
-        "seq": seq, "seq_no": seq, "state_version": seq,
+        "scene_id": scene, "session_id": "ss_m11b4", "entity_id": "e_m11b4",
-        "created_at": created_at, "source": "m11b4_selftest",
+        "actor_id": "a_m11b4", "event_type": event_type, "causation_id": uid,
+        "seq": seq, "seq_no": seq, "state_version": seq, "state": "active",
+        "tx_group": "tx-" + uid, "broadcast": 1, "created_by": "m11b4_selftest",
+        "created_at": created_at, "updated_at": created_at,
+        "occurred_at": created_at, "source": "m11b4_selftest",
+        "payload": "{}", "payload_json": "{}",
     })
     marks = ", ".join(["?"] * len(cols))
     conn.execute("INSERT INTO %s (%s) VALUES (%s)"
@@ -161,8 +187,11 @@ def test_watermark_format_guard(dbpath):
           % normalize_watermark("2026-09-22 03:00:00"))
     fb = PollingFallback(factory_for(dbpath))
     ok_iso = fb.set_watermark("sc_guard", "2026-09-22T03:00:00Z")
+    wm_after_iso = fb.watermark("sc_guard")          # 必须在下一步之前取值
+    check("AC1c.1 ISO commit_ts_text 被拒绝",
+          ok_iso is False and wm_after_iso == "",
+          "ret=%r watermark=%r" % (ok_iso, wm_after_iso))
     ok_sp = fb.set_watermark("sc_guard", "2026-09-22 03:00:00")
-    check("AC1c.1 ISO commit_ts_text 被拒绝", ok_iso is False and fb.watermark("sc_guard") == "")
     check("AC1c.2 库内空格格式被接受", ok_sp is True)
     check("AC1c.3 水位线单调不减（旧值被忽略）",
           fb.set_watermark("sc_guard", "2026-09-22 02:00:00") is False)
@@ -207,9 +236,10 @@ def test_explain_uses_index(dbpath):
     for r in plan:
         print("    %s" % json.dumps(r, ensure_ascii=False))
     blob = json.dumps(plan, ensure_ascii=False)
-    uses_idx = ("ix_scene_created" in blob) or ("USING INDEX" in blob and "scene_id" in blob)
+    uses_idx = "ix_scene_created" in blob
     check("AC2.1 sqlite EXPLAIN 走 ix_scene_created（非 SCAN 全表）",
-          uses_idx and "SCAN " + ps.EVENT_TABLE not in blob, blob[:160])
+          uses_idx and ("SCAN " + ps.EVENT_TABLE) not in blob
+          and "TEMP B-TREE" not in blob, blob[:220])
 
     # MySQL/MariaDB 侧：若本机测试库可达，跑真 EXPLAIN 并粘贴 type/key/rows
     my = try_mysql_explain()

world_sync/pbl_runtime_polling.py

@@ -183,6 +183,8 @@ class PollingFallback:
         self.timeout = float(timeout)
         self.limit = int(limit)
         self._lock = threading.Lock()
+        # 广播到点唤醒闸：notify_broadcast 置位 → run() 立即结束等待（不必等满 3s）
+        self._wake = threading.Event()
         # scene_id -> {"watermark": str, "notified": bool, "last_notify_ts": float}
         self._scenes: Dict[str, Dict[str, Any]] = {}
         # scene_id -> [event_uid]（保序，用于 NOT IN 去重）
@@ -253,6 +255,8 @@ class PollingFallback:
             st["last_notify_ts"] = time.time()
             info = {"scene_id": scene_id, "notified": True,
                     "watermark": st["watermark"], "watermark_advanced": advanced}
+        # 唤醒正在等待的兜底循环：广播正常路径下消费端不应再等满一个轮询周期
+        self._wake.set()
         return info
 
     def was_notified(self, scene_id: str) -> bool:
@@ -376,6 +380,18 @@ class PollingFallback:
                 "found": found, "notified": self.was_notified(scene_id)}
 
     # ---------------- 兜底循环 ----------------
+    def _wait_interval(self, scene_id: str, interval: float,
+                       sleep: Callable[[float], None]) -> None:
+        """等待一个轮询周期；期间广播到点则立即返回（不白等满 3s）。
+
+        注入 ``sleep``（测试假时钟）时退化为整段等待，保持可测性；默认走
+        :class:`threading.Event` 闸，零轮询开销。
+        """
+        if sleep is not time.sleep:
+            sleep(interval)
+            return
+        self._wake.wait(interval)
+
     def run(self, scene_id: str, wait_for: Optional[str] = None,
             timeout: Optional[float] = None, interval: Optional[float] = None,
             on_recovered: Optional[Callable[[Dict[str, Any]], None]] = None,
@@ -389,13 +405,14 @@ class PollingFallback:
         interval = float(interval if interval is not None else self.interval)
         timeout = float(timeout if timeout is not None else self.timeout)
         self.clear_notified(scene_id)
+        self._wake.clear()
         deadline = time.monotonic() + timeout
         polls = 0
         recovered: List[Dict[str, Any]] = []
         stopped_by = "timeout"
         found = False
         while time.monotonic() < deadline:
-            sleep(interval)
+            self._wait_interval(scene_id, interval, sleep)
             polls += 1
             res = self.poll_once(scene_id, wait_for=wait_for)
             for row in res["recovered"]: