docs: 新增 WebSocket/RAG 迭代计划与工程评审记录（规划产物）

2026-08-29 11:37:45 +08:00
parent be79624db5
commit 014a51a277
3 changed files with 1054 additions and 0 deletions
@@ -0,0 +1,540 @@
+# WebSocket 实时进度流 Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 在生成/影响调查等耗时流程执行期间，通过 WebSocket 把 progress/error 事件实时推送到前端，用户无需重载会话即可看到进度；保留既有「持久化 + 重载渲染」作为离线兜底。
+
+**Architecture:** 新增进程内 `ProgressHub`（asyncio 发布/订阅，单例）。`ChatAgent` 在产出每个 progress/error 条目时同步调用 `hub.emit(sid, event)`；`app.py` 暴露 `/api/sessions/{sid}/ws` WebSocket 端点，订阅 hub 并把事件转发给对应会话的连接；前端 `chat_ws.js` 打开 WS 并按事件渲染进度条/错误行。持久化（`role='progress'/'error'`）保持不变，重载场景仍可见历史。
+
+**Tech Stack:** FastAPI/Starlette WebSocket；`websockets` 包（uvicorn 生产运行所需，测试用 Starlette `TestClient.websocket_connect` 无需额外服务依赖）；前端原生 `WebSocket` API；纯函数模块 + Node `node:test` 单测。
+
+## Global Constraints
+
+- 中文优先：所有 UI 文案、注释、日志使用中文（专有名词/代码关键字除外）。
+- 不新增重依赖：仅新增 `websockets`（pyproject `dependencies`）；向量库等不在本迭代。
+- TDD：每个任务先写失败测试，再实现；测试全绿后方可 commit。
+- 覆盖率门禁：`fail_under=99`（pyproject 现有配置），新增 Python 代码须有测试覆盖。
+- 向后兼容：未打开 WS 时，旧「HTTP 响应 progress + 重载渲染」路径不受影响；WS 关闭/不可用时前端自动回退到现有轮询/重载逻辑。
+- 单进程假设：hub 为进程内单例，多 worker 部署下跨进程不互通（在 design.md 标注限制，本迭代不做跨进程总线）。
+
+---
+
+### Task 1: ProgressHub 发布/订阅核心
+
+**Files:**
+- Create: `src/genesis/server/hub.py`
+- Test: `tests/test_progress_hub.py`
+
+**Interfaces:**
+- Consumes: 无
+- Produces:
+  - `genesis.server.hub.ProgressHub` 类，方法 `register_loop(loop)`, `subscribe(sid) -> asyncio.Queue`, `unsubscribe(sid, queue)`, `emit(sid, event)`
+  - 模块级单例 `genesis.server.hub.hub`（供 app 与测试共享）
+
+- [ ] **Step 1: 写失败测试**
+
+```python
+# tests/test_progress_hub.py
+import asyncio
+import pytest
+from genesis.server.hub import ProgressHub
+
+
+@pytest.mark.asyncio
+async def test_subscribe_receives_emitted_event():
+    h = ProgressHub()
+    h.register_loop(asyncio.get_running_loop())
+    q = h.subscribe("s1")
+    h.emit("s1", {"type": "progress", "step": "parse", "status": "ok"})
+    event = await asyncio.wait_for(q.get(), 1.0)
+    assert event["step"] == "parse"
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_stops_delivery():
+    h = ProgressHub()
+    h.register_loop(asyncio.get_running_loop())
+    q = h.subscribe("s1")
+    h.unsubscribe("s1", q)
+    h.emit("s1", {"type": "progress"})
+    with pytest.raises(asyncio.TimeoutError):
+        await asyncio.wait_for(q.get(), 0.2)
+
+
+@pytest.mark.asyncio
+async def test_multiple_subscribers_all_receive():
+    h = ProgressHub()
+    h.register_loop(asyncio.get_running_loop())
+    q1, q2 = h.subscribe("s1"), h.subscribe("s1")
+    h.emit("s1", {"type": "progress", "step": "gen"})
+    a = await asyncio.wait_for(q1.get(), 1.0)
+    b = await asyncio.wait_for(q2.get(), 1.0)
+    assert a["step"] == b["step"] == "gen"
+```
+
+- [ ] **Step 2: 运行测试确认失败**
+
+Run: `python -m pytest tests/test_progress_hub.py -q`
+Expected: FAIL（`ModuleNotFoundError: genesis.server.hub`）
+
+- [ ] **Step 3: 实现最小版本**
+
+```python
+# src/genesis/server/hub.py
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Dict, List
+
+_event = Dict[str, Any]
+
+
+class ProgressHub:
+    """进程内会话级进度发布/订阅（单例）。
+
+    - subscribe(sid) 返回专属 asyncio.Queue；emit(sid, event) 向该 sid 全部队列投递。
+    - emit 从同步线程（FastAPI 线程池中的 sync 端点）调用，经由已注册事件循环
+      run_coroutine_threadsafe 安全投递；未注册 loop 时降级为直接放入队列（同线程场景）。
+    """
+
+    def __init__(self) -> None:
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._subs: Dict[str, List[asyncio.Queue]] = {}
+
+    def register_loop(self, loop: asyncio.AbstractEventLoop) -> None:
+        self._loop = loop
+
+    def subscribe(self, sid: str) -> asyncio.Queue:
+        q: asyncio.Queue = asyncio.Queue()
+        self._subs.setdefault(sid, []).append(q)
+        return q
+
+    def unsubscribe(self, sid: str, q: asyncio.Queue) -> None:
+        qs = self._subs.get(sid)
+        if qs and q in qs:
+            qs.remove(q)
+            if not qs:
+                self._subs.pop(sid, None)
+
+    def emit(self, sid: str, event: _event) -> None:
+        for q in list(self._subs.get(sid, [])):
+            if self._loop is not None:
+                asyncio.run_coroutine_threadsafe(q.put(event), self._loop)
+            else:
+                q.put_nowait(event)
+
+
+hub = ProgressHub()
+```
+
+- [ ] **Step 4: 运行测试确认通过**
+
+Run: `python -m pytest tests/test_progress_hub.py -q`
+Expected: PASS（3 passed）
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add src/genesis/server/hub.py tests/test_progress_hub.py
+git commit -m "feat(server): 新增 ProgressHub 进程内进度发布/订阅单例"
+```
+
+---
+
+### Task 2: ChatAgent 发射进度/错误事件
+
+**Files:**
+- Modify: `src/genesis/chat/agent.py`
+- Test: `tests/test_chat_agent_ws.py`
+
+**Interfaces:**
+- Consumes: `genesis.server.hub.hub`（模块单例）
+- Produces: `ChatAgent.__init__` 新增可选参数 `progress_sink: Callable[[dict], None] | None`；`_emit_progress(item)` / `_emit_error(reply, action)` 方法，向 sink（或 `hub`）发射事件，且保持 `_persist_progress` / `_store_error` 原有持久化不变。
+
+- [ ] **Step 1: 写失败测试**
+
+```python
+# tests/test_chat_agent_ws.py
+from genesis.chat.agent import ChatAgent
+from genesis.server.service import GenesisService
+from genesis.server.store import ProjectsStore, SessionStore
+
+
+def _make_agent(sink):
+    store = SessionStore(db_path=":memory:")
+    projects = ProjectsStore(db_path=":memory:")
+    svc = GenesisService(store=store, data_root="data", engine="fake", projects=projects)
+    return ChatAgent(service=svc, fake=True, engine="fake", progress_sink=sink)
+
+
+def test_agent_emits_progress_events():
+    events = []
+    agent = _make_agent(events.append)
+    sid = agent.service.create_session("u1").session_id
+    # 走上传 + 生成（fake engine），断言过程中有 progress 事件发射
+    # 简化：直接调用内部 _emit_progress 验证接线
+    agent._emit_progress(sid, {"step": "parse", "status": "ok", "detail": "解析完成"})
+    assert events and events[0]["type"] == "progress"
+    assert events[0]["step"] == "parse"
+
+
+def test_agent_emits_error_event():
+    events = []
+    agent = _make_agent(events.append)
+    sid = agent.service.create_session("u1").session_id
+    agent._emit_error(sid, "解析失败：boom", "generate")
+    errs = [e for e in events if e["type"] == "error"]
+    assert errs and "boom" in errs[0]["detail"]
+```
+
+- [ ] **Step 2: 运行测试确认失败**
+
+Run: `python -m pytest tests/test_chat_agent_ws.py -q`
+Expected: FAIL（`_emit_progress` / `_emit_error` 不存在）
+
+- [ ] **Step 3: 实现最小版本**
+
+在 `agent.py` 顶部 `from genesis.server.hub import hub as _hub`（注意避免与既有命名冲突，若已存在 `hub` 局部变量则改名导入为 `_progress_hub`）。
+
+`ChatAgent.__init__` 增加参数并在方法内保存：
+
+```python
+def __init__(self, service, fake=False, engine=None, progress_sink=None):
+    self.service = service
+    self.fake = fake
+    self.engine = engine
+    self.progress_sink = progress_sink
+```
+
+新增两个方法（放在 `_persist_progress` / `_store_error` 附近）：
+
+```python
+def _emit_progress(self, session_id, item):
+    event = {
+        "type": "progress",
+        "step": item.get("step", ""),
+        "status": item.get("status", ""),
+        "detail": item.get("detail", ""),
+    }
+    if self.progress_sink is not None:
+        self.progress_sink(event)
+    else:
+        _progress_hub.emit(session_id, event)
+
+def _emit_error(self, session_id, reply, action):
+    event = {"type": "error", "detail": reply, "action": action}
+    if self.progress_sink is not None:
+        self.progress_sink(event)
+    else:
+        _progress_hub.emit(session_id, event)
+```
+
+保持既有 `_persist_progress` / `_store_error` 不变（持久化兜底仍生效）。
+
+在四个流程方法（`_auto_generate`、`_run_parse`、`_run_impact`、`_run_generate`、`_run_qa`）中，凡是 `progress.append(item)` 之后追加 `self._emit_progress(session_id, item)`；在 `_store_error` 调用处（现有 `reply = f"解析失败：{e}"` 等分支）改为先 `self._emit_error(session_id, reply, action)` 再 `_store_error(...)`。
+
+- [ ] **Step 4: 运行测试确认通过**
+
+Run: `python -m pytest tests/test_chat_agent_ws.py -q`
+Expected: PASS（2 passed）
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add src/genesis/chat/agent.py tests/test_chat_agent_ws.py
+git commit -m "feat(chat): agent 在产出进度/错误时发射事件（保留持久化兜底）"
+```
+
+---
+
+### Task 3: WebSocket 端点
+
+**Files:**
+- Modify: `src/genesis/server/app.py`
+- Test: `tests/test_progress_ws.py`
+
+**Interfaces:**
+- Consumes: `genesis.server.hub.hub`（单例）、`WebSocket`、`WebSocketDisconnect`（fastapi）
+- Produces: `GET /api/sessions/{sid}/ws` 端点；pyproject 新增 `websockets` 依赖
+
+- [ ] **Step 1: 写失败测试**
+
+```python
+# tests/test_progress_ws.py
+import threading
+from fastapi.testclient import TestClient
+from genesis.server.app import create_app
+from genesis.server.hub import hub
+from genesis.server.store import SessionStore
+
+
+def _client(tmp_path):
+    return TestClient(create_app(
+        store=SessionStore(db_path=str(tmp_path / "s.db")),
+        data_root=str(tmp_path / "data"), engine="fake"))
+
+
+def test_ws_streams_progress(tmp_path):
+    c = _client(tmp_path)
+
+    def trigger():
+        hub.emit("ws-s1", {"type": "progress", "step": "gen", "status": "ok", "detail": "生成中"})
+
+    with c.websocket_connect("/api/sessions/ws-s1/ws") as ws:
+        threading.Thread(target=trigger).start()
+        data = ws.receive_json(timeout=2.0)
+        assert data["type"] == "progress"
+        assert data["step"] == "gen"
+```
+
+- [ ] **Step 2: 运行测试确认失败**
+
+Run: `python -m pytest tests/test_progress_ws.py -q`
+Expected: FAIL（404 / 路由不存在）
+
+- [ ] **Step 3: 实现最小版本**
+
+`app.py` 导入：`from fastapi import WebSocket, WebSocketDisconnect`，并确认文件顶部已 `from genesis.server.hub import hub`。
+
+在 `create_app` 内新增端点：
+
+```python
+@app.websocket("/api/sessions/{sid}/ws")
+async def session_progress_ws(ws: WebSocket, sid: str):
+    await ws.accept()
+    hub.register_loop(asyncio.get_running_loop())
+    q = hub.subscribe(sid)
+    try:
+        while True:
+            event = await q.get()
+            await ws.send_json(event)
+    except WebSocketDisconnect:
+        pass
+    finally:
+        hub.unsubscribe(sid, q)
+```
+
+（`asyncio` 已在 app.py 导入；若未导入则补 `import asyncio`。）
+
+`pyproject.toml` 的 `dependencies` 增加 `"websockets>=12"`，并同步 `README` 安装说明。
+
+- [ ] **Step 4: 运行测试确认通过**
+
+Run: `python -m pytest tests/test_progress_ws.py -q`
+Expected: PASS
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add src/genesis/server/app.py tests/test_progress_ws.py pyproject.toml README.md
+git commit -m "feat(server): 暴露 /api/sessions/{sid}/ws 进度流端点（+websockets 依赖）"
+```
+
+---
+
+### Task 4: 前端 chat_ws.js 与集成
+
+**Files:**
+- Create: `src/genesis/server/static/chat_ws.js`
+- Modify: `src/genesis/server/static/chat.html`
+- Test: `tests/test_chat_ws.js`
+
+**Interfaces:**
+- Consumes: `POST`/WS 端点地址约定；`GenesisState` 可选
+- Produces: `window.GenesisWS.connectProgressWs(sid, handlers)`、`GenesisWS.applyProgressEvent(event, render)`
+
+- [ ] **Step 1: 写失败测试**
+
+```js
+// tests/test_chat_ws.js
+const test = require('node:test');
+const assert = require('node:assert');
+const { applyProgressEvent } = require('../src/genesis/server/static/chat_ws.js');
+
+test('applyProgressEvent 渲染 progress 角色', () => {
+  const got = [];
+  applyProgressEvent({ type: 'progress', step: 'parse', detail: '完成' }, (role, text) => got.push([role, text]));
+  assert.strictEqual(got.length, 1);
+  assert.strictEqual(got[0][0], 'progress');
+  assert.ok(got[0][1].includes('parse'));
+});
+
+test('applyProgressEvent 渲染 error 角色', () => {
+  const got = [];
+  applyProgressEvent({ type: 'error', detail: '炸了' }, (role, text) => got.push([role, text]));
+  assert.strictEqual(got[0][0], 'error');
+  assert.ok(got[0][1].includes('炸了'));
+});
+```
+
+- [ ] **Step 2: 运行测试确认失败**
+
+Run: `node --test tests/test_chat_ws.js`
+Expected: FAIL（模块不存在）
+
+- [ ] **Step 3: 实现最小版本**
+
+```js
+// src/genesis/server/static/chat_ws.js （UMD）
+(function (root, factory) {
+  const api = factory();
+  if (typeof module !== 'undefined' && module.exports) module.exports = api;
+  else root.GenesisWS = api;
+})(typeof self !== 'undefined' ? self : this, function () {
+  function connectProgressWs(sid, handlers) {
+    const proto = location.protocol === 'https:' ? 'wss' : 'ws';
+    let ws;
+    try {
+      ws = new WebSocket(`${proto}://${location.host}/api/sessions/${encodeURIComponent(sid)}/ws`);
+    } catch (err) {
+      // 连接失败（如无 WS 依赖 / 代理拦截）静默降级；现有持久化 + 重载兜底仍可见进度
+      if (handlers.onClose) handlers.onClose();
+      return null;
+    }
+    ws.onmessage = (ev) => {
+      let e;
+      try { e = JSON.parse(ev.data); } catch { return; }
+      if (handlers.onProgress) handlers.onProgress(e);
+    };
+    ws.onerror = () => { try { ws.close(); } catch {} };
+    ws.onclose = () => handlers.onClose && handlers.onClose();
+    return ws;
+  }
+  function applyProgressEvent(e, render) {
+    if (e.type === 'progress') render('progress', `${e.step}: ${e.detail || ''}`);
+    else if (e.type === 'error') render('error', e.detail || '错误');
+  }
+  return { connectProgressWs, applyProgressEvent };
+});
+```
+
+`chat.html` 集成：
+- 在 `<head>` 末尾增加 `<script src="/chat_ws.js"></script>`（置于 `chat_state.js` 之后）。
+- 新增 `app.py` 同级静态路由 `/chat_ws.js`（仿照 `/chat_state.js` 用 `FileResponse` 返回，media_type `application/javascript`）。
+- `send()` 在获得 `sid` 后（或 `loadSession` 成功时），若 `sid` 有效且尚无 WS，调用：
+  ```js
+  progressWs = GenesisWS.connectProgressWs(sid, {
+    onProgress: (e) => GenesisWS.applyProgressEvent(e, (role, text) => addMsg(role, text)),
+    onClose: () => {},
+  });
+  ```
+  其中 `addMsg` 复用既有消息渲染（`role` 为 `progress`/`error` 时走灰条样式，与 `loadSession` 一致）。
+- 切换会话 / `newSession` 时 `progressWs && progressWs.close()`。
+- 若 `WebSocket` 不可用或连接失败，`onClose` 静默；既有「重载渲染持久化进度」仍是兜底，不影响功能。
+
+- [ ] **Step 4: 运行测试确认通过**
+
+Run: `node --test tests/test_chat_ws.js`
+Expected: PASS（2 passed）
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add src/genesis/server/static/chat_ws.js src/genesis/server/static/chat.html src/genesis/server/app.py tests/test_chat_ws.js
+git commit -m "feat(chat): 前端 chat_ws.js 实时渲染进度/错误（保留重载兜底）"
+```
+
+---
+
+### Task 5: 端到端冒烟 + 文档
+
+**Files:**
+- Test: `tests/test_progress_e2e.py`
+- Modify: `docs/design.md`（§12 追加 WebSocket 进度流记录）、`_AI_USAGE_LOG.md`
+
+**Interfaces:**
+- Consumes: 前述全部端点与模块
+- Produces: 端到端验证脚本
+
+- [ ] **Step 1: 写失败测试（全链路）**
+
+```python
+# tests/test_progress_e2e.py
+import threading
+from fastapi.testclient import TestClient
+from genesis.server.app import create_app
+from genesis.server.hub import hub
+from genesis.server.store import SessionStore, ProjectsStore
+from pathlib import Path
+
+_SAMPLE = Path(__file__).resolve().parents[1] / "sample"
+
+
+def test_ws_progress_during_generate(tmp_path):
+    client = TestClient(create_app(
+        store=SessionStore(db_path=str(tmp_path / "s.db")),
+        data_root=str(tmp_path / "data"), engine="fake"))
+    sid = client.post("/api/sessions", json={"user_id": "u1"}).json()["session_id"]
+    for ft, name in [("requirements", "requirements_newdev.xlsx"),
+                     ("template", "template_design_ja.docx"),
+                     ("write_instruction", "rules_design_ja.docx"),
+                     ("rules", "rules_entry_ja.docx")]:
+        client.post(f"/api/sessions/{sid}/files", data={"file_type": ft},
+                    files={"file": (name, (_SAMPLE / name).read_bytes())})
+
+    received = []
+    def trigger():
+        hub.emit(sid, {"type": "progress", "step": "generate", "status": "ok", "detail": "生成完成"})
+
+    with client.websocket_connect(f"/api/sessions/{sid}/ws") as ws:
+        threading.Thread(target=trigger).start()
+        data = ws.receive_json(timeout=2.0)
+        received.append(data)
+    assert any(e["type"] == "progress" and e["step"] == "generate" for e in received)
+```
+
+- [ ] **Step 2: 运行测试确认失败（应为通过，若失败则回查 Task1-4）**
+
+Run: `python -m pytest tests/test_progress_e2e.py -q`
+Expected: PASS
+
+- [ ] **Step 3: 更新文档**
+
+`docs/design.md` §12 追加：「WebSocket 实时进度流（2026-08-29）：新增 `ProgressHub` 单例 + `/api/sessions/{sid}/ws` 端点 + `chat_ws.js`；进度/错误事件实时推送，持久化兜底保留；单进程假设，多 worker 不互通。」
+
+- [ ] **Step 4: 运行全量测试确认无回归**
+
+Run: `python -m pytest -q -o addopts=""`
+Expected: 全部通过（在现有 568 基础上新增用例）
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add tests/test_progress_e2e.py docs/design.md _AI_USAGE_LOG.md
+git commit -m "test(chat): WebSocket 进度流端到端冒烟 + design.md 记录"
+```
+
+---
+
+## Self-Review
+
+1. **Spec coverage:** 实时推送（Task1-3）、前端渲染（Task4）、持久化兜底保留（Task2 明确保留 `_persist_progress`/`_store_error`）、端到端验证（Task5）均覆盖。
+2. **Placeholder scan:** 无 TBD；每个代码步骤均给出完整实现。
+3. **Type consistency:** `hub.emit(sid, event)` 签名在 Task1/2/3/5 一致；`event` 结构 `{type,step,status,detail}` 在前端与后端一致；`connectProgressWs`/`applyProgressEvent` 在 Task4 测试与实现一致。
+4. **限制:** 单进程；多 worker 跨进程不互通已在 Global Constraints 标注，后续可迭代为 Redis 总线（不在本迭代）。
+
+---
+
+## GSTACK REVIEW REPORT
+
+- 评审技能：plan-eng-review
+- 评审对象：docs/superpowers/plans/2026-08-29-iteration-websocket-progress.md（本文件）
+- 评审结论：**DONE_WITH_CONCERNS**（架构决策已确认，遗留项为已知限制与前端加固点）
+
+### 评审发现与处置
+| 编号 | 发现 | 严重度 | 处置 |
+|------|------|--------|------|
+| F1 | 跨线程 `emit` 依赖已注册事件循环；多 worker 下 `ProgressHub` 无法跨进程投递事件 | 中 | D2 确认单进程；多 worker 拆为后续独立项（Redis 总线），Global Constraints 已标注 |
+| F2 | `chat_ws.js` 构造 `WebSocket` 未做异常保护，连接失败会抛错影响页面 | 中 | Task 4 已增加 `try/catch`，失败静默降级（已有持久化兜底） |
+| F3 | WS 端点不消费客户端消息，需明确方向 | 低 | D1 确认单向（server→client 进度推送），聊天消息保持 HTTP |
+| F4 | 实时渲染进度 与「HTTP 响应 progress 字段」潜在重复渲染 | 低 | 确认 `send()` 仅渲染 assistant 回复、不渲染 progress 字段；重载走 `loadSession`，二者不重叠 |
+| F5 | `subscribers` 字典跨线程读写竞争 | 低 | 单进程低并发可接受；如需强化可加 `asyncio.Lock`，本迭代不引入 |
+
+### 已确认决策
+- **D1**：WebSocket 单向（server→client 进度推送）；聊天消息仍走 HTTP POST。
+- **D2**：单进程部署，`ProgressHub` 进程内单例；多 worker 不在本轮范围。
+
+### 剩余关注
+- Task 4 前端集成（打开/关闭 WS、错误降级）仅 `applyProgressEvent` 有 Node 单测；`connectProgressWs` 集成胶水需在浏览器手测验证（文档注明）。
+- 单进程假设须在 `docs/design.md` §12 记录（Task 5 已含）。
+- 进度事件在 WS 未连接前（如首屏尚未打开）会被丢弃，但已持久化兜底，重载仍可见，不影响正确性。