diff --git a/.gitignore b/.gitignore index 36e584a..e09e899 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,6 @@ kb-libraries/aqua/build/ kb-libraries/aqua/registry/* !kb-libraries/aqua/registry/README.md kb-libraries/aqua/data/ + +# 18 阶段拓展分析测试用例数据(2.6G,不入库;测试时从本机同步) +tests/extension_cases_18/ diff --git a/bioflow_service/app/routers/bioflow.py b/bioflow_service/app/routers/bioflow.py index 3024edc..7f6d724 100644 --- a/bioflow_service/app/routers/bioflow.py +++ b/bioflow_service/app/routers/bioflow.py @@ -177,6 +177,46 @@ async def chat_stream(body: ChatStreamRequest) -> StreamingResponse: }}) return + # 0.5. 扩展意图两层判定(契约 §5;先于阶段规划,漏判安全)。 + # 命中 → 发 extension_* 引导帧后返回(不进入正常生信阶段规划)。 + from ..services.bioflow import extensions as ext_mod + from ..services.bioflow.intent import detect_extension_intent + try: + ext_store = ext_mod.get_store() + published = ext_store.list_extensions() + ext_intent = detect_extension_intent(body.message, published) + layer = str(ext_intent.get("layer") or "none") + if layer == "create": + yield _sse({"type": "extension_upload", "seq": -2, "data": { + "layer": "create", "message": body.message, + }}) + return + if layer == "run": + candidates = [ + { + "ext_id": x.get("ext_id"), + "display_name": x.get("display_name"), + "purpose": x.get("analysis_type") or "", + "score": x.get("score"), + "reasons": x.get("reasons") or [], + } + for x in (ext_intent.get("candidates") or []) + ] + yield _sse({"type": "extension_candidates", "seq": -2, "data": { + "candidates": candidates, + "high_confidence": len(candidates) == 1, + }}) + return + if layer == "create_ambiguous": + yield _sse({"type": "extension_upload", "seq": -2, "data": { + "layer": "create_ambiguous", + "message": "没有匹配的已发布扩展,是否创建?", + "raw_message": body.message, + }}) + return + except Exception as exc: # noqa: BLE001 - 扩展链路异常不回退生信主链 + logger.warning("chat_stream: extension intent check failed: %s", exc) + # 1. 规划（候选路径；复用/LLM/降级） try: plan: PlanResult = await ctx.planner.plan(body.message) diff --git a/bioflow_service/app/services/bioflow/extensions.py b/bioflow_service/app/services/bioflow/extensions.py index 95d8afd..52fa937 100644 --- a/bioflow_service/app/services/bioflow/extensions.py +++ b/bioflow_service/app/services/bioflow/extensions.py @@ -213,11 +213,50 @@ class ExtensionExecutor: cpu=spec_in.cpu, memory=spec_in.memory, ) - raw = await _call_executor(real, req) + # 分派(TL 修复):execute 方法 = 旧 Dev-A 契约(入参 ExtensionRunSpec, + # 返回 ExtensionRunResult——测试 fake 与历史实现); + # 无 execute 的 Dev-B 真实执行器走 run(request) + dict 结果归一化。 + if hasattr(real, "execute"): + raw = await _call_executor(real, spec_in) + else: + raw = await real.run(req) if isinstance(raw, dict): - return _result_from_dict(raw, run_id=spec_in.log_path or new_run_id()) + return self._to_result(spec_in, raw) return raw + def _to_result(self, spec_in: ExtensionRunSpec, raw: dict[str, Any]) -> ExtensionRunResult: + """Dev-B run() 返回 {exit_code, timed_out, log_tail, cmd} → Dev-A 结果契约。""" + out_dir = Path(spec_in.output_dir) + outputs: list[str] = [] + try: + if out_dir.is_dir(): + outputs = sorted( + str(p.relative_to(out_dir)) + for p in out_dir.rglob("*") + if p.is_file() + ) + except OSError: + outputs = [] + exit_code = int(raw.get("exit_code") or 0) + if exit_code == spec_in.output_contract_exit: + error = "输出契约校验失败:输出目录为空" + elif raw.get("timed_out"): + error = "扩展执行超时" + elif exit_code != 0: + error = raw.get("log_tail") and ("命令失败:见日志") or "命令失败" + else: + error = "" + return ExtensionRunResult( + run_id=raw.get("run_id") or new_run_id(), + exit_code=exit_code, + log_tail=str(raw.get("log_tail") or ""), + outputs=outputs, + error=error, + timed_out=bool(raw.get("timed_out")), + started_at=raw.get("started_at") or "", + finished_at=raw.get("finished_at") or "", + ) + # ---- 扩展生命周期状态机 ---- @@ -920,10 +959,16 @@ def get_store() -> ExtensionStore: def inject_real_executor(real: Any, store: ExtensionStore | None = None) -> None: - """把 Dev-B extension_exec.ExtensionExecutor 注入扩展执行器（app 启动接线）。""" + """把 Dev-B extension_exec.ExtensionExecutor 注入扩展执行器（app 启动接线）。 + + store.executor 字段默认 None（占位执行器惰性创建）——注入时先补适配器实例 + 再挂真实执行器（TL 修复:此前直接 .inject_real 在 None 上崩溃）。 + """ target = store if store is not None else _active_store if target is None: return + if target.executor is None: + target.executor = ExtensionExecutor() target.executor.inject_real(real) diff --git a/bioflow_service/tests/conftest.py b/bioflow_service/tests/conftest.py index 3f71a45..0aa65a8 100644 --- a/bioflow_service/tests/conftest.py +++ b/bioflow_service/tests/conftest.py @@ -244,6 +244,7 @@ def test_settings(tmp_path, monkeypatch): _env_file=None, data_root=str(tmp_path / "data"), jobs_dir=str(tmp_path / "jobs"), + extensions_dir=str(tmp_path / "extensions"), db_path=str(tmp_path / "bioflow.db"), archives_dir=str(tmp_path / "archives"), image_map_path=str(tmp_path / "stage_image_map.json"), @@ -254,7 +255,7 @@ def test_settings(tmp_path, monkeypatch): ) Path(s.image_map_abs).parent.mkdir(parents=True, exist_ok=True) Path(s.image_map_abs).write_text(json.dumps(STAGE_MAP_JSON), encoding="utf-8") - for d in (s.data_root_abs, s.jobs_dir_abs, s.archives_dir_abs, s.flow_docs_dir_abs): + for d in (s.data_root_abs, s.jobs_dir_abs, s.archives_dir_abs, s.flow_docs_dir_abs, s.extensions_dir_abs): Path(d).mkdir(parents=True, exist_ok=True) monkeypatch.setenv("BIOFLOW_TEST_UID", "1234") @@ -278,6 +279,7 @@ def _isolated_env(tmp_path, monkeypatch): _env_file=None, data_root=str(tmp_path / "data"), jobs_dir=str(tmp_path / "jobs"), + extensions_dir=str(tmp_path / "extensions"), db_path=str(tmp_path / "bioflow.db"), archives_dir=str(tmp_path / "archives"), image_map_path=str(tmp_path / "stage_image_map.json"),