feat(broker): 外置 Redis pub/sub——蓝绿跨实例 SSE/cancel 可达(bump 0.42.0)

用户量上升重评落地(DESIGN §7.0):web/broker.py 拆 LocalRunBroker(原进程内
实现,默认)/ RedisRunBroker 同接口双实现,ZCBOT_REDIS_URL env 开关。

- event:PUBLISH zcbot:ev:<tid> + 单条 async pubsub reader 路由本地订阅 queue
  (Redis 只管跨进程一跳,fan-out 最后一公里仍在进程内)
- done:SETEX 60s key + publish 双通道;订阅先挂 channel 再查 key,晚到不漏
- cancel:SETEX/GET/DEL key,loop chunk 间 poll,跨实例可达——蓝绿切换窗口
  「刷新丢直播 / 停止失灵」两个边缘消失;稳态双实例分流只差 nginx 配置
- 容错:启动 ping 不通 fail-fast(同 sandbox init 纪律);运行中失败 10s 节流
  log + 降级丢帧,run 不受影响;reader 1s 退避重连并重挂活跃 channel
- lifespan finally 加 broker.shutdown();sinks.py 解开对旧类名的 import
- tests/test_broker_redis.py 9 例(fakeredis:跨实例 emit/cancel、晚订阅 done、
  降级不抛、fail-fast);requirements 加 redis/fakeredis
- RUN.md:env 段 + 蓝绿节 + bootstrap(apt install redis-server)+ 故障兜底 3 行

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

DESIGN.md

