diff --git a/bioflow_service/app/context.py b/bioflow_service/app/context.py index c82ae75..4301325 100644 --- a/bioflow_service/app/context.py +++ b/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 diff --git a/bioflow_service/app/services/bioflow/events.py b/bioflow_service/app/services/bioflow/events.py index 578962f..67f3a4b 100644 --- a/bioflow_service/app/services/bioflow/events.py +++ b/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" diff --git a/bioflow_service/tests/test_context_wiring.py b/bioflow_service/tests/test_context_wiring.py new file mode 100644 index 0000000..219fac5 --- /dev/null +++ b/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 diff --git a/bioflow_service/tests/test_events.py b/bioflow_service/tests/test_events.py index d7a0f58..9c29f3f 100644 --- a/bioflow_service/tests/test_events.py +++ b/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"}) diff --git a/docs/team/report_bioflow_dev2.md b/docs/team/report_bioflow_dev2.md index a805a6f..45036a3 100644 --- a/docs/team/report_bioflow_dev2.md +++ b/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 全绿。 diff --git a/sdk/app/routers/bioflow_proxy.py b/sdk/app/routers/bioflow_proxy.py index dd0371f..d10adf2 100644 --- a/sdk/app/routers/bioflow_proxy.py +++ b/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)