wjq/PlantAI-rag-new
1
0
Fork 0
forked from admin/PlantAI-rag-new
Code Pull Requests Activity

fix(extension): 测试隔离+注入惰性适配+chat_stream 意图接线

Browse Source 
TL 测试循环修复:
- conftest 两处 Settings 补 extensions_dir=tmp 隔离(此前测试扩展
  落仓库默认目录,跨测试污染致列表排序断言失败)
- inject_real_executor 惰性创建适配器(store.executor 默认 None)
- chat_stream 接扩展意图两层判定:create→extension_upload 帧、
  run→extension_candidates 帧、create_ambiguous→引导帧;
  扩展链路异常不回退生信主链
- .gitignore 排除 tests/extension_cases_18/(2.6G 测试数据不入库)

验证:bioflow_service 324/324、内核 15/15、ui 122/122+373/373、
真实 docker 冒烟 SMOKE_OK。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
admin
2026-09-14 15:09:15 +08:00
co-authored by Claude Fable 5
parent 3b16397baf
commit 812decf310
4 changed files with 94 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -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/
bioflow_service/app/routers/bioflow.py
+40
View File
@@ -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)
bioflow_service/app/services/bioflow/extensions.py
+48 -3
View File
@@ -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)
bioflow_service/tests/conftest.py
+3 -1
View File
@@ -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"),