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

2026-09-21 13:25:24 +08:00 · 2026-09-21 13:25:24 +08:00 · a4c91c869e
commit a4c91c869e
parent a20287a10a
5 changed files with 1540 additions and 1 deletions
--- a/scripts/sync_code_to_app.py
+++ b/scripts/sync_code_to_app.py
@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""M11b-3 代码镜像同步器：modules/world_sync/ 是唯一真源，
+apps/scense/pkgs/world_sync/ 是打包镜像（宿主 build.sh 从应用仓库安装，禁止手工双写）。
+
+与 sync_models_to_app.py（只管 models/）互补，本脚本同步 **Python 包与测试**：
+  - world_sync/*.py（包源码，含 pbl_runtime_broadcast.py）
+  - tests/*.py（含 test_m11b3_post_commit_broadcast.py）
+  - skill/SKILL.md、README.md
+
+用法:
+    python modules/world_sync/scripts/sync_code_to_app.py           # 同步
+    python modules/world_sync/scripts/sync_code_to_app.py --check   # 只校验 sha256
+
+行为：逐文件以同一份字节覆盖镜像落点，打印两两 sha256 比对；--check 不一致退出码 1。
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import os
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+WS_ROOT = os.path.abspath(os.path.join(HERE, "..", "..", ".."))
+
+SRC_ROOT = os.path.join("modules", "world_sync")
+DST_ROOT = os.path.join("apps", "scense", "pkgs", "world_sync")
+
+#: 需要镜像的相对路径（真源 = modules/world_sync/）
+MIRROR_SUBDIRS = ("world_sync", "tests")
+MIRROR_FILES = ("README.md", os.path.join("skill", "SKILL.md"))
+#: 不同步的产物/缓存
+SKIP_SUFFIX = ("__pycache__", ".pyc", ".egg-info", ".pytest_cache")
+
+
+def sha256_of(path: str) -> str:
+    h = hashlib.sha256()
+    with open(path, "rb") as fh:
+        for chunk in iter(lambda: fh.read(65536), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def collect() -> list:
+    rels = []
+    for sub in MIRROR_SUBDIRS:
+        base = os.path.join(SRC_ROOT, sub)
+        if not os.path.isdir(base):
+            continue
+        for name in sorted(os.listdir(base)):
+            if any(s in name for s in SKIP_SUFFIX):
+                continue
+            if not name.endswith(".py"):
+                continue
+            rels.append(os.path.join(sub, name))
+    for rel in MIRROR_FILES:
+        if os.path.isfile(os.path.join(SRC_ROOT, rel)):
+            rels.append(rel)
+    return rels
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--check", action="store_true", help="只校验，不写入")
+    args = ap.parse_args()
+
+    bad = 0
+    for rel in collect():
+        src = os.path.join(WS_ROOT, SRC_ROOT, rel)
+        dst = os.path.join(WS_ROOT, DST_ROOT, rel)
+        src_sha = sha256_of(src)
+        same = os.path.isfile(dst) and sha256_of(dst) == src_sha
+        if args.check:
+            print("%-58s %s" % (rel, "OK" if same else "DRIFT"))
+            if not same:
+                bad += 1
+            continue
+        os.makedirs(os.path.dirname(dst), exist_ok=True)
+        with open(src, "rb") as fh:
+            data = fh.read()
+        with open(dst, "wb") as fh:
+            fh.write(data)
+        ok = sha256_of(dst) == src_sha
+        print("%-58s %s sha256=%s" % (rel, "SYNCED" if ok else "FAIL", src_sha[:12]))
+        if not ok:
+            bad += 1
+    if bad:
+        print("RESULT: FAIL (%d 项不一致)" % bad)
+        return 1
+    print("RESULT: PASS (镜像与真源逐文件 sha256 一致)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
--- a/tests/test_m11b3_post_commit_broadcast.py
+++ b/tests/test_m11b3_post_commit_broadcast.py
@ -0,0 +1,432 @@
+# -*- coding: utf-8 -*-
+"""M11b-3 事务提交后广播机制 —— 单元测试。
+
+覆盖验收标准：
+1. 事务 COMMIT 后广播消息在 ≤300ms 内发出（P99 基准）；
+2. 广播失败（通道不可用）不影响已提交数据，且有日志记录；
+3. 广播不阻塞 write_event_with_state 的返回（异步）；
+4. 广播内容至少含 event_id / scene_id / entity_id / event_type / payload 摘要；
+5. 事务未提交（抛异常回滚）时**不广播**；
+6. 不引入轮询兜底（源码静态审计：无 sleep 轮询循环、无 DB 读取）。
+
+测试用内存假事务（不依赖真实 DB / 网络），因此在无数据库的 CI 环境同样可跑。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import re
+import threading
+import time
+
+import pytest
+
+import world_sync as ws
+from world_sync.pbl_runtime_broadcast import (
+    BROADCAST_LATENCY_SLO_MS,
+    BaseChannel,
+    Broadcaster,
+    ChannelUnavailable,
+    InProcessChannel,
+    WebhookChannel,
+    build_broadcast_message,
+    emit_post_commit_broadcast,
+    install_broadcast_hook,
+    reset_broadcaster,
+    subscribe_broadcast,
+    unsubscribe_broadcast,
+    wrap_with_post_commit_broadcast,
+)
+
+
+# ---------------------------------------------------------------------------
+# fixtures / helpers
+# ---------------------------------------------------------------------------
+class RecordingChannel(BaseChannel):
+    """记录型通道：保存收到的消息，可配置为持续失败。"""
+
+    name = "recording"
+
+    def __init__(self, fail: bool = False, delay_s: float = 0.0) -> None:
+        self.messages = []
+        self.fail = fail
+        self.delay_s = delay_s
+        self.calls = 0
+        self._lock = threading.Lock()
+
+    def publish(self, message):
+        self.calls += 1
+        if self.delay_s:
+            time.sleep(self.delay_s)
+        if self.fail:
+            raise ChannelUnavailable("channel down (test)")
+        with self._lock:
+            self.messages.append(message)
+
+
+class BlockingChannel(BaseChannel):
+    """首次 publish 阻塞，直到事件被 set —— 用于证明广播不阻塞事务返回。"""
+
+    name = "blocking"
+
+    def __init__(self) -> None:
+        self.entered = threading.Event()
+        self.release = threading.Event()
+        self.messages = []
+
+    def publish(self, message):
+        self.entered.set()
+        self.release.wait(timeout=5.0)
+        self.messages.append(message)
+
+
+def fake_write_result(**over):
+    """模拟 M11b-2 write_event_with_state 提交成功后的返回值。"""
+    res = {
+        "event_uid": "evt-uid-0001",
+        "event_id": "evt-uid-0001",
+        "tenant_id": "tenant-A",
+        "world_id": "world-77",
+        "session_id": "sess-9",
+        "entity_id": "ent-42",
+        "state_version": 4,
+        "action": "updated",
+        "created_at": "2026-09-21 10:00:00",
+        "dialect": "sqlite",
+        "elapsed_ms": 12.3,
+    }
+    res.update(over)
+    return res
+
+
+def fake_event_data(**over):
+    ev = {
+        "tenant_id": "tenant-A",
+        "world_id": "world-77",
+        "scene_id": "scene-77",
+        "session_id": "sess-9",
+        "entity_id": "ent-42",
+        "event_type": "state.updated",
+        "payload": {"hp": 100, "pos": [1, 2, 3]},
+    }
+    ev.update(over)
+    return ev
+
+
+@pytest.fixture
+def broadcaster():
+    chan = RecordingChannel()
+    b = Broadcaster(channels=[chan], retry_attempts=1, retry_backoff_s=0.001)
+    yield b, chan
+    b.shutdown(timeout=1.0)
+
+
+@pytest.fixture(autouse=True)
+def _clean_global():
+    yield
+    reset_broadcaster(None, shutdown_old=True)
+
+
+def wait_until(predicate, timeout=3.0, interval=0.005):
+    """等待条件成立（仅测试同步用，非生产轮询）。"""
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if predicate():
+            return True
+        time.sleep(interval)
+    return predicate()
+
+
+# ---------------------------------------------------------------------------
+# 1. 广播消息内容
+# ---------------------------------------------------------------------------
+def test_build_message_contains_required_fields():
+    msg = build_broadcast_message(
+        fake_write_result(),
+        commit_ts=time.time(),
+        context={"event_data": fake_event_data(), "state_update": {"state": {"hp": 100}}},
+    )
+    assert msg["event_id"] == "evt-uid-0001"
+    assert msg["scene_id"] == "scene-77"          # 显式 scene_id
+    assert msg["entity_id"] == "ent-42"
+    assert msg["event_type"] == "state.updated"
+    assert msg["payload_summary"]                  # 简要 payload 摘要
+    assert "hp" in msg["payload_summary"]
+    assert len(msg["payload_summary"]) <= 256
+    assert msg["state_version"] == 4
+    assert msg["tenant_id"] == "tenant-A"
+    assert json.dumps(msg, ensure_ascii=False)     # 可序列化（webhook/redis 通道要求）
+
+
+def test_scene_id_falls_back_to_world_id():
+    ev = fake_event_data()
+    del ev["scene_id"]
+    msg = build_broadcast_message(fake_write_result(), context={"event_data": ev})
+    assert msg["scene_id"] == "world-77"
+
+
+def test_payload_summary_is_truncated():
+    big = {"blob": "x" * 5000}
+    msg = build_broadcast_message(fake_write_result(), context={"event_data": {"payload": big}})
+    assert len(msg["payload_summary"]) <= 256
+    assert msg["payload_summary"].endswith("...")
+
+
+def test_build_message_never_raises_on_garbage():
+    for junk in (None, 123, "str", object(), [1, 2]):
+        msg = build_broadcast_message(junk)
+        assert "event_id" in msg and "payload_summary" in msg
+
+
+# ---------------------------------------------------------------------------
+# 2. COMMIT 后广播被触发（post-commit hook）
+# ---------------------------------------------------------------------------
+def test_hook_fires_only_after_commit(broadcaster):
+    b, chan = broadcaster
+    calls = {"n": 0}
+
+    def write_fn(event_data=None, state_update=None):
+        calls["n"] += 1
+        return fake_write_result()
+
+    wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b)
+    res = wrapped(event_data=fake_event_data(), state_update={"state": {}})
+    assert res["event_id"] == "evt-uid-0001"       # 原返回值透传
+    assert calls["n"] == 1
+    assert wait_until(lambda: len(chan.messages) == 1)
+    msg = chan.messages[0]
+    assert msg["event_id"] == "evt-uid-0001"
+    assert msg["entity_id"] == "ent-42"
+    assert msg["commit_ts"] <= msg["submit_ts"]
+
+
+def test_no_broadcast_when_transaction_raises(broadcaster):
+    """事务回滚（抛异常）→ 绝不广播，异常原样上抛。"""
+    b, chan = broadcaster
+
+    def write_fn(*a, **kw):
+        raise RuntimeError("deadlock, transaction rolled back")
+
+    wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b)
+    with pytest.raises(RuntimeError):
+        wrapped(event_data=fake_event_data())
+    time.sleep(0.15)
+    assert chan.messages == []
+    assert chan.calls == 0
+
+
+def test_install_hook_wraps_module_symbol_and_is_idempotent():
+    import world_sync.pbl_runtime_tx_sqlor as txs
+    original = txs.write_event_with_state
+    underlying = getattr(original, "_m11b3_underlying", original)
+    try:
+        wrapped1 = install_broadcast_hook()
+        wrapped2 = install_broadcast_hook()          # 幂等，不重复包装
+        assert getattr(wrapped1, "_m11b3_broadcast_wrapped", False)
+        assert txs.write_event_with_state is wrapped1
+        assert wrapped1 is wrapped2
+        assert getattr(wrapped1, "_m11b3_underlying") is underlying
+        assert ws.is_broadcast_hook_installed() is True
+    finally:
+        txs.write_event_with_state = underlying
+        reset_broadcaster(None, shutdown_old=True)
+
+
+# ---------------------------------------------------------------------------
+# 3. 广播失败不回滚已提交数据 + 有日志
+# ---------------------------------------------------------------------------
+def test_broadcast_failure_does_not_affect_committed_data(broadcaster, caplog):
+    chan = RecordingChannel(fail=True)
+    b = Broadcaster(channels=[chan], retry_attempts=2, retry_backoff_s=0.001)
+    try:
+        def write_fn(*a, **kw):
+            return fake_write_result()             # 事务已成功提交
+
+        wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b)
+        with caplog.at_level(logging.ERROR, logger="world_sync.broadcast"):
+            res = wrapped(event_data=fake_event_data())
+        # 调用方拿到正常返回值，无异常 → 已提交数据不受影响
+        assert res["event_id"] == "evt-uid-0001"
+        assert wait_until(lambda: b.stats()["failed"] >= 1)
+        assert b.stats()["published"] == 0
+        assert b.stats()["last_error"]
+        text = "\n".join(r.getMessage() for r in caplog.records)
+        assert "广播失败" in text and "事务已提交" in text
+    finally:
+        b.shutdown(timeout=1.0)
+
+
+def test_failing_channel_does_not_propagate_to_caller():
+    """emit_post_commit_broadcast 自身永不抛异常（即使消息构造/通道全炸）。"""
+    chan = RecordingChannel(fail=True)
+    b = Broadcaster(channels=[chan], retry_attempts=0)
+    try:
+        msg = emit_post_commit_broadcast(fake_write_result(), broadcaster=b)
+        assert msg is not None and msg["event_id"]
+        assert wait_until(lambda: b.stats()["failed"] >= 1)
+    finally:
+        b.shutdown(timeout=1.0)
+
+
+def test_partial_channel_failure_keeps_good_channel():
+    good = RecordingChannel()
+    bad = RecordingChannel(fail=True)
+    b = Broadcaster(channels=[bad, good], retry_attempts=1, retry_backoff_s=0.001)
+    try:
+        emit_post_commit_broadcast(fake_write_result(), broadcaster=b)
+        assert wait_until(lambda: len(good.messages) == 1)
+        assert b.stats()["published"] == 1
+    finally:
+        b.shutdown(timeout=1.0)
+
+
+def test_subscriber_exception_is_swallowed():
+    chan = InProcessChannel()
+
+    def boom(message):
+        raise ValueError("subscriber bug")
+
+    chan.subscribe(boom)
+    b = Broadcaster(channels=[chan], retry_attempts=0)
+    try:
+        emit_post_commit_broadcast(fake_write_result(), broadcaster=b)
+        assert wait_until(lambda: b.stats()["published"] >= 1)
+        assert b.stats()["failed"] == 0
+    finally:
+        b.shutdown(timeout=1.0)
+
+
+# ---------------------------------------------------------------------------
+# 4. 异步：不阻塞事务返回
+# ---------------------------------------------------------------------------
+def test_broadcast_does_not_block_write_return():
+    slow = BlockingChannel()
+    b = Broadcaster(channels=[slow], retry_attempts=0)
+    try:
+        def write_fn(*a, **kw):
+            return fake_write_result()
+
+        wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b)
+        t0 = time.perf_counter()
+        res = wrapped(event_data=fake_event_data())   # 不应被 slow 通道卡住
+        elapsed_ms = (time.perf_counter() - t0) * 1000.0
+        assert res["event_id"]
+        assert elapsed_ms < 100.0, "广播阻塞了事务返回: %.1fms" % elapsed_ms
+        # 稍后 worker 线程仍会发出
+        slow.release.set()
+        assert wait_until(lambda: len(slow.messages) == 1)
+    finally:
+        slow.release.set()
+        b.shutdown(timeout=1.5)
+
+
+def test_submit_returns_immediately_for_many_messages():
+    chan = RecordingChannel()
+    b = Broadcaster(channels=[chan], queue_size=100, retry_attempts=0)
+    try:
+        t0 = time.perf_counter()
+        for i in range(100):
+            b.submit(build_broadcast_message(fake_write_result(event_id="e%d" % i)))
+        elapsed_ms = (time.perf_counter() - t0) * 1000.0
+        assert elapsed_ms < 100.0
+        assert b.drain(timeout=5.0)
+        assert len(chan.messages) == 100
+    finally:
+        b.shutdown(timeout=1.0)
+
+
+def test_queue_full_falls_back_to_direct_publish_without_blocking():
+    """队列满时不阻塞调用方（退化为同步直发），消息不丢。"""
+    chan = RecordingChannel()
+    b = Broadcaster(channels=[chan], queue_size=1, retry_attempts=0)
+    try:
+        for i in range(20):
+            b.submit(build_broadcast_message(fake_write_result(event_id="e%d" % i)))
+        assert b.drain(timeout=5.0)
+        assert len(chan.messages) == 20
+        assert b.stats()["published"] == 20
+    finally:
+        b.shutdown(timeout=1.0)
+
+
+# ---------------------------------------------------------------------------
+# 5. 延迟 SLO ≤ 300ms（P99）
+# ---------------------------------------------------------------------------
+def test_broadcast_latency_p99_within_300ms():
+    chan = RecordingChannel()
+    b = Broadcaster(channels=[chan], retry_attempts=2, retry_backoff_s=0.001)
+    n = 200
+    try:
+        for i in range(n):
+            emit_post_commit_broadcast(
+                fake_write_result(event_id="lat-%d" % i),
+                broadcaster=b,
+                context={"event_data": fake_event_data()},
+            )
+        assert b.drain(timeout=15.0), "广播未在超时内全部发出"
+        assert len(chan.messages) == n
+        p99 = b.latency_percentile_ms(99.0)
+        p50 = b.latency_percentile_ms(50.0)
+        stats = b.stats()
+        print("\n[M11b-3 延迟基准] n=%d p50=%.3fms p99=%.3fms max=%.3fms SLO=%.0fms"
+              % (n, p50, p99, stats["max_latency_ms"], BROADCAST_LATENCY_SLO_MS))
+        assert p99 <= BROADCAST_LATENCY_SLO_MS, "P99 超阈值: %.3fms" % p99
+        assert stats["failed"] == 0
+    finally:
+        b.shutdown(timeout=1.0)
+
+
+def test_latency_slo_constant_matches_spec():
+    assert BROADCAST_LATENCY_SLO_MS == 300.0
+    assert ws.BROADCAST_LATENCY_SLO_MS == 300.0
+
+
+# ---------------------------------------------------------------------------
+# 6. 进程内订阅 API（宿主接线）
+# ---------------------------------------------------------------------------
+def test_subscribe_and_unsubscribe_global():
+    reset_broadcaster(Broadcaster(channels=[InProcessChannel()], retry_attempts=0))
+    got = []
+    handler = subscribe_broadcast(got.append)
+    emit_post_commit_broadcast(fake_write_result(event_id="sub-1"))
+    assert wait_until(lambda: len(got) == 1)
+    assert got[0]["event_id"] == "sub-1"
+    assert unsubscribe_broadcast(handler) is True
+    emit_post_commit_broadcast(fake_write_result(event_id="sub-2"))
+    time.sleep(0.15)
+    assert len(got) == 1                            # 取消后不再收到
+    assert ws.broadcast_stats()["published"] >= 2
+
+
+def test_configure_broadcast_replaces_channels():
+    chan = RecordingChannel()
+    b = ws.configure_broadcast(channels=[chan], retry_attempts=0)
+    emit_post_commit_broadcast(fake_write_result(event_id="cfg-1"))
+    assert wait_until(lambda: len(chan.messages) == 1)
+    assert b.stats()["channels"] == ["recording"]
+
+
+# ---------------------------------------------------------------------------
+# 7. 无轮询兜底（静态审计）
+# ---------------------------------------------------------------------------
+def test_no_polling_loop_in_broadcast_source():
+    src_path = os.path.join(os.path.dirname(ws.__file__), "pbl_runtime_broadcast.py")
+    with open(src_path, encoding="utf-8") as fh:
+        lines = fh.readlines()
+    code = "".join(l for l in lines if not l.lstrip().startswith("#"))
+    # 生产代码里不得出现 sleep 轮询（time.sleep 只允许出现在重试退避一处）
+    sleeps = [i for i, l in enumerate(code.splitlines(), 1) if "time.sleep(" in l]
+    assert len(sleeps) <= 1, "疑似轮询 sleep: %s" % sleeps
+    # 不得有 while True + sleep 的轮询形态
+    assert not re.search(r"while\s+(True|1)\s*:\s*(#.*)?\n\s*(try:\s*\n\s*)?time\.sleep", code), \
+        "存在 while True + sleep 轮询兜底"
+    # 不得读数据库（无 SQL 关键字 / 无 conn_factory / 无 sor 调用）
+    for token in ("SELECT ", "UPDATE ", "INSERT INTO", "conn_factory", "sqlorContext", "DBPools"):
+        assert token not in code, "广播层不应触碰 DB: %s" % token
+
+
+def test_broadcast_channel_name_exported():
+    assert ws.BROADCAST_CHANNEL == "pbl.runtime.event"
+    assert ws.PAYLOAD_SUMMARY_MAX_CHARS == 256
--- a/world_sync/init.py
+++ b/world_sync/init.py
@ -120,6 +120,36 @@ from .pbl_runtime_tx_sqlor import (
    write_event_with_state,
 )

+# --- M11b-3 事务提交后广播（post-commit broadcast，真源：pbl_runtime_broadcast.py） ---
+from .pbl_runtime_broadcast import (
+    BROADCAST_CHANNEL,
+    BROADCAST_LATENCY_SLO_MS,
+    DEFAULT_EVENT_CHANNEL,
+    PAYLOAD_SUMMARY_MAX_CHARS,
+    BaseChannel,
+    Broadcaster,
+    ChannelUnavailable,
+    InProcessChannel,
+    NoopChannel,
+    RedisPubSubChannel,
+    WebhookChannel,
+    broadcast_latency_percentile_ms,
+    broadcast_last_message,
+    broadcast_stats,
+    build_broadcast_message,
+    configure_broadcast,
+    drain_broadcast,
+    emit_post_commit_broadcast,
+    get_broadcaster,
+    install_broadcast_hook,
+    is_broadcast_hook_installed,
+    reset_broadcaster,
+    shutdown_broadcast,
+    subscribe_broadcast,
+    unsubscribe_broadcast,
+    wrap_with_post_commit_broadcast,
+)
+
 #: 契约解析优先级键（与 pbl_runtime_tx_env.SERVER_ENV_KEYS 一致，单一来源）
 assert tuple(SERVER_ENV_CONN_FACTORY_KEYS) == tuple(SERVER_ENV_KEYS), \
    "world_sync: SERVER_ENV_CONN_FACTORY_KEYS 与 pbl_runtime_tx_env.SERVER_ENV_KEYS 分叉"
@ -190,6 +220,33 @@ __all__ = [
    "InvalidWriteRequest",
    "ConnectionUnavailable",
    "TransactionAborted",
+    # M11b-3 事务提交后广播（post-commit hook / 通道 / 指标）
+    "BROADCAST_CHANNEL",
+    "BROADCAST_LATENCY_SLO_MS",
+    "DEFAULT_EVENT_CHANNEL",
+    "PAYLOAD_SUMMARY_MAX_CHARS",
+    "BaseChannel",
+    "Broadcaster",
+    "ChannelUnavailable",
+    "InProcessChannel",
+    "NoopChannel",
+    "RedisPubSubChannel",
+    "WebhookChannel",
+    "build_broadcast_message",
+    "emit_post_commit_broadcast",
+    "wrap_with_post_commit_broadcast",
+    "install_broadcast_hook",
+    "is_broadcast_hook_installed",
+    "configure_broadcast",
+    "get_broadcaster",
+    "reset_broadcaster",
+    "subscribe_broadcast",
+    "unsubscribe_broadcast",
+    "broadcast_stats",
+    "broadcast_last_message",
+    "broadcast_latency_percentile_ms",
+    "drain_broadcast",
+    "shutdown_broadcast",
    # 宿主基础包可用性诊断
    "HOST_DEPS_AVAILABLE",
    "HOST_DEPS_ERROR",
--- a/world_sync/init.py
+++ b/world_sync/init.py
@ -148,9 +148,39 @@ from .pbl_runtime_tx_sqlor import (
    write_event_with_state,
 )

+# --- M11b-3: 事务提交后广播（post-commit broadcast）注册接线 ---
+from .pbl_runtime_broadcast import (
+    BROADCAST_CHANNEL,
+    BROADCAST_LATENCY_SLO_MS,
+    DEFAULT_EVENT_CHANNEL,
+    PAYLOAD_SUMMARY_MAX_CHARS,
+    BaseChannel,
+    Broadcaster,
+    ChannelUnavailable,
+    InProcessChannel,
+    NoopChannel,
+    RedisPubSubChannel,
+    WebhookChannel,
+    broadcast_latency_percentile_ms,
+    broadcast_last_message,
+    broadcast_stats,
+    build_broadcast_message,
+    configure_broadcast,
+    drain_broadcast,
+    emit_post_commit_broadcast,
+    get_broadcaster,
+    install_broadcast_hook,
+    is_broadcast_hook_installed,
+    reset_broadcaster,
+    shutdown_broadcast,
+    subscribe_broadcast,
+    unsubscribe_broadcast,
+    wrap_with_post_commit_broadcast,
+)
+

 MODULE_NAME = 'world_sync'
-MODULE_VERSION = '0.2.0-m11b2a'
+MODULE_VERSION = '0.3.0-m11b3'
 #: 宿主 ServerEnv 上登记 conn_factory 的键（与 pbl_runtime_tx_env.SERVER_ENV_KEYS 一致）
 SERVER_ENV_CONN_FACTORY_KEYS = SERVER_ENV_KEYS

@ -480,6 +510,25 @@ def load_world_sync(server_env=None, conn_factory=None):
    env.pbl_runtime_conn_factory_keys = list(SERVER_ENV_KEYS)
    env.pbl_runtime_pool_keys = list(POOL_KEYS)
    env.pbl_runtime_env_name_keys = list(ENV_NAME_KEYS)
+    # --- M11b-3: 事务提交后广播（异步 fire-and-forget，SLO <=300ms P99） ---
+    env.world_sync_broadcast_channel = BROADCAST_CHANNEL
+    env.world_sync_broadcast_latency_slo_ms = BROADCAST_LATENCY_SLO_MS
+    env.build_broadcast_message = build_broadcast_message
+    env.emit_post_commit_broadcast = emit_post_commit_broadcast
+    env.wrap_with_post_commit_broadcast = wrap_with_post_commit_broadcast
+    env.configure_broadcast = configure_broadcast
+    env.get_broadcast_broadcaster = get_broadcaster
+    env.subscribe_broadcast = subscribe_broadcast
+    env.unsubscribe_broadcast = unsubscribe_broadcast
+    env.broadcast_stats = broadcast_stats
+    env.broadcast_last_message = broadcast_last_message
+    env.broadcast_latency_percentile_ms = broadcast_latency_percentile_ms
+    env.drain_broadcast = drain_broadcast
+    env.shutdown_broadcast = shutdown_broadcast
+    env.reset_broadcast = reset_broadcaster
+    # post-commit hook：把 write_event_with_state 包成「COMMIT 后触发广播」版本，
+    # 并同步刷新三处引用（pbl_runtime_tx_sqlor / world_sync 包命名空间 / env）。
+    env.write_event_with_state = install_broadcast_hook(env)
    _configure_runtime_writer_from_env(env)
    _register_tables_to_env(env)
    return env
--- a/world_sync/pbl_runtime_broadcast.py
+++ b/world_sync/pbl_runtime_broadcast.py
@ -0,0 +1,904 @@
+# -*- coding: utf-8 -*-
+"""world_sync M11b-3 —— 事务提交后广播（post-commit broadcast）。
+
+范围（严格对齐任务书）
+====================
+只负责 **M11b-2 ``write_event_with_state`` 事务 COMMIT 成功之后** 的广播通知：
+
+1. COMMIT 成功后由 post-commit hook 触发广播（:func:`wrap_with_post_commit_broadcast`
+   / :func:`emit_post_commit_broadcast`），广播内容至少含
+   ``event_id / scene_id / entity_id / event_type / payload_summary``；
+2. 通道可插拔：进程内 :class:`InProcessChannel`、HTTP webhook
+   :class:`WebhookChannel`、Redis pub/sub :class:`RedisPubSubChannel`；
+3. **异步 fire-and-forget**：广播在独立 worker 线程执行，绝不阻塞事务返回，
+   广播失败只记日志 + 计数，**绝不回滚已提交事务**（异常不外抛）；
+4. 延迟 SLO：COMMIT → 广播消息发出 ≤ **300ms（P99）**，实测见
+   :func:`broadcast_latency_percentile_ms`；
+5. **不含轮询兜底**：本模块无任何 ``sleep`` 轮询循环、无任何 SQL/DB 读取，
+   队列消费是阻塞式条件变量等待（``Queue.join`` / ``Queue.get(block=True)``）。
+
+宿主接线::
+
+    from world_sync import load_world_sync, configure_broadcast
+    load_world_sync()                       # 已自动 install_broadcast_hook()
+    configure_broadcast(webhook_urls=["http://127.0.0.1:8099/pbl/event"])
+
+    from world_sync import subscribe_broadcast
+    subscribe_broadcast(lambda msg: print(msg["event_id"]))
+
+线程模型：单 dispatcher 线程 + 有界 ``queue.Queue``；``shutdown_broadcast()``
+投递哨兵优雅退出。所有共享状态由 ``threading.RLock`` 保护。
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import queue
+import threading
+import time
+import urllib.error
+import urllib.request
+from collections import deque
+from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple
+
+__all__ = [
+    # 常量
+    "BROADCAST_LATENCY_SLO_MS",
+    "BROADCAST_CHANNEL",
+    "DEFAULT_EVENT_CHANNEL",
+    "PAYLOAD_SUMMARY_MAX_CHARS",
+    # 通道
+    "ChannelUnavailable",
+    "BaseChannel",
+    "InProcessChannel",
+    "WebhookChannel",
+    "RedisPubSubChannel",
+    "NoopChannel",
+    # 核心 API
+    "build_broadcast_message",
+    "emit_post_commit_broadcast",
+    "wrap_with_post_commit_broadcast",
+    "install_broadcast_hook",
+    "configure_broadcast",
+    "get_broadcaster",
+    "reset_broadcaster",
+    "subscribe_broadcast",
+    "unsubscribe_broadcast",
+    "broadcast_stats",
+    "broadcast_last_message",
+    "broadcast_latency_percentile_ms",
+    "drain_broadcast",
+    "shutdown_broadcast",
+    "is_broadcast_hook_installed",
+]
+
+_LOG = logging.getLogger("world_sync.broadcast")
+
+# ---------------------------------------------------------------------------
+# 配置常量（单一真源，测试与文档引用同一份）
+# ---------------------------------------------------------------------------
+#: 任务书 SLO：COMMIT → 广播消息发出 ≤ 300ms（P99）
+BROADCAST_LATENCY_SLO_MS: float = 300.0
+#: 默认通道名（Redis channel / webhook 逻辑主题）
+DEFAULT_EVENT_CHANNEL: str = "pbl.runtime.event"
+#: 别名（与既有 world_sync 命名风格一致）
+BROADCAST_CHANNEL: str = DEFAULT_EVENT_CHANNEL
+#: payload 摘要最大字符数（广播体保持轻量，不带全量 payload）
+PAYLOAD_SUMMARY_MAX_CHARS: int = 256
+#: 分发队列容量（满时降级为「调用线程内直接发布」，不丢消息也不阻塞事务）
+DEFAULT_QUEUE_SIZE: int = 2000
+#: 单次投递失败重试次数（不含首次尝试）
+DEFAULT_RETRY_ATTEMPTS: int = 2
+#: 重试退避基数（秒）
+DEFAULT_RETRY_BACKOFF_S: float = 0.02
+#: webhook 超时（秒）——远小于 SLO，保证不会把广播线程拖死
+WEBHOOK_TIMEOUT_S: float = 0.25
+#: 延迟样本保留数
+LATENCY_SAMPLE_WINDOW: int = 5000
+
+_SENTINEL = object()
+
+
+class ChannelUnavailable(RuntimeError):
+    """广播通道不可用（仅内部记录/计数，绝不向事务调用方外抛）。"""
+
+
+# ---------------------------------------------------------------------------
+# 小工具
+# ---------------------------------------------------------------------------
+def _field(obj: Any, name: str, default: Any = None) -> Any:
+    """兼容 dict / DictObject / dataclass 的宽松取值（不抛异常）。"""
+    if obj is None:
+        return default
+    if isinstance(obj, dict):
+        val = obj.get(name, default)
+        return default if val is None else val
+    try:
+        val = getattr(obj, name)
+    except Exception:
+        val = None
+    if val is None:
+        try:
+            val = obj[name]  # type: ignore[index]
+        except Exception:
+            val = None
+    return default if val is None else val
+
+
+def _percentile(values: Sequence[float], pct: float) -> float:
+    """最近秩法百分位（不依赖 numpy）。空序列返回 0.0。"""
+    vals = sorted(float(v) for v in values)
+    if not vals:
+        return 0.0
+    if pct <= 0:
+        return vals[0]
+    if pct >= 100:
+        return vals[-1]
+    idx = int(round((pct / 100.0) * (len(vals) - 1)))
+    idx = min(max(0, idx), len(vals) - 1)
+    return vals[idx]
+
+
+def _summarize_payload(payload: Any, limit: int = PAYLOAD_SUMMARY_MAX_CHARS) -> str:
+    """生成简要 payload 摘要（稳定、可截断、永不抛异常）。"""
+    if payload is None:
+        return ""
+    try:
+        if isinstance(payload, str):
+            text = payload
+        else:
+            text = json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str)
+    except Exception:
+        text = str(payload)
+    if len(text) > limit:
+        return text[: limit - 3] + "..."
+    return text
+
+
+def _payload_digest(payload: Any) -> str:
+    """payload 的 sha256 前 16 位（订阅方可据此判断是否需要回查详情）。"""
+    try:
+        if isinstance(payload, (bytes, bytearray)):
+            raw = bytes(payload)
+        elif payload is None:
+            raw = b""
+        else:
+            raw = json.dumps(
+                payload, ensure_ascii=False, sort_keys=True, default=str
+            ).encode("utf-8")
+    except Exception:
+        raw = str(payload).encode("utf-8", "replace")
+    return hashlib.sha256(raw).hexdigest()[:16]
+
+
+def _iso(ts: float) -> str:
+    try:
+        return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(ts)) + \
+            ".%03dZ" % int((ts % 1) * 1000)
+    except Exception:
+        return ""
+
+
+def _truthy_env(name: str, default: bool = False) -> bool:
+    raw = os.environ.get(name)
+    if raw is None or raw == "":
+        return default
+    return raw.strip().lower() in ("1", "true", "yes", "y", "on")
+
+
+# ---------------------------------------------------------------------------
+# 通道实现
+# ---------------------------------------------------------------------------
+class BaseChannel:
+    """广播通道基类。子类实现 :meth:`publish`，失败请抛异常（由 Broadcaster 兜住）。"""
+
+    name: str = "base"
+
+    def publish(self, message: Dict[str, Any]) -> None:  # pragma: no cover - 接口
+        raise NotImplementedError
+
+    def close(self) -> None:
+        return None
+
+
+class NoopChannel(BaseChannel):
+    """空通道（广播关闭或未配置真实通道时使用，仅计数）。"""
+
+    name = "noop"
+
+    def __init__(self) -> None:
+        self.published: List[Dict[str, Any]] = []
+
+    def publish(self, message: Dict[str, Any]) -> None:
+        self.published.append(message)
+
+
+class InProcessChannel(BaseChannel):
+    """进程内广播通道：订阅者回调列表，线程安全。"""
+
+    name = "in_process"
+
+    def __init__(self) -> None:
+        self._lock = threading.RLock()
+        self._subscribers: List[Callable[[Dict[str, Any]], None]] = []
+        self.published: List[Dict[str, Any]] = []
+
+    def subscribe(self, handler: Callable[[Dict[str, Any]], None]) -> Callable[[Dict[str, Any]], None]:
+        if not callable(handler):
+            raise TypeError("subscribe_broadcast(handler): handler 必须可调用")
+        with self._lock:
+            self._subscribers.append(handler)
+        return handler
+
+    def unsubscribe(self, handler: Callable[[Dict[str, Any]], None]) -> bool:
+        with self._lock:
+            if handler in self._subscribers:
+                self._subscribers.remove(handler)
+                return True
+        return False
+
+    @property
+    def subscriber_count(self) -> int:
+        with self._lock:
+            return len(self._subscribers)
+
+    def publish(self, message: Dict[str, Any]) -> None:
+        with self._lock:
+            handlers = list(self._subscribers)
+            self.published.append(message)
+            if len(self.published) > LATENCY_SAMPLE_WINDOW:
+                self.published = self.published[-LATENCY_SAMPLE_WINDOW:]
+        errors: List[str] = []
+        for handler in handlers:
+            try:
+                handler(message)
+            except Exception as exc:  # 订阅者异常不影响其他订阅者
+                errors.append("%s: %s" % (type(exc).__name__, exc))
+                _LOG.warning("world_sync.broadcast 订阅者异常已吞掉: %s", exc)
+        if errors:
+            self._last_errors = errors
+
+    def close(self) -> None:
+        with self._lock:
+            self._subscribers.clear()
+
+
+class WebhookChannel(BaseChannel):
+    """HTTP webhook 通道：POST JSON，短超时，失败抛 :class:`ChannelUnavailable`。"""
+
+    name = "webhook"
+
+    def __init__(self, urls: Sequence[str], timeout_s: float = WEBHOOK_TIMEOUT_S) -> None:
+        self.urls: List[str] = [str(u) for u in (urls or []) if u]
+        if not self.urls:
+            raise ChannelUnavailable("WebhookChannel 需要至少一个 url")
+        self.timeout_s = float(timeout_s)
+
+    def publish(self, message: Dict[str, Any]) -> None:
+        body = json.dumps(message, ensure_ascii=False, default=str).encode("utf-8")
+        failures: List[str] = []
+        for url in self.urls:
+            try:
+                req = urllib.request.Request(
+                    url,
+                    data=body,
+                    headers={
+                        "Content-Type": "application/json; charset=utf-8",
+                        "X-Pbl-Event-Id": str(message.get("event_id", "")),
+                    },
+                    method="POST",
+                )
+                with urllib.request.urlopen(req, timeout=self.timeout_s) as resp:
+                    code = int(getattr(resp, "status", 200) or 200)
+                    if code >= 400:
+                        failures.append("%s -> HTTP %d" % (url, code))
+            except urllib.error.HTTPError as exc:
+                failures.append("%s -> HTTP %s" % (url, getattr(exc, "code", "?")))
+            except Exception as exc:
+                failures.append("%s -> %s" % (url, exc))
+        if failures:
+            raise ChannelUnavailable("; ".join(failures))
+
+
+class RedisPubSubChannel(BaseChannel):
+    """Redis pub/sub 通道（可选依赖：``redis``）。构造时即判定可用性。"""
+
+    name = "redis_pubsub"
+
+    def __init__(self, redis_url: Optional[str] = None, channel: str = DEFAULT_EVENT_CHANNEL) -> None:
+        try:
+            import redis  # type: ignore
+        except Exception as exc:  # pragma: no cover - 环境无 redis
+            raise ChannelUnavailable("redis 客户端不可用: %s" % exc) from exc
+        url = redis_url or os.environ.get("PBL_BROADCAST_REDIS_URL") or ""
+        if not url:
+            raise ChannelUnavailable("未配置 redis_url / PBL_BROADCAST_REDIS_URL")
+        try:
+            self._client = redis.Redis.from_url(url, socket_connect_timeout=WEBHOOK_TIMEOUT_S,
+                                                socket_timeout=WEBHOOK_TIMEOUT_S)
+        except Exception as exc:  # pragma: no cover
+            raise ChannelUnavailable("redis 连接创建失败: %s" % exc) from exc
+        self.channel = channel
+
+    def publish(self, message: Dict[str, Any]) -> None:
+        payload = json.dumps(message, ensure_ascii=False, default=str)
+        try:
+            self._client.publish(self.channel, payload)
+        except Exception as exc:
+            raise ChannelUnavailable("redis publish 失败: %s" % exc) from exc
+
+    def close(self) -> None:
+        try:
+            self._client.close()
+        except Exception:
+            pass
+
+
+# ---------------------------------------------------------------------------
+# Broadcaster：异步分发 + 重试 + 指标
+# ---------------------------------------------------------------------------
+class Broadcaster:
+    """事务提交后的异步广播器（fire-and-forget，绝不向调用方抛异常）。"""
+
+    def __init__(
+        self,
+        channels: Optional[Iterable[BaseChannel]] = None,
+        queue_size: int = DEFAULT_QUEUE_SIZE,
+        retry_attempts: int = DEFAULT_RETRY_ATTEMPTS,
+        retry_backoff_s: float = DEFAULT_RETRY_BACKOFF_S,
+        enabled: bool = True,
+    ) -> None:
+        self._lock = threading.RLock()
+        self.channels: List[BaseChannel] = list(channels) if channels else [InProcessChannel()]
+        self.retry_attempts = int(retry_attempts)
+        self.retry_backoff_s = float(retry_backoff_s)
+        self.enabled = bool(enabled)
+        self._seq = 0
+        self._latency: deque = deque(maxlen=LATENCY_SAMPLE_WINDOW)
+        self._stats: Dict[str, Any] = {
+            "published": 0,
+            "dropped_disabled": 0,
+            "failed": 0,
+            "retried": 0,
+            "direct_publish": 0,
+            "queued": 0,
+            "subscriber_errors": 0,
+            "max_latency_ms": 0.0,
+        }
+        self._last_error: Optional[str] = None
+        self._last_message: Optional[Dict[str, Any]] = None
+        self._queue: "queue.Queue" = queue.Queue(maxsize=max(1, int(queue_size)))
+        self._closed = False
+        self._worker: Optional[threading.Thread] = None
+        self._ensure_worker()
+
+    # -- worker 生命周期 -------------------------------------------------
+    def _ensure_worker(self) -> None:
+        with self._lock:
+            if self._closed or (self._worker is not None and self._worker.is_alive()):
+                return
+            self._worker = threading.Thread(
+                target=self._dispatch_loop, name="world-sync-broadcast", daemon=True
+            )
+            self._worker.start()
+
+    def _dispatch_loop(self) -> None:
+        while True:
+            item = self._queue.get(block=True)  # 条件变量阻塞等待，非轮询
+            try:
+                if item is _SENTINEL:
+                    return
+                self._publish_now(item)
+            except Exception as exc:  # pragma: no cover - _publish_now 已兜住
+                _LOG.error("world_sync.broadcast 分发线程异常已吞掉: %s", exc)
+            finally:
+                self._queue.task_done()
+
+    # -- 指标 ------------------------------------------------------------
+    @property
+    def in_flight(self) -> int:
+        return self._queue.qsize()
+
+    def stats(self) -> Dict[str, Any]:
+        with self._lock:
+            snap = dict(self._stats)
+            snap["in_flight"] = self._queue.qsize()
+            snap["latency_samples"] = len(self._latency)
+            snap["last_error"] = self._last_error
+            snap["enabled"] = self.enabled
+            snap["channels"] = [c.name for c in self.channels]
+            snap["p50_latency_ms"] = round(_percentile(self._latency, 50), 3)
+            snap["p99_latency_ms"] = round(_percentile(self._latency, 99), 3)
+            return snap
+
+    def latency_percentile_ms(self, pct: float = 99.0) -> float:
+        with self._lock:
+            return round(_percentile(self._latency, pct), 3)
+
+    @property
+    def last_message(self) -> Optional[Dict[str, Any]]:
+        with self._lock:
+            return self._last_message
+
+    # -- 提交入口（调用方线程，O(1) 非阻塞） -----------------------------
+    def submit(self, message: Dict[str, Any]) -> bool:
+        """把广播消息交给 dispatcher。永不阻塞、永不抛异常。"""
+        if not self.enabled:
+            with self._lock:
+                self._stats["dropped_disabled"] += 1
+            return False
+        self._ensure_worker()
+        try:
+            self._queue.put_nowait(message)
+            with self._lock:
+                self._stats["queued"] += 1
+            return True
+        except Exception:
+            # 队列满：退化为调用线程内直接发布（宁可不阻塞事务也要发出广播）；
+            # _publish_now 内部已 try/except，绝不外抛。
+            with self._lock:
+                self._stats["direct_publish"] += 1
+            self._publish_now(message)
+            return True
+
+    # -- 实际发布（worker 线程） ------------------------------------------
+    def _publish_now(self, message: Dict[str, Any]) -> bool:
+        commit_ts = float(message.get("commit_ts") or time.time())
+        ok_any = False
+        errors: List[str] = []
+        attempts = max(0, self.retry_attempts) + 1
+        for channel in list(self.channels):
+            last_exc: Optional[BaseException] = None
+            sent = False
+            for attempt in range(attempts):
+                try:
+                    channel.publish(message)
+                    sent = True
+                    break
+                except Exception as exc:
+                    last_exc = exc
+                    if attempt + 1 < attempts:
+                        with self._lock:
+                            self._stats["retried"] += 1
+                        if self.retry_backoff_s > 0:
+                            time.sleep(self.retry_backoff_s * (attempt + 1))
+            if sent:
+                ok_any = True
+            else:
+                errors.append("%s: %s" % (channel.name, last_exc))
+        latency_ms = max(0.0, (time.time() - commit_ts) * 1000.0)
+        with self._lock:
+            self._latency.append(latency_ms)
+            self._last_message = message
+            if latency_ms > float(self._stats["max_latency_ms"]):
+                self._stats["max_latency_ms"] = round(latency_ms, 3)
+            if ok_any:
+                self._stats["published"] += 1
+            else:
+                self._stats["failed"] += 1
+            if errors:
+                self._last_error = "; ".join(errors)
+        if not ok_any:
+            # 关键不变式：事务已提交，广播失败只记日志，绝不回滚、绝不外抛。
+            _LOG.error(
+                "world_sync.broadcast 广播失败（事务已提交，数据不受影响）"
+                " event_id=%s scene_id=%s entity_id=%s err=%s",
+                message.get("event_id"), message.get("scene_id"),
+                message.get("entity_id"), "; ".join(errors),
+            )
+        return ok_any
+
+    # -- 订阅（进程内通道） ------------------------------------------------
+    def subscribe(self, handler: Callable[[Dict[str, Any]], None]) -> Callable[[Dict[str, Any]], None]:
+        chans = [c for c in self.channels if isinstance(c, InProcessChannel)]
+        if not chans:
+            chans = [InProcessChannel()]
+            with self._lock:
+                self.channels.append(chans[0])
+        for ch in chans:
+            ch.subscribe(handler)
+        return handler
+
+    def unsubscribe(self, handler: Callable[[Dict[str, Any]], None]) -> bool:
+        hit = False
+        for ch in [c for c in self.channels if isinstance(c, InProcessChannel)]:
+            hit = ch.unsubscribe(handler) or hit
+        return hit
+
+    # -- 收尾 ------------------------------------------------------------
+    def drain(self, timeout: float = 2.0) -> bool:
+        """等待已入队广播全部发送完成。
+
+        用 ``Queue.all_tasks_done`` 条件变量阻塞等待（worker 调 ``task_done`` 时唤醒），
+        **不是轮询**：没有任何 sleep 循环，也不查询数据库。
+        """
+        try:
+            with self._queue.all_tasks_done:
+                deadline = time.time() + float(timeout)
+                while self._queue.unfinished_tasks:
+                    remaining = deadline - time.time()
+                    if remaining <= 0:
+                        return False
+                    self._queue.all_tasks_done.wait(remaining)
+            return True
+        except Exception as exc:  # pragma: no cover
+            _LOG.warning("world_sync.broadcast drain 异常: %s", exc)
+            return False
+
+    def shutdown(self, timeout: float = 1.0) -> None:
+        with self._lock:
+            if self._closed:
+                return
+            self._closed = True
+            worker = self._worker
+        try:
+            self.drain(timeout=timeout)
+        except Exception:
+            pass
+        try:
+            self._queue.put(_SENTINEL)
+        except Exception:
+            pass
+        if worker is not None and worker.is_alive():
+            worker.join(timeout=timeout)
+        for channel in list(self.channels):
+            try:
+                channel.close()
+            except Exception:
+                pass
+
+
+# ---------------------------------------------------------------------------
+# 消息构造 + post-commit hook
+# ---------------------------------------------------------------------------
+def build_broadcast_message(
+    write_result: Any,
+    commit_ts: Optional[float] = None,
+    channel: str = DEFAULT_EVENT_CHANNEL,
+    extra: Optional[Dict[str, Any]] = None,
+    context: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    """从 ``write_event_with_state`` 的返回值构造广播消息（永不抛异常）。
+
+    至少包含 ``event_id / scene_id / entity_id / event_type / payload_summary``；
+    兼容返回值为 dict、DictObject、或 ``{"event": {...}, "state": {...}}`` 嵌套形态。
+
+    :param context: 可选的原始调用上下文 ``{"event_data":..., "state_update":...}``。
+        M11b-2 的返回值不含 ``event_type`` / ``payload``，由 post-commit hook 把调用
+        入参传进来补齐；取值优先级为「事务返回值 > 事件对象 > 状态对象 > 调用入参」。
+    """
+    ts = float(commit_ts) if commit_ts else time.time()
+    ctx = context or {}
+    event = _field(write_result, "event", None) or {}
+    state = _field(write_result, "state", None) or {}
+    src: List[Any] = [
+        write_result, event, state,
+        ctx.get("event_data"), ctx.get("state_update"),
+    ]
+
+    def pick(*names: str) -> Any:
+        for holder in src:
+            for nm in names:
+                val = _field(holder, nm)
+                if val is not None and val != "":
+                    return val
+        return None
+
+    payload = pick("payload", "event_payload", "data")
+    world_id = pick("world_id", "worldid")
+    scene_id = pick("scene_id", "sceneid") or world_id
+    message: Dict[str, Any] = {
+        "schema_version": "1",
+        "channel": channel,
+        "event_id": pick("event_id", "eventid", "id"),
+        "event_uid": pick("event_uid", "uid"),
+        "tenant_id": pick("tenant_id", "tenantid"),
+        "world_id": world_id,
+        "scene_id": scene_id,
+        "entity_id": pick("entity_id", "entityid"),
+        "event_type": pick("event_type", "eventtype") or "state.updated",
+        "session_id": pick("session_id", "sessionid"),
+        "source": pick("source") or "world_sync",
+        "state_version": pick("state_version", "version", "new_version"),
+        "payload_summary": _summarize_payload(payload),
+        "payload_digest": _payload_digest(payload),
+        "commit_ts": ts,
+        "commit_ts_text": _iso(ts),
+        "submit_ts": time.time(),
+    }
+    if extra:
+        try:
+            message.update(extra)
+        except Exception:
+            pass
+    return message
+
+
+#: 模块级单例（宿主可 configure_broadcast 替换）
+_BROADCASTER: Optional[Broadcaster] = None
+_LOCK = threading.RLock()
+_HOOK_INSTALLED = False
+
+
+def get_broadcaster() -> Broadcaster:
+    """取（惰性创建）全局广播器。默认通道：进程内 + 可选 webhook（环境变量）。"""
+    global _BROADCASTER
+    with _LOCK:
+        if _BROADCASTER is None or getattr(_BROADCASTER, "_closed", False):
+            _BROADCASTER = _default_broadcaster()
+        return _BROADCASTER
+
+
+def _default_broadcaster() -> Broadcaster:
+    channels: List[BaseChannel] = [InProcessChannel()]
+    urls = [u.strip() for u in (os.environ.get("PBL_BROADCAST_WEBHOOK_URLS") or "").split(",") if u.strip()]
+    if urls:
+        try:
+            channels.append(WebhookChannel(urls))
+        except Exception as exc:  # pragma: no cover
+            _LOG.warning("world_sync.broadcast webhook 通道构造失败，降级为进程内: %s", exc)
+    if _truthy_env("PBL_BROADCAST_REDIS_ENABLED", False):
+        try:
+            channels.append(RedisPubSubChannel())
+        except Exception as exc:  # pragma: no cover
+            _LOG.warning("world_sync.broadcast redis 通道不可用，已跳过: %s", exc)
+    enabled = not _truthy_env("PBL_BROADCAST_DISABLED", False)
+    return Broadcaster(channels=channels, enabled=enabled)
+
+
+def configure_broadcast(
+    channels: Optional[Iterable[BaseChannel]] = None,
+    enabled: Optional[bool] = None,
+    queue_size: int = DEFAULT_QUEUE_SIZE,
+    retry_attempts: int = DEFAULT_RETRY_ATTEMPTS,
+    retry_backoff_s: float = DEFAULT_RETRY_BACKOFF_S,
+    webhook_urls: Optional[Sequence[str]] = None,
+    redis_url: Optional[str] = None,
+    keep_subscribers: bool = True,
+) -> Broadcaster:
+    """替换全局广播器配置（宿主启动时调用一次即可）。"""
+    global _BROADCASTER
+    with _LOCK:
+        old = _BROADCASTER
+        chosen: Optional[List[BaseChannel]] = list(channels) if channels else None
+        if chosen is None:
+            chosen = [InProcessChannel()]
+            if webhook_urls:
+                try:
+                    chosen.append(WebhookChannel(webhook_urls))
+                except Exception as exc:
+                    _LOG.warning("configure_broadcast webhook 通道失败: %s", exc)
+            if redis_url:
+                try:
+                    chosen.append(RedisPubSubChannel(redis_url=redis_url))
+                except Exception as exc:
+                    _LOG.warning("configure_broadcast redis 通道失败: %s", exc)
+        elif webhook_urls:
+            chosen.extend(c for c in _try_webhook(webhook_urls))
+        elif redis_url:
+            chosen.extend(c for c in _try_redis(redis_url))
+        if keep_subscribers and old is not None:
+            for handler in _collect_subscribers(old):
+                for ch in chosen:
+                    if isinstance(ch, InProcessChannel):
+                        ch.subscribe(handler)
+        new = Broadcaster(
+            channels=chosen,
+            queue_size=queue_size,
+            retry_attempts=retry_attempts,
+            retry_backoff_s=retry_backoff_s,
+            enabled=old.enabled if enabled is None and old is not None else
+                    (enabled if enabled is not None else True),
+        )
+        _BROADCASTER = new
+    if old is not None and old is not new:
+        try:
+            old.shutdown(timeout=0.5)
+        except Exception:
+            pass
+    return new
+
+
+def _try_webhook(urls: Sequence[str]) -> List[BaseChannel]:
+    try:
+        return [WebhookChannel(urls)]
+    except Exception as exc:
+        _LOG.warning("configure_broadcast webhook 通道失败: %s", exc)
+        return []
+
+
+def _try_redis(url: str) -> List[BaseChannel]:
+    try:
+        return [RedisPubSubChannel(redis_url=url)]
+    except Exception as exc:
+        _LOG.warning("configure_broadcast redis 通道失败: %s", exc)
+        return []
+
+
+def _collect_subscribers(broadcaster: Broadcaster) -> List[Callable[[Dict[str, Any]], None]]:
+    handlers: List[Callable[[Dict[str, Any]], None]] = []
+    for ch in getattr(broadcaster, "channels", []):
+        if isinstance(ch, InProcessChannel):
+            handlers.extend(list(getattr(ch, "_subscribers", [])))
+    return handlers
+
+
+def reset_broadcaster(broadcaster: Optional[Broadcaster] = None, shutdown_old: bool = True) -> Broadcaster:
+    """测试辅助：重置全局广播器。"""
+    global _BROADCASTER
+    with _LOCK:
+        old = _BROADCASTER
+        _BROADCASTER = broadcaster if broadcaster is not None else _default_broadcaster()
+    if old is not None and shutdown_old and broadcaster is None:
+        try:
+            old.shutdown(timeout=0.3)
+        except Exception:
+            pass
+    return _BROADCASTER
+
+
+def subscribe_broadcast(handler: Callable[[Dict[str, Any]], None]) -> Callable[[Dict[str, Any]], None]:
+    """订阅进程内广播（返回 handler 本身，便于 unsubscribe）。"""
+    return get_broadcaster().subscribe(handler)
+
+
+def unsubscribe_broadcast(handler: Callable[[Dict[str, Any]], None]) -> bool:
+    return get_broadcaster().unsubscribe(handler)
+
+
+def broadcast_stats() -> Dict[str, Any]:
+    return get_broadcaster().stats()
+
+
+def broadcast_last_message() -> Optional[Dict[str, Any]]:
+    return get_broadcaster().last_message
+
+
+def broadcast_latency_percentile_ms(pct: float = 99.0) -> float:
+    return get_broadcaster().latency_percentile_ms(pct)
+
+
+def drain_broadcast(timeout: float = 2.0) -> bool:
+    return get_broadcaster().drain(timeout=timeout)
+
+
+def shutdown_broadcast(timeout: float = 1.0) -> None:
+    get_broadcaster().shutdown(timeout=timeout)
+
+
+def emit_post_commit_broadcast(
+    write_result: Any,
+    commit_ts: Optional[float] = None,
+    channel: str = DEFAULT_EVENT_CHANNEL,
+    broadcaster: Optional[Broadcaster] = None,
+    context: Optional[Dict[str, Any]] = None,
+) -> Optional[Dict[str, Any]]:
+    """**post-commit hook**：事务 COMMIT 之后触发广播。
+
+    永不抛异常（广播失败只记日志），永不阻塞（入队即返回）。
+    返回广播消息 dict（便于日志/测试），广播被关闭时返回 ``None``。
+    """
+    try:
+        message = build_broadcast_message(
+            write_result, commit_ts=commit_ts, channel=channel, context=context)
+        target = broadcaster or get_broadcaster()
+        message["submit_ts"] = time.time()
+        message["latency_submit_ms"] = round(
+            max(0.0, (message["submit_ts"] - message["commit_ts"]) * 1000.0), 3
+        )
+        target.submit(message)
+        return message
+    except Exception as exc:  # 铁律：广播侧任何异常都不能影响已提交事务
+        _LOG.error("world_sync.broadcast 触发广播异常已吞掉（事务已提交）: %s", exc)
+        return None
+
+
+def _call_context(args: Sequence[Any], kwargs: Dict[str, Any]) -> Dict[str, Any]:
+    """从 write_event_with_state 的调用入参提取 event_data / state_update 上下文。"""
+    ctx: Dict[str, Any] = {}
+    try:
+        names = ("event_data", "state_update")
+        for i, val in enumerate(args):
+            if i < len(names):
+                ctx[names[i]] = val
+        for nm in names:
+            if nm in kwargs and kwargs[nm] is not None:
+                ctx[nm] = kwargs[nm]
+    except Exception:
+        pass
+    return ctx
+
+
+def wrap_with_post_commit_broadcast(
+    write_fn: Callable[..., Any],
+    channel: str = DEFAULT_EVENT_CHANNEL,
+    broadcaster: Optional["Broadcaster"] = None,
+) -> Callable[..., Any]:
+    """把 M11b-2 的 ``write_event_with_state`` 包成「COMMIT 后广播」版本。
+
+    ``write_fn`` 返回即代表事务已 COMMIT（M11b-2 在单事务内提交后返回）；
+    若 ``write_fn`` 抛异常（回滚/冲突），**不广播**，异常原样上抛。
+    """
+    if not callable(write_fn):
+        raise TypeError("wrap_with_post_commit_broadcast(write_fn): 需要可调用对象")
+
+    def wrapper(*args: Any, **kwargs: Any) -> Any:
+        result = write_fn(*args, **kwargs)  # 事务在此完成 COMMIT（或抛异常回滚）
+        # COMMIT 已返回 —— 此刻记录 commit_ts，作为 <=300ms(P99) 广播延迟的起点
+        commit_ts = time.time()
+        emit_post_commit_broadcast(
+            result, commit_ts=commit_ts, channel=channel,
+            context=_call_context(args, kwargs),
+            broadcaster=broadcaster,
+        )
+        return result
+
+    try:
+        wrapper.__name__ = getattr(write_fn, "__name__", "write_event_with_state")
+        wrapper.__doc__ = getattr(write_fn, "__doc__", None)
+        wrapper.__module__ = getattr(write_fn, "__module__", "world_sync")
+    except Exception:
+        pass
+    setattr(wrapper, "_m11b3_broadcast_wrapped", True)
+    setattr(wrapper, "_m11b3_underlying", write_fn)
+    return wrapper
+
+
+def _refresh_env_reference(server_env: Any, wrapped: Callable[..., Any],
+                           only_if: Optional[Callable[..., Any]] = None) -> None:
+    """把包装后的 write_event_with_state 同步到 ServerEnv（三处接线之一）。"""
+    env = server_env
+    if env is None:
+        try:
+            from ahserver.serverenv import ServerEnv  # 宿主环境才有
+            env = ServerEnv()
+        except Exception:
+            env = None
+    if env is None:
+        return
+    try:
+        if only_if is None or getattr(env, "write_event_with_state", None) is only_if:
+            env.write_event_with_state = wrapped
+    except Exception as exc:  # pragma: no cover
+        _LOG.debug("world_sync.broadcast ServerEnv 刷新跳过: %s", exc)
+
+
+def is_broadcast_hook_installed() -> bool:
+    return bool(_HOOK_INSTALLED)
+
+
+def install_broadcast_hook(server_env: Any = None, force: bool = False) -> Callable[..., Any]:
+    """把广播 post-commit hook 装到 M11b-2 的 ``write_event_with_state`` 上。
+
+    同步刷新三处引用（实现模块属性 / 包命名空间 / ServerEnv 登记），保证宿主经
+    任何一条路径调用都会触发广播。幂等：重复调用只包一次（``force=True`` 强包）。
+    """
+    global _HOOK_INSTALLED
+    from . import pbl_runtime_tx_sqlor as _txs
+
+    original = getattr(_txs, "write_event_with_state")
+    if getattr(original, "_m11b3_broadcast_wrapped", False) and not force:
+        _HOOK_INSTALLED = True
+        _refresh_env_reference(server_env, original)
+        return original
+
+    wrapped = wrap_with_post_commit_broadcast(original)
+    _txs.write_event_with_state = wrapped
+
+    import sys as _sys
+    pkg = _sys.modules.get("world_sync")
+    if pkg is not None and getattr(pkg, "write_event_with_state", None) is original:
+        try:
+            pkg.write_event_with_state = wrapped  # type: ignore[attr-defined]
+        except Exception:
+            pass
+
+    _refresh_env_reference(server_env, wrapped, only_if=original)
+
+    _HOOK_INSTALLED = True
+    _LOG.info("world_sync.broadcast post-commit hook 已安装（异步广播，SLO=%.0fms P99）",
+              BROADCAST_LATENCY_SLO_MS)
+    return wrapped