fix(bioflow): R1 缺陷修复——反代 aread 时序(uvloop 大响应体 502)/flow_store 启动接线(ask 引用块)/events snapshot dict 保真

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

bioflow_service/app/context.py

@@ -144,7 +144,13 @@ def _make_archive_finder(database: Database):
 
 
 async def init_context(db_path: str | None = None) -> Context:
-    """应用启动：初始化全局 context（幂等；db_path 供测试替换）。"""
+    """应用启动：初始化全局 context（幂等；db_path 供测试替换）。
+
+    BUG-R1-03 修复：生产启动链接线 FlowStore——此前 build_context 恒不传
+    flow_store（默认 None），部署态 ask/chat_qa 引用块恒空（citations=[]）。
+    现在按配置 flow_docs_dir 加载本地实例（vstore/data/flow_docs 已入库 724
+    块）；缺失/失败自动降级 None（planner 走 llm_only，服务不阻断）。
+    """
     global context
     from .db import init_db
 
@@ -153,7 +159,8 @@ async def init_context(db_path: str | None = None) -> Context:
         database = await init_db(db_path)
     elif context.db is None:
         database = await init_db()
-    context = build_context(db=database)
+    flow_store = init_flow_store(get_settings().flow_docs_dir_abs)
+    context = build_context(db=database, flow_store=flow_store)
     return context
 
 

bioflow_service/app/services/bioflow/events.py

@@ -44,6 +44,24 @@ class Subscriber:
     job_filter: str | None = None
 
 
+def _stage_key(s: Any) -> str:
+    """job_created stages 元素 → 阶段名键（str 或 dict 统一；BUG-R1-04）。"""
+    if isinstance(s, dict):
+        return str(s.get("stage") or "")
+    return str(s)
+
+
+def _stage_entry(s: Any) -> dict[str, Any]:
+    """job_created stages 元素 → 干净 {stage, status}（dict 保真透传）。
+
+    runner 发布的 job_created.stages 元素为 {stage, status} dict；此前直接
+    str(s) 会把 dict repr 串当阶段名（前端进度卡乱串，BUG-R1-04）。
+    """
+    if isinstance(s, dict):
+        return {"stage": str(s.get("stage") or ""), "status": str(s.get("status") or "waiting")}
+    return {"stage": str(s), "status": "waiting"}
+
+
 class EventBus:
     """进程内事件总线：publish → 订阅队列广播 + 内存环形缓冲 + jsonl 持久日志。"""
 
@@ -144,7 +162,9 @@ class EventBus:
             if t == "job_created":
                 job_status = "created"
                 for s in d.get("stages") or []:
-                    stage_states.setdefault(str(s), {"stage": str(s), "status": "waiting"})
+                    key = _stage_key(s)
+                    if key:
+                        stage_states.setdefault(key, _stage_entry(s))
             elif t == "stage_started":
                 st = stage_states.setdefault(str(d.get("stage")), {"stage": str(d.get("stage"))})
                 st["status"] = "running"

bioflow_service/tests/test_context_wiring.py

