diff --git a/_AI_USAGE_LOG.md b/_AI_USAGE_LOG.md index 0829e3b..b441a14 100644 --- a/_AI_USAGE_LOG.md +++ b/_AI_USAGE_LOG.md @@ -138,3 +138,4 @@ | 2026-08-28 15:33 | 反馈迭代 | Task2 删除 header h1,侧边栏品牌名加副标题「概要设计书做成Agent」 | src/genesis/server/static/chat.html | deepseek-chat | | 2026-08-28 15:33 | 反馈迭代 | Task3 折叠态切换按钮移至左下角避免遮挡 logo + 会话列表底部留白 | src/genesis/server/static/chat.html | deepseek-chat | | 2026-08-29 14:20 | 测试验证 | 前端审计修复(docs/frontend_audit_plan.md,经 plan-eng-review 评审定稿):静态抽取 chat_state.js 纯函数模块 + Node 单测(11 passed);chat.html 统一 currentProject→draftProject、新会话继承已选项目(§4.2)、loadProjects 自动绑定、上传区显隐/类型映射/抽屉脏检测改用 GenesisState、loadSession 按 role 渲染 progress/error、加 h1 标题;app.py 预览改 HTMLResponse(返回渲染 HTML 而非 {html} JSON)、新增 /chat_state.js 静态路由、预览缺失返回 404(RESULT_NOT_FOUND);agent.py 新增 _persist_progress/_store_error 将进度与错误以 role 入库(重载可见);pytest 新增 test_frontend_audit_fixes.py(5) 并修正 test_server_api 中旧的 JSON 契约断言;全量 pytest 568 passed / 99.06% 达标 | docs/frontend_audit_plan.md; src/genesis/server/static/chat_state.js; tests/test_chat_state.js; src/genesis/server/static/chat.html; src/genesis/server/app.py; src/genesis/chat/agent.py; tests/test_frontend_audit_fixes.py; tests/test_server_api.py; _AI_USAGE_LOG.md | hy3-free | +| 2026-08-29 14:40 | 架构设计 | 将前端审计 §4 剩余两项(WebSocket 实时进度流、影响调查 RAG 化)拆为两个独立迭代并产出实施计划:docs/superpowers/plans/2026-08-29-iteration-websocket-progress.md(ProgressHub 单例 + /api/sessions/{sid}/ws 端点 + chat_ws.js,保留持久化兜底)、docs/superpowers/plans/2026-08-29-iteration-impact-rag.md(genesis/rag: Embedder/FakeEmbedder + RagStore SQLite 向量 + ImpactRAG + ImpactAgent.use_rag 默认关闭);均遵循 writing-plans 格式(TDD/无占位/频繁提交/逐任务验收) | docs/superpowers/plans/2026-08-29-iteration-websocket-progress.md; docs/superpowers/plans/2026-08-29-iteration-impact-rag.md; _AI_USAGE_LOG.md | hy3-free | diff --git a/docs/superpowers/plans/2026-08-29-iteration-impact-rag.md b/docs/superpowers/plans/2026-08-29-iteration-impact-rag.md new file mode 100644 index 0000000..f4c8536 --- /dev/null +++ b/docs/superpowers/plans/2026-08-29-iteration-impact-rag.md @@ -0,0 +1,513 @@ +# 影响调查 RAG 化 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:** 在影响调查(impact investigation)阶段,对项目的既有系统源码/设计文档/规则做切块与向量检索(RAG),把与当前需求/表最相关的片段作为上下文喂给 LLM,提升受影响模块/字段的召回率;在无法检索或无嵌入能力时回退到现有确定性交叉引用(DesignReference),保证行为不退化。 + +**Architecture:** 新增轻量本地 RAG 栈——`Embedder`(优先复用推理引擎的 embed 能力,缺省用确定性的词袋 FakeEmbedder 以保证测试与离线运行)、`RagStore`(SQLite 持久化 chunks + 余弦检索,复用项目既有 SQLite 习惯,不引入 chromadb/faiss 等新重依赖)、`ImpactRAG`(index/retrieve)。`ImpactAgent` 增加 `use_rag` 开关:开启时把检索到的 top-k 片段拼入影响调查 prompt;关闭时走原路径。`GenesisService` 透传 `use_rag`,默认 `False`(向后兼容)。 + +**Tech Stack:** Python `sqlite3` + `json`(向量存储,无新依赖);现有 `inference` 引擎抽象(取 embed);`node:test` 仅用于前端(本迭代无前端改动,故无 Node 测试)。 + +## Global Constraints + +- 中文优先:所有 UI 文案、注释、日志使用中文(专有名词/代码关键字除外)。 +- 不新增重依赖:RAG 存储用标准库 `sqlite3` + `json`;Embedder 优先复用推理引擎,缺省 `FakeEmbedder`(纯标准库)。禁止引入 chromadb / faiss / sentence-transformers 等重依赖。 +- TDD:每个任务先写失败测试再实现;全绿方可 commit。 +- 覆盖率门禁:`fail_under=99`(pyproject 现有),新增 Python 代码须有测试覆盖。 +- 向后兼容:`use_rag` 默认 `False`;未开启时影响调查输出与现状完全一致(DesignReference 路径不变)。 +- 增量索引:同一 scope 重复 index 先清后写,避免重复累积。 + +--- + +### Task 1: Embedder 抽象(含离线 FakeEmbedder) + +**Files:** +- Create: `src/genesis/rag/embeddings.py` +- Test: `tests/test_rag_embeddings.py` + +**Interfaces:** +- Consumes: `genesis.inference`(可选,用于真实 embed) +- Produces: + - `genesis.rag.embeddings.Embedder` 协议:`embed(texts: list[str]) -> list[list[float]]` + - `genesis.rag.embeddings.FakeEmbedder`(词袋确定性,dim=64,归一化) + - `genesis.rag.embeddings.get_embedder(engine) -> Embedder`(有 embed 能力用真实,否则 FakeEmbedder) + +- [ ] **Step 1: 写失败测试** + +```python +# tests/test_rag_embeddings.py +from genesis.rag.embeddings import FakeEmbedder, get_embedder + + +def test_fake_embedder_deterministic(): + e = FakeEmbedder() + a = e.embed(["OrderController 创建订单"])[0] + b = e.embed(["OrderController 创建订单"])[0] + assert a == b + + +def test_fake_embedder_similar_closer_than_unrelated(): + e = FakeEmbedder() + base = e.embed(["OrderController 处理创建订单请求"])[0] + sim = e.embed(["OrderController 保存订单到数据库"])[0] + dif = e.embed(["用户登录认证模块"])[0] + import math + def cos(x, y): + dot = sum(p * q for p, q in zip(x, y)) + nx = math.sqrt(sum(p * p for p in x)); ny = math.sqrt(sum(q * q for q in y)) + return dot / (nx * ny or 1.0) + assert cos(base, sim) > cos(base, dif) + + +def test_get_embedder_fake_engine_returns_fake(): + e = get_embedder("fake") + assert e.__class__.__name__ == "FakeEmbedder" +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `python -m pytest tests/test_rag_embeddings.py -q` +Expected: FAIL(`ModuleNotFoundError`) + +- [ ] **Step 3: 实现最小版本** + +```python +# src/genesis/rag/embeddings.py +from __future__ import annotations + +import hashlib +import math +from typing import List, Protocol + + +class Embedder(Protocol): + def embed(self, texts: List[str]) -> List[List[float]]: ... + + +_DIM = 64 + + +def _tokenize(text: str) -> List[str]: + # 简易分词:去标点、按空白与驼峰边界切分,转小写 + toks = [] + cur = "" + for ch in text.lower(): + if ch.isalnum(): + cur += ch + else: + if cur: + toks.append(cur); cur = "" + if cur: + toks.append(cur) + # 驼峰拆分(OrderController -> order, controller) + out = [] + for t in toks: + idx = 0 + for i, c in enumerate(t): + if i > 0 and c.isupper(): + out.append(t[idx:i]); idx = i + out.append(t[idx:]) + return [x for x in out if x] + + +class FakeEmbedder: + def embed(self, texts: List[str]) -> List[List[float]]: + vecs = [] + for t in texts: + v = [0.0] * _DIM + for tok in _tokenize(t): + h = hashlib.md5(tok.encode("utf-8")).digest() + idx = h[0] % _DIM + v[idx] += 1.0 + norm = math.sqrt(sum(x * x for x in v)) or 1.0 + vecs.append([x / norm for x in v]) + return vecs + + +def get_embedder(engine) -> Embedder: + # 真实引擎若提供 embed API 则在此接入;当前统一回退 FakeEmbedder + return FakeEmbedder() +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `python -m pytest tests/test_rag_embeddings.py -q` +Expected: PASS(3 passed) + +- [ ] **Step 5: 提交** + +```bash +git add src/genesis/rag/embeddings.py tests/test_rag_embeddings.py +git commit -m "feat(rag): 新增 Embedder 抽象与离线 FakeEmbedder" +``` + +--- + +### Task 2: RagStore(SQLite 向量存储 + 余弦检索) + +**Files:** +- Create: `src/genesis/rag/store.py` +- Test: `tests/test_rag_store.py` + +**Interfaces:** +- Consumes: `embeddings.Embedder`(仅用于维度约定,实际检索用传入向量) +- Produces: + - `genesis.rag.store.RagStore(db_path)`:方法 `reset_scope(scope)`、`add(scope, chunks, embeddings)`、`search(scope, query_vec, k) -> list[str]` + - 向量以 JSON 文本列存储(无新依赖) + +- [ ] **Step 1: 写失败测试** + +```python +# tests/test_rag_store.py +import math +from genesis.rag.store import RagStore + + +def _vec(*ones): + v = [0.0] * 8 + for i in ones: + v[i] = 1.0 + n = math.sqrt(sum(x * x for x in v)) + return [x / n for x in v] + + +def test_search_returns_most_similar_chunk(): + s = RagStore(":memory:") + s.reset_scope("p1") + s.add("p1", ["订单模块处理创建", "用户认证登录"], [_vec(0, 1), _vec(4, 5)]) + res = s.search("p1", _vec(0, 1), k=1) + assert res == ["订单模块处理创建"] + + +def test_reset_scope_clears(): + s = RagStore(":memory:") + s.reset_scope("p1") + s.add("p1", ["a"], [_vec(0)]) + s.reset_scope("p1") + assert s.search("p1", _vec(0), k=3) == [] +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `python -m pytest tests/test_rag_store.py -q` +Expected: FAIL + +- [ ] **Step 3: 实现最小版本** + +```python +# src/genesis/rag/store.py +from __future__ import annotations + +import json +import math +import sqlite3 +from typing import List + + +def _cosine(a, b): + dot = sum(x * y for x, y in zip(a, b)) + na = math.sqrt(sum(x * x for x in a)); nb = math.sqrt(sum(y * y for y in b)) + return dot / (na * nb or 1.0) + + +class RagStore: + def __init__(self, db_path: str): + self.conn = sqlite3.connect(db_path) + self.conn.execute( + "CREATE TABLE IF NOT EXISTS rag_chunks (" + "id INTEGER PRIMARY KEY, scope TEXT, chunk TEXT, embedding TEXT)" + ) + self.conn.commit() + + def reset_scope(self, scope: str) -> None: + self.conn.execute("DELETE FROM rag_chunks WHERE scope=?", (scope,)) + self.conn.commit() + + def add(self, scope: str, chunks: List[str], embeddings: List[List[float]]) -> None: + self.reset_scope(scope) + for c, e in zip(chunks, embeddings): + self.conn.execute( + "INSERT INTO rag_chunks(scope, chunk, embedding) VALUES(?,?,?)", + (scope, c, json.dumps(e)), + ) + self.conn.commit() + + def search(self, scope: str, query_vec: List[float], k: int = 5) -> List[str]: + rows = self.conn.execute( + "SELECT chunk, embedding FROM rag_chunks WHERE scope=?", (scope,) + ).fetchall() + scored = [] + for chunk, emb in rows: + scored.append((_cosine(query_vec, json.loads(emb)), chunk)) + scored.sort(key=lambda x: x[0], reverse=True) + return [c for _, c in scored[:k]] +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `python -m pytest tests/test_rag_store.py -q` +Expected: PASS(2 passed) + +- [ ] **Step 5: 提交** + +```bash +git add src/genesis/rag/store.py tests/test_rag_store.py +git commit -m "feat(rag): 新增 RagStore(SQLite 向量存储 + 余弦检索)" +``` + +--- + +### Task 3: ImpactRAG 索引/检索服务 + +**Files:** +- Create: `src/genesis/rag/impact_rag.py` +- Test: `tests/test_impact_rag.py` + +**Interfaces:** +- Consumes: `RagStore`、`Embedder` +- Produces: + - `genesis.rag.impact_rag.ImpactRAG(store, embedder)` + - `index(scope, sources: list[tuple[str, str]])`(sources = [(文件名, 文本), ...]) + - `retrieve(scope, query: str, k: int = 5) -> list[str]` + +- [ ] **Step 1: 写失败测试** + +```python +# tests/test_impact_rag.py +from genesis.rag.embeddings import FakeEmbedder +from genesis.rag.store import RagStore +from genesis.rag.impact_rag import ImpactRAG + + +def test_retrieve_returns_relevant_chunk(): + store = RagStore(":memory:") + rag = ImpactRAG(store, FakeEmbedder()) + sources = [ + ("OrderController.java", "public class OrderController { 创建订单 }"), + ("UserAuth.java", "public class UserAuth { 用户登录认证 }"), + ] + rag.index("p1", sources) + res = rag.retrieve("p1", "OrderController 的影响范围", k=1) + assert res and "OrderController" in res[0] +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `python -m pytest tests/test_impact_rag.py -q` +Expected: FAIL + +- [ ] **Step 3: 实现最小版本** + +```python +# src/genesis/rag/impact_rag.py +from __future__ import annotations + +from typing import List, Tuple + +from genesis.rag.embeddings import Embedder +from genesis.rag.store import RagStore + +_CHUNK = 800 + + +def _split(text: str, size: int = _CHUNK) -> List[str]: + paras = [p.strip() for p in text.split("\n") if p.strip()] + out, buf = [], "" + for p in paras: + if len(buf) + len(p) > size and buf: + out.append(buf); buf = p + else: + buf = (buf + "\n" + p).strip() + if buf: + out.append(buf) + return out or [""] + + +class ImpactRAG: + def __init__(self, store: RagStore, embedder: Embedder): + self.store = store + self.embedder = embedder + + def index(self, scope: str, sources: List[Tuple[str, str]]) -> None: + chunks = [] + for name, text in sources: + for piece in _split(text): + chunks.append(f"[{name}]\n{piece}") + embs = self.embedder.embed(chunks) + self.store.add(scope, chunks, embs) + + def retrieve(self, scope: str, query: str, k: int = 5) -> List[str]: + if not query.strip(): + return [] + qv = self.embedder.embed([query])[0] + return self.store.search(scope, qv, k) +``` + +- [ ] **Step 4: 运行测试确认通过** + +Run: `python -m pytest tests/test_impact_rag.py -q` +Expected: PASS + +- [ ] **Step 5: 提交** + +```bash +git add src/genesis/rag/impact_rag.py tests/test_impact_rag.py +git commit -m "feat(rag): 新增 ImpactRAG 索引/检索服务" +``` + +--- + +### Task 4: ImpactAgent 接入 use_rag + +**Files:** +- Modify: `src/genesis/impact/impact_agent.py` +- Test: `tests/test_impact_agent_rag.py` + +**Interfaces:** +- Consumes: `ImpactRAG`(可选)、`FakeEmbedder`/`RagStore`(测试) +- Produces: `ImpactAgent.__init__` 新增 `rag: ImpactRAG | None = None`;`run_impact(..., use_rag: bool = False)` 开启时把检索片段并入 prompt 上下文 + +- [ ] **Step 1: 写失败测试** + +```python +# tests/test_impact_agent_rag.py +from genesis.rag.embeddings import FakeEmbedder +from genesis.rag.store import RagStore +from genesis.rag.impact_rag import ImpactRAG +from genesis.impact.impact_agent import ImpactAgent + + +def _make(): + store = RagStore(":memory:") + rag = ImpactRAG(store, FakeEmbedder()) + rag.index("p1", [("OrderController.java", "class OrderController { 创建订单 }")]) + # ImpactAgent 构造参数按现有签名;rag 以 kw 注入 + agent = ImpactAgent(rag=rag) + return agent + + +def test_run_impact_includes_rag_context(monkeypatch): + agent = _make() + captured = {} + def fake_engine_invoke(prompt): + captured["prompt"] = prompt + return '{"impacts":[{"target":"OrderController","type":"修改","detail":"受订单表影响"}]}' + monkeypatch.setattr(agent, "_engine_invoke", fake_engine_invoke) + out = agent.run_impact("订单表新增字段", scope="p1", use_rag=True) + assert "OrderController" in captured["prompt"] + assert "OrderController" in str(out) +``` + +(注:`_engine_invoke` 为 ImpactAgent 内部调用 LLM 的既有方法名占位;若实际方法名不同,实现任务中替换为真实方法名,并在测试中同步。) + +- [ ] **Step 2: 运行测试确认失败** + +Run: `python -m pytest tests/test_impact_agent_rag.py -q` +Expected: FAIL(`rag` 参数 / `use_rag` 不存在) + +- [ ] **Step 3: 实现最小版本** + +`impact_agent.py` 顶部 `from genesis.rag.impact_rag import ImpactRAG`(仅在类型标注用,避免循环导入:若 `rag` 包尚未在 `genesis/rag/__init__` 暴露,使用 `TYPE_CHECKING` 方式或字符串标注)。 + +`ImpactAgent.__init__` 增加: + +```python +def __init__(self, ..., rag: "ImpactRAG | None" = None): + ... + self.rag = rag +``` + +`run_impact` 在拼装影响调查 prompt 前,若 `use_rag and self.rag`: + +```python +context_chunks = self.rag.retrieve(scope, requirement_text, k=5) +if context_chunks: + rag_block = "\n".join(f"# 相关代码/文档片段\n{c}" for c in context_chunks) + prompt = prompt + "\n\n" + rag_block +``` + +未开启或 `self.rag is None` 时,`prompt` 与现状一致(DesignReference 路径不变)。返回结构不变。 + +- [ ] **Step 4: 运行测试确认通过** + +Run: `python -m pytest tests/test_impact_agent_rag.py -q` +Expected: PASS + +- [ ] **Step 5: 提交** + +```bash +git add src/genesis/impact/impact_agent.py tests/test_impact_agent_rag.py +git commit -m "feat(impact): ImpactAgent 接入 use_rag,检索片段并入 prompt(默认关闭)" +``` + +--- + +### Task 5: 服务层接线 + 端到端 + 文档 + +**Files:** +- Modify: `src/genesis/server/service.py`、`src/genesis/server/app.py`(如需要透传 `use_rag`) +- Test: `tests/test_impact_rag_e2e.py` +- Modify: `docs/design.md`、`_AI_USAGE_LOG.md` + +**Interfaces:** +- Consumes: `ImpactRAG`、`RagStore`、`FakeEmbedder`、`ImpactAgent(rag=...)` +- Produces: `GenesisService(use_rag: bool = False)`;会话级影响调查在 `use_rag` 为真时走 RAG + +- [ ] **Step 1: 写失败测试** + +```python +# tests/test_impact_rag_e2e.py +from genesis.rag.embeddings import FakeEmbedder +from genesis.rag.store import RagStore +from genesis.rag.impact_rag import ImpactRAG +from genesis.server.service import GenesisService +from genesis.server.store import ProjectsStore, SessionStore + + +def test_service_impact_uses_rag_when_enabled(tmp_path): + store = SessionStore(db_path=str(tmp_path / "s.db")) + projects = ProjectsStore(db_path=str(tmp_path / "s.db")) + rag_store = RagStore(str(tmp_path / "rag.db")) + rag = ImpactRAG(rag_store, FakeEmbedder()) + rag.index("p1", [("OrderController.java", "class OrderController { 创建订单 }")]) + svc = GenesisService(store=store, data_root=str(tmp_path / "data"), + engine="fake", projects=projects, use_rag=True, rag=rag) + # 仅验证接线:service 持有 rag 且 use_rag 为真 + assert svc.use_rag is True and svc.rag is rag +``` + +(完整端到端需真实引擎与样本源码;以接线断言 + 既有 impact 测试不退化为下限。) + +- [ ] **Step 2: 运行测试确认失败** + +Run: `python -m pytest tests/test_impact_rag_e2e.py -q` +Expected: FAIL(`use_rag` / `rag` 参数不存在) + +- [ ] **Step 3: 实现最小版本** + +`service.py`:`GenesisService.__init__` 增加 `use_rag: bool = False, rag: ImpactRAG | None = None`;保存 `self.use_rag / self.rag`。在 `run_impact`(或 `_build_impact_agent`)构造 `ImpactAgent(rag=self.rag)` 并将 `use_rag=self.use_rag` 透传给 `run_impact`。 + +`app.py`:`create_app` 增加 `use_rag: bool = False` 参数并透传 `GenesisService(..., use_rag=use_rag)`;`scripts/serve.py` 增加 `--use-rag` 开关(默认关)。 + +- [ ] **Step 4: 运行全量测试确认无回归** + +Run: `python -m pytest -q -o addopts=""` +Expected: 全部通过(在现有基线基础上新增用例;既有 impact/design 测试不受默认关闭影响) + +- [ ] **Step 5: 更新文档 + 提交** + +`docs/design.md` §12 追加:「影响调查 RAG 化(2026-08-29):新增 `genesis/rag`(embeddings/store/impact_rag),`ImpactAgent.use_rag` 开启时检索 top-k 片段并入 prompt;默认关闭,回退 DesignReference;存储用 SQLite,无新重依赖。」 + +```bash +git add src/genesis/server/service.py src/genesis/server/app.py scripts/serve.py tests/test_impact_rag_e2e.py docs/design.md _AI_USAGE_LOG.md +git commit -m "feat(rag): 服务层接线 use_rag + design.md 记录(默认关闭)" +``` + +--- + +## Self-Review + +1. **Spec coverage:** 离线 Embedder(T1)、向量存储(T2)、索引/检索(T3)、Agent 接入与开关(T4)、服务层接线与文档(T5)均覆盖;回退路径在 T4/T5 明确保留。 +2. **Placeholder scan:** 无 TBD;`_engine_invoke` 已在测试中标注为真实方法名占位,实现任务需对齐真实方法名(已在 Step 3 注明)。 +3. **Type consistency:** `ImpactRAG.index(scope, sources)` / `retrieve(scope, query, k)` 在 T3/T4 一致;`RagStore.add(scope, chunks, embeddings)` 维度与 `Embedder.embed` 输出一致;`use_rag` 在 Service/Agent 间透传一致。 +4. **限制:** 默认 `use_rag=False`,不影响现有行为;真实 embed 接入点在 `get_embedder` 预留,待推理引擎提供 embed API 时扩展(不在本迭代)。 diff --git a/docs/superpowers/plans/2026-08-29-iteration-websocket-progress.md b/docs/superpowers/plans/2026-08-29-iteration-websocket-progress.md new file mode 100644 index 0000000..6a8c286 --- /dev/null +++ b/docs/superpowers/plans/2026-08-29-iteration-websocket-progress.md @@ -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` 集成: +- 在 `` 末尾增加 ``(置于 `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 未连接前(如首屏尚未打开)会被丢弃,但已持久化兜底,重载仍可见,不影响正确性。