@@ -127,7 +127,7 @@ Session = 消息列表,ORM 直写 PG `messages`(append-only,jsonb 存 LiteLLM
 
 **无感部署(蓝绿,0.41)**:生产双 systemd 实例(`ZCBOT_INSTANCE=blue/green`)+ nginx upstream 切流(对外仍 8765),取代单实例 restart 的 503 窗口。双实例并存的互踩由三件事消解:`tasks.run_owner`(0020)让 reaper 只收自己色、sandbox 容器名/label 带实例色各管各的、微信长轮询 PG advisory lock 选主;定时任务 claim 本就 SKIP LOCKED 天然安全。部署编排在 `deploy/update_bluegreen.sh`(流程/兜底见 RUN)。
 
-**broker 外置(Redis pub/sub)**:0.41 落地时 event/cancel broker 留在进程内,代价是切换窗口内"刷新看不到旧实例 run 直播 / 停止送达不到"两个边缘。**2026-07-06 重评(用户量已非个位数):由"搁置到真实投诉"转为"推荐实施"** —— ①部署时总有 in-flight run,窗口边缘从偶发变常态;②单进程逼近天花板后,最近的扩容手段是稳态双实例同时接流量,broker 外置是它的硬前提(否则 POST 与 SSE 落不同实例直播全瞎)。选 Redis 不选 PG LISTEN/NOTIFY(token 级 delta 全过主库太吵:NOTIFY 全局队列 + 8KB payload 限制,把实时路径耦到 PG)、不选 nginx sticky hash(upstream 变更时 hash 重排,恰在部署窗口失效,治标)。方案:`ZCBOT_REDIS_URL` env 开关,不设即回退进程内(dev 零影响);done 语义改查 DB run_status + done key(天然跨实例);cancel 改 Redis key poll。Redis 不解决的:run 本体仍绑进程(进程死 run 死,drain/reaper 不变)、线程池上限(调 `ZCBOT_RUN_MAX_WORKERS` 的事)。
+**broker 外置(Redis pub/sub,✅ 0.42)**:0.41 落地时 event/cancel broker 留在进程内,代价是切换窗口内"刷新看不到旧实例 run 直播 / 停止送达不到"两个边缘。2026-07-06 重评(用户量已非个位数)决定实施 —— ①部署时总有 in-flight run,窗口边缘从偶发变常态;②单进程逼近天花板后,最近的扩容手段是稳态双实例同时接流量,broker 外置是它的硬前提(否则 POST 与 SSE 落不同实例直播全瞎)。选 Redis 不选 PG LISTEN/NOTIFY(token 级 delta 全过主库太吵:NOTIFY 全局队列 + 8KB payload 限制,把实时路径耦到 PG)、不选 nginx sticky hash(upstream 变更时 hash 重排,恰在部署窗口失效,治标)。实现(`web/broker.py`,LocalRunBroker/RedisRunBroker 同接口鸭子类型):`ZCBOT_REDIS_URL` env 开关,不设即进程内(dev 零影响);event 走 `PUBLISH zcbot:ev:<tid>` + 单条 async pubsub reader 路由到本地订阅 queue(Redis 只管跨进程一跳,fan-out 最后一公里仍在进程内);done = SETEX key(60s)+ publish 双通道,订阅先挂 channel 再查 key 不漏;cancel = SETEX/GET/DEL key,loop 在 chunk 间 poll(localhost ~0.1ms)。容错纪律:启动 ping 不通 fail-fast(同 sandbox init),运行中失败 10s 节流 log + 降级丢帧,reader 1s 退避重连。Redis 不解决的:run 本体仍绑进程(进程死 run 死,drain/reaper 不变)、线程池上限(调 `ZCBOT_RUN_MAX_WORKERS` 的事)。
 
 ### 7.1 心智模型:Task 一等公民 + Dir 文件副视图
 
@@ -264,7 +264,7 @@ scheduled_jobs(§8.5)   channel_bindings(§8.7,判别列+JSONB)
 
 ### 8.4 运维监控 / 无感更新(监控 ✅ / 换版 design)
 
-**优雅 drain 已是单实例上限**(SIGTERM 拒新 run + 等收尾);先撞的瓶颈是线程池(每活跃 run 占 1 线程)。落地排序:① 轻量监控(显式 executor + 60s `[stats]` 周期日志——要历史峰值不是快照);② 按数据决策扩容;③ `--reload` 缩 503 窗。**不做监控界面**:运维健康是少数标量,日志够;业务分析走 DB SQL。界面阶梯:日志 → /v1/stats → Grafana → 只读 dashboard,现停第一级。无感换版已由蓝绿落地(0.41,见 §7.0);broker 外置 Redis 2026-07-06 重评为推荐实施(§7.0),扩容路径:调大线程池 → 稳态双实例分流(broker 外置后纯 nginx 配置动作)。
+**优雅 drain 已是单实例上限**(SIGTERM 拒新 run + 等收尾);先撞的瓶颈是线程池(每活跃 run 占 1 线程)。落地排序:① 轻量监控(显式 executor + 60s `[stats]` 周期日志——要历史峰值不是快照);② 按数据决策扩容;③ `--reload` 缩 503 窗。**不做监控界面**:运维健康是少数标量,日志够;业务分析走 DB SQL。界面阶梯:日志 → /v1/stats → Grafana → 只读 dashboard,现停第一级。无感换版已由蓝绿落地(0.41)、broker 外置 Redis 已实施(0.42,均见 §7.0);扩容路径:调大线程池 → 稳态双实例分流(broker 已外置,纯 nginx 配置动作)。
 
 ### 8.5 定时任务(✅ 2026-06-18)
 

PROGRESS.md

@@ -2,7 +2,7 @@
 
 > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。
 
-最后更新:2026-07-06(规模设定重评:broker 外置 Redis 转推荐实施,bump 0.41.2)
+最后更新:2026-07-06(broker 外置 Redis 落地,bump 0.42.0)
 
 ---
 
@@ -23,6 +23,7 @@
 
 ### 2026-07
 
+- **07-06 / 0.42.0 broker 外置 Redis(用户量上升重评落地)**:`web/broker.py` 拆 LocalRunBroker(原进程内实现)/ RedisRunBroker 同接口双实现,`ZCBOT_REDIS_URL` env 开关(不设 = 进程内,dev/单实例零影响)。Redis 版:event 走 PUBLISH + 单条 async pubsub reader 路由本地 queue,done = SETEX(60s)+publish 双通道(订阅先挂 channel 再查 key 不漏),cancel = key GET poll 跨实例可达——蓝绿切换窗口「刷新丢直播/停止失灵」两边缘消失,稳态双实例分流从此只差 nginx 配置。容错:启动 ping fail-fast、运行中降级丢帧(10s 节流 log)、reader 1s 退避重连。`tests/test_broker_redis.py` 9 例(fakeredis,含跨实例/晚订阅/降级);deploy bootstrap 加 redis-server 步骤。关键判断:排除 PG LISTEN/NOTIFY(token delta 耦主库)与 nginx sticky(部署窗 hash 重排),Redis 只管跨进程一跳、fan-out 最后一公里仍在进程内。
 - **07-06 / 0.41.1 蓝绿端口方案反转**:nginx 接管原对外端口 8765(客户端 / platform 嵌入零改动),实例挪 8766(blue)/ 8767(green) 只绑本机——对外契约(URL:8765)不动优先于"实例沿用旧端口"的省事;顺带把「切换窗口内对旧实例 run 点停止送达不到」补进已知边缘文档。
 - **07-06 / 0.41.0 蓝绿无感部署(RUN.md B 档落地)**:生产部署零 503 窗口 + 在跑对话 SSE 直播不断。`zcbot@blue/green` 双 systemd 实例(模板 unit + `deploy/instances/*.env`)+ nginx upstream 单行文件切流,编排在 `deploy/update_bluegreen.sh`(复用 `update.sh --prepare-only`;切流后轮询 DB 等旧色 run 自然清空再 stop——直接 stop 会被 uvicorn `timeout_graceful_shutdown=5` 掐 SSE)。双实例并存互踩三修:`tasks.run_owner`(0020)reaper 只收自己色、sandbox 容器名/label 带实例色各清各的、微信 ClawBot 长轮询 PG advisory lock 选主(detach 连接持锁,断连即让位);定时任务 claim 本就 SKIP LOCKED 免改。关键判断:broker 刻意不出进程(不上 Redis),接受"切换瞬间刷新的用户看不到旧实例直播"这一边缘。单实例部署 `ZCBOT_INSTANCE` 未设全部退回原行为,`update.sh` 保留不废。
 - **07-06 / 0.40.0 ppt 伪导出根治(三层)**:0.35.1 复发(模型照禁令里的"配方"装 cairosvg 手搓整页贴图导出)复盘——①平台层机检门落地(`core/pptx_guard.py` + loop 钩子:shell/run_python 新产出 .pptx 命中"≥80% 页整版位图 + 全 deck 零原生文本"即注入 `[产物机检 ERROR]` 当场逼返工;判定刻意保守,`tests/test_pptx_guard.py` 8 例);②禁令去菜谱:ppt/brief SKILL.md 否定式指令改"正面窄门"(唯一入口 + 报错就修/上报),删掉可照做的违规配方;③官方脚本 5 处 `pip install` hint 改环境说明(pptx_media/svg_preview/crop_images/fix_image_aspect/pptx_cli)——打印给 agent 的安装指令就是执行指令。详 DESIGN §8.9。

RUN.md

@@ -62,6 +62,12 @@
   # SMTP_PASSWORD=<授权码/应用专用密码,非登录密码>
   # SMTP_FROM=you@qq.com         # 可选,默认取 SMTP_USER
   # SMTP_FROM_NAME=总院科研辅助智能体   # 可选,发件人显示名,默"总院科研辅助智能体"(不暴露内部代号)
+  # broker 外置(DESIGN §7.0,蓝绿双实例跨进程 SSE/cancel):可选。设了 → event/cancel
+  # 走 Redis pub/sub(蓝绿两实例读同一份 .env,天然同 URL);不设 → 进程内 broker
+  # (dev / 单实例,零依赖不用起 redis)。设了但连不上 → 启动 fail-fast(报错见故障兜底)。
+  # 蓝绿部署强烈建议设:消掉切换窗口「刷新丢直播 / 停止失灵」两个边缘,也是将来稳态
+  # 双实例分流的前提。
+  # ZCBOT_REDIS_URL=redis://127.0.0.1:6379/0
   # 定时任务守护循环(DESIGN §8.5,随 web 进程起,plain-asyncio 仿 _disk_scanner):
   # ZCBOT_DISABLE_SCHEDULER=1    # 可选,整体关掉调度(对照 Claude Code CLAUDE_CODE_DISABLE_CRON)
   # ZCBOT_SCHEDULER_TICK_SECONDS=10   # 可选,扫描间隔,默 10s(只决定最坏延迟≤1tick,不影响会否漏)
@@ -321,7 +327,9 @@ sudo bash /opt/zcbot/deploy/update_bluegreen.sh
 
 脚本流程:`update.sh --prepare-only`(拉码 / pip / migration / build,全复用单实例脚本)→ 起空闲色实例 → healthz 验活(核对 `instance` 字段)→ 改写 upstream + nginx reload(**新请求全走新实例,零 503**)→ **轮询 DB 等旧色在跑 run 自然清空**(上限 `DRAIN_WAIT_SECONDS`,期间旧 SSE 继续直播不断流 —— 直接 stop 的话 uvicorn `timeout_graceful_shutdown=5` 会 5s 掐长连接)→ stop 旧实例(残留 run 走 SIGTERM drain / 协作式 cancel)→ 旧色 SIGKILL 残留的 running 行标 error → enable 新色 / disable 旧色(重启机器只拉起 live 色)。
 
-效果 vs 单实例 `update.sh`:**没有 503 窗口,在跑对话的 SSE 直播也不断**。已知边缘(接受,仅存在于切换后旧色 drain 的几分钟窗口内、且 run 恰好还在旧实例上):① 用户刷新页面 → SSE 连到新实例,broker 没有该 run 的事件,直播看不到(run 照常在旧实例跑完落库,完了刷新看历史);② 点「停止」→ cancel 信号进的是新实例的进程内 broker,旧实例收不到,等 run 自然收尾即可。两者根因同一个:event/cancel broker 是进程内的(`web/broker.py`)不跨实例。**2026-07-06 重评(用户量上升):broker 外置 Redis 已转为推荐实施**(消掉这两个边缘 + 为稳态双实例分流铺路,取舍与方案见 DESIGN §7.0);实施落地前此边缘仍在。
+效果 vs 单实例 `update.sh`:**没有 503 窗口,在跑对话的 SSE 直播也不断**。
+
+**broker 外置(`ZCBOT_REDIS_URL`,0.42 起,蓝绿强烈建议设)**:不设时 event/cancel broker 是进程内的(`web/broker.py`),切换后旧色 drain 的几分钟窗口内有两个边缘 —— ① 刷新页面 SSE 连到新实例、看不到旧实例上 run 的直播(run 照常落库,完了刷新看历史);② 点「停止」送达不到旧实例上的 run(等自然收尾)。`.env` 设 `ZCBOT_REDIS_URL=redis://127.0.0.1:6379/0`(两实例共用同一 .env,自动同源)后 event/cancel 走 Redis pub/sub,**这两个边缘消失**,SSE 订阅与停止按钮跨实例可达;顺带打开将来稳态双实例分流(nginx upstream 挂两行)的大门。取舍与方案见 DESIGN §7.0;redis 运行中挂掉只降级直播丢帧(10s 节流 log),run 不受影响,恢复后 1s 内自愈。
 
 蓝绿并存期的互踩问题已在代码层解决(单实例部署 `ZCBOT_INSTANCE` 未设,以下全部退回原行为):
 - **启动 reaper**:`tasks.run_owner`(migration 0020)记 run 归属实例,实例启动只收自己色(+ NULL 遗留)的孤儿 run,不误杀另一实例在跑的
@@ -346,6 +354,11 @@ sudo apt-get install -y nginx
 sudo cp deploy/nginx/zcbot.conf.example /etc/nginx/conf.d/zcbot.conf
 echo 'server 127.0.0.1:8766;' | sudo tee /etc/nginx/zcbot_upstream.conf
 
+# 3.5) 装 redis(broker 外置,强烈建议)+ .env 加一行;默认绑 127.0.0.1 即可
+sudo apt-get install -y redis-server
+redis-cli ping    # 应回 PONG
+echo 'ZCBOT_REDIS_URL=redis://127.0.0.1:6379/0' | sudo tee -a /opt/zcbot/.env
+
 # 4) 停旧单实例服务(优雅 drain,释放 8765)→ 清无实例 label 的旧 sandbox 容器
 #    → 起 blue(8766)→ nginx 接管 8765。停旧到 nginx 起来之间是本次唯一的中断窗口
 sudo systemctl disable --now zcbot
@@ -841,6 +854,9 @@ sudo xfs_quota -x -c "limit -p bhard=10g zcbot_<user_uuid>" /opt
 | 蓝绿切换后用户直播不动、但任务其实在跑 | 用户恰在切换窗口刷新页面,run 还在旧实例上而 SSE 连到了新实例(broker 进程内不跨实例)。run 照常跑完落库,等旧实例日志安静后让用户刷新看历史;非 bug,已知边缘 |
 | 蓝绿并存期微信 ClawBot 没反应 | 入站长轮询走 PG advisory lock 选主。两实例 journalctl 里找 `[wechat] inbound leader lock acquired`——都没有 → 检查 DB 连通;锁在已停实例上不可能(连接断锁即释),新实例最长 15s 接管 |
 | 蓝绿部署后旧色 sandbox 容器残留 | 正常:旧实例 stop 时 `shutdown_all` 只清自己色,失败/超时的靠下次该色启动兜底。手动清:`docker rm -f $(docker ps -aq -f label=zcbot.instance=<色>)` |
+| 启动报 `ZCBOT_REDIS_URL 已设但 Redis 连不上` | redis-server 没起 / URL 错。`systemctl status redis-server` + `redis-cli ping` 验;或从 `.env` 删掉该行回退进程内 broker(蓝绿切换边缘会回来,见「broker 外置」段) |
+| 日志刷 `[broker] redis publish failed`(10s 节流) | redis 运行中挂了。降级:直播丢帧、SSE 卡 ping,run 本身照常跑完落库;`systemctl restart redis-server` 恢复后 pubsub reader 1s 退避自动重连,无需重启 zcbot |
+| 蓝绿切换窗口刷新丢直播 / 停止失灵(设了 ZCBOT_REDIS_URL 仍出现) | 两实例 journalctl 找 `[broker] redis pub/sub enabled` —— 没有 = 该实例没读到 env(改完 .env 没 restart 实例);有但仍异常看是否伴随 publish failed(redis 本身故障) |
 | 长任务跑到一半停下、提示「已达单轮步数上限…回复『继续』可接着跑」 | **预期行为非崩溃**:单个 run 自主步数到 backstop(`config/models/*.yaml` 的 `max_iterations`,flash 120 / pro 150)就主动停,回 `[reached max iterations]`。直接回复「继续」即续跑。经常撞顶=任务确实大,可调高对应 variant 的 `max_iterations` 或换 pro |
 | 任务停下提示「连续 N 步无净产出…已自动停止」(回 `[stopped: no progress]`) | **空转熔断**:连续 `_STALL_LIMIT`(loop.py,默认 8)步所有 tool 都只返 `[Error]`/重复结果/被拦 = 没在推进,主动停以免空烧 token。说明模型卡死在某个错上——**换思路 / 补充信息再回复「继续」**,别原样重发(会再次撞同一墙) |
 

core/__init__.py

@@ -1,3 +1,3 @@
 # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。
 # 改版本只动这一行。
-__version__ = "0.41.2"
+__version__ = "0.42.0"

requirements.txt

@@ -27,6 +27,11 @@ croniter>=2.0
 segno>=1.6
 cryptography>=42.0
 
+# broker 外置(§7.0,蓝绿双实例跨进程 SSE/cancel):ZCBOT_REDIS_URL 设了才用,
+# 不设走进程内 broker(dev 不需要起 redis);fakeredis 供单测(无需真 redis 服务)
+redis>=5.0
+fakeredis>=2.23
+
 # §7 B 阶段: Storage 落 PG
 sqlalchemy>=2.0.0
 psycopg[binary]>=3.1.0

tests/test_broker_redis.py

@@ -0,0 +1,177 @@
+"""RedisRunBroker 单测(fakeredis,无需真 redis 服务)。
+
+覆盖跨进程语义的关键面:
+- publish → 订阅 queue 收到(经 pubsub reader 路由)
+- close 的 done key + publish 双通道:晚到订阅者(错过 publish)靠 key 立即收 done
+- start 清 done key(新 run 的订阅者不被上一轮 done 打断)
+- cancel key 的 request/is/clear 三态(跨"实例"可见:第二个 broker 实例读得到)
+- emit 在 redis 挂掉时降级不抛(丢帧,run 不受影响)
+
+用两个 broker 实例共享同一 FakeServer 模拟蓝绿两进程。
+"""
+import asyncio
+import unittest
+from uuid import uuid4
+
+import fakeredis
+import fakeredis.aioredis
+
+from web.broker import LocalRunBroker, RedisRunBroker, _make_broker
+
+
+def _make_pair(server):
+    """一个共享 FakeServer 的 RedisRunBroker(sync + async 两个客户端注入)。"""
+    sync_c = fakeredis.FakeRedis(server=server, decode_responses=True)
+    async_c = fakeredis.aioredis.FakeRedis(server=server, decode_responses=True)
+    return RedisRunBroker("redis://fake:6379/0", sync_client=sync_c, async_client=async_c)
+
+
+async def _drain_until(q: asyncio.Queue, timeout: float = 2.0):
+    return await asyncio.wait_for(q.get(), timeout=timeout)
+
+
+class TestRedisBroker(unittest.TestCase):
+    def _run(self, coro):
+        return asyncio.run(coro)
+
+    def test_emit_reaches_subscriber(self):
+        async def main():
+            server = fakeredis.FakeServer()
+            b = _make_pair(server)
+            b.bind_loop(asyncio.get_running_loop())
+            tid = uuid4()
+            q = b.subscribe(tid)
+            await asyncio.sleep(0.1)  # 等 channel 挂载
+            b.emit(tid, {"type": "text", "delta": "你好"})
+            ev = await _drain_until(q)
+            self.assertEqual(ev, {"type": "text", "delta": "你好"})
+            b.unsubscribe(tid, q)
+            await b.shutdown()
+        self._run(main())
+
+    def test_cross_instance_emit(self):
+        """模拟蓝绿:A 实例上的 run emit,B 实例上的 SSE 订阅者收到。"""
+        async def main():
+            server = fakeredis.FakeServer()
+            b_a = _make_pair(server)   # 跑 run 的实例
+            b_b = _make_pair(server)   # 挂 SSE 的实例
+            loop = asyncio.get_running_loop()
+            b_a.bind_loop(loop)
+            b_b.bind_loop(loop)
+            tid = uuid4()
+            q = b_b.subscribe(tid)
+            await asyncio.sleep(0.1)
+            b_a.emit(tid, {"type": "llm_start"})
+            ev = await _drain_until(q)
+            self.assertEqual(ev["type"], "llm_start")
+            await b_a.shutdown()
+            await b_b.shutdown()
+        self._run(main())
+
+    def test_late_subscriber_gets_done_via_key(self):
+        """close 之后才来的订阅者:publish 早已错过,靠 done key 立即断流。"""
+        async def main():
+            server = fakeredis.FakeServer()
+            b = _make_pair(server)
+            b.bind_loop(asyncio.get_running_loop())
+            tid = uuid4()
+            b.close(tid)                     # SETEX done + publish(无人订阅,publish 丢)
+            q = b.subscribe(tid)             # 晚到
+            ev = await _drain_until(q)
+            self.assertEqual(ev, {"type": "done"})
+            self.assertTrue(b.is_done(tid))
+            await b.shutdown()
+        self._run(main())
+
+    def test_start_clears_done(self):
+        async def main():
+            server = fakeredis.FakeServer()
+            b = _make_pair(server)
+            b.bind_loop(asyncio.get_running_loop())
+            tid = uuid4()
+            b.close(tid)
+            self.assertTrue(b.is_done(tid))
+            b.start(tid)                     # 新 run
+            self.assertFalse(b.is_done(tid))
+            q = b.subscribe(tid)
+            await asyncio.sleep(0.1)
+            self.assertTrue(q.empty())       # 不该被上一轮 done 打断
+            b.emit(tid, {"type": "run_start"})
+            ev = await _drain_until(q)
+            self.assertEqual(ev["type"], "run_start")
+            await b.shutdown()
+        self._run(main())
+
+    def test_cancel_cross_instance(self):
+        """A 实例(SSE/cancel 端点)request_cancel,B 实例(跑 run 的)poll 到。"""
+        async def main():
+            server = fakeredis.FakeServer()
+            b_a = _make_pair(server)
+            b_b = _make_pair(server)
+            b_a.bind_loop(asyncio.get_running_loop())
+            b_b.bind_loop(asyncio.get_running_loop())
+            tid = uuid4()
+            self.assertFalse(b_b.is_cancelled(tid))
+            b_a.request_cancel(tid)
+            self.assertTrue(b_b.is_cancelled(tid))    # 跨实例可见
+            b_b.clear_cancel(tid)
+            self.assertFalse(b_a.is_cancelled(tid))
+            await b_a.shutdown()
+            await b_b.shutdown()
+        self._run(main())
+
+    def test_emit_degrades_when_redis_down(self):
+        """redis 操作抛异常 → emit/is_cancelled 降级不抛(log 节流另测无必要)。"""
+        class Boom:
+            def __getattr__(self, name):
+                def _raise(*a, **k):
+                    raise ConnectionError("redis down")
+                return _raise
+
+        b = RedisRunBroker("redis://fake:6379/0", sync_client=Boom(),
+                           async_client=fakeredis.aioredis.FakeRedis(decode_responses=True))
+        tid = uuid4()
+        b.emit(tid, {"type": "text", "delta": "x"})       # 不抛
+        self.assertFalse(b.is_cancelled(tid))              # 降级 False
+        b.request_cancel(tid)                              # 不抛
+        b.close(tid)                                       # 不抛
+
+    def test_bind_loop_fail_fast_when_unreachable(self):
+        class Boom:
+            def ping(self):
+                raise ConnectionError("refused")
+
+        b = RedisRunBroker("redis://fake:6379/0", sync_client=Boom(),
+                           async_client=fakeredis.aioredis.FakeRedis(decode_responses=True))
+
+        async def main():
+            with self.assertRaises(RuntimeError):
+                b.bind_loop(asyncio.get_running_loop())
+        self._run(main())
+
+
+class TestFactory(unittest.TestCase):
+    def test_default_is_local(self):
+        import os
+        old = os.environ.pop("ZCBOT_REDIS_URL", None)
+        try:
+            self.assertIsInstance(_make_broker(), LocalRunBroker)
+        finally:
+            if old is not None:
+                os.environ["ZCBOT_REDIS_URL"] = old
+
+    def test_env_selects_redis(self):
+        import os
+        old = os.environ.get("ZCBOT_REDIS_URL")
+        os.environ["ZCBOT_REDIS_URL"] = "redis://127.0.0.1:6379/0"
+        try:
+            self.assertIsInstance(_make_broker(), RedisRunBroker)
+        finally:
+            if old is None:
+                del os.environ["ZCBOT_REDIS_URL"]
+            else:
+                os.environ["ZCBOT_REDIS_URL"] = old
+
+
+if __name__ == "__main__":
+    unittest.main()

web/app.py

@@ -1257,6 +1257,13 @@ def create_app() -> FastAPI:
                     except Exception as e:
                         print(f"[shutdown] sandbox shutdown_all error: {type(e).__name__}: {e}")
 
+            # broker 收尾(redis 版停 pubsub reader + 断连;local no-op)。放在 drain
+            # 之后 —— drain 期间 run 还要 emit/close。
+            try:
+                await broker.shutdown()
+            except Exception as e:
+                print(f"[shutdown] broker shutdown error: {type(e).__name__}: {e}")
+
             # drain 已 await inflight 收尾、run 线程退完;非阻塞关池(进程在退出,保守清理)
             run_executor.shutdown(wait=False)
 

web/broker.py

@@ -1,38 +1,50 @@
-"""RunBroker:in-process pub/sub,把 agent run 产生的 event fan-out 给所有 SSE 订阅者。
+"""RunBroker:pub/sub,把 agent run 产生的 event fan-out 给所有 SSE 订阅者。
 
 DESIGN §7 简化(0004 一并)—— 单活 run 形态下 run_id 是冗余的(同 task 同时
 最多 1 个活 run);broker 内部全用 task_id 索引,客户端只需要 task_id 即可
 订阅 / cancel,不再需要先拿 run_id。
 
-设计:
+两个实现,env `ZCBOT_REDIS_URL` 决定(import 时定,进程生命周期不变):
-- emit() 从工作线程调(agent.run 在 to_thread 跑),用 loop.call_soon_threadsafe
+- **LocalRunBroker**(默认,env 不设):进程内 asyncio queue,零依赖 —— 本地 dev /
-  桥到 asyncio queue;SSE generator await queue.get() 拉出来推流。
+  单实例部署用。多进程不共享:蓝绿并存期跨实例订阅/cancel 送达不到(RUN.md B 档边缘)。
-- 同一 task 多个订阅者(刷新页面 / 多 tab / 桌面+移动)— 每个订阅 1 个独立 queue。
+- **RedisRunBroker**(env 设了):event 走 Redis PUB/SUB(`zcbot:ev:<task_id>`),
-- run 结束 → broker.close(task_id) 给所有订阅者派一条 done;新订阅者(在 done
+  done/cancel 走带 TTL 的 key —— 蓝绿双实例跨进程可达,也是稳态双实例分流的前提
-  后到的)立即收到 done 并断流(不漏不挂)。
+  (DESIGN §7.0)。启动 ping 不通 fail-fast(同 sandbox init 纪律);运行中 publish
-- 同 task 起新 run → broker.start(task_id) 清掉 _done 标记;否则上一轮 done
+  失败 log + 丢帧(直播降级,run 不受影响)。
-  会让新订阅者立刻断流看不到流式。
-- 进程内单实例 / 多进程不共享 — 个人 SaaS 单 worker 够用;真要扩多 worker 再上 Redis。
-- 不持久化 event — messages 已落 PG,刷新页面走 GET /v1/tasks/{id}/messages 看历史;
-  真要"刷新继续看实时流"未来加 event log 表 + backfill。
 
-线程模型:
+公共语义(两实现一致):
-- broker.bind_loop(loop) 在 FastAPI startup 调一次,记录 asyncio loop 引用。
+- emit() 从工作线程调(agent.run 在 to_thread 跑);SSE generator await queue.get()。
-- emit() 调用方可能在任意线程;put_nowait 是 thread-unsafe(asyncio.Queue 设计前提
+- 同一 task 多个订阅者(刷新页面 / 多 tab)— 每个订阅 1 个独立 queue。
-  是单 loop),所以走 call_soon_threadsafe 跨回 loop 线程再 put。
+- run 结束 → close(task_id) 给所有订阅者派 done;晚到的订阅者立即收 done 断流
-- subscribe / unsubscribe / close / start 也都用 call_soon_threadsafe 包,避免 race
+  (local 用 _done set;redis 用 `zcbot:done:<tid>` 60s TTL key —— 主保护是 SSE
-  (实测 SSE generator 在 finally 里 unsubscribe,这个就在 loop 线程,直接调也行)。
+  端点先查 DB run_status,key 只兜"查完 DB 到订阅完成之间 run 恰好收尾"的窗口)。
+- 同 task 起新 run → start(task_id) 清 done 标记。
+- cancel:request_cancel 置位,AgentLoop 在 chunk 间 poll is_cancelled() 即退;
+  redis 版是 `zcbot:cancel:<tid>` key(GET ~0.1ms/次,localhost 可忽略),跨实例可达。
+- 不持久化 event — messages 已落 PG,刷新走 GET /messages 看历史;订阅建立前
+  emit 的帧两个实现都会丢(设计选择)。
 """
 from __future__ import annotations
 
 import asyncio
+import json
+import os
 import threading
+import time
 from collections import defaultdict
 from typing import Any, Optional
 from uuid import UUID
 
 
-class RunBroker:
+class LocalRunBroker:
+    """进程内实现(原 RunBroker)。
+
+    线程模型:
+    - bind_loop(loop) 在 FastAPI startup 调一次,记录 asyncio loop 引用。
+    - emit() 调用方可能在任意线程;put_nowait 是 thread-unsafe(asyncio.Queue 设计
+      前提是单 loop),所以走 call_soon_threadsafe 跨回 loop 线程再 put。
+    """
+
     def __init__(self) -> None:
         self._subs: dict[UUID, set[asyncio.Queue]] = defaultdict(set)
         # 已经发完 done 的 task — 后来订阅者直接收到 done,避免无限等
@@ -47,6 +59,9 @@ class RunBroker:
         """FastAPI startup 调一次。"""
         self._loop = loop
 
+    async def shutdown(self) -> None:
+        """lifespan finally 调;local 无外部资源,no-op。"""
+
     def start(self, task_id: UUID) -> None:
         """同 task 起新 run 时调:清 _done 标记,让新订阅者能看到流式。
         cancel flag 在 finally 里 clear_cancel 清,这里不动(避免擦掉刚刚 request_cancel 的请求)。
@@ -98,7 +113,7 @@ class RunBroker:
         return len(self._subs.get(task_id, set()))
 
     def total_subscribers(self) -> int:
-        """全局 SSE 订阅者总数(所有 task 累加),供 §8.4 并发监控用。"""
+        """本进程 SSE 订阅者总数(所有 task 累加),供 §8.4 并发监控用。"""
         return sum(len(s) for s in self._subs.values())
 
     def is_done(self, task_id: UUID) -> bool:
@@ -121,5 +136,243 @@ class RunBroker:
         self._cancel_flags.pop(task_id, None)
 
 
+class RedisRunBroker:
+    """Redis pub/sub 实现(蓝绿双实例跨进程,DESIGN §7.0)。
+
+    - event:`PUBLISH zcbot:ev:<task_id>`(JSON);单条 async pubsub reader 任务
+      把消息路由到本进程各订阅 queue(订阅表 _subs 仍在本地 —— fan-out 的"最后
+      一公里"还是进程内,Redis 只负责跨进程那一跳)。
+    - done:close 先 `SETEX zcbot:done:<tid> 60`再 publish done;subscribe 先挂
+      channel 再查 key,晚到订阅者不漏。
+    - cancel:`SETEX zcbot:cancel:<tid> 3600` / GET / DEL,跨实例可达。
+    - 容错:bind_loop ping 不通直接 raise(fail-fast);运行中 redis 操作失败
+      log(10s 节流)+ 降级(emit 丢帧、is_cancelled 返 False),run 不受影响;
+      reader 断线 1s 退避重连并重挂全部活跃 channel。
+    """
+
+    _EV_PREFIX = "zcbot:ev:"
+    _DONE_PREFIX = "zcbot:done:"
+    _CANCEL_PREFIX = "zcbot:cancel:"
+    _CTL_CHANNEL = "zcbot:ev:__ctl__"   # 常挂空 channel,保 pubsub 始终有订阅可 get_message
+    _DONE_TTL = 60
+    _CANCEL_TTL = 3600
+
+    def __init__(self, url: str, *, sync_client=None, async_client=None) -> None:
+        import redis as _redis
+        import redis.asyncio as _aredis
+        self._url = url
+        # sync client:emit/close/cancel 从工作线程调(redis-py 自带连接池,线程安全);
+        # start/request_cancel 也会在 loop 线程上调 —— localhost 单次 ~0.1ms,超时收紧
+        # 到 2s 防 redis 挂掉时把 event loop 卡死太久。
+        self._redis = sync_client if sync_client is not None else _redis.Redis.from_url(
+            url, decode_responses=True, socket_timeout=2.0, socket_connect_timeout=2.0,
+        )
+        self._aredis = async_client if async_client is not None else _aredis.from_url(
+            url, decode_responses=True,
+        )
+        self._pubsub = None
+        self._reader_task: Optional[asyncio.Task] = None
+        self._subs: dict[UUID, set[asyncio.Queue]] = defaultdict(set)
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self._last_err_log = 0.0
+
+    # ─────────────── helpers ───────────────
+
+    def _chan(self, task_id: UUID) -> str:
+        return f"{self._EV_PREFIX}{task_id}"
+
+    def _log_err(self, what: str, e: Exception) -> None:
+        """redis 故障时 emit/is_cancelled 每 chunk 都会撞,10s 节流防刷屏。"""
+        now = time.monotonic()
+        if now - self._last_err_log >= 10:
+            self._last_err_log = now
+            print(f"[broker] redis {what} failed: {type(e).__name__}: {e} (降级继续,10s 内同类不再报)")
+
+    # ─────────────── lifecycle ───────────────
+
+    def bind_loop(self, loop: asyncio.AbstractEventLoop) -> None:
+        """FastAPI startup 调一次:ping 验活(不通 fail-fast)+ 起 pubsub reader。"""
+        self._loop = loop
+        try:
+            self._redis.ping()
+        except Exception as e:
+            raise RuntimeError(
+                f"ZCBOT_REDIS_URL 已设但 Redis 连不上({type(e).__name__}: {e})——"
+                f"确认 redis-server 在跑 / URL 对,或删掉该 env 回退进程内 broker"
+            )
+        self._reader_task = loop.create_task(self._reader(), name="redis-broker-reader")
+        host = self._url.rsplit("@", 1)[-1]  # 不把密码打进日志
+        print(f"[broker] redis pub/sub enabled ({host})")
+
+    async def shutdown(self) -> None:
+        """lifespan finally 调:停 reader、断两个客户端。"""
+        if self._reader_task is not None:
+            self._reader_task.cancel()
+            try:
+                await self._reader_task
+            except (asyncio.CancelledError, Exception):
+                pass
+        if self._pubsub is not None:
+            try:
+                await self._pubsub.aclose()
+            except Exception:
+                pass
+        try:
+            await self._aredis.aclose()
+        except Exception:
+            pass
+        try:
+            self._redis.close()
+        except Exception:
+            pass
+
+    async def _reader(self) -> None:
+        """常驻:单条 pubsub 连接收所有活跃 channel 的消息,路由到本地订阅 queue。
+        断线 1s 退避重连,并把 _subs 里所有活跃 channel 重挂上(订阅状态不丢)。"""
+        while True:
+            try:
+                self._pubsub = self._aredis.pubsub()
+                channels = [self._CTL_CHANNEL] + [self._chan(t) for t in self._subs]
+                await self._pubsub.subscribe(*channels)
+                while True:
+                    msg = await self._pubsub.get_message(
+                        ignore_subscribe_messages=True, timeout=5.0
+                    )
+                    if msg is None or msg.get("type") != "message":
+                        continue
+                    ch = msg.get("channel") or ""
+                    if ch == self._CTL_CHANNEL:
+                        continue
+                    try:
+                        tid = UUID(ch[len(self._EV_PREFIX):])
+                        event = json.loads(msg["data"])
+                    except (ValueError, TypeError):
+                        continue  # 不认识的 channel / 坏 payload,丢
+                    for q in list(self._subs.get(tid, ())):
+                        q.put_nowait(event)
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                self._log_err("reader", e)
+                try:
+                    if self._pubsub is not None:
+                        await self._pubsub.aclose()
+                except Exception:
+                    pass
+                await asyncio.sleep(1)
+
+    # ─────────────── pub/sub ───────────────
+
+    def start(self, task_id: UUID) -> None:
+        try:
+            self._redis.delete(f"{self._DONE_PREFIX}{task_id}")
+        except Exception as e:
+            self._log_err("start/del-done", e)
+
+    def subscribe(self, task_id: UUID) -> asyncio.Queue:
+        """调用方:SSE handler(loop 线程)。queue 立即返回;channel 挂载与 done-key
+        补查异步完成 —— 挂载完成前 publish 的帧会丢(与 local 版"订阅前的帧丢"同类,
+        SSE 端点已先查过 DB run_status 兜大头)。"""
+        q: asyncio.Queue = asyncio.Queue()
+        first = not self._subs[task_id]
+        self._subs[task_id].add(q)
+
+        async def _setup() -> None:
+            try:
+                if first and self._pubsub is not None:
+                    await self._pubsub.subscribe(self._chan(task_id))
+                # 先挂 channel 再查 done key:close() 是先 SETEX 再 publish,
+                # 两边一交叉,晚到订阅者要么收到 publish、要么查到 key,不漏
+                if await self._aredis.exists(f"{self._DONE_PREFIX}{task_id}"):
+                    q.put_nowait({"type": "done"})
+            except Exception as e:
+                self._log_err("subscribe-setup", e)
+                q.put_nowait({"type": "done"})  # 保守断流,客户端自动重连
+
+        if self._loop is not None:
+            self._loop.create_task(_setup())
+        return q
+
+    def unsubscribe(self, task_id: UUID, q: asyncio.Queue) -> None:
+        subs = self._subs.get(task_id)
+        if subs is None:
+            return
+        subs.discard(q)
+        if not subs:
+            del self._subs[task_id]
+            ch = self._chan(task_id)
+
+            async def _unsub() -> None:
+                try:
+                    if self._pubsub is not None:
+                        await self._pubsub.unsubscribe(ch)
+                except Exception:
+                    pass  # reader 重连时按 _subs 重挂,残留订阅无害
+
+            if self._loop is not None:
+                self._loop.create_task(_unsub())
+
+    def emit(self, task_id: UUID, event: dict[str, Any]) -> None:
+        try:
+            self._redis.publish(
+                self._chan(task_id), json.dumps(event, ensure_ascii=False, default=str)
+            )
+        except Exception as e:
+            self._log_err("publish", e)  # 丢帧降级,run 不受影响
+
+    def close(self, task_id: UUID) -> None:
+        try:
+            self._redis.set(f"{self._DONE_PREFIX}{task_id}", "1", ex=self._DONE_TTL)
+        except Exception as e:
+            self._log_err("close/set-done", e)
+        self.emit(task_id, {"type": "done"})
+
+    def n_subscribers(self, task_id: UUID) -> int:
+        return len(self._subs.get(task_id, set()))
+
+    def total_subscribers(self) -> int:
+        """本进程的订阅数(蓝绿双实例各报各的,admin 面板看单实例)。"""
+        return sum(len(s) for s in self._subs.values())
+
+    def is_done(self, task_id: UUID) -> bool:
+        try:
+            return bool(self._redis.exists(f"{self._DONE_PREFIX}{task_id}"))
+        except Exception as e:
+            self._log_err("is_done", e)
+            return False
+
+    # ─────────────── cancel signaling ───────────────
+
+    def request_cancel(self, task_id: UUID) -> None:
+        try:
+            self._redis.set(f"{self._CANCEL_PREFIX}{task_id}", "1", ex=self._CANCEL_TTL)
+        except Exception as e:
+            self._log_err("request_cancel", e)
+
+    def is_cancelled(self, task_id: UUID) -> bool:
+        """AgentLoop 在 stream chunk 间 poll(工作线程);redis 挂了返 False 降级
+        (cancel 暂不可达,等价于 redis 前的跨实例行为,run 不受影响)。"""
+        try:
+            return self._redis.get(f"{self._CANCEL_PREFIX}{task_id}") is not None
+        except Exception as e:
+            self._log_err("is_cancelled", e)
+            return False
+
+    def clear_cancel(self, task_id: UUID) -> None:
+        try:
+            self._redis.delete(f"{self._CANCEL_PREFIX}{task_id}")
+        except Exception as e:
+            self._log_err("clear_cancel", e)
+
+
+def _make_broker():
+    """env `ZCBOT_REDIS_URL` 在 → Redis 版(蓝绿跨进程);不在 → 进程内版(dev/单实例)。
+    import 时定一次,与 ZCBOT_INSTANCE 同款读法。"""
+    url = os.getenv("ZCBOT_REDIS_URL", "").strip()
+    if url:
+        return RedisRunBroker(url)
+    return LocalRunBroker()
+
+
 # 进程内单例 — FastAPI lifespan 里 bind_loop;agent / sink / SSE handler 共享。
-broker = RunBroker()
+broker = _make_broker()

web/sinks.py

@@ -9,11 +9,9 @@ from __future__ import annotations
 from typing import Any
 from uuid import UUID
 
-from .broker import RunBroker
-
-
 class WebEventSink:
-    def __init__(self, broker: RunBroker, task_id: UUID) -> None:
+    def __init__(self, broker, task_id: UUID) -> None:
+        # broker: LocalRunBroker | RedisRunBroker(鸭子类型同接口,web/broker.py)
         self._broker = broker
         self._task_id = task_id