@@ -0,0 +1,207 @@
+"""tests/test_context_wiring.py —— 生产启动链接线（BUG-R1-03 回归锁定）。
+
+此前 init_context→build_context 不传 flow_store（恒 None），部署态 ask/
+chat_qa 引用块恒空。本文件锁定：
+1. init_context() 真实接线：临时 flow_docs 目录 + 假嵌入 → context.flow_store
+   非 None 且可检索。
+2. ask 端点经接线后的 context 返回引用块（citations 非空）。
+3. flow_docs 缺失时降级 None 不阻断启动（init_flow_store 契约保持）。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.bioflow_unit
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def _seed_flow_docs(flow_dir: Path) -> list[dict]:
+    """临时 flow_docs：用真实 FlowStore.add_chunk 入库 2 块（假嵌入注入）。
+
+    假嵌入必须产 1024 维（RetrievalIndex IndexFlatIP 固定维，对齐 vstore
+    tests/fake_embed_utils.DIM=1024；词级 shingle 哈希到 1024 维）。
+    """
+    import hashlib
+    import re
+
+    sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
+    from vstore.flow import FlowStore, set_embed_fn
+
+    DIM = 1024
+
+    def fake_embed(texts):
+        vectors = []
+        for t in texts:
+            vec = [0.0] * DIM
+            # 中文字符 2-gram + ASCII 词 shingle 哈希到维度（主题粗向量替代）
+            tokens = re.findall(r"[A-Za-z0-9_\-]+", t)
+            for tok in tokens:
+                for i in range(max(1, len(tok) - 3)):
+                    sh = tok[i:i + 4].lower()
+                    idx = int(hashlib.md5(sh.encode()).hexdigest()[:8], 16) % DIM
+                    vec[idx] += 1.0
+            for m in re.finditer(r"[一-鿿]{2,}", t):
+                seg = m.group(0)
+                for i in range(len(seg) - 1):
+                    sh = seg[i:i + 2]
+                    idx = int(hashlib.md5(sh.encode()).hexdigest()[:8], 16) % DIM
+                    vec[idx] += 1.0
+            norm = sum(v * v for v in vec) ** 0.5
+            if norm > 0:
+                vec = [v / norm for v in vec]
+            vectors.append(vec)
+        return vectors
+
+    set_embed_fn(fake_embed)
+    store = FlowStore(flow_dir)
+    store.add_chunk(
+        {
+            "source_id": "contract_identify", "chunk_key": "contract_identify__abc123",
+            "source": "stage_contract", "doc_kind": "contract",
+            "title": "Identify Stage Contract", "title_path": "Identify | ## 1",
+            "heading": "## 1. 输入", "stage_name": "identify",
+            "path": "gfa_core/schemas/stages/identify_contract_v1.md",
+            "cite": {"anchor": "## 1", "line": 1},
+        },
+        "Identify 阶段需要全蛋白 FASTA 与 HMM 文件作为输入。",
+        vector=None,
+    )
+    store.add_chunk(
+        {
+            "source_id": "corpus_main", "chunk_key": "corpus_main__def456",
+            "source": "corpus", "doc_kind": "corpus",
+            "title": "全流程文档", "title_path": "第一部分 | ## 4. identify",
+            "heading": "## 4. identify：家族成员初筛",
+            "stage_name": "",
+            "path": "docs/bioflow_corpus/PlantAI_现有分析全流程.md",
+            "cite": {"anchor": "## 4. identify：家族成员初筛"},
+        },
+        "identify 阶段对输入蛋白序列进行 HMMER 搜索与结构域确认。",
+        vector=None,
+    )
+    return [
+        {"chunk_key": "contract_identify__abc123", "id": "contract_identify__abc123",
+         "source": "stage_contract", "doc_kind": "contract", "stage_name": "identify",
+         "title": "Identify Stage Contract", "title_path": "Identify | ## 1",
+         "path": "gfa_core/schemas/stages/identify_contract_v1.md",
+         "cite": {"anchor": "## 1", "line": 1}, "text": "Identify 阶段需要全蛋白 FASTA。"},
+    ]
+
+
+def test_init_context_wires_flow_store(tmp_path, monkeypatch):
+    """BUG-R1-03 回归：init_context() 接线后 context.flow_store 非 None。"""
+    flow_dir = tmp_path / "flow_docs"
+    _seed_flow_docs(flow_dir)
+
+    import app.config as config_mod
+    import app.context as context_mod
+
+    s = config_mod.Settings(
+        _env_file=None,
+        data_root=str(tmp_path / "data"),
+        jobs_dir=str(tmp_path / "jobs"),
+        db_path=str(tmp_path / "bioflow.db"),
+        archives_dir=str(tmp_path / "archives"),
+        image_map_path=str(tmp_path / "stage_image_map.json"),
+        flow_docs_dir=str(flow_dir),
+    )
+    # 真实映射表避免 load_kernel 依赖测试表：指向仓库 A1 交付物
+    s2 = config_mod.Settings(
+        _env_file=None,
+        data_root=str(tmp_path / "data"),
+        jobs_dir=str(tmp_path / "jobs"),
+        db_path=str(tmp_path / "bioflow.db"),
+        archives_dir=str(tmp_path / "archives"),
+        image_map_path=str(Path(__file__).resolve().parent.parent.parent
+                           / "bioflow_kernel" / "stage_image_map.json"),
+        flow_docs_dir=str(flow_dir),
+    )
+    monkeypatch.setattr(config_mod, "get_settings", lambda: s2)
+    monkeypatch.setattr(context_mod, "get_settings", lambda: s2)
+
+    ctx = _run(context_mod.init_context(db_path=str(tmp_path / "bioflow.db")))
+    try:
+        assert ctx.flow_store is not None, "init_context 未接线 flow_store（BUG-R1-03）"
+        # 真实检索：引用块可召回
+        result = ctx.flow_store.search("identify 需要什么输入", top_k=4)
+        assert result.get("items"), "接线后的 FlowStore 检索应为非空"
+    finally:
+        _run(context_mod.shutdown_context())
+        from app.db import close_db
+
+        _run(close_db())
+        context_mod.context = context_mod.Context()  # 还原全局（后续用例隔离）
+
+
+def test_flow_ask_returns_citations_after_wiring(tmp_path, monkeypatch):
+    """BUG-R1-03 回归：接线后 ask 端点返回引用块（citations 非空）。"""
+    flow_dir = tmp_path / "flow_docs"
+    _seed_flow_docs(flow_dir)
+    sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
+    from vstore.flow import set_embed_fn
+
+    import app.config as config_mod
+    import app.context as context_mod
+    from app.routers import bioflow as r_flow
+    from app.schemas import FlowAskRequest
+
+    s = config_mod.Settings(
+        _env_file=None,
+        data_root=str(tmp_path / "data"),
+        jobs_dir=str(tmp_path / "jobs"),
+        db_path=str(tmp_path / "bioflow.db"),
+        archives_dir=str(tmp_path / "archives"),
+        image_map_path=str(Path(__file__).resolve().parent.parent.parent
+                           / "bioflow_kernel" / "stage_image_map.json"),
+        flow_docs_dir=str(flow_dir),
+    )
+    monkeypatch.setattr(config_mod, "get_settings", lambda: s)
+    monkeypatch.setattr(context_mod, "get_settings", lambda: s)
+
+    ctx = _run(context_mod.init_context(db_path=str(tmp_path / "bioflow.db")))
+    try:
+        monkeypatch.setattr(r_flow, "context", ctx)
+        resp = _run(r_flow.flow_ask(FlowAskRequest(
+            session_id="s1", question="identify 需要什么输入")))
+        assert resp["ok"] is True
+        citations = resp["answer"]["citations"]
+        assert citations, "接线后 ask 应返回引用块（BUG-R1-03）"
+        assert citations[0]["path"]
+        assert citations[0]["cite"]
+        assert resp["answer"]["degraded"] is False
+    finally:
+        _run(context_mod.shutdown_context())
+        from app.db import close_db
+
+        _run(close_db())
+        context_mod.context = context_mod.Context()
+
+
+def test_init_flow_store_missing_dir_degrades(tmp_path, monkeypatch):
+    """flow_docs 缺失 → init_flow_store 返回 None（降级不阻断）。"""
+    import app.config as config_mod
+    import app.context as context_mod
+
+    s = config_mod.Settings(
+        _env_file=None,
+        data_root=str(tmp_path / "data"),
+        jobs_dir=str(tmp_path / "jobs"),
+        db_path=str(tmp_path / "bioflow.db"),
+        archives_dir=str(tmp_path / "archives"),
+        image_map_path=str(tmp_path / "stage_image_map.json"),
+        flow_docs_dir=str(tmp_path / "no_such_flow_docs"),
+    )
+    monkeypatch.setattr(config_mod, "get_settings", lambda: s)
+    monkeypatch.setattr(context_mod, "get_settings", lambda: s)
+    store = context_mod.init_flow_store(s.flow_docs_dir_abs)
+    # 目录缺失时 FlowStore 可能空载成功也可能抛错降级——两态均不得阻断
+    # （关键契约：init_flow_store 绝不抛异常上浮）
+    assert store is None or store.stats().get("total_chunks", 0) == 0

bioflow_service/tests/test_events.py

@@ -91,6 +91,34 @@ def test_snapshot_job_done(tmp_path):
     assert bus.snapshot("j1")["job"]["status"] == "done"
 
 
+def test_snapshot_dict_stages_clean(tmp_path):
+    """BUG-R1-04 回归锁定：job_created.stages 为 dict（runner 真实发布形态）时，
+    快照 stages 逐阶段必须是干净 {stage, status} 结构——不得出现 dict repr 串
+    阶段名（此前 str(s) 会把 {stage:..., status:...} 串当阶段名，前端进度卡乱串）。
+    """
+    bus = EventBus(tmp_path / "events")
+    bus.publish("job_created", {
+        "job_id": "j1", "title": "t",
+        "stages": [
+            {"stage": "identify", "status": "waiting"},
+            {"stage": "phylogeny", "status": "waiting"},
+        ],
+    })
+    bus.publish("stage_started", {"job_id": "j1", "stage": "identify"})
+    bus.publish("stage_done", {"job_id": "j1", "stage": "identify", "exit_code": 0})
+    snap = bus.snapshot("j1")
+    assert snap["type"] == "snapshot"
+    assert len(snap["stages"]) == 2
+    stages = {s["stage"]: s["status"] for s in snap["stages"]}
+    # 干净阶段名（非 dict repr 串）
+    assert stages["identify"] == "done"
+    assert stages["phylogeny"] == "waiting"
+    for s in snap["stages"]:
+        assert isinstance(s["stage"], str)
+        assert not s["stage"].startswith("{")
+        assert isinstance(s["status"], str)
+
+
 def test_memory_replay_after_seq(tmp_path):
     bus = EventBus(tmp_path / "events")
     e1 = bus.publish("stage_started", {"job_id": "j1", "stage": "a"})

docs/team/report_bioflow_dev2.md

@@ -165,3 +165,51 @@
 | sdk/app/routers/bioflow_proxy.py | BFF 流式反代 + 报告登记 |
 | sdk/app/main.py（+2 行）/ sdk/app/config.py（+2 键） | BFF 最小改动 |
 | docs/team/report_bioflow_dev2.md | 本报告 |
+
+## 6. R1 缺陷修复记录（T2 轮次测试，2026-09-06）
+
+### BUG-R1-01（阻断）BFF 反代大响应体 502 —— 已修复
+- **根因**：`sdk/app/routers/bioflow_proxy.py` `_proxy()` 中 `upstream.aread()`
+  在 `async with httpx.AsyncClient` 上下文退出后才调用；生产事件循环（uvloop）下
+  连接已随 client 关闭，大响应体（全量 catalog 55,905B）首读未缓冲 → httpx.ReadError
+  → 502；小响应体（health 80B 等）因首读已缓冲幸存。
+- **修法**：client 生命周期手工管理（非 async with）——非 SSE 分支在 client 存活期内
+  完成 `aread()`，随后 `upstream.aclose()` + `client.aclose()`；SSE 分支在 stream()
+  的 finally 中一并关闭两对象（client 活到流消费完毕）。send 失败/aread 失败均
+  aclose 后返回 503/502 信封。
+- **验证**：修复前 uvloop_probe 复现 5/5 = 502（含 AREAD EXC 堆栈）；修复后同一
+  脚本 catalog 5/5 = 200（bodylen=55905）、health 200，无异常。
+
+### BUG-R1-03（严重）flow_store 未接线 —— 已修复
+- **根因**：`init_flow_store` 已实现但 `init_context()` 调 `build_context` 时不传
+  flow_store（恒 None）→ 部署态 ask/chat_qa 引用块恒空。L2 测试注入 FakeFlowStore
+  掩盖了生产链路缺口。
+- **修法**：`init_context()` 接线 `flow_store = init_flow_store(get_settings()
+  .flow_docs_dir_abs)` 并传入 `build_context(flow_store=...)`；缺失/失败照旧降级
+  None 不阻断启动。
+- **新增测试**：`tests/test_context_wiring.py`
+  - `test_init_context_wires_flow_store`（临时 flow_docs + 1024 维假嵌入 → 接线后
+    context.flow_store 非 None 且真实 FlowStore.search 检索非空）
+  - `test_flow_ask_returns_citations_after_wiring`（接线后 ask 端点返回 citations
+    非空且 degraded=false）
+  - `test_init_flow_store_missing_dir_degrades`（缺失目录降级不阻断）
+- **E2E 验证**：真实生产 flow_docs（724 块）经 init_flow_store 加载成功，检索
+  "identify 需要什么输入" 4 hits（stage_contract/skill 引用块正常）。
+
+### BUG-R1-04（一般）events snapshot 脏结构 —— 已修复
+- **根因**：`snapshot()` 对 job_created 载荷 stages 元素直接 `str(s)`；runner 实际
+  发布 `{stage, status}` dict → 快照首帧阶段名变成 dict repr 串（前端进度卡乱串），
+  直到后续回放事件覆盖。单测用字符串列表 stages 未触发。
+- **修法**：新增 `_stage_key()`/`_stage_entry()`——dict 元素保真透传为干净
+  `{stage, status}`（取 stage 字段为键），字符串元素照旧兼容；空键跳过。
+- **新增测试**：`tests/test_events.py::test_snapshot_dict_stages_clean`（dict
+  stages 发布 → 快照逐阶段 {stage, status} 干净结构，阶段名非 dict repr 串）。
+
+### 自测结果（修复后）
+| 命令 | 结果 |
+|---|---|
+| `python -m pytest tests/ -q` | **170 passed**（15.8s，0 failed） |
+| `-m bioflow_unit` | 131 passed |
+| `-m bioflow_api` | 39 passed |
+
+修复前基线 166 passed；新增 4 条回归用例（wiring 3 + dict-stages 1）后 170 全绿。

sdk/app/routers/bioflow_proxy.py

@@ -59,13 +59,19 @@ async def _proxy(request: Request, path: str) -> Any:
                if k.lower() not in ("host", "content-length", "connection")}
     timeout = httpx.Timeout(connect=PROXY_CONNECT_TIMEOUT, read=_proxy_timeout(path),
                             write=PROXY_UPLOAD_TIMEOUT, pool=5.0)
+    # 手工管理 client 生命周期（不用 async with）：
+    # - 非 SSE：必须在 client 存活期内完成 aread()（BUG-R1-01 根因——此前
+    #   aread 在 client 上下文退出后调用，uvloop 下连接已随 client 关闭 →
+    #   httpx.ReadError → 502；大响应体首读未缓冲必现）。
+    # - SSE：client 须活到流消费完毕，故在 stream() finally 中一并 aclose。
+    client = httpx.AsyncClient(timeout=timeout)
     try:
-        async with httpx.AsyncClient(timeout=timeout) as client:
-            req = client.build_request(
-                request.method, url, content=body, headers=headers,
-            )
-            upstream = await client.send(req, stream=True)
+        req = client.build_request(
+            request.method, url, content=body, headers=headers,
+        )
+        upstream = await client.send(req, stream=True)
     except httpx.HTTPError as exc:
+        await client.aclose()
         logger.warning("bioflow_proxy: 生信服务不可达 %s: %s", url, exc)
         payload, status = envelope("UPSTREAM_DOWN", message="生信服务当前不可用，请稍后重试",
                                    degraded=True, status=503)
@@ -80,6 +86,7 @@ async def _proxy(request: Request, path: str) -> Any:
                     yield chunk
             finally:
                 await upstream.aclose()
+                await client.aclose()
 
         resp_headers = {
             "Cache-Control": "no-cache",
@@ -89,15 +96,18 @@ async def _proxy(request: Request, path: str) -> Any:
         return StreamingResponse(stream(), status_code=upstream.status_code,
                                  media_type="text/event-stream", headers=resp_headers)
 
+    # 非 SSE：client 存活期内完成读取（BUG-R1-01 修复）
     try:
         content = await upstream.aread()
-        await upstream.aclose()
     except httpx.HTTPError:
+        await client.aclose()
         payload, status = envelope("UPSTREAM_DOWN", message="生信服务响应中断",
                                    degraded=True, status=502)
         return JSONResponse(status_code=status, content=payload)
+    await upstream.aclose()
+    await client.aclose()
     # 非 SSE：纯透传（content 是 bytes；不能 JSONResponse——bytes 序列化会 500）
-    media_type = upstream.headers.get("content-type") or "application/json"
+    media_type = content_type or "application/json"
     return Response(status_code=upstream.status_code, content=content,
                     media_type=media_type)