董宏凯andCursor
99642fca6c
Backend:
- Conversation API at /api/maf-roundtable/conversations: create/list/detail,
paginated messages, atomic send+run start, owner-only isolation,
admin read-only diagnostic access
- SSE real-time event stream (GET /conversations/{id}/stream) backed by
ConversationEventHub + conversation_events store; multi-worker deployment doc
- Deterministic server-owned stage-3 dispatch contracts (dispatch_contracts.py):
routing resolved from trusted MAF deployment config instead of manager-chosen
agent ids; config.yaml keeps stage-3 routing deployment-owned
(default_manager_agent_id / seat ids / business_routes / max_seats_per_run)
- Conversation persistence (conversations/messages stores) + v2 migration script
- Tests: conversations, conversation stream, migration, phase3 conversation dispatch
Frontend:
- Vitest test infra for MAF (vitest.maf.config.ts + test:maf script) with
component tests for ApprovalCard/RunHeader/SessionsPage and use-maf-events
- Session form lib, flow-graph dialog / run-detail / sessions page updates,
routes and nav tweaks
- Docs: R0 legacy interaction baseline mapping, updated implementation plan
Co-authored-by: Cursor <cursoragent@cursor.com>
113 lines
3.9 KiB
Markdown
113 lines
3.9 KiB
Markdown
# MAF 会商 SSE 多 Worker 部署说明
|
||
|
||
> 状态：Phase 2 (R2) — 单 Worker 已就绪，多 Worker 需补 Redis Pub/Sub
|
||
> 更新时间：2026-08-03
|
||
|
||
---
|
||
|
||
## 1. 当前架构（单 Worker）
|
||
|
||
`ConversationEventHub`（`app/maf_roundtable/conversation_event_hub.py`）是一个
|
||
**进程内** 的 `asyncio.Event` 通知器：
|
||
|
||
- 每个 worker 进程有自己独立的 hub 实例（`app.state.maf_conversation_event_hub`）
|
||
- 事件写入者（`ConversationService.send_message`）在事务提交后调用 `hub.publish()`
|
||
- SSE 订阅者通过 `hub.subscribe()` 注册一个 `asyncio.Event`，被 `publish()` 唤醒后
|
||
从数据库查询最新事件（通知-only，不缓存事件内容）
|
||
- 心跳：15 秒超时 → `: heartbeat\n\n` SSE 注释帧
|
||
|
||
**适用场景：** `uvicorn --workers 1` / `gunicorn -w 1`，开发环境、小规模灰度。
|
||
|
||
---
|
||
|
||
## 2. 多 Worker 限制
|
||
|
||
当使用 `--workers 2` 或更多 worker 时：
|
||
|
||
| 问题 | 原因 | 影响 |
|
||
|------|------|------|
|
||
| **跨 Worker 不通知** | Worker A 写入事件后 `hub.publish()` 只通知 A 自己的订阅者；Worker B 的 SSE 客户端收不到实时推送 | SSE 流"卡住"直到客户端超时重连，重连后从 DB 重放才能补到事件 |
|
||
| **无跨进程共享状态** | `asyncio.Event` 是进程内对象，不能跨进程通信 | 需要 Redis Pub/Sub 或等价中间件 |
|
||
|
||
**在接入 Redis Pub/Sub 之前，禁止多 Worker 灰度上线。**
|
||
|
||
---
|
||
|
||
## 3. 健康检查建议
|
||
|
||
建议在 `GET /api/maf-roundtable/health` 中增加：
|
||
|
||
```json
|
||
{
|
||
"sse_hub_type": "in_memory",
|
||
"sse_multi_worker": false,
|
||
"sse_warning": "Single-worker only. Multi-worker requires Redis Pub/Sub."
|
||
}
|
||
```
|
||
|
||
当前版本已通过 `sse_multi_worker: false` 明确标注不支持多 Worker。
|
||
部署脚本应检查此项，在多 Worker 模式下发出告警。
|
||
|
||
---
|
||
|
||
## 4. Redis Pub/Sub 接入方案（后续阶段）
|
||
|
||
当需要多 Worker 时，替换 `ConversationEventHub` 为 Redis-backed 实现：
|
||
|
||
### 4.1 接口不变
|
||
|
||
```python
|
||
class ConversationEventHub(abc.ABC):
|
||
async def publish(self, conversation_id: str) -> None: ...
|
||
async def subscribe(self, conversation_id: str, ...) -> AsyncIterator[bool]: ...
|
||
```
|
||
|
||
### 4.2 Redis 实现
|
||
|
||
```python
|
||
class RedisConversationEventHub(ConversationEventHub):
|
||
def __init__(self, redis: Redis, *, heartbeat_interval: float = 15.0):
|
||
self._redis = redis
|
||
self._heartbeat_interval = heartbeat_interval
|
||
|
||
async def publish(self, conversation_id: str) -> None:
|
||
await self._redis.publish(f"maf:conv:{conversation_id}", "1")
|
||
|
||
async def subscribe(self, conversation_id: str, ...):
|
||
pubsub = self._redis.pubsub()
|
||
await pubsub.subscribe(f"maf:conv:{conversation_id}")
|
||
try:
|
||
while True:
|
||
msg = await asyncio.wait_for(
|
||
pubsub.get_message(ignore_subscribe_messages=True, timeout=...),
|
||
timeout=self._heartbeat_interval,
|
||
)
|
||
if msg is None:
|
||
yield False # heartbeat
|
||
else:
|
||
yield True # new events might be available
|
||
finally:
|
||
await pubsub.unsubscribe(f"maf:conv:{conversation_id}")
|
||
```
|
||
|
||
### 4.3 选择逻辑
|
||
|
||
```python
|
||
# deps.py
|
||
if config.database and config.database.redis_url:
|
||
hub = RedisConversationEventHub(create_redis(config.database.redis_url))
|
||
else:
|
||
hub = ConversationEventHub() # in-memory, single-worker
|
||
```
|
||
|
||
---
|
||
|
||
## 5. 部署检查清单
|
||
|
||
- [ ] 确认 `uvicorn --workers 1`（或等价单进程配置）
|
||
- [ ] 确认 SSE 端点 `GET /api/maf-roundtable/conversations/{id}/stream` 可访问
|
||
- [ ] 确认 nginx / 反向代理未缓冲 SSE（`X-Accel-Buffering: no` 生效）
|
||
- [ ] 确认心跳间隔 15s 足够长不会误触发代理超时（proxy_read_timeout 应 ≥ 60s）
|
||
- [ ] 确认 `after_seq` 重连不产生重复气泡（前端 reducer 按 seq 去重）
|
||
- [ ] 多 Worker 上线前：接入 Redis Pub/Sub 并更新 `sse_multi_worker: true`
|