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

feat(bioflow): extension lifecycle backend (P0) - spec whitelist, state machine, AI build, intent layers

Browse Source 
服务端生命周期部分（契约 §1/§2/§4/§5/§6；Dev-A）：
- extension_spec.py: spec_v2 白名单 12 占位符 + {output_dir} 必含 + 白名单外拒绝
- extensions.py: 生命周期状态机(契约 §2 枚举) + {extensions}/{ext_id} 文件存储
  + ExtensionExecutor 接口占位(Dev-B 实现 extension_exec.py, inject_executor 注入)
  + extension_* 事件帧经 EventBus 发布(events.py 注册, 不改 job_* 语义)
- extension_llm.py: build/repair AI 转换(DOCGEN_LLM_* 优先, BIOFLOW_EXTENSION_* 别名)
  LLM 不可用/坏输出一律回退规则版(_detect_requirements/_command_to_template 移植), 绝不 500
- routers/extension.py: 契约 §6 端点全集; trial/run 执行器未接入返回 503
- intent.py: 契约 §5 两层判定纯函数(A 创建词库 × B 已发布扩展打分 + 歧义消解 + 漏判安全)
  chat/stream 接线留给 Dev-C 主分支整合
- config/schemas/main: upload_ext_allow 加 .zip/.py/.r/.sh; extensions_dir +
  extension_max_repair_rounds(20) + LLM 键注释; 模型; router 注册
- tests: 5 个测试文件 71 用例; 全仓 286 用例通过

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
admin
2026-09-14 13:48:37 +08:00
co-authored by Claude Fable 5
parent f0a1711ebc
commit 4866b0408a
15 changed files with 3582 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bioflow_service/app/config.py
+11 -1
View File
@@ -54,7 +54,13 @@ class Settings(BaseSettings):
default_memory: str = "32g"
log_tail_lines: int = 80 # manifest 接口日志尾行数（≤80）
max_upload_gb: float = 100.0 # 上传大小上限（GB；超限 413）
upload_ext_allow: str = ".fasta,.fa,.faa,.fna,.pep,.cds,.gz,.gff,.gff3,.gtf,.txt,.tsv,.csv,.xlsx,.xls,.bed,.vcf,.vcf.gz,.bam,.cram,.sam,.hmm,.nwk,.newick,.treefile,.json,.yaml,.yml,.pdf"
upload_ext_allow: str = ".fasta,.fa,.faa,.fna,.pep,.cds,.gz,.gff,.gff3,.gtf,.txt,.tsv,.csv,.xlsx,.xls,.bed,.vcf,.vcf.gz,.bam,.cram,.sam,.hmm,.nwk,.newick,.treefile,.json,.yaml,.yml,.pdf,.zip,.py,.r,.sh"
# ---- 用户自定义扩展分析（并行链路；存储根 + AI 转换 LLM 键 + 修复上限）----
extensions_dir: str = "bioflow_data/extensions" # 扩展存储根（{ext_id}/extension.json...）
extension_max_repair_rounds: int = 20 # AI 修复循环上限（契约 §6 可配）
# 扩展转换 LLM 键：DOCGEN_LLM_* 直接复用（docgen 直连模式），
# BIOFLOW_EXTENSION_* 为可选别名（extension_llm.llm_config 兜底读取）。
# ---- 规划（planner）----
llm_plan_timeout: float = 60.0 # 规划 LLM 单次调用超时（秒）
@@ -120,6 +126,10 @@ class Settings(BaseSettings):
def flow_docs_dir_abs(self) -> str:
return _resolve_root(self.flow_docs_dir)
@property
def extensions_dir_abs(self) -> str:
return _resolve_root(self.extensions_dir)
@property
def upload_ext_set(self) -> set[str]:
return {ext.strip().lower() for ext in self.upload_ext_allow.split(",") if ext.strip()}
bioflow_service/app/main.py
+2 -1
View File
@@ -17,7 +17,7 @@ from fastapi.middleware.cors import CORSMiddleware
from .config import get_settings
from .context import context, init_context, shutdown_context
from .db import close_db, init_db
from .routers import bioarchives, bioflow, biojobs
from .routers import bioarchives, bioflow, biojobs, extension
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@@ -74,6 +74,7 @@ def create_app() -> FastAPI:
app.include_router(bioflow.router)
app.include_router(biojobs.router)
app.include_router(bioarchives.router)
app.include_router(extension.router)
return app
bioflow_service/app/routers/extension.py
+363
View File
@@ -0,0 +1,363 @@
"""app/routers/extension.py —— 用户自定义扩展分析端点（契约 §6 全集）。
- POST /api/bioflow/extensions 上传 zip（multipart）+ display_name
+ command + runtime_image? + golden?
- GET /api/bioflow/extensions 列表（published 优先）
- GET /api/bioflow/extensions/{id} 详情
- DELETE /api/bioflow/extensions/{id} 删除
- POST /api/bioflow/extensions/{id}/build AI 转换（LLM 失败回退规则版）
- POST /api/bioflow/extensions/{id}/trial-run 试运行（docker 执行占位 Executor）
- POST /api/bioflow/extensions/{id}/validate 金标/试运行校验
- POST /api/bioflow/extensions/{id}/confirm 用户确认/参数调整（nl_triggers 可编辑）
- POST /api/bioflow/extensions/{id}/repair LLM 修复（上限 20 轮可配）
- POST /api/bioflow/extensions/{id}/publish 发布（仅 published 可被 NL 触发）
- POST /api/bioflow/extensions/{id}/run 正式运行（同步返回 run_result；
SSE 由 Dev-B 执行对接）
- GET /api/bioflow/extensions/{id}/runs 运行历史
- GET /api/bioflow/extensions/{id}/runs/{run_id}/events SSE（占位：先返回
已完成 run 的快照帧；live 流由
Dev-B 执行链对接后补齐）
docker 执行：trial/run 接 extensions.ExtensionExecutor 接口（Dev-B 的
extension_exec.py 实现；inject_executor 注入）。未接入时返回 503（不是 500）。
服务层全部经 services/bioflow/extensions.py 状态机；LLM 转换不可用自动降级
规则版（extension_llm.py 保证），端点绝不因 LLM 挂掉而 500。
"""
from __future__ import annotations
import base64
import io
import json
import logging
import zipfile
from pathlib import Path
from typing import Any
from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile
from fastapi.responses import JSONResponse, StreamingResponse
from ..schemas import ExtensionConfirmRequest, ExtensionRunRequest
from ..services.bioflow import extensions as ext_mod
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/bioflow/extensions", tags=["bioflow-extensions"])
_MAX_ZIP_ENTRIES = 500 # zip 条目数上限（防炸弹）
_MAX_ZIP_TOTAL = 200 * 1024 * 1024 # 解压总量上限 200MB
def _err(code: str, message: str, status: int) -> JSONResponse:
return JSONResponse(status_code=status, content={
"ok": False,
"error": {"code": code, "message": message, "degraded": False},
})
def _store() -> ext_mod.ExtensionStore:
return ext_mod.get_store()
def _ext_or_404(ext_id: str) -> dict[str, Any]:
try:
return _store().load_extension(ext_id)
except ext_mod.ExtensionNotFoundError:
raise HTTPException(status_code=404, detail={
"ok": False,
"error": {"code": "NOT_FOUND", "message": "扩展不存在", "degraded": False},
}) from None
def _state_error(exc: ext_mod.ExtensionError, status: int = 400) -> JSONResponse:
code = "EXT_STATUS" if isinstance(exc, ext_mod.ExtensionStatusError) else (
"EXT_BUILD" if isinstance(exc, ext_mod.ExtensionBuildError) else (
"EXT_RUN" if isinstance(exc, ext_mod.ExtensionRunError) else "EXT_ERROR"))
return _err(code, str(exc), status)
# ---------------------------------------------------------------------------
# 上传/列表/详情/删除
# ---------------------------------------------------------------------------
@router.post("")
async def create_extension(
request: Request,
display_name: str | None = Form(default=None),
command: str | None = Form(default=None),
runtime_image: str = Form(default=""),
golden: str | None = Form(default=None),
code_zip: UploadFile | None = File(default=None),
test_file: UploadFile | None = File(default=None),
) -> JSONResponse:
"""上传 zip（代码包）+ 可选测试数据/金标 → 创建 draft 扩展（契约 §6）。
multipart：display_name/command 为表单字段；JSON body：同名字段在 JSON 内
（Form 参数可选化以兼容两种形态——fastapi 在 multipart 缺失表单时不再 422）。
golden 为 JSON 字符串（{test_input, expected_outputs}），测试数据可随
test_file 上传；code_files 为 base64 清单（Dev-C 前端上传卡可裁剪）。
"""
code_files: list[tuple[str, bytes]] = []
data_files: list[tuple[str, bytes]] = []
golden_dict: dict[str, Any] | None = None
content_type = request.headers.get("content-type", "")
try:
if "multipart/form-data" in content_type:
if golden:
try:
golden_dict = json.loads(golden)
except ValueError:
return _err("VALIDATION", "golden 非合法 JSON", 422)
if test_file is not None and test_file.filename:
data_files.append((test_file.filename, await test_file.read()))
if code_zip is not None and code_zip.filename:
code_files = _read_zip(await code_zip.read())
else:
# JSON 形态（前端可先不传 zip；代码文件经 base64 清单）
try:
body = await request.json()
except Exception: # noqa: BLE001
return _err("VALIDATION", "请求体须为 multipart 或 JSON", 400)
display_name = str(body.get("display_name") or display_name or "")
command = str(body.get("command") or command or "")
runtime_image = str(body.get("runtime_image") or runtime_image)
golden_dict = body.get("golden")
for item in body.get("code_files") or []:
if not isinstance(item, dict) or not item.get("path"):
continue
try:
blob = base64.b64decode(str(item.get("data") or ""))
except Exception: # noqa: BLE001
continue
code_files.append((str(item["path"]), blob))
for item in body.get("data_files") or []:
if not isinstance(item, dict) or not item.get("path"):
continue
try:
blob = base64.b64decode(str(item.get("data") or ""))
except Exception: # noqa: BLE001
continue
data_files.append((str(item["path"]), blob))
except ext_mod.ExtensionError as exc:
return _state_error(exc, 422)
if not code_files:
return _err("VALIDATION", "扩展代码包不能为空（zip 解压无代码文件）", 422)
try:
ext = _store().create_extension(
display_name=str(display_name or ""),
command=str(command or ""),
runtime_image=str(runtime_image),
code_files=code_files,
data_files=data_files,
golden=golden_dict if isinstance(golden_dict, dict) else None,
)
except ext_mod.ExtensionError as exc:
return _state_error(exc, 422)
return JSONResponse(status_code=201, content={"ok": True, "data": ext})
def _read_zip(blob: bytes) -> list[tuple[str, bytes]]:
"""zip 解压（安全护栏：条目数/总量/路径净化；失败抛 ExtensionError 语义由调用方处理）。"""
try:
zf = zipfile.ZipFile(io.BytesIO(blob))
except zipfile.BadZipFile:
raise ext_mod.ExtensionError("zip 文件损坏") from None
items: list[tuple[str, bytes]] = []
total = 0
with zf:
infos = zf.infolist()
if len(infos) > _MAX_ZIP_ENTRIES:
raise ext_mod.ExtensionError(f"zip 条目过多（> {_MAX_ZIP_ENTRIES}）")
for info in infos:
if info.is_dir():
continue
total += info.file_size
if total > _MAX_ZIP_TOTAL:
raise ext_mod.ExtensionError("zip 解压总量超限")
name = Path(info.filename).name
if not name:
continue
items.append((name, zf.read(info)))
return items
@router.get("")
async def list_extensions() -> dict[str, Any]:
items = _store().list_extensions()
return {"ok": True, "data": {"items": items, "total": len(items)}}
@router.get("/{ext_id}")
async def get_extension(ext_id: str) -> dict[str, Any]:
ext = _ext_or_404(ext_id)
return {"ok": True, "data": ext}
@router.delete("/{ext_id}")
async def delete_extension(ext_id: str) -> JSONResponse:
try:
_store().delete(ext_id)
except ext_mod.ExtensionNotFoundError:
raise HTTPException(status_code=404, detail={
"ok": False, "error": {"code": "NOT_FOUND", "message": "扩展不存在", "degraded": False},
}) from None
except ext_mod.ExtensionStatusError as exc:
return _state_error(exc, 409)
return JSONResponse(status_code=204, content=None)
# ---------------------------------------------------------------------------
# 生命周期动作（build/trial/validate/repair/confirm/publish）
# ---------------------------------------------------------------------------
@router.post("/{ext_id}/build")
async def build_extension(ext_id: str) -> JSONResponse:
_ext_or_404(ext_id)
try:
ext = await _store().build_extension(ext_id)
except ext_mod.ExtensionBuildError as exc:
return _state_error(exc, 400)
except ext_mod.ExtensionStatusError as exc:
return _state_error(exc, 409)
return JSONResponse(content={"ok": True, "data": ext})
@router.post("/{ext_id}/trial-run")
async def trial_run(ext_id: str) -> JSONResponse:
_ext_or_404(ext_id)
try:
ext = await _store().trial_run(ext_id)
except ext_mod.ExtensionStatusError as exc:
return _state_error(exc, 409)
except ext_mod.ExtensionRunError as exc:
return _state_error(exc, 503)
return JSONResponse(content={"ok": True, "data": ext})
@router.post("/{ext_id}/validate")
async def validate(ext_id: str) -> JSONResponse:
_ext_or_404(ext_id)
try:
ext = _store().validate(ext_id)
except ext_mod.ExtensionStatusError as exc:
return _state_error(exc, 409)
return JSONResponse(content={"ok": True, "data": ext})
@router.post("/{ext_id}/repair")
async def repair(ext_id: str) -> JSONResponse:
_ext_or_404(ext_id)
try:
ext = await _store().repair(ext_id)
except ext_mod.ExtensionBuildError as exc:
return _state_error(exc, 400)
except ext_mod.ExtensionStatusError as exc:
return _state_error(exc, 409)
return JSONResponse(content={"ok": True, "data": ext})
@router.post("/{ext_id}/confirm")
async def confirm(ext_id: str, body: ExtensionConfirmRequest) -> JSONResponse:
_ext_or_404(ext_id)
try:
ext = _store().confirm(
ext_id,
nl_triggers=body.nl_triggers,
parameters=body.parameters,
runtime_image=body.runtime_image,
)
except ext_mod.ExtensionStatusError as exc:
return _state_error(exc, 409)
return JSONResponse(content={"ok": True, "data": ext})
@router.post("/{ext_id}/publish")
async def publish(ext_id: str) -> JSONResponse:
_ext_or_404(ext_id)
try:
ext = _store().publish(ext_id)
except ext_mod.ExtensionStatusError as exc:
return _state_error(exc, 409)
return JSONResponse(content={"ok": True, "data": ext})
# ---------------------------------------------------------------------------
# 运行（run/runs/events）
# ---------------------------------------------------------------------------
@router.post("/{ext_id}/run")
async def run(ext_id: str, body: ExtensionRunRequest) -> JSONResponse:
"""正式运行：同步执行并返回 run_result（run 事件 SSE 由 Dev-B 执行对接）。"""
_ext_or_404(ext_id)
try:
result = await _store().run(ext_id, params=body.params, files=body.files)
except ext_mod.ExtensionStatusError as exc:
return _state_error(exc, 409)
except ext_mod.ExtensionRunError as exc:
return _state_error(exc, 400 if "不存在" in str(exc) or "白名单" in str(exc) else 503)
return JSONResponse(content={"ok": True, "data": result})
@router.get("/{ext_id}/runs")
async def list_runs(ext_id: str) -> dict[str, Any]:
_ext_or_404(ext_id)
items = _store().list_runs(ext_id)
return {"ok": True, "data": {"items": items, "total": len(items)}}
@router.get("/{ext_id}/runs/{run_id}/events")
async def run_events(ext_id: str, run_id: str, request: Request) -> StreamingResponse:
"""运行事件 SSE（契约 §6；首帧快照）。
P0 占位实现：run 为同步执行，此处返回已完成 run 的快照帧
（run_started/run_log/run_done|run_failed）后关闭；live 进度流在 Dev-B
把执行改造为后台任务 + 事件推送后，本端点改为订阅事件总线。
"""
ext = _ext_or_404(ext_id)
result_file = _store().runs_dir(ext_id) / run_id / "run_result.json"
log_file = _store().runs_dir(ext_id) / run_id / "run.log"
def _frames():
result: dict[str, Any] = {}
if result_file.is_file():
try:
result = json.loads(result_file.read_text(encoding="utf-8"))
except (ValueError, OSError):
result = {}
log_text = ""
if log_file.is_file():
try:
log_text = log_file.read_text(encoding="utf-8")
except OSError:
log_text = ""
yield _sse_frame("extension_run_started", {
"run_id": run_id, "ext_id": ext_id,
"started_at": result.get("started_at") or "",
})
if log_text:
yield _sse_frame("extension_run_log", {"run_id": run_id, "log": log_text})
if result.get("status") == "done":
yield _sse_frame("extension_run_done", {
"run_id": run_id, "outputs": result.get("outputs") or [],
})
elif result:
yield _sse_frame("extension_run_failed", {
"run_id": run_id, "exit_code": result.get("exit_code"),
"error": result.get("error") or "", "log_tail": log_text[-2000:],
})
return StreamingResponse(
_frames(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
def _sse_frame(event: str, data: dict[str, Any]) -> bytes:
return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n".encode("utf-8")
bioflow_service/app/schemas.py
+23
View File
@@ -110,3 +110,26 @@ class ReportReviseRequest(BaseModel):
target: str | None = Field(default=None, max_length=500)
instruction: str = Field(min_length=1, max_length=2000)
params: dict[str, Any] = Field(default_factory=dict)
# ---- extensions（用户自定义扩展分析；契约 §2/§6）----
EXTENSION_STATUSES = (
"draft", "building", "built", "trial_running", "trial_passed", "trial_failed",
"validating", "validated", "repairing", "confirming", "published", "build_failed",
)
ExtensionStatus = Literal[
"draft", "building", "built", "trial_running", "trial_passed", "trial_failed",
"validating", "validated", "repairing", "confirming", "published", "build_failed",
]
class ExtensionRunRequest(BaseModel):
params: dict[str, Any] = Field(default_factory=dict)
files: list[str] = Field(default_factory=list, max_length=50)
class ExtensionConfirmRequest(BaseModel):
nl_triggers: list[str] | None = Field(default=None, max_length=50)
parameters: list[dict[str, Any]] | None = None
runtime_image: str | None = Field(default=None, max_length=128)
bioflow_service/app/services/bioflow/events.py
+5
View File
@@ -26,6 +26,11 @@ EVENT_TYPES = (
"job_created", "plan_ready", "stage_started", "stage_log", "stage_done",
"stage_failed", "stage_skipped", "stage_await_confirm", "stage_await_files",
"job_done", "job_paused", "job_archived", "job_recovered",
# 用户自定义扩展分析事件帧（契约 §4；并行链路，不触碰现有 job_* 语义）
"extension_created", "extension_build", "extension_trial", "extension_repair",
"extension_validate", "extension_published", "extension_candidates",
"extension_confirm", "extension_run_started", "extension_run_log",
"extension_run_done", "extension_run_failed",
)
# 断线重连时需要回放的关键状态事件（含历史事件全量）
bioflow_service/app/services/bioflow/extension_llm.py
+511
View File
@@ -0,0 +1,511 @@
"""app/services/bioflow/extension_llm.py —— 扩展 build/repair 的 AI 转换（DeepSeek 直连）。
- LLM 配置键：DOCGEN_LLM_*（沿用 docgen 直连模式）；BIOFLOW_EXTENSION_*
为可选别名（缺 DOCGEN 值时尝试）。
- LLM 不可用/失败：一律回退规则版（_command_to_template/_detect_requirements/
_infer_language 概念移植自 gfa_core/extension_stage/manager.py），**绝不抛 500**；
回退无法完成才抛 LLMUnavailable（上层收敛 build_failed）。
- 复用 app/services/bioflow/llm.py 的 complete_json（httpx 直连 + JSON 修复重试）。
服务层纪律：纯 stdlib + httpx，无 pydantic。
"""
from __future__ import annotations
import json
import logging
import os
import re
import shlex
from pathlib import Path
from typing import Any
from . import extension_spec as spec
logger = logging.getLogger(__name__)
class LLMUnavailable(RuntimeError):
"""LLM 不可用且规则版无法完成转换（上层收敛 build_failed，不冒 500）。"""
# 标准库白名单（_detect_requirements 移植：这些 import 不视为第三方依赖）
STDLIB_MODULES = {
"os", "sys", "re", "json", "csv", "argparse", "pathlib", "math",
"statistics", "subprocess", "time", "datetime", "collections",
"itertools", "functools", "glob", "gzip", "shutil", "logging", "random",
"string", "typing", "io", "tempfile", "warnings", "hashlib", "urllib",
}
CODE_SUFFIXES = {".py", ".r", ".sh", ".bash", ".command"}
_LLM_SYSTEM = (
"You convert an already-runnable bioinformatics script package into a "
"PlantAI custom extension. Reply with strict JSON only, no markdown fences."
)
def llm_config() -> dict[str, Any]:
"""DOCGEN_LLM_* 优先，BIOFLOW_EXTENSION_* 别名兜底（base/key/model/timeout）。"""
cfg: dict[str, Any] = {}
for field, keys in (
("base_url", ("DOCGEN_LLM_BASE_URL", "BIOFLOW_EXTENSION_BASE_URL")),
("api_key", ("DOCGEN_LLM_API_KEY", "BIOFLOW_EXTENSION_API_KEY")),
("model", ("DOCGEN_LLM_MODEL", "BIOFLOW_EXTENSION_MODEL")),
("timeout", ("DOCGEN_LLM_TIMEOUT", "BIOFLOW_EXTENSION_TIMEOUT")),
):
value = ""
for key in keys:
raw = os.environ.get(key, "")
if raw:
value = raw
break
cfg[field] = value
cfg["base_url"] = (cfg["base_url"] or "https://api.deepseek.com/v1").rstrip("/")
cfg["model"] = cfg["model"] or "deepseek-chat"
try:
cfg["timeout"] = float(cfg["timeout"] or "300")
except ValueError:
cfg["timeout"] = 300.0
return cfg
# ---------------------------------------------------------------------------
# 规则版探测（_detect_requirements/_infer_language 概念移植，纯函数）
# ---------------------------------------------------------------------------
def infer_language(path: str) -> str:
suffix = Path(str(path)).suffix.lower()
if suffix == ".py":
return "python"
if suffix == ".r":
return "r"
if suffix in {".sh", ".bash", ".command"}:
return "shell"
return suffix.lstrip(".") or "unknown"
def detect_requirements(code_dir: str | Path) -> dict[str, list[str]]:
"""规则版依赖探测：grep .py import / requirements.txt / .r library / .sh 命令。
返回 {"pip": [...], "apt": [...]}（契约 §2 requirements 形态）。
"""
pip: set[str] = set()
apt: set[str] = set()
root = Path(code_dir)
if not root.is_dir():
return {"pip": [], "apt": []}
for path in sorted(root.rglob("*")):
if not path.is_file():
continue
name = path.name.lower()
if name in {"requirements.txt", "pip_requirements.txt"}:
for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
dep = re.split(r"[=<>!~\s]", line)[0].strip()
if dep:
pip.add(dep)
continue
if path.suffix.lower() == ".py":
text = path.read_text(encoding="utf-8", errors="ignore")
for match in re.finditer(r"^\s*(?:import|from)\s+([A-Za-z_][A-Za-z0-9_]*)", text, re.M):
mod = match.group(1)
if mod not in STDLIB_MODULES:
pip.add(mod)
continue
if path.suffix.lower() == ".r":
text = path.read_text(encoding="utf-8", errors="ignore")
for match in re.finditer(r"\b(?:library|require)\(([^)]+)\)", text):
pkg = match.group(1).strip().strip("'\"")
if pkg:
pip.add(pkg)
continue
if path.suffix.lower() in {".sh", ".bash", ".command"}:
text = path.read_text(encoding="utf-8", errors="ignore")
apt.update(_shell_commands(text))
return {"pip": sorted(pip), "apt": sorted(apt)}
_SHELL_COMMAND_WHITELIST = {
"python", "python3", "rscript", "perl", "bash", "sh", "awk", "sed", "grep",
"cut", "sort", "uniq", "head", "tail", "cat", "echo", "cd", "mkdir", "rm",
"cp", "mv", "ls", "wc", "tr", "paste", "join", "find", "xargs", "tar", "gzip",
"gunzip", "zcat", "chmod", "export", "conda", "source", "time", "date", "ln",
"set", "if", "then", "else", "fi", "for", "while", "do", "done", "exit",
"return", "case", "esac", "unset", "shift", "read",
}
def _shell_commands(text: str) -> set[str]:
""".sh 脚本中的可执行命令探测（_detect_shell_commands 移植裁剪）。
白名单外且形如命令名者计为 apt 依赖（如 fastqc/samtools）。
"""
commands: set[str] = set()
for raw in text.splitlines():
line = raw.strip()
if not line or line.startswith("#"):
continue
try:
parts = shlex.split(line, comments=True)
except ValueError:
continue
if not parts:
continue
cmd = Path(parts[0]).name.lower()
if not re.fullmatch(r"[a-z][a-z0-9_+.-]*", cmd or ""):
continue
if cmd in _SHELL_COMMAND_WHITELIST:
continue
commands.add(cmd)
return commands
# ---------------------------------------------------------------------------
# 规则版命令模板化（LLM 失败回退；_command_to_template 概念移植）
# ---------------------------------------------------------------------------
def command_to_template(command: str, code_dir: str | Path, *, data_files: list[str] | None = None) -> dict[str, Any]:
"""规则版：把 original_command 转为 command_template + 依赖探测 + 参数派生。
步骤：
1. 已知脚本路径片段 → {code_dir}/<脚本>；
2. 输入文件 token → 语义占位符（扩展名优先，其次 {input_file}）；
3. results/ 类输出目录 → {output_dir}；裸 --out/--output_dir 类 token 补 {output_dir}；
4. 无 {output_dir} 时追加 " > {output_dir}/output.txt"（output_dir 必含硬约束）。
"""
command = str(command or "").strip()
code_dir = Path(code_dir)
data_files = data_files or []
# 1. 代码脚本 → {code_dir}/<name>
template = command
for path in code_dir.rglob("*"):
if not path.is_file() or path.suffix.lower() not in CODE_SUFFIXES:
continue
name = path.name
template = re.sub(
rf"(?<![A-Za-z0-9_./-]){re.escape(name)}\b",
f"{{code_dir}}/{name}",
template,
)
template = re.sub(r"\{code_dir\}/?\{code_dir\}", "{code_dir}", template)
# 2. 输入文件 → 语义占位符（扩展名优先）
known_inputs = {Path(x).name for x in data_files}
for fname in sorted(known_inputs, key=len, reverse=True):
placeholder = _placeholder_for_filename(fname)
template = re.sub(
rf"(?<![A-Za-z0-9_./-]){re.escape(fname)}\b",
placeholder,
template,
)
# 3. 输出目录 → {output_dir}
for token in ("results/", "result/", "out_dir/", "output_dir/", "outputs/", "out/", "output/"):
template = template.replace(token, "{output_dir}/")
template = re.sub(r"\{output_dir\}+/+", "{output_dir}/", template)
template = template.replace("{output_dir}//", "{output_dir}/")
# 4. 末尾裸目录 token（--out xxx 等）→ {output_dir}
for token in ("--out", "--outdir", "--output", "--output_dir", "--output-dir", "-o"):
template = re.sub(
rf"({re.escape(token)})\s+([A-Za-z0-9_./-]+/?)",
lambda m: f"{m.group(1)} {{output_dir}}",
template,
)
# 5. 无 {output_dir}：追加重定向兜底（output_dir 必含硬约束）
if "{output_dir}" not in template:
template = f"{template.rstrip()} > {{output_dir}}/output.txt"
# 6. 依赖探测 + 参数派生
requirements = detect_requirements(code_dir)
parameters: list[dict[str, Any]] = []
seen: set[str] = set()
for match in re.finditer(r"\{param_([A-Za-z_][A-Za-z0-9_]*)\}", template):
name = match.group(1)
if name not in seen:
seen.add(name)
parameters.append({"name": name, "type": "string", "required": True,
"description": "", "default": None})
return {
"command_template": template,
"requirements": requirements,
"parameters": parameters,
"source": "rules",
}
def _placeholder_for_filename(fname: str) -> str:
lower = fname.lower()
if lower.endswith(".hmm"):
return "{input_hmm}"
if lower.endswith((".pep.fasta", ".pep.fa", ".pep")):
return "{input_pep_fasta}"
if lower.endswith((".cds.fasta", ".cds.fa")):
return "{input_cds_fasta}"
if lower.endswith((".fasta", ".fa", ".faa", ".fna")):
return "{input_fasta}"
if lower.endswith((".id", ".ids", ".geneid", ".geneids", ".list", ".lst")):
return "{input_id_file}"
return "{input_file}"
# ---------------------------------------------------------------------------
# build：LLM 转换（失败回退规则版）
# ---------------------------------------------------------------------------
def _sanitize_conversion(raw: dict[str, Any]) -> dict[str, Any]:
"""LLM 输出净化：字段形态归一 + 白名单校验（spec_v2 sanitize 概念裁剪）。"""
out: dict[str, Any] = {}
template = str(raw.get("command_template") or "").strip()
spec.validate_command_template(template) # 不通过抛 ExtensionSpecError
out["command_template"] = template
requirements: dict[str, Any] = {}
raw_req = raw.get("requirements") or {}
if isinstance(raw_req, dict):
for key in ("pip", "apt"):
values = raw_req.get(key) or []
if isinstance(values, str):
values = [values]
requirements[key] = sorted({str(x).strip() for x in values if str(x).strip()})
out["requirements"] = {"pip": requirements.get("pip") or [], "apt": requirements.get("apt") or []}
parameters: list[dict[str, Any]] = []
raw_params = raw.get("parameters") or []
if isinstance(raw_params, dict):
raw_params = [{"name": k, **({"description": v} if isinstance(v, str) else v)} for k, v in raw_params.items()]
if isinstance(raw_params, list):
for item in raw_params:
if not isinstance(item, dict):
continue
name = spec.safe_parameter_name(item.get("name") or "")
if not name:
continue
ptype = str(item.get("type") or "string")
if ptype not in {"string", "file", "int", "float", "bool"}:
ptype = "string"
parameters.append({
"name": name,
"type": ptype,
"required": bool(item.get("required", False)),
"description": str(item.get("description") or ""),
"default": item.get("default"),
})
# 模板 {param_*} 未在 parameters 声明的一律补齐（契约完整性）
out["parameters"] = spec.template_parameters(template, parameters)
input_roles = raw.get("input_roles")
out["input_roles"] = input_roles if isinstance(input_roles, dict) else {}
nl_triggers = raw.get("nl_triggers") or []
if isinstance(nl_triggers, str):
nl_triggers = [nl_triggers]
out["nl_triggers"] = [str(x).strip() for x in nl_triggers if str(x).strip()][:50]
out["analysis_type"] = spec.normalize_analysis_type(str(raw.get("analysis_type") or ""))
runtime_image = str(raw.get("runtime_image") or raw.get("runtime_env") or "")
out["runtime_image"] = spec.RUNTIME_IMAGE_MAP.get(runtime_image, runtime_image)
out["source"] = str(raw.get("source") or "llm")
return out
async def _llm_build(ext: dict[str, Any], store: Any) -> dict[str, Any]:
"""LLM 转换：把 original_command 转 command_template + 元数据。"""
from ...config import get_settings
s = get_settings()
cfg = llm_config()
if not cfg["api_key"]:
raise LLMUnavailable("扩展转换 LLM 未配置（DOCGEN_LLM_API_KEY 为空）")
from .llm import complete_json, LLMError, LLMTimeout, LLMUnavailable as PlanLLMUnavailable
raw_code = store.ext_dir(ext["ext_id"]) / "raw_code"
code_files = _inventory(raw_code)
data_files = _inventory(store.ext_dir(ext["ext_id"]) / "raw_test_data")
user = _build_prompt(ext, code_files=code_files, data_files=data_files)
try:
payload = await complete_json(
_LLM_SYSTEM,
user,
temperature=0,
max_tokens=6000,
timeout=float(s.docgen_llm_timeout),
)
except (PlanLLMUnavailable, LLMTimeout, LLMError, ValueError):
raise
return _sanitize_conversion(payload if isinstance(payload, dict) else {})
def _build_prompt(ext: dict[str, Any], *, code_files: list[str], data_files: list[str]) -> str:
lines = [
"Convert the following extension upload into an extension spec JSON.",
"The JSON must contain exactly these keys:",
" command_template: the original_command rewritten with placeholders only from",
" the whitelist: {code_dir} {input_dir} {input_file} {input_hmm} {input_fasta}",
" {input_query_fasta} {input_reference_fasta} {input_pep_fasta} {input_cds_fasta}",
" {input_id_file} {output_dir} {work_dir}, or {param_<name>} for CLI parameters.",
" {output_dir} is REQUIRED (the extension must write outputs below it).",
" requirements: {\"pip\": [...], \"apt\": [...]} - third-party dependencies only",
" (stdlib imports are not dependencies).",
" parameters: [{\"name\", \"type\": string|file|int|float|bool, \"required\",",
" \"description\", \"default\"}] for each {param_*} used.",
" input_roles: {\"<placeholder>\": \"<role>\"} for semantic placeholders used.",
" nl_triggers: 3-6 short Chinese phrases users would say to invoke this",
" extension in conversation (plus the display_name itself).",
" analysis_type: one of " + ", ".join(sorted(spec.ANALYSIS_TYPES)) + ".",
" runtime_image: one of " + ", ".join(sorted(spec.RUNTIME_IMAGE_MAP.values())) + ".",
"",
f"display_name: {ext.get('display_name')}",
f"original_command: {ext.get('original_command')}",
f"runtime_image (uploaded): {ext.get('runtime_image')}",
"code files: " + (", ".join(code_files[:30]) or "(none)"),
"test data files: " + (", ".join(data_files[:20]) or "(none)"),
]
return "\n".join(lines)
def _inventory(root: Path) -> list[str]:
if not root.is_dir():
return []
return sorted(
str(p.relative_to(root)) for p in root.rglob("*") if p.is_file()
)[:200]
def _rules_build(ext: dict[str, Any], store: Any) -> dict[str, Any]:
"""规则版回退：command_to_template + 依赖探测 + analysis_type 推断。"""
raw_code = store.ext_dir(ext["ext_id"]) / "raw_code"
data_files = _inventory(store.ext_dir(ext["ext_id"]) / "raw_test_data")
converted = command_to_template(
ext.get("original_command") or "",
raw_code,
data_files=[Path(x).name for x in data_files],
)
spec.validate_command_template(converted["command_template"])
converted["parameters"] = spec.template_parameters(converted["command_template"], converted.get("parameters"))
converted["input_roles"] = {}
converted["nl_triggers"] = _default_triggers(ext, converted["command_template"])
converted["analysis_type"] = spec.infer_analysis_type(
template=converted["command_template"],
display_name=str(ext.get("display_name") or ""),
requirements=converted["requirements"],
)
converted["runtime_image"] = str(ext.get("runtime_image") or spec.DEFAULT_RUNTIME_IMAGE)
converted["source"] = "rules"
return converted
def _default_triggers(ext: dict[str, Any], template: str) -> list[str]:
"""无 LLM 时的 nl_triggers 兜底：display_name + analysis_type 关键词。"""
name = str(ext.get("display_name") or "").strip()
triggers = [name] if name else []
atype = spec.infer_analysis_type(
template=template, display_name=name,
requirements=ext.get("requirements") or {},
)
keywords = {
"blast_search": "blast 比对",
"hmmer_search": "hmm 搜索",
"table_filter": "表格过滤",
"sequence_extraction": "序列提取",
"plot_render": "绘图",
}
if atype in keywords and keywords[atype] not in triggers:
triggers.append(keywords[atype])
return [t for t in triggers if t]
async def build_conversion(ext: dict[str, Any], *, store: Any) -> dict[str, Any]:
"""build 主入口：LLM 优先，失败回退规则版；两者皆失败抛 LLMUnavailable。"""
try:
return await _llm_build(ext, store)
except Exception as exc: # noqa: BLE001 - 全部 LLM 失败路径（连接/超时/鉴权/坏 JSON）
logger.warning("extensions: LLM build 失败，回退规则版: %s", exc)
try:
return _rules_build(ext, store)
except Exception: # noqa: BLE001
logger.exception("extensions: 规则版 build 也失败")
raise LLMUnavailable("扩展转换失败：LLM 与规则版回退均无法完成") from None
# ---------------------------------------------------------------------------
# repair：失败上下文 → LLM 修复（失败回退规则版）
# ---------------------------------------------------------------------------
async def repair_conversion(ext: dict[str, Any], *, failure_context: dict[str, Any], store: Any) -> dict[str, Any]:
"""repair：把 trial 失败日志交给 LLM 修复 command_template（或 requirements）。
LLM 失败回退规则版（log_tail 里出现 ModuleNotFoundError 时把缺失模块并入
requirements——B2「AI 修复循环兜底」的最小规则实现；镜像重建由 Dev-B）。
"""
cfg = llm_config()
if cfg["api_key"]:
from ...config import get_settings
from .llm import complete_json
log_tail = str(failure_context.get("log_tail") or "")[-4000:]
user = (
"The extension trial run failed. Repair the extension spec JSON "
"(same keys as build: command_template with whitelist placeholders "
"and REQUIRED {output_dir}, requirements, parameters, input_roles, "
"nl_triggers, analysis_type, runtime_image, plus a \"suggestion\" "
"string describing the fix in Chinese).\n\n"
f"current extension.json: {json.dumps(_repair_view(ext), ensure_ascii=False)}\n"
f"trial log tail:\n{log_tail}\n"
)
try:
payload = await complete_json(
_LLM_SYSTEM,
user,
temperature=0,
max_tokens=6000,
timeout=float(get_settings().docgen_llm_timeout),
)
if isinstance(payload, dict) and payload.get("command_template"):
converted = _sanitize_conversion(payload)
converted["suggestion"] = str(payload.get("suggestion") or "")
converted["source"] = "repair_llm"
return converted
except Exception as exc: # noqa: BLE001
logger.warning("extensions: LLM repair 失败，回退规则版: %s", exc)
# 规则版：ModuleNotFoundError → requirements 补缺；模板保持
template = str(ext.get("command_template") or "")
if not template:
raise LLMUnavailable("repair 需要已有 command_template")
log_tail = str(failure_context.get("log_tail") or "")
added: list[str] = []
for match in re.finditer(r"No module named ['\"]?([A-Za-z_][A-Za-z0-9_]*)", log_tail):
added.append(match.group(1))
requirements = dict(ext.get("requirements") or {"pip": [], "apt": []})
requirements["pip"] = sorted({str(x) for x in (requirements.get("pip") or []) if str(x)} | set(added))
if not added:
raise LLMUnavailable("repair 失败：LLM 不可用且日志中未发现缺失依赖，无法规则修复")
return {
"command_template": template,
"requirements": {"pip": requirements.get("pip") or [], "apt": requirements.get("apt") or []},
"parameters": ext.get("parameters") or [],
"input_roles": ext.get("input_roles") or {},
"nl_triggers": ext.get("nl_triggers") or [],
"analysis_type": ext.get("analysis_type") or spec.DEFAULT_ANALYSIS_TYPE,
"runtime_image": ext.get("runtime_image") or spec.DEFAULT_RUNTIME_IMAGE,
"suggestion": "已并入缺失依赖: " + ", ".join(added),
"source": "repair_rules",
}
def _repair_view(ext: dict[str, Any]) -> dict[str, Any]:
"""传给 LLM 的扩展视图（剔除易超长字段：source_files 明细等）。"""
keys = ("ext_id", "display_name", "status", "command_template", "original_command",
"runtime_image", "requirements", "parameters", "input_roles",
"nl_triggers", "analysis_type", "repair_count")
return {k: ext.get(k) for k in keys}
bioflow_service/app/services/bioflow/extension_spec.py
+184
View File
@@ -0,0 +1,184 @@
"""app/services/bioflow/extension_spec.py —— 扩展 command_template 白名单契约（spec_v2 移植）。
从 gfa_core/extension_stage/spec_v2.py 只移植契约常量与纯校验函数（旧模块只读）：
- 占位符白名单 12 个 + {param_<name>} 受控参数占位符；
- 硬约束：{output_dir} 必含；白名单外占位符拒绝；
- analysis_type 枚举（发布扩展的 intent 打分维度之一）；
- template → parameters 派生（未配置 parameters 时从 {param_*} 占位符生成）。
服务层纪律：纯 stdlib，无 pydantic、无 httpx。
"""
from __future__ import annotations
import re
from typing import Any
SPEC_VERSION = "extension_stage_v1"
# ---- 占位符白名单（契约 §2，spec_v2.PLACEHOLDERS 原表 12 个）----
PLACEHOLDERS = {
"code_dir",
"input_dir",
"input_file",
"input_hmm",
"input_fasta",
"input_query_fasta",
"input_reference_fasta",
"input_pep_fasta",
"input_cds_fasta",
"input_id_file",
"output_dir",
"work_dir",
}
# 语义占位符（与旧版语义同表：绑定具体输入文件的角色化占位符）
SEMANTIC_PLACEHOLDERS = {
"input_hmm",
"input_fasta",
"input_query_fasta",
"input_reference_fasta",
"input_pep_fasta",
"input_cds_fasta",
"input_id_file",
}
# ---- analysis_type 枚举（spec_v2.ANALYSIS_TYPES 移植）----
ANALYSIS_TYPES = {
"sequence_extraction",
"blast_search",
"hmmer_search",
"table_filter",
"id_cleaning",
"fasta_header_cleaning",
"gff_processing",
"plot_render",
"enrichment_or_stats",
"generic_cli",
}
DEFAULT_ANALYSIS_TYPE = "generic_cli"
# 默认运行时镜像（B2 决策：工具链最全的 bioflow-gf）
DEFAULT_RUNTIME_IMAGE = "bioflow-gf"
# 运行时环境 → 镜像映射（B2 定稿；上传/LLM 推荐时翻译旧 conda 环境名）
RUNTIME_IMAGE_MAP = {
"test4py311": "bioflow-gf",
"gfpy312": "bioflow-gfpy",
"rnaseq": "bioflow-rnaseq",
"rnaR": "bioflow-rnaR",
"ASAW": "bioflow-wgs",
"interproscanpy311": "bioflow-interproscan",
"panfamily": "bioflow-pan",
"r_de_analysis": "bioflow-rde",
}
_PARAM_RE = re.compile(r"\{([A-Za-z_][A-Za-z0-9_]*)\}")
class ExtensionSpecError(RuntimeError):
"""command_template 违反扩展契约（白名单/必含约束）。"""
def used_placeholders(template: str) -> set[str]:
"""模板中出现的全部 {占位符} 名（不含 {param_*} 前缀剥离）。"""
return set(_PARAM_RE.findall(template or ""))
def unsupported_placeholders(template: str) -> set[str]:
"""白名单外且非 {param_*} 的占位符名集合。"""
names = used_placeholders(template)
return {name for name in names if name not in PLACEHOLDERS and not name.startswith("param_")}
def param_placeholder_names(template: str) -> set[str]:
"""{param_<name>} 占位符的参数名集合（剔除前缀）。"""
return {
name[len("param_"):]
for name in used_placeholders(template)
if name.startswith("param_") and len(name) > len("param_")
}
def validate_command_template(template: str) -> None:
"""command_template 契约校验：{output_dir} 必含；白名单外占位符拒绝。
通过无异常；违反抛 ExtensionSpecError（spec_v2.normalize_spec_v2 同序）。
"""
if not (template or "").strip():
raise ExtensionSpecError("command_template 不能为空")
if "{output_dir}" not in template:
raise ExtensionSpecError("command_template 必须包含 {output_dir}（输出契约硬约束）")
unsupported = sorted(unsupported_placeholders(template))
if unsupported:
raise ExtensionSpecError("command_template 含白名单外占位符: " + ", ".join(unsupported))
def normalize_analysis_type(value: str | None) -> str:
"""analysis_type 白名单归一：非法值回落 generic_cli（漏判安全）。"""
v = str(value or "").strip().lower()
return v if v in ANALYSIS_TYPES else DEFAULT_ANALYSIS_TYPE
def infer_analysis_type(*, template: str, display_name: str, requirements: dict[str, Any] | None = None) -> str:
"""规则版 analysis_type 探测（spec_v2.infer_analysis_type 移植裁剪：模板/依赖/名称关键词）。"""
t = (template or "").lower()
pip_deps = " ".join(
str(x).lower() for x in ((requirements or {}).get("pip") or [])
)
name = (display_name or "").lower()
if "blastp" in t or "blastp" in pip_deps:
return "blast_search"
if any(k in t or k in pip_deps for k in ("hmmsearch", "hmmscan", "hmmbuild", "hmmer")):
return "hmmer_search"
if any(k in t for k in ("awk", "cut ", "grep ")):
return "table_filter"
if any(k in name for k in ("clean_id", "id_clean", "header_clean")):
return "fasta_header_cleaning"
if "seqtk" in t and "subseq" in t:
return "sequence_extraction"
if "matplotlib" in t or "plot" in name or any(k in name for k in ("画图", "绘图", "作图")):
return "plot_render"
return DEFAULT_ANALYSIS_TYPE
def template_parameters(template: str, existing: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]:
"""由 {param_*} 占位符派生/补齐 parameters（已配置的定义优先，缺省的补齐为空定义）。"""
merged: dict[str, dict[str, Any]] = {}
for item in existing or []:
if not isinstance(item, dict):
continue
name = safe_parameter_name(item.get("name") or "")
if not name:
continue
merged[name] = {
"name": name,
"type": str(item.get("type") or "string"),
"required": bool(item.get("required", False)),
"description": str(item.get("description") or ""),
"default": item.get("default"),
}
for name in sorted(param_placeholder_names(template)):
merged.setdefault(name, {
"name": name,
"type": "string",
"required": False,
"description": "",
"default": None,
})
return [merged[name] for name in sorted(merged)]
def safe_parameter_name(value: Any) -> str:
"""参数名净化（spec_v2.safe_parameter_name 同语义：非字母数字下划线→_，去 param_ 前缀）。"""
text = re.sub(r"[^A-Za-z0-9_]+", "_", str(value or "").strip())
text = re.sub(r"_+", "_", text).strip("_")
if not text:
return ""
if text[0].isdigit():
text = "p_" + text
if text.startswith("param_"):
text = text[len("param_"):]
return text
bioflow_service/app/services/bioflow/extensions.py
+884
View File
@@ -0,0 +1,884 @@
"""app/services/bioflow/extensions.py —— 扩展生命周期状态机与文件存储（契约 §1/§2/§4 服务端）。
从 gfa_core/extension_stage/manager.py 只取生命周期概念（draft→build→trial→validate
→confirm→repair→publish→run），按现行服务风格重写：纯 stdlib + dataclass/dict，
零新增依赖。旧模块只读，不做任何 import。
职责：
- 存储：{data_root}/extensions/{ext_id}/{extension.json, raw_code/, raw_test_data/,
converted_code/, validation/, runs/}（契约 §1 布局）。
- extension.json 唯一事实源（契约 §2 schema）；写盘 = 读-合并-原子替换，
同进程多协程交错写不丢字段。
- 状态机：契约 §2 status 枚举（合法流转表 _TRANSITIONS，非法流转 409）。
- docker 执行占位：trial/run 接 ExtensionExecutor 接口（extension_exec.py 由
Dev-B 实现；本模块定义接口签名并注入 ctx.executor）。
- build/repair 的 LLM 转换委托 extension_llm.py（LLM 不可用时规则版回退，
绝不抛 500）。
服务层纪律：纯 dataclass/dict/stdlib，无 pydantic、无 httpx。
"""
from __future__ import annotations
import inspect
import json
import logging
import shutil
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from ...db import now_iso
from . import extension_spec as spec
logger = logging.getLogger(__name__)
# ---- 契约 §2 状态枚举 ----
STATUS_DRAFT = "draft"
STATUS_BUILDING = "building"
STATUS_BUILT = "built"
STATUS_TRIAL_RUNNING = "trial_running"
STATUS_TRIAL_PASSED = "trial_passed"
STATUS_TRIAL_FAILED = "trial_failed"
STATUS_VALIDATING = "validating"
STATUS_VALIDATED = "validated"
STATUS_REPAIRING = "repairing"
STATUS_CONFIRMING = "confirming"
STATUS_PUBLISHED = "published"
STATUS_BUILD_FAILED = "build_failed"
EXTENSION_STATUSES = {
STATUS_DRAFT, STATUS_BUILDING, STATUS_BUILT, STATUS_TRIAL_RUNNING,
STATUS_TRIAL_PASSED, STATUS_TRIAL_FAILED, STATUS_VALIDATING, STATUS_VALIDATED,
STATUS_REPAIRING, STATUS_CONFIRMING, STATUS_PUBLISHED, STATUS_BUILD_FAILED,
}
# 试运行可进入的状态（含重试）
TRIAL_RETRYABLE = {
STATUS_BUILT, STATUS_TRIAL_RUNNING, STATUS_TRIAL_FAILED, STATUS_TRIAL_PASSED,
STATUS_VALIDATED,
}
# 正式运行状态
RUN_DONE = "done"
RUN_FAILED = "failed"
# 输出契约默认：输出目录非空即通过（契约 §3；run_extension.py 退出码 3）
OUTPUT_CONTRACT_EXIT = 3
class ExtensionError(RuntimeError):
"""扩展状态机业务错误（路由层转 4xx/503）。"""
class ExtensionNotFoundError(ExtensionError):
"""扩展不存在（404）。"""
class ExtensionStatusError(ExtensionError):
"""状态不可流转（409）。"""
class ExtensionBuildError(ExtensionError):
"""build/repair 转换失败（含规则版回退后的最终失败；状态已收敛 build_failed）。"""
class ExtensionRunError(ExtensionError):
"""执行器未接入/运行前置失败（503/400）。"""
_ID_COUNTER = 0
_ID_LOCK = threading.Lock()
def _short_id() -> str:
"""唯一短 id：时间戳 + 线程号 + 进程内自增计数（同秒同线程不冲突）。"""
global _ID_COUNTER
with _ID_LOCK:
_ID_COUNTER = (_ID_COUNTER + 1) % 1000
counter = _ID_COUNTER
return f"{int(time.time()) % 1000000:06d}{threading.get_ident() % 1000:03d}{counter:03d}"
def new_ext_id() -> str:
return f"ext_{_short_id()}"
def new_run_id() -> str:
return f"run_{_short_id()}"
def safe_filename(value: str, fallback: str = "item") -> str:
"""zip 条目名净化（去路径穿越；保留中文；空名给 fallback）。"""
name = Path(str(value or "")).name or fallback
return name.replace("\x00", "") or fallback
def _merge(target: dict[str, Any], patch: dict[str, Any]) -> None:
for key, value in patch.items():
target[key] = value
async def _call_executor(executor: Any, spec_in: ExtensionRunSpec) -> Any:
"""执行器统一调用：兼容 async 方法/协程函数/同步函数三类实现。
Dev-B 的真实执行器为 async（docker subprocess）；测试 fake 可 sync。
"""
raw = executor.execute(spec_in)
if inspect.isawaitable(raw):
raw = await raw
return raw
class _JsonLock:
"""extension.json 写锁（RLock：_apply_and_write 的读-改-写全程持锁）。"""
def __init__(self) -> None:
self._lock = threading.RLock()
def __enter__(self) -> "threading.RLock":
self._lock.acquire()
return self._lock
def __exit__(self, *exc: Any) -> None:
self._lock.release()
# ---- Executor 接口（Dev-B 实现 extension_exec.py；签名即契约）----
@dataclass
class ExtensionRunSpec:
"""ExtensionExecutor.execute 入参（Dev-B 拼装 docker 命令的唯一依据，契约 §3）。"""
ext_dir: str # {data_root}/extensions/{ext_id}（宿主路径）
input_dir: str # 输入目录（宿主路径；可为空目录）
output_dir: str # 输出目录（宿主路径，{output_dir} 渲染目标）
output_contract_exit: int = OUTPUT_CONTRACT_EXIT # 输出非空校验退出码
params: dict[str, Any] = field(default_factory=dict) # {param_*} 与运行时参数
cpu: str | None = None
memory: str | None = None
runtime_image: str = spec.DEFAULT_RUNTIME_IMAGE
log_path: str | None = None # 可选：全量日志落点（宿主路径）
@dataclass
class ExtensionRunResult:
"""ExtensionExecutor.execute 返回。
Executor 承诺：exit_code=0 表示命令成功且输出目录非空；非 0/超时/启动
失败一律算失败并给 error 文案；log_tail 为截尾日志文本。
"""
run_id: str
exit_code: int
log_tail: str
outputs: list[str] = field(default_factory=list) # 输出文件相对 output_dir 路径
error: str = "" # 失败原因（命令失败/启动失败）
timed_out: bool = False
started_at: str = ""
finished_at: str = ""
class ExtensionExecutor:
"""docker 执行占位（Dev-B 的 extension_exec.py 替换实现）。
本模块经 inject_executor(executor) 注入；路由层从 context 取。
占位实现 = 留痕执行（不走 docker，也不假装成功）。
"""
async def execute(self, spec_in: ExtensionRunSpec) -> ExtensionRunResult:
raise ExtensionRunError(
"扩展执行器未接入（extension_exec.py 由 Dev-B 实现；"
"接口签名见 extensions.ExtensionExecutor）"
)
# ---- 扩展生命周期状态机 ----
# 合法流转表（契约 §2 status 枚举；模块级常量，dataclass 外）
TRANSITIONS: dict[str, set[str]] = {
STATUS_DRAFT: {STATUS_BUILDING, STATUS_BUILD_FAILED},
STATUS_BUILDING: {STATUS_BUILT, STATUS_BUILD_FAILED},
STATUS_BUILT: {STATUS_TRIAL_RUNNING, STATUS_BUILDING},
STATUS_TRIAL_RUNNING: {STATUS_TRIAL_PASSED, STATUS_TRIAL_FAILED},
STATUS_TRIAL_PASSED: {STATUS_VALIDATING, STATUS_TRIAL_RUNNING, STATUS_CONFIRMING, STATUS_BUILDING},
STATUS_TRIAL_FAILED: {STATUS_REPAIRING, STATUS_TRIAL_RUNNING, STATUS_BUILDING},
STATUS_VALIDATING: {STATUS_VALIDATED, STATUS_TRIAL_FAILED},
STATUS_VALIDATED: {STATUS_CONFIRMING, STATUS_PUBLISHED, STATUS_BUILDING, STATUS_TRIAL_RUNNING},
STATUS_REPAIRING: {STATUS_BUILT, STATUS_TRIAL_RUNNING, STATUS_TRIAL_FAILED, STATUS_BUILD_FAILED},
STATUS_CONFIRMING: {STATUS_PUBLISHED, STATUS_VALIDATED, STATUS_TRIAL_RUNNING, STATUS_TRIAL_PASSED},
STATUS_PUBLISHED: set(), # 终态：只可 run/删除
STATUS_BUILD_FAILED: {STATUS_BUILDING, STATUS_REPAIRING, STATUS_DRAFT},
}
@dataclass
class ExtensionStore:
"""扩展存储（文件布局 §1）。"""
base_dir: str # {data_root}/extensions
executor: ExtensionExecutor | None = None # 注入的执行器（Dev-B 替换点）
max_repair_rounds: int = 20 # repair 上限（契约 §6 可配）
def __post_init__(self) -> None:
Path(self.base_dir).mkdir(parents=True, exist_ok=True)
self._locks: dict[str, _JsonLock] = {}
self._lock_guard = threading.Lock()
# ---- 路径 ----
def ext_dir(self, ext_id: str) -> Path:
return Path(self.base_dir) / str(ext_id)
def manifest_path(self, ext_id: str) -> Path:
return self.ext_dir(ext_id) / "extension.json"
def runs_dir(self, ext_id: str) -> Path:
return self.ext_dir(ext_id) / "runs"
def validation_dir(self, ext_id: str) -> Path:
return self.ext_dir(ext_id) / "validation"
def _lock_for(self, ext_id: str) -> _JsonLock:
with self._lock_guard:
return self._locks.setdefault(str(ext_id), _JsonLock())
# ---- 持久化 ----
def save_extension(self, ext: dict[str, Any]) -> dict[str, Any]:
"""extension.json 落盘：读-合并-原子替换（本次字段覆盖磁盘版本）。"""
ext_id = str(ext.get("ext_id") or "")
ext["updated_at"] = now_iso()
path = self.manifest_path(ext_id)
if path.exists():
try:
on_disk = json.loads(path.read_text(encoding="utf-8"))
if isinstance(on_disk, dict):
merged = dict(on_disk)
_merge(merged, ext)
ext = merged
except (ValueError, OSError):
pass
with self._lock_for(ext_id):
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(".json.tmp")
tmp.write_text(json.dumps(ext, ensure_ascii=False, indent=2), encoding="utf-8")
tmp.replace(path)
return ext
def load_extension(self, ext_id: str, *, required: bool = True) -> dict[str, Any] | None:
path = self.manifest_path(ext_id)
if not path.exists():
if required:
raise ExtensionNotFoundError(f"扩展不存在: {ext_id}")
return None
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (ValueError, OSError) as exc:
if required:
raise ExtensionError(f"extension.json 损坏: {ext_id}: {exc}") from exc
return None
if not isinstance(data, dict):
if required:
raise ExtensionError(f"extension.json 非对象: {ext_id}")
return None
return data
# ---- 状态机（合法流转表；契约 §2）----
def _set_status(self, ext: dict[str, Any], new_status: str) -> None:
old = str(ext.get("status") or STATUS_DRAFT)
if new_status not in EXTENSION_STATUSES:
raise ExtensionStatusError(f"未知状态: {new_status}")
if old not in TRANSITIONS:
raise ExtensionStatusError(f"未知当前状态: {old}")
if new_status != old and new_status not in TRANSITIONS[old]:
raise ExtensionStatusError(f"状态不可流转: {old} → {new_status}")
ext["status"] = new_status
def _apply_and_write(self, ext_id: str, mutate: Any) -> dict[str, Any]:
"""读-改-写骨架（全程持该扩展锁）：mutate(fresh) 内可 _set_status。"""
with self._lock_for(ext_id):
fresh = self.load_extension(ext_id)
mutate(fresh)
return self.save_extension(fresh)
# ---- 创建（POST /extensions）----
def create_extension(
self,
*,
display_name: str,
command: str,
runtime_image: str = "",
code_files: list[tuple[str, bytes]] | None = None,
data_files: list[tuple[str, bytes]] | None = None,
golden: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""创建扩展（draft）：保存代码/测试数据包，登记 extension.json。
code_files: [(rel_path, bytes)] zip 解压条目（路由层完成解压；服务层
只做落盘与净化）；data_files: 测试数据/金标输入。
"""
name = str(display_name or "").strip()
if not name:
raise ExtensionError("display_name 不能为空")
if not (command or "").strip():
raise ExtensionError("运行命令（command）不能为空")
if not code_files:
raise ExtensionError("扩展代码包不能为空")
ext_id = new_ext_id()
ext_dir = self.ext_dir(ext_id)
raw_code = ext_dir / "raw_code"
raw_test_data = ext_dir / "raw_test_data"
converted_code = ext_dir / "converted_code"
for path in (raw_code, raw_test_data, converted_code, self.validation_dir(ext_id)):
path.mkdir(parents=True, exist_ok=True)
saved_code: list[str] = []
for rel, blob in code_files:
dest = raw_code / safe_filename(rel)
dest.write_bytes(blob)
saved_code.append(str(dest.relative_to(ext_dir)))
mirror = converted_code / dest.name
mirror.write_bytes(blob) # converted_code 初始 = raw_code 拷贝（175 同款）
saved_data: list[str] = []
for rel, blob in data_files or []:
dest = raw_test_data / safe_filename(rel)
dest.write_bytes(blob)
saved_data.append(str(dest.relative_to(ext_dir)))
runtime_image = spec.RUNTIME_IMAGE_MAP.get(runtime_image, runtime_image) or spec.DEFAULT_RUNTIME_IMAGE
now = now_iso()
ext: dict[str, Any] = {
"schema_version": spec.SPEC_VERSION,
"ext_id": ext_id,
"display_name": name,
"status": STATUS_DRAFT,
"command_template": "",
"original_command": str(command).strip(),
"runtime_image": runtime_image,
"requirements": {"pip": [], "apt": []},
"parameters": [],
"input_roles": {},
"nl_triggers": [],
"analysis_type": spec.DEFAULT_ANALYSIS_TYPE,
"golden": golden if isinstance(golden, dict) else None,
"repair_count": 0,
"created_at": now,
"updated_at": now,
"trial": None,
"validation": None,
"last_build": None,
"source_files": {"code": saved_code, "test_data": saved_data},
}
ext = self.save_extension(ext)
self._publish("extension_created", {
"ext_id": ext_id, "display_name": name, "status": STATUS_DRAFT,
})
logger.info("extensions: created %s (%s)", ext_id, name)
return ext
# ---- build（POST /{id}/build：AI 转换）----
async def build_extension(self, ext_id: str) -> dict[str, Any]:
self._apply_and_write(ext_id, lambda ext: self._set_status(ext, STATUS_BUILDING))
# LLM 转换委托 extension_llm（内部保证：LLM 失败回退规则版，绝不抛 500）
from .extension_llm import build_conversion
try:
conversion = await build_conversion(self.load_extension(ext_id), store=self)
except ExtensionError as exc:
# 失败收敛 build_failed（契约状态枚举）再抛出（路由层 4xx，非 500）
try:
self._apply_and_write(ext_id, lambda ext: self._set_status(ext, STATUS_BUILD_FAILED))
except ExtensionStatusError:
pass
raise ExtensionBuildError(str(exc)) from exc
except Exception as exc: # noqa: BLE001 - 防御：未知异常同样收敛，绝不 500
try:
self._apply_and_write(ext_id, lambda ext: self._set_status(ext, STATUS_BUILD_FAILED))
except ExtensionStatusError:
pass
raise ExtensionBuildError(f"扩展转换失败: {exc}") from exc
template = str(conversion.get("command_template") or "")
spec.validate_command_template(template) # 白名单契约终检（不通过抛 → build_failed）
fresh = self.load_extension(ext_id)
fresh["command_template"] = template
fresh["requirements"] = conversion.get("requirements") or {"pip": [], "apt": []}
fresh["runtime_image"] = str(conversion.get("runtime_image") or fresh.get("runtime_image") or spec.DEFAULT_RUNTIME_IMAGE)
fresh["parameters"] = spec.template_parameters(
template, conversion.get("parameters") or fresh.get("parameters") or []
)
fresh["input_roles"] = conversion.get("input_roles") or {}
fresh["nl_triggers"] = [str(x).strip() for x in (conversion.get("nl_triggers") or []) if str(x).strip()]
fresh["analysis_type"] = spec.normalize_analysis_type(conversion.get("analysis_type"))
fresh["last_build"] = {"source": conversion.get("source", "unknown"), "at": now_iso()}
self._set_status(fresh, STATUS_BUILT)
fresh = self.save_extension(fresh)
self._publish("extension_build", {
"ext_id": ext_id,
"command_template": fresh["command_template"],
"requirements": fresh["requirements"],
"runtime_image": fresh["runtime_image"],
"nl_triggers": fresh["nl_triggers"],
"analysis_type": fresh["analysis_type"],
"parameters": fresh["parameters"],
"status": fresh["status"],
})
logger.info("extensions: build ok %s -> %s", ext_id, template)
return fresh
# ---- trial-run（POST /{id}/trial-run）----
async def trial_run(self, ext_id: str) -> dict[str, Any]:
ext = self.load_extension(ext_id)
status = str(ext.get("status") or STATUS_DRAFT)
if status not in TRIAL_RETRYABLE:
raise ExtensionStatusError(f"状态不可试运行: {status}")
if not (ext.get("command_template") or "").strip():
raise ExtensionStatusError("试运行前必须先 build（command_template 为空）")
trial = ext.get("trial") if isinstance(ext.get("trial"), dict) else {}
n = max(0, int(trial.get("n") or 0)) + 1
validation = self.validation_dir(ext_id)
log_path = validation / f"trial_{n}.log"
out_dir = validation / f"trial_{n}.outputs"
out_dir.mkdir(parents=True, exist_ok=True)
input_dir = self.ext_dir(ext_id) / "raw_test_data"
if not input_dir.exists() or not any(input_dir.iterdir()):
input_dir = self.runs_dir(ext_id) / "inputs" # 无测试数据 → 空输入目录
input_dir.mkdir(parents=True, exist_ok=True)
run_id = f"trial_{n}"
self._apply_and_write(ext_id, lambda ext: (
self._set_status(ext, STATUS_TRIAL_RUNNING),
ext.__setitem__("trial", {"n": n, "run_id": run_id,
"log_path": str(log_path.relative_to(self.ext_dir(ext_id)))}),
))
executor = self.executor or ExtensionExecutor()
if isinstance(executor, ExtensionExecutor) and type(executor) is ExtensionExecutor:
# 占位执行器（Dev-B 未接入）：明确 503 语义，绝不伪装成 trial_failed 触发修复循环
raise ExtensionRunError(
"扩展执行器未接入（extension_exec.py 由 Dev-B 实现；"
"接口签名见 extensions.ExtensionExecutor）"
)
try:
result = await _call_executor(executor, ExtensionRunSpec(
ext_dir=str(self.ext_dir(ext_id)),
input_dir=str(input_dir),
output_dir=str(out_dir),
runtime_image=str(ext.get("runtime_image") or spec.DEFAULT_RUNTIME_IMAGE),
log_path=str(log_path),
))
except Exception as exc: # noqa: BLE001 - 执行器异常 → trial_failed 语义（可修复），绝不 500
result = ExtensionRunResult(
run_id=run_id, exit_code=1, log_tail="", outputs=[],
error=str(exc) or "扩展执行失败",
)
if isinstance(result, dict): # 容错：dict 形态 executor 也接受
result = _result_from_dict(result, run_id=run_id)
result.started_at = result.started_at or now_iso()
result.finished_at = result.finished_at or now_iso()
ok = result.exit_code == 0 and not result.error
fresh = self.load_extension(ext_id)
fresh["trial"] = {
"n": n,
"run_id": run_id,
"ok": ok,
"exit_code": result.exit_code,
"log_tail": result.log_tail,
"outputs": result.outputs,
"error": result.error,
}
self._set_status(fresh, STATUS_TRIAL_PASSED if ok else STATUS_TRIAL_FAILED)
fresh = self.save_extension(fresh)
self._publish("extension_trial", {
"ext_id": ext_id,
"status": "done" if ok else "failed",
"log_tail": result.log_tail,
"exit_code": result.exit_code,
"error": result.error,
})
if ok:
logger.info("extensions: trial ok %s (run %s)", ext_id, run_id)
else:
logger.warning("extensions: trial failed %s (run %s): %s", ext_id, run_id, result.error)
return fresh
# ---- validate（POST /{id}/validate；有金标时对比，无金标按 trial 结果放行）----
def validate(self, ext_id: str) -> dict[str, Any]:
ext = self.load_extension(ext_id)
self._apply_and_write(ext_id, lambda e: self._set_status(e, STATUS_VALIDATING))
ext = self.load_extension(ext_id)
golden = ext.get("golden")
report: dict[str, Any] = {"golden": golden, "compared": False}
if isinstance(golden, dict) and (golden.get("expected_outputs") or golden.get("test_input")):
# 金标校验（文件名/存在性级；全量内容对比由 Dev-B 执行链扩展 comparators）
test_input = str(golden.get("test_input") or "")
input_dir = self.ext_dir(ext_id) / "raw_test_data"
missing_input = bool(test_input) and not (input_dir / safe_filename(test_input)).exists()
expected = [str(x) for x in (golden.get("expected_outputs") or []) if str(x)]
trial = ext.get("trial") if isinstance(ext.get("trial"), dict) else {}
produced = {Path(str(x)).name for x in (trial.get("outputs") or [])}
missing_outputs = [x for x in expected if Path(x).name not in produced]
ok = not missing_input and not missing_outputs and bool(trial.get("ok"))
report.update({
"compared": True,
"ok": ok,
"missing_input": bool(missing_input),
"missing_outputs": missing_outputs,
})
fresh = self.load_extension(ext_id)
fresh["validation"] = report
self._set_status(fresh, STATUS_VALIDATED if ok else STATUS_TRIAL_FAILED)
return self.save_extension(fresh)
# 无金标：trial 通过即 validated
trial = ext.get("trial") if isinstance(ext.get("trial"), dict) else {}
ok = bool(trial.get("ok"))
report["ok"] = ok
fresh = self.load_extension(ext_id)
fresh["validation"] = report
self._set_status(fresh, STATUS_VALIDATED if ok else STATUS_TRIAL_FAILED)
return self.save_extension(fresh)
# ---- repair（POST /{id}/repair：LLM 修复，上限 max_repair_rounds）----
async def repair(self, ext_id: str) -> dict[str, Any]:
ext = self.load_extension(ext_id)
status = str(ext.get("status") or STATUS_DRAFT)
if status not in {STATUS_TRIAL_FAILED, STATUS_BUILD_FAILED}:
raise ExtensionStatusError(f"状态不可修复: {status}")
max_rounds = max(1, int(self.max_repair_rounds))
repair_count = max(0, int(ext.get("repair_count") or 0))
if repair_count >= max_rounds:
raise ExtensionStatusError(f"修复轮次已达上限（{max_rounds}）")
trial = ext.get("trial") if isinstance(ext.get("trial"), dict) else {}
context = {
"log_tail": trial.get("log_tail") or "",
"error": trial.get("error") or "",
"exit_code": trial.get("exit_code"),
}
self._apply_and_write(ext_id, lambda e: (
self._set_status(e, STATUS_REPAIRING),
e.__setitem__("repair_count", repair_count + 1),
))
from .extension_llm import repair_conversion
try:
conversion = await repair_conversion(self.load_extension(ext_id), failure_context=context, store=self)
except Exception as exc: # noqa: BLE001 - 修复失败收敛回 trial_failed/build_failed
try:
def _back(ext: dict[str, Any]) -> None:
if status == STATUS_BUILD_FAILED:
self._set_status(ext, STATUS_BUILD_FAILED)
else:
self._set_status(ext, STATUS_TRIAL_FAILED)
self._apply_and_write(ext_id, _back)
except ExtensionStatusError:
pass
raise ExtensionBuildError(f"扩展修复失败: {exc}") from exc
template = str(conversion.get("command_template") or "")
spec.validate_command_template(template)
fresh = self.load_extension(ext_id)
fresh["command_template"] = template
if conversion.get("requirements"):
fresh["requirements"] = conversion["requirements"]
fresh["repair_suggestion"] = str(conversion.get("suggestion") or "")
fresh["repair_count"] = repair_count + 1
self._set_status(fresh, STATUS_BUILT)
fresh = self.save_extension(fresh)
self._publish("extension_repair", {
"ext_id": ext_id,
"repair_count": fresh["repair_count"],
"suggestion": fresh.get("repair_suggestion") or "",
"status": fresh["status"],
})
logger.info("extensions: repair ok %s (round %d)", ext_id, repair_count + 1)
return fresh
# ---- confirm（用户确认/参数调整 → confirming）----
def confirm(self, ext_id: str, *, nl_triggers: list[str] | None = None,
parameters: list[dict[str, Any]] | None = None,
runtime_image: str | None = None) -> dict[str, Any]:
ext = self.load_extension(ext_id)
status = str(ext.get("status") or STATUS_DRAFT)
if status not in {STATUS_VALIDATED, STATUS_CONFIRMING, STATUS_TRIAL_PASSED, STATUS_PUBLISHED}:
raise ExtensionStatusError(f"状态不可确认: {status}")
def _apply(fresh: dict[str, Any]) -> None:
if nl_triggers is not None:
fresh["nl_triggers"] = [str(x).strip() for x in nl_triggers if str(x).strip()][:50]
if parameters is not None:
template = fresh.get("command_template") or ""
# 用户编辑后的 parameters 仍需覆盖模板中全部 {param_*}（漏参补齐）
fresh["parameters"] = spec.template_parameters(template, parameters)
if runtime_image:
fresh["runtime_image"] = spec.RUNTIME_IMAGE_MAP.get(runtime_image, runtime_image)
if status != STATUS_PUBLISHED:
self._set_status(fresh, STATUS_CONFIRMING)
return self._apply_and_write(ext_id, _apply)
# ---- publish（POST /{id}/publish）----
def publish(self, ext_id: str) -> dict[str, Any]:
ext = self.load_extension(ext_id)
status = str(ext.get("status") or STATUS_DRAFT)
if status not in {STATUS_CONFIRMING, STATUS_VALIDATED, STATUS_TRIAL_PASSED, STATUS_PUBLISHED}:
raise ExtensionStatusError(f"状态不可发布: {status}")
def _apply(fresh: dict[str, Any]) -> None:
if not fresh.get("nl_triggers"):
fresh["nl_triggers"] = [str(fresh.get("display_name") or "")]
if status != STATUS_PUBLISHED:
self._set_status(fresh, STATUS_PUBLISHED)
fresh = self._apply_and_write(ext_id, _apply)
self._publish("extension_published", {
"ext_id": ext_id,
"display_name": fresh.get("display_name"),
"nl_triggers": fresh.get("nl_triggers"),
})
logger.info("extensions: published %s", ext_id)
return fresh
# ---- run（POST /{id}/run：正式运行；同步返回 run_result，SSE 由 Dev-B 对接）----
async def run(self, ext_id: str, *, params: dict[str, Any] | None = None,
files: list[str] | None = None) -> dict[str, Any]:
ext = self.load_extension(ext_id)
if str(ext.get("status") or "") != STATUS_PUBLISHED:
raise ExtensionStatusError(f"仅已发布扩展可运行（当前 {ext.get('status')}）")
run_id = new_run_id()
run_dir = self.runs_dir(ext_id) / run_id
input_dir = run_dir / "inputs"
output_dir = run_dir / "outputs"
for path in (input_dir, output_dir):
path.mkdir(parents=True, exist_ok=True)
# 文件登记：路径须在白名单根内（data_root/扩展目录），拷贝入 runs/{id}/inputs
registered: list[dict[str, Any]] = []
for src in files or []:
p = Path(str(src)).expanduser()
if not p.is_file():
raise ExtensionRunError(f"运行输入文件不存在: {src}")
resolved = p.resolve()
allowed_roots = (Path(self.base_dir).resolve().parent, Path(self.base_dir).resolve())
if not any(resolved.is_relative_to(root) for root in allowed_roots):
raise ExtensionRunError(f"输入路径不在白名单根内: {src}")
dest = input_dir / safe_filename(p.name)
if resolved != dest.resolve():
shutil.copy2(p, dest)
registered.append({"name": dest.name, "size": p.stat().st_size})
if registered:
logger.info("extensions: run %s inputs=%s", run_id, [x["name"] for x in registered])
executor = self.executor or ExtensionExecutor()
started = now_iso()
if isinstance(executor, ExtensionExecutor) and type(executor) is ExtensionExecutor:
raise ExtensionRunError(
"扩展执行器未接入（extension_exec.py 由 Dev-B 实现；"
"接口签名见 extensions.ExtensionExecutor）"
)
try:
result = await _call_executor(executor, ExtensionRunSpec(
ext_dir=str(self.ext_dir(ext_id)),
input_dir=str(input_dir),
output_dir=str(output_dir),
runtime_image=str(ext.get("runtime_image") or spec.DEFAULT_RUNTIME_IMAGE),
params=params or {},
))
except Exception as exc: # noqa: BLE001 - 执行器异常 → failed 语义，绝不 500
result = ExtensionRunResult(
run_id=run_id, exit_code=1, log_tail="", outputs=[],
error=str(exc) or "扩展执行失败",
)
if isinstance(result, dict):
result = _result_from_dict(result, run_id=run_id)
result.started_at = result.started_at or started
result.finished_at = result.finished_at or now_iso()
outputs = []
if output_dir.is_dir():
outputs = sorted(
str(p.relative_to(output_dir)) for p in output_dir.rglob("*") if p.is_file()
)
ok = result.exit_code == 0 and not result.error
run_result: dict[str, Any] = {
"run_id": run_id,
"ext_id": ext_id,
"exit_code": result.exit_code,
"started_at": result.started_at,
"finished_at": result.finished_at,
"outputs": outputs or result.outputs,
"error": result.error,
"status": RUN_DONE if ok else RUN_FAILED,
}
(run_dir / "run_result.json").write_text(
json.dumps(run_result, ensure_ascii=False, indent=2), encoding="utf-8"
)
if result.log_tail:
(run_dir / "run.log").write_text(str(result.log_tail), encoding="utf-8")
# 事件帧（契约 §4；同步执行 → 先补帧再返回，Dev-B 后台化后改实时推送）
# 全部帧都带 ext_id：EventBus 按 job_id(=ext_id) 落同一 jsonl 可回放
self._publish("extension_run_started", {
"run_id": run_id, "ext_id": ext_id, "started_at": result.started_at,
})
if result.log_tail:
self._publish("extension_run_log", {"run_id": run_id, "ext_id": ext_id, "log": result.log_tail})
if ok:
self._publish("extension_run_done", {
"run_id": run_id, "ext_id": ext_id, "outputs": run_result["outputs"],
})
else:
self._publish("extension_run_failed", {
"run_id": run_id, "ext_id": ext_id, "exit_code": result.exit_code,
"error": result.error, "log_tail": result.log_tail,
})
return run_result
def list_runs(self, ext_id: str) -> list[dict[str, Any]]:
runs_dir = self.runs_dir(ext_id)
if not runs_dir.is_dir():
return []
items: list[dict[str, Any]] = []
for path in sorted(runs_dir.iterdir(), key=lambda p: p.name, reverse=True):
result_file = path / "run_result.json"
if result_file.is_file():
try:
items.append(json.loads(result_file.read_text(encoding="utf-8")))
except (ValueError, OSError):
continue
return items
# ---- 事件总线（契约 §4 事件帧；惰性取 context.events，缺失静默降级）----
def _event_bus(self) -> Any:
"""事件总线：优先测试/部署注入的 self._bus 覆盖，否则惰性取 context.events。"""
override = getattr(self, "_bus_override", None)
if override is not None:
return override
try:
from ... import context as context_mod
return context_mod.context.events
except Exception: # noqa: BLE001
return None
def _publish(self, event_type: str, data: dict[str, Any]) -> None:
"""发布 extension_* 事件帧；总线未初始化时静默降级（不阻断业务）。
data 统一补 job_id=ext_id：EventBus 持久日志按 job_id 落文件
（{ext_id}.jsonl），replay/snapshot 按扩展维度可回放（契约 §4 帧）。
"""
bus = self._event_bus()
if bus is None:
return
payload = dict(data)
if not payload.get("job_id"):
payload["job_id"] = payload.get("ext_id") or payload.get("run_id") or ""
try:
bus.publish(event_type, payload)
except Exception: # noqa: BLE001
pass
# ---- 列表 / 删除 ----
def list_extensions(self) -> list[dict[str, Any]]:
"""全部扩展列表（published 优先，其余按 updated_at 倒序；契约 §6）。"""
items: list[dict[str, Any]] = []
for path in Path(self.base_dir).glob("ext_*"):
manifest = path / "extension.json"
if not manifest.is_file():
continue
try:
data = json.loads(manifest.read_text(encoding="utf-8"))
except (ValueError, OSError):
continue
if isinstance(data, dict) and data.get("ext_id"):
items.append(data)
return sorted(
items,
key=lambda x: (str(x.get("status")) != STATUS_PUBLISHED,
str(x.get("updated_at") or x.get("created_at") or "")),
)
def delete(self, ext_id: str) -> None:
ext = self.load_extension(ext_id)
if str(ext.get("status")) == STATUS_TRIAL_RUNNING:
raise ExtensionStatusError("试运行中不可删除")
shutil.rmtree(self.ext_dir(ext_id), ignore_errors=True)
def _result_from_dict(data: dict[str, Any], *, run_id: str) -> ExtensionRunResult:
return ExtensionRunResult(
run_id=str(data.get("run_id") or run_id),
exit_code=int(data.get("exit_code") or 0),
log_tail=str(data.get("log_tail") or ""),
outputs=[str(x) for x in (data.get("outputs") or [])],
error=str(data.get("error") or ""),
timed_out=bool(data.get("timed_out")),
started_at=str(data.get("started_at") or ""),
finished_at=str(data.get("finished_at") or ""),
)
# ---- 模块级默认实例（路由直接使用；测试可构造独立实例）----
_store: ExtensionStore | None = None
def get_store() -> ExtensionStore:
"""惰性单例：数据根来自配置（测试经 monkeypatch get_settings 生效）。"""
global _store
if _store is None:
from ...config import get_settings
s = get_settings()
# 扩展存储根来自配置 BIOFLOW_EXTENSIONS_DIR（默认 bioflow_data/extensions，
# 与 data_root 同根的契约 §1 布局；生产可按需指向 data_root 外可写区）
_store = ExtensionStore(
base_dir=str(Path(s.extensions_dir_abs)),
max_repair_rounds=int(s.extension_max_repair_rounds),
)
inject_executor(_store)
return _store
def inject_executor(store: ExtensionStore | None = None) -> None:
"""Dev-B 替换点：extension_exec.py 实现后调用本函数注入真执行器。
executor 为 ExtensionExecutor 实例或可调用对象（协程：
execute(ExtensionRunSpec) -> ExtensionRunResult | dict）。
"""
target = store or get_store()
try:
from ... import context as context_mod
executor = context_mod.context.executor
if executor is not None:
target.executor = executor
except Exception: # noqa: BLE001
pass
def reset_store() -> None:
"""测试重置单例（config monkeypatch 后重建）。"""
global _store
_store = None
bioflow_service/app/services/bioflow/intent.py
+175
View File
@@ -358,3 +358,178 @@ def _family_hit(text_lower: str) -> bool:
if not tail_stripped.strip("，。！？；：,.!?;:（）() "):
return True
return False
# ---------------------------------------------------------------------------
# 用户自定义扩展分析意图（契约 §5 两层判定；Dev-A 纯函数导出）
#
# 匹配位置 = 生信服务 chat/stream 入口（先于 LLM 阶段规划）；chat/stream 的
# 接线由 Dev-C 联调时在主分支整合，本节只导出纯函数。
#
# A 层创建：动作词 × 目标词同句命中 → 创建流程（上传引导卡）。
# B 层运行：仅 published 扩展参与打分（display_name 模糊 + nl_triggers +
# analysis_type 加权，动作词加分）。
# 歧义消解：目标词命中但无动作词 → 有 B 层命中走运行；无命中回
# 「没有匹配的已发布扩展，是否创建?」引导卡。
# 漏判安全：无命中且非创建 → None 路径（回落正常生信阶段规划，零影响）。
# ---------------------------------------------------------------------------
# 创建意图目标词（契约 §5 A 层）
EXTENSION_TARGET_WORD = ["拓展分析", "分析模板", "分析脚本", "分析工具", "扩展"]
# 创建意图动作词（契约 §5 A 层；全词边界匹配——「建」不误命中「建议」等）
EXTENSION_ACTION_WORD = ["创建", "新建", "上传", "自定义", "做一个", "建", "写", "弄"]
# 运行意图动作词（契约 §5 B 层加分）
EXTENSION_RUN_ACTION_WORD = ["帮我跑", "帮我用", "做一下", "分析一下", "跑一下", "用一下"]
# 运行打分阈值：高于此分才算 B 层命中（漏判安全）
EXTENSION_SCORE_THRESHOLD = 2.0
# 单字动作词的后随禁用字（防「建议/建设/写作」等复合词误命中；
# 契约例「建个分析模板」→「个」不在禁用集 → 命中）
_ACTION_STOP_SUFFIX: dict[str, set[str]] = {
"建": {"议", "设", "筑", "造", "国"},
"写": {"作", "字"},
"弄": set(),
}
def _create_action_hit(text_lower: str) -> bool:
"""创建动作词命中（长词子串 + 单字词禁用后缀检查）。"""
for w in sorted(EXTENSION_ACTION_WORD, key=len, reverse=True):
if len(w) > 1:
if w in text_lower:
return True
continue
start = 0
while True:
idx = text_lower.find(w, start)
if idx < 0:
break
nxt = text_lower[idx + 1] if idx + 1 < len(text_lower) else ""
if nxt not in _ACTION_STOP_SUFFIX.get(w, set()):
return True
start = idx + 1
return False
def extension_create_intent(text: str) -> bool:
"""A 层创建意图：动作词 × 目标词同句命中（契约 §5）。"""
tl = (text or "").lower()
if not _contains_any_longest_first(tl, EXTENSION_TARGET_WORD):
return False
return _create_action_hit(tl)
def extension_run_actions(text: str) -> bool:
"""句子中是否含运行动作词（B 层加分项）。"""
return _contains_any_longest_first((text or "").lower(), EXTENSION_RUN_ACTION_WORD)
_ANALYSIS_TYPE_KEYWORDS: dict[str, tuple[str, ...]] = {
"blast_search": ("blast", "比对"),
"hmmer_search": ("hmm",),
"table_filter": ("过滤", "筛选", "表格"),
"sequence_extraction": ("提取", "抽取"),
"plot_render": ("绘图", "画图", "作图"),
"enrichment_or_stats": ("富集", "统计"),
"gff_processing": ("gff",),
"id_cleaning": ("去重", "清洗"),
}
def score_extension(text: str, ext: dict[str, object]) -> float:
"""单扩展打分（纯函数）：display_name 模糊 + nl_triggers + analysis_type，
动作词加分。分项权重与 detect 内注释一致。"""
score, _reasons = _score_extension(text, ext)
return score
def _score_extension(text: str, ext: dict[str, object]) -> tuple[float, list[str]]:
tl = (text or "").lower()
score = 0.0
reasons: list[str] = []
display = str(ext.get("display_name") or "").strip().lower()
if display:
if display in tl: # display_name 全文命中（最强信号）
score += 4.0
reasons.append("display_name")
else:
# 字符 2-gram 覆盖度（2-6 字名称模糊匹配；漏判安全：过半才计分）
grams = {display[i:i + 2] for i in range(len(display) - 1)} if len(display) >= 2 else {display}
hit = sum(1 for g in grams if g in tl)
if hit == len(grams):
score += 3.0
reasons.append("display_fuzzy")
elif hit >= max(1, len(grams) // 2):
score += 1.5
reasons.append("display_partial")
triggers = [str(t).strip().lower() for t in (ext.get("nl_triggers") or []) if str(t).strip()]
for t in triggers: # nl_triggers 关键词（只计一次最高）
if t and t in tl:
score += 2.5
reasons.append(f"trigger:{t}")
break
for kw in _ANALYSIS_TYPE_KEYWORDS.get(str(ext.get("analysis_type") or ""), ()):
if kw in tl:
score += 1.5
reasons.append(f"analysis_type:{kw}")
break
if extension_run_actions(tl):
score += 1.0
reasons.append("action_verb")
return score, reasons
def match_extensions(
text: str,
extensions: list[dict[str, object]],
*,
threshold: float = EXTENSION_SCORE_THRESHOLD,
) -> list[dict[str, object]]:
"""B 层运行意图打分：仅 published 扩展参与，得分 > 阈值，按分降序。
返回 [{ext_id, display_name, analysis_type, score, reasons, ext}]。
"""
out: list[dict[str, object]] = []
for ext in extensions or []:
if str(ext.get("status") or "") != "published":
continue
score, reasons = _score_extension(text, ext)
if score > threshold:
out.append({
"ext_id": ext.get("ext_id"),
"display_name": ext.get("display_name"),
"analysis_type": ext.get("analysis_type"),
"score": score,
"reasons": reasons,
"ext": ext,
})
out.sort(key=lambda x: float(x.get("score") or 0.0), reverse=True)
return out
def detect_extension_intent(
text: str,
extensions: list[dict[str, object]],
) -> dict[str, object]:
"""扩展意图两层判定主入口（契约 §5）。
返回（Dev-C 据此发卡）：
- {"layer": "create", ...} 创建流程（上传引导卡）
- {"layer": "run", "candidates": [...]} 运行意图（高置信单命中/多命中候选）
- {"layer": "create_ambiguous", ...} 歧义引导卡（「没有匹配的已发布扩展，是否创建?」）
- {"layer": "none", ...} 漏判安全：回落正常生信阶段规划
"""
has_target = _contains_any_longest_first((text or "").lower(), EXTENSION_TARGET_WORD)
has_create_action = _create_action_hit((text or "").lower())
if has_target and has_create_action:
return {"layer": "create", "create": True, "candidates": [], "ambiguous": False}
candidates = match_extensions(text, extensions)
if candidates:
return {"layer": "run", "create": False, "candidates": candidates, "ambiguous": False}
if has_target:
# 目标词命中、无创建动作词、无 B 层命中 → 歧义引导卡（不误判为创建）
return {"layer": "create_ambiguous", "create": False, "candidates": [], "ambiguous": True}
return {"layer": "none", "create": False, "candidates": [], "ambiguous": False}
bioflow_service/tests/test_extension_api.py
+356
View File
@@ -0,0 +1,356 @@
"""tests/test_extension_api.py —— 扩展端点（L2；契约 §6 全集）。
mock LLM（无 key → 规则版回退）/mock 执行器（注入 store.executor）。
SSE 用例端点直调迭代 body_iterator（同 test_api.py 纪律）。
"""
from __future__ import annotations
import asyncio
import io
import json
import zipfile
import pytest
pytestmark = pytest.mark.bioflow_api
from app.services.bioflow import extensions as ext_mod # noqa: E402
# ---------------------------------------------------------------------------
# 传输辅助（同 test_api.py ApiSession 模式）
# ---------------------------------------------------------------------------
class ApiSession:
_instance = None
def __init__(self):
import httpx
from app.main import app as _app
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
transport = httpx.ASGITransport(app=_app)
self._client = httpx.AsyncClient(transport=transport, base_url="http://t")
@classmethod
def shared(cls) -> "ApiSession":
if cls._instance is None:
cls._instance = cls()
asyncio.set_event_loop(cls._instance.loop)
return cls._instance
def request(self, method, url, **kw):
resp = self.loop.run_until_complete(self._client.request(method, url, **kw))
return resp
def get(self, url, **kw):
return self.request("GET", url, **kw)
def post(self, url, **kw):
return self.request("POST", url, **kw)
def delete(self, url, **kw):
return self.request("DELETE", url, **kw)
# ---------------------------------------------------------------------------
# fixtures
# ---------------------------------------------------------------------------
class FakeExtExecutor:
"""async 执行器：可编程退出码/输出；记录调用。"""
def __init__(self, exit_code: int = 0, log: str = "trial ok\n", outputs=("result.txt",)):
self.exit_code = exit_code
self.log = log
self.outputs = outputs
self.calls = []
async def execute(self, spec_in):
from pathlib import Path
self.calls.append(spec_in)
out_dir = Path(spec_in.output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
for name in self.outputs:
(out_dir / name).write_text("data", encoding="utf-8")
if spec_in.log_path:
Path(spec_in.log_path).parent.mkdir(parents=True, exist_ok=True)
Path(spec_in.log_path).write_text(self.log, encoding="utf-8")
return ext_mod.ExtensionRunResult(
run_id="", exit_code=self.exit_code, log_tail=self.log,
outputs=list(self.outputs),
error="" if self.exit_code == 0 else "命令失败",
)
def _make_zip(entries: dict[str, bytes]) -> bytes:
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
for name, blob in entries.items():
zf.writestr(name, blob)
return buf.getvalue()
@pytest.fixture()
def ext_api(test_settings, monkeypatch):
"""扩展 API 会话：settings 指 tmp、store 重置、无 LLM key。"""
monkeypatch.delenv("DOCGEN_LLM_API_KEY", raising=False)
monkeypatch.delenv("BIOFLOW_EXTENSION_API_KEY", raising=False)
ext_mod.reset_store()
session = ApiSession.shared()
yield session
ext_mod.reset_store()
def _store():
return ext_mod.get_store()
def _post_ext(session, name="测序质量统计", command="python quality_stats.py --fastq input.fastq --out results/",
code=None, test_blob=None, golden=None, runtime_image="test4py311"):
files = {"code_zip": ("code.zip", _make_zip(code or {"quality_stats.py": b"import sys\nprint('hi')\n"}), "application/zip")}
data = {"display_name": name, "command": command, "runtime_image": runtime_image}
if golden:
data["golden"] = json.dumps(golden)
if test_blob is not None:
files["test_file"] = ("input.fastq", test_blob, "application/octet-stream")
return session.post("/api/bioflow/extensions", files=files, data=data)
def _body(resp) -> dict:
try:
return json.loads(resp.content.decode("utf-8"))
except Exception: # noqa: BLE001
return {}
# ---------------------------------------------------------------------------
# 创建 / 列表 / 详情 / 删除
# ---------------------------------------------------------------------------
def test_create_multipart_zip(ext_api):
resp = _post_ext(ext_api, test_blob=b">s1\nACGT\n")
assert resp.status_code == 201
body = _body(resp)
assert body["ok"] is True
ext = body["data"]
assert ext["status"] == "draft"
assert ext["runtime_image"] == "bioflow-gf" # test4py311 映射
root = _store().ext_dir(ext["ext_id"])
assert (root / "raw_code" / "quality_stats.py").is_file()
assert (root / "raw_test_data" / "input.fastq").is_file()
assert (root / "converted_code" / "quality_stats.py").is_file()
def test_create_json_form(ext_api):
import base64
body = {
"display_name": "json 扩展",
"command": "python a.py --out results/",
"runtime_image": "bioflow-gf",
"code_files": [{"path": "a.py", "data": base64.b64encode(b"print(1)\n").decode()}],
}
resp = ext_api.post("/api/bioflow/extensions", json=body)
assert resp.status_code == 201
assert _body(resp)["data"]["display_name"] == "json 扩展"
def test_create_bad_zip(ext_api):
files = {"code_zip": ("bad.zip", b"not a zip", "application/zip")}
resp = ext_api.post(
"/api/bioflow/extensions",
files=files,
data={"display_name": "x", "command": "python a.py"},
)
assert resp.status_code == 422
assert "zip" in _body(resp)["error"]["message"]
def test_create_missing_fields(ext_api):
resp = ext_api.post(
"/api/bioflow/extensions",
data={"display_name": "", "command": ""},
)
assert resp.status_code in (422, 400)
def test_get_404(ext_api):
resp = ext_api.get("/api/bioflow/extensions/ext_nope")
assert resp.status_code == 404
def test_list_published_first(ext_api):
a = _body(_post_ext(ext_api, name="扩展 A"))["data"]
b = _body(_post_ext(ext_api, name="扩展 B"))["data"]
_publish_quick(ext_api, b["ext_id"], executor=FakeExtExecutor(0))
resp = ext_api.get("/api/bioflow/extensions")
items = _body(resp)["data"]["items"]
assert items[0]["ext_id"] == b["ext_id"]
def test_delete(ext_api):
e = _body(_post_ext(ext_api))["data"]
resp = ext_api.delete(f"/api/bioflow/extensions/{e['ext_id']}")
assert resp.status_code == 204
assert ext_api.get(f"/api/bioflow/extensions/{e['ext_id']}").status_code == 404
# ---------------------------------------------------------------------------
# 生命周期动作（build/trial/validate/repair/confirm/publish）
# ---------------------------------------------------------------------------
def _publish_quick(session, ext_id, *, executor=None):
"""build → trial → validate → confirm → publish 快速链。"""
if executor is not None:
_store().executor = executor
r = session.post(f"/api/bioflow/extensions/{ext_id}/build")
assert r.status_code == 200, _body(r)
r = session.post(f"/api/bioflow/extensions/{ext_id}/trial-run")
assert r.status_code == 200, _body(r)
r = session.post(f"/api/bioflow/extensions/{ext_id}/validate")
assert r.status_code == 200, _body(r)
r = session.post(f"/api/bioflow/extensions/{ext_id}/confirm", json={"nl_triggers": ["质量统计"]})
assert r.status_code == 200, _body(r)
r = session.post(f"/api/bioflow/extensions/{ext_id}/publish")
assert r.status_code == 200, _body(r)
return _body(r)["data"]
def test_build_rules_fallback(ext_api):
e = _body(_post_ext(ext_api))["data"]
resp = ext_api.post(f"/api/bioflow/extensions/{e['ext_id']}/build")
assert resp.status_code == 200
ext = _body(resp)["data"]
assert ext["status"] == "built"
assert ext["last_build"]["source"] == "rules"
assert "{code_dir}" in ext["command_template"]
assert "{output_dir}" in ext["command_template"]
def test_build_wrong_state_409(ext_api):
e = _body(_post_ext(ext_api))["data"]
# draft → publish 非法（须 build/trial/validate/confirm 链）
resp = ext_api.post(f"/api/bioflow/extensions/{e['ext_id']}/publish")
assert resp.status_code == 409
assert _body(resp)["error"]["code"] == "EXT_STATUS"
def test_trial_without_executor_503(ext_api):
e = _body(_post_ext(ext_api))["data"]
ext_api.post(f"/api/bioflow/extensions/{e['ext_id']}/build")
resp = ext_api.post(f"/api/bioflow/extensions/{e['ext_id']}/trial-run")
assert resp.status_code == 503
def test_full_lifecycle(ext_api):
e = _body(_post_ext(ext_api, test_blob=b">s\nACGT\n",
golden={"test_input": "input.fastq", "expected_outputs": ["result.txt"]}))["data"]
ext_id = e["ext_id"]
_publish_quick(ext_api, ext_id, executor=FakeExtExecutor(0, "ok\n", ("result.txt",)))
ext = _body(ext_api.get(f"/api/bioflow/extensions/{ext_id}"))["data"]
assert ext["status"] == "published"
assert ext["nl_triggers"] == ["质量统计"]
# 执行器收到 ext_dir/output_dir（Dev-B 契约字段）
assert _store().executor.calls[-1].ext_dir.endswith(ext_id)
assert _store().executor.calls[-1].output_dir
def test_trial_fail_then_repair(ext_api):
e = _body(_post_ext(ext_api))["data"]
ext_id = e["ext_id"]
_store().executor = FakeExtExecutor(1, "No module named 'pandas'\n", ())
r = ext_api.post(f"/api/bioflow/extensions/{ext_id}/build")
assert r.status_code == 200
r = ext_api.post(f"/api/bioflow/extensions/{ext_id}/trial-run")
assert _body(r)["data"]["status"] == "trial_failed"
r = ext_api.post(f"/api/bioflow/extensions/{ext_id}/repair")
assert r.status_code == 200
ext = _body(r)["data"]
assert ext["status"] == "built"
assert "pandas" in ext["requirements"]["pip"]
def test_repair_limit_reached(ext_api):
e = _body(_post_ext(ext_api))["data"]
ext_id = e["ext_id"]
_store().max_repair_rounds = 1
_store().executor = FakeExtExecutor(1, "No module named 'pandas'\n", ())
ext_api.post(f"/api/bioflow/extensions/{ext_id}/build")
ext_api.post(f"/api/bioflow/extensions/{ext_id}/trial-run")
assert ext_api.post(f"/api/bioflow/extensions/{ext_id}/repair").status_code == 200
ext_api.post(f"/api/bioflow/extensions/{ext_id}/trial-run") # 再失败
resp = ext_api.post(f"/api/bioflow/extensions/{ext_id}/repair")
assert resp.status_code == 409
assert "上限" in _body(resp)["error"]["message"]
# ---------------------------------------------------------------------------
# 运行（run/runs/events）
# ---------------------------------------------------------------------------
def test_run_not_published_409(ext_api):
e = _body(_post_ext(ext_api))["data"]
resp = ext_api.post(f"/api/bioflow/extensions/{e['ext_id']}/run", json={"params": {}})
assert resp.status_code == 409
def test_run_and_events(ext_api):
e = _body(_post_ext(ext_api))["data"]
ext_id = e["ext_id"]
_publish_quick(ext_api, ext_id, executor=FakeExtExecutor(0, "run ok\n", ("out.tsv",)))
resp = ext_api.post(f"/api/bioflow/extensions/{ext_id}/run", json={"params": {"cpu": "4"}})
assert resp.status_code == 200
result = _body(resp)["data"]
assert result["status"] == "done"
assert "out.tsv" in result["outputs"]
runs = _body(ext_api.get(f"/api/bioflow/extensions/{ext_id}/runs"))["data"]
assert runs["total"] == 1
assert runs["items"][0]["run_id"] == result["run_id"]
# events SSE（占位实现：已完成 run 的快照帧）
from app.routers.extension import run_events
async def _collect():
resp = await run_events(ext_id, result["run_id"], request=None)
chunks = []
async for chunk in resp.body_iterator:
chunks.append(chunk)
return b"".join(c.encode("utf-8") if isinstance(c, str) else c for c in chunks).decode("utf-8")
raw = asyncio.get_event_loop().run_until_complete(_collect())
assert "extension_run_started" in raw
assert "extension_run_done" in raw
assert "run ok" in raw
def test_run_failed_events(ext_api):
e = _body(_post_ext(ext_api))["data"]
ext_id = e["ext_id"]
_publish_quick(ext_api, ext_id, executor=FakeExtExecutor(0, "trial ok\n"))
_store().executor = FakeExtExecutor(3, "output contract failed\n", ())
resp = ext_api.post(f"/api/bioflow/extensions/{ext_id}/run", json={"params": {}})
result = _body(resp)["data"]
assert result["status"] == "failed"
assert result["exit_code"] == 3
from app.routers.extension import run_events
async def _collect():
resp = await run_events(ext_id, result["run_id"], request=None)
chunks = []
async for chunk in resp.body_iterator:
chunks.append(chunk)
return b"".join(c.encode("utf-8") if isinstance(c, str) else c for c in chunks).decode("utf-8")
raw = asyncio.get_event_loop().run_until_complete(_collect())
assert "extension_run_failed" in raw
bioflow_service/tests/test_extension_intent.py
+168
View File
@@ -0,0 +1,168 @@
"""tests/test_extension_intent.py —— 扩展意图两层判定（契约 §5）。
覆盖 A 层创建（动作词×目标词）/ B 层运行打分（仅 published、display_name
模糊/nl_triggers/analysis_type/动作词加分）/ 歧义消解（目标词无动作词 →
有命中走运行，无命中回 create_ambiguous）/ 漏判安全（无命中且非创建 →
layer none，不影响现有 route() 返回路径）。
"""
from __future__ import annotations
import pytest
from app.services.bioflow.intent import (
detect_extension_intent,
extension_create_intent,
match_extensions,
route,
)
pytestmark = pytest.mark.bioflow_unit
def _pub_ext(ext_id="ext_1", name="测序质量统计", triggers=None, analysis_type="table_filter"):
return {
"ext_id": ext_id,
"display_name": name,
"status": "published",
"nl_triggers": triggers or ["质量统计", "fastq 统计"],
"analysis_type": analysis_type,
}
def _draft_ext(ext_id="ext_9", name="未发布扩展"):
return {"ext_id": ext_id, "display_name": name, "status": "draft",
"nl_triggers": ["未发布"], "analysis_type": "generic_cli"}
# ---- A 层创建 ----
@pytest.mark.parametrize("text", [
"帮我创建一个扩展：统计 fastq 质量",
"建个分析模板",
"新建一个分析脚本",
"上传我的分析工具",
"自定义一个扩展",
"写一个分析模板",
"弄个扩展统计 fastq",
])
def test_create_intent_hits(text):
assert extension_create_intent(text) is True
@pytest.mark.parametrize("text", [
"帮我跑水稻 SPL 家族鉴定", # 无目标词
"质量统计", # 目标词+运行词，无创建动作词（消歧走 B 层）
"做拓展分析 统计 fastq 质量", # 契约歧义消解句：无创建动作词 → 引导卡而非直接创建
"建议新建文件夹", # 「建」后随「议」→ 不误命中
"这是写作练习", # 「写」后随「作」→ 不误命中
"扩展名为 fasta 的文件", # 「扩展名」≠ 目标词语义（漏判安全，不误入创建）
"今天天气不错",
])
def test_create_intent_misses(text):
assert extension_create_intent(text) is False
# ---- B 层打分 ----
def test_match_only_published():
exts = [_pub_ext(), _draft_ext()]
hits = match_extensions("帮我做一下质量统计", exts)
assert [h["ext_id"] for h in hits] == ["ext_1"]
def test_match_display_name_fuzzy():
hits = match_extensions("测序质量统计", [_pub_ext()])
assert len(hits) == 1
assert hits[0]["ext_id"] == "ext_1"
assert hits[0]["score"] >= 4.0
def test_match_nl_trigger_keyword():
hits = match_extensions("帮我把这个 fastq 做一下 fastq 统计", [_pub_ext()])
assert len(hits) == 1
assert any(r.startswith("trigger:") for r in hits[0]["reasons"])
def test_match_analysis_type_keyword():
ext = _pub_ext(ext_id="ext_2", name="表达量统计", triggers=["表达量"],
analysis_type="enrichment_or_stats")
# 动作词 + analysis_type 关键词 = 1.0 + 1.5 = 2.5 > 阈值 → 命中
hits = match_extensions("帮我做一下富集分析", [ext])
assert len(hits) == 1
assert hits[0]["ext_id"] == "ext_2"
assert any(r == "analysis_type:富集" for r in hits[0]["reasons"])
# 漏判安全：仅 analysis_type 关键词（无动作词/触发词）→ 1.5 ≤ 阈值 → 不命中
assert match_extensions("富集分析", [ext]) == []
def test_match_action_verb_bonus():
ext = _pub_ext(ext_id="ext_3", name="表格过滤工具", triggers=["过滤表格"],
analysis_type="table_filter")
# 无动作词：nl_trigger 2.5 + analysis_type 1.5 = 4.0（过阈值 2.0）
base = match_extensions("过滤表格", [ext])
assert len(base) == 1
# 动作词加分后更高
boosted = match_extensions("帮我跑过滤表格", [ext])
assert boosted[0]["score"] > base[0]["score"]
def test_match_no_hit_empty():
assert match_extensions("帮我跑水稻 SPL 家族鉴定", [_pub_ext()]) == []
assert match_extensions("随便说点什么", [_pub_ext()]) == []
# ---- 两层判定主入口 ----
def test_detect_create_layer():
r = detect_extension_intent("帮我创建一个扩展：统计 fastq 质量", [_pub_ext()])
assert r["layer"] == "create"
assert r["create"] is True
def test_detect_run_layer_high_confidence():
r = detect_extension_intent("帮我把这个 fastq 做一下质量统计", [_pub_ext()])
assert r["layer"] == "run"
assert len(r["candidates"]) == 1
assert r["candidates"][0]["ext_id"] == "ext_1"
def test_detect_run_layer_multi_candidates():
exts = [
_pub_ext("ext_1", "测序质量统计", ["质量统计"]),
_pub_ext("ext_2", "表达量统计", ["统计"]),
]
r = detect_extension_intent("帮我做个统计分析", exts)
assert r["layer"] == "run"
assert len(r["candidates"]) >= 1
def test_detect_ambiguity_no_published():
"""目标词命中、无创建动作词、无 B 层命中 → create_ambiguous（引导卡）。
契约歧义消解句「做拓展分析 质量统计」：无创建动作词 → 走消解而非直接创建。
"""
r = detect_extension_intent("做拓展分析 质量统计", [])
assert r["layer"] == "create_ambiguous"
assert r["ambiguous"] is True
def test_detect_ambiguity_target_no_action_but_run_hit():
"""歧义消解：目标词命中但无创建动作词 → 有 B 层命中走运行。"""
r = detect_extension_intent("做拓展分析 质量统计", [_pub_ext()])
assert r["layer"] == "run"
def test_detect_none_fallback():
"""漏判安全：无命中且非创建 → layer none（回落正常生信阶段规划）。"""
r = detect_extension_intent("帮我跑水稻 SPL 家族鉴定", [_pub_ext()])
assert r["layer"] == "none"
# ---- 漏判安全：现有 route() 返回路径零改动 ----
def test_route_unchanged_by_extension_words():
"""含「扩展」等的句子不得改变现有意图分流结论（扩展判定是独立纯函数层）。"""
assert route("帮我跑水稻 SPL 家族鉴定") == "bioflow"
assert route("今天天气不错") == "qa"
# 「建个分析模板」不命中生信任务动词表 → 仍 qa（创建流程由扩展层接管）
assert route("建个分析模板") == "qa"
bioflow_service/tests/test_extension_llm.py
+273
View File
@@ -0,0 +1,273 @@
"""tests/test_extension_llm.py —— 扩展 AI 转换（LLM 优先 + 规则版回退，绝不 500）。
覆盖：llm_config（DOCGEN 优先/BIOFLOW_EXTENSION 别名兜底）/
detect_requirements（import/requirements.txt/R library/sh 命令）/
command_to_template（脚本路径/输入占位/output_dir 必含兜底）/
build_conversion（无 key 回退 rules；LLM 坏输出回退 rules）/
repair_conversion（ModuleNotFoundError 规则并入 pip；无可修复抛 LLMUnavailable）。
"""
from __future__ import annotations
from pathlib import Path
import pytest
from app.services.bioflow import extension_llm as extllm
from app.services.bioflow import extension_spec as spec
pytestmark = pytest.mark.bioflow_unit
# ---- llm_config ----
def test_llm_config_docgen_priority(monkeypatch):
monkeypatch.setenv("DOCGEN_LLM_API_KEY", "docgen-key")
monkeypatch.setenv("BIOFLOW_EXTENSION_API_KEY", "ext-key")
monkeypatch.setenv("DOCGEN_LLM_BASE_URL", "https://doc.example/v1")
monkeypatch.setenv("BIOFLOW_EXTENSION_BASE_URL", "https://ext.example/v1")
cfg = extllm.llm_config()
assert cfg["api_key"] == "docgen-key"
assert cfg["base_url"] == "https://doc.example/v1"
assert cfg["model"] == "deepseek-chat"
def test_llm_config_alias_fallback(monkeypatch):
monkeypatch.delenv("DOCGEN_LLM_API_KEY", raising=False)
monkeypatch.delenv("DOCGEN_LLM_MODEL", raising=False)
monkeypatch.setenv("BIOFLOW_EXTENSION_API_KEY", "ext-key")
monkeypatch.setenv("BIOFLOW_EXTENSION_MODEL", "deepseek-chat-v2")
cfg = extllm.llm_config()
assert cfg["api_key"] == "ext-key"
assert cfg["model"] == "deepseek-chat-v2"
assert cfg["base_url"] == "https://api.deepseek.com/v1"
# ---- 规则版依赖探测 ----
def _make_code_dir(tmp_path, files: dict[str, str]) -> Path:
root = tmp_path / "code"
root.mkdir(parents=True, exist_ok=True)
for name, text in files.items():
(root / name).write_text(text, encoding="utf-8")
return root
def test_detect_requirements_python(tmp_path):
root = _make_code_dir(tmp_path, {
"a.py": "import pandas\nfrom Bio import SeqIO\nimport os\nfrom sklearn import svm\n",
})
req = extllm.detect_requirements(root)
assert set(req["pip"]) == {"pandas", "Bio", "sklearn"} # os 标准库剔除
assert req["apt"] == []
def test_detect_requirements_file(tmp_path):
root = _make_code_dir(tmp_path, {
"requirements.txt": "pandas>=1.0\nnumpy==1.2\n# 注释\n",
})
req = extllm.detect_requirements(root)
assert set(req["pip"]) == {"pandas", "numpy"}
def test_detect_requirements_r_and_shell(tmp_path):
root = _make_code_dir(tmp_path, {
"a.r": "library(ggplot2)\nrequire(dplyr)\n",
"run.sh": "#!/bin/bash\nset -e\npython a.py\nsamtools view -b input.bam\nfastqc input.fastq\ngrep foo\n",
})
req = extllm.detect_requirements(root)
assert set(req["pip"]) == {"ggplot2", "dplyr"}
assert set(req["apt"]) == {"samtools", "fastqc"} # 白名单命令（python/grep）不计
def test_infer_language():
assert extllm.infer_language("a.py") == "python"
assert extllm.infer_language("a.R") == "r"
assert extllm.infer_language("run.sh") == "shell"
assert extllm.infer_language("a.pl") == "pl"
# ---- 规则版模板化 ----
def test_command_to_template_full(tmp_path):
root = _make_code_dir(tmp_path, {"quality_stats.py": "print('x')\n"})
out = extllm.command_to_template(
"python quality_stats.py --fastq input.fastq --out results/",
root,
data_files=["input.fastq"],
)
# fastq 非 fasta 系扩展名 → 契约 §2 示例语义 {input_file}
assert out["command_template"] == (
"python {code_dir}/quality_stats.py --fastq {input_file} --out {output_dir}/"
)
spec.validate_command_template(out["command_template"])
assert out["source"] == "rules"
def test_command_to_template_no_output_appends_redirect(tmp_path):
root = _make_code_dir(tmp_path, {"a.py": "print('x')\n"})
out = extllm.command_to_template("python a.py --in x.txt", root)
assert "{output_dir}" in out["command_template"]
spec.validate_command_template(out["command_template"])
def test_command_to_template_param_derived(tmp_path):
root = _make_code_dir(tmp_path, {"a.py": "print('x')\n"})
out = extllm.command_to_template(
"python a.py --cpu {param_cpu} --out results/",
root,
)
assert "{param_cpu}" in out["command_template"]
params = spec.template_parameters(out["command_template"], out["parameters"])
assert {p["name"] for p in params} == {"cpu"}
# ---- build_conversion 回退 ----
def _make_ext(store, name="测序质量统计"):
return store.create_extension(
display_name=name,
command="python quality_stats.py --fastq input.fastq --out results/",
code_files=[("quality_stats.py", b"import sys\nprint('hi')\n")],
data_files=[("input.fastq", b">s\nACGT\n")],
)
async def test_build_conversion_no_key_falls_back_rules(store, monkeypatch):
monkeypatch.delenv("DOCGEN_LLM_API_KEY", raising=False)
monkeypatch.delenv("BIOFLOW_EXTENSION_API_KEY", raising=False)
e = _make_ext(store)
out = await extllm.build_conversion(e, store=store)
assert out["source"] == "rules"
assert "{code_dir}" in out["command_template"]
assert "{output_dir}" in out["command_template"]
assert out["requirements"]["pip"] == []
assert out["nl_triggers"] # display_name 兜底
assert out["analysis_type"] in spec.ANALYSIS_TYPES
spec.validate_command_template(out["command_template"])
async def test_build_conversion_llm_ok(store, monkeypatch):
monkeypatch.setenv("DOCGEN_LLM_API_KEY", "k")
monkeypatch.setattr("app.services.bioflow.extension_llm.llm_config",
lambda: {"base_url": "http://x", "api_key": "k", "model": "m", "timeout": 1.0})
async def fake_complete_json(system, user, **kw):
return {
"command_template": "python {code_dir}/quality_stats.py --fastq {input_fasta} --out {output_dir}",
"requirements": {"pip": ["pandas"], "apt": []},
"parameters": [{"name": "fastq", "type": "file", "required": True, "description": "输入"}],
"input_roles": {"input_fasta": "generic_input_file"},
"nl_triggers": ["质量统计", "fastq 统计"],
"analysis_type": "table_filter",
"runtime_image": "bioflow-gf",
}
monkeypatch.setattr("app.services.bioflow.llm.complete_json", fake_complete_json)
e = _make_ext(store)
out = await extllm.build_conversion(e, store=store)
assert out["source"] == "llm"
assert out["command_template"].startswith("python {code_dir}/")
assert out["requirements"]["pip"] == ["pandas"]
assert out["nl_triggers"] == ["质量统计", "fastq 统计"]
assert out["analysis_type"] == "table_filter"
assert {p["name"] for p in out["parameters"]} == {"fastq"}
async def test_build_conversion_llm_bad_output_falls_back(store, monkeypatch):
"""LLM 返回白名单外占位符 → sanitize 抛 → 回退规则版（绝不 500）。"""
monkeypatch.setenv("DOCGEN_LLM_API_KEY", "k")
monkeypatch.setattr("app.services.bioflow.extension_llm.llm_config",
lambda: {"base_url": "http://x", "api_key": "k", "model": "m", "timeout": 1.0})
async def fake_complete_json(system, user, **kw):
return {"command_template": "python {code_dir}/a.py --out {evil}"}
monkeypatch.setattr("app.services.bioflow.llm.complete_json", fake_complete_json)
e = _make_ext(store)
out = await extllm.build_conversion(e, store=store)
assert out["source"] == "rules"
spec.validate_command_template(out["command_template"])
async def test_build_conversion_llm_unavailable_falls_back(store, monkeypatch):
from app.services.bioflow.llm import LLMUnavailable as PlanLLMUnavailable
monkeypatch.setenv("DOCGEN_LLM_API_KEY", "k")
monkeypatch.setattr("app.services.bioflow.extension_llm.llm_config",
lambda: {"base_url": "http://x", "api_key": "k", "model": "m", "timeout": 1.0})
async def fake_complete_json(system, user, **kw):
raise PlanLLMUnavailable("down")
monkeypatch.setattr("app.services.bioflow.llm.complete_json", fake_complete_json)
e = _make_ext(store)
out = await extllm.build_conversion(e, store=store)
assert out["source"] == "rules"
# ---- repair ----
async def _built_ext(store):
e = _make_ext(store)
await store.build_extension(e["ext_id"])
return e["ext_id"]
async def test_repair_rules_module_not_found(store, monkeypatch):
monkeypatch.delenv("DOCGEN_LLM_API_KEY", raising=False)
monkeypatch.delenv("BIOFLOW_EXTENSION_API_KEY", raising=False)
ext_id = await _built_ext(store)
ext = store.load_extension(ext_id)
out = await extllm.repair_conversion(
ext,
failure_context={"log_tail": "Traceback: ModuleNotFoundError: No module named 'pandas'", "error": "x"},
store=store,
)
assert "pandas" in out["requirements"]["pip"]
assert out["source"] == "repair_rules"
assert out["command_template"] == ext["command_template"]
async def test_repair_no_fixable_hint_raises(store, monkeypatch):
monkeypatch.delenv("DOCGEN_LLM_API_KEY", raising=False)
monkeypatch.delenv("BIOFLOW_EXTENSION_API_KEY", raising=False)
ext_id = await _built_ext(store)
ext = store.load_extension(ext_id)
with pytest.raises(extllm.LLMUnavailable):
await extllm.repair_conversion(
ext,
failure_context={"log_tail": "exit 1: 未知错误", "error": "x"},
store=store,
)
async def test_repair_llm_ok(store, monkeypatch):
monkeypatch.setenv("DOCGEN_LLM_API_KEY", "k")
monkeypatch.setattr("app.services.bioflow.extension_llm.llm_config",
lambda: {"base_url": "http://x", "api_key": "k", "model": "m", "timeout": 1.0})
async def fake_complete_json(system, user, **kw):
return {
"command_template": "python {code_dir}/quality_stats.py --fastq {input_fasta} --out {output_dir}",
"requirements": {"pip": ["pandas"], "apt": []},
"parameters": [],
"nl_triggers": ["质量统计"],
"analysis_type": "table_filter",
"runtime_image": "bioflow-gf",
"suggestion": "把输入参数改为 {input_fasta}",
}
monkeypatch.setattr("app.services.bioflow.llm.complete_json", fake_complete_json)
ext_id = await _built_ext(store)
ext = store.load_extension(ext_id)
out = await extllm.repair_conversion(ext, failure_context={"log_tail": "boom"}, store=store)
assert out["source"] == "repair_llm"
assert "pandas" in out["requirements"]["pip"]
assert "input_fasta" in out["suggestion"]
@pytest.fixture()
def store(tmp_path):
from app.services.bioflow import extensions as ext_mod
return ext_mod.ExtensionStore(base_dir=str(tmp_path / "data" / "extensions"))
bioflow_service/tests/test_extension_spec.py
+110
View File
@@ -0,0 +1,110 @@
"""tests/test_extension_spec.py —— 扩展模板白名单契约（契约 §2 spec_v2 移植）。
覆盖：占位符白名单 12 个 / output_dir 必含 / 白名单外拒绝 / param_* 放行 /
analysis_type 归一 / 规则版 analysis_type 探测 / template_parameters 派生补齐。
"""
from __future__ import annotations
import pytest
from app.services.bioflow import extension_spec as spec
pytestmark = pytest.mark.bioflow_unit
# ---- 白名单 ----
def test_placeholders_whitelist_has_12():
assert len(spec.PLACEHOLDERS) == 12
assert spec.PLACEHOLDERS == {
"code_dir", "input_dir", "input_file", "input_hmm", "input_fasta",
"input_query_fasta", "input_reference_fasta", "input_pep_fasta",
"input_cds_fasta", "input_id_file", "output_dir", "work_dir",
}
def test_output_dir_required():
with pytest.raises(spec.ExtensionSpecError, match="output_dir"):
spec.validate_command_template("python {code_dir}/a.py --in {input_file}")
def test_empty_template_rejected():
with pytest.raises(spec.ExtensionSpecError):
spec.validate_command_template("")
def test_unknown_placeholder_rejected():
with pytest.raises(spec.ExtensionSpecError, match="白名单外"):
spec.validate_command_template("python {code_dir}/a.py --out {output_dir} --x {input_bam}")
def test_whitelist_all_accepted():
tpl = (
"python {code_dir}/run.py --dir {input_dir} --file {input_file} "
"--hmm {input_hmm} --fa {input_fasta} --q {input_query_fasta} "
"--ref {input_reference_fasta} --pep {input_pep_fasta} "
"--cds {input_cds_fasta} --ids {input_id_file} --out {output_dir} "
"--work {work_dir}"
)
spec.validate_command_template(tpl) # 不抛即通过
def test_param_placeholders_allowed():
tpl = "python {code_dir}/a.py --in {input_file} --out {output_dir} --cpu {param_cpu} --mem {param_memory}"
spec.validate_command_template(tpl)
assert spec.param_placeholder_names(tpl) == {"cpu", "memory"}
def test_unsupported_detection():
tpl = "python {code_dir}/a.py --out {output_dir} --db {database}"
assert spec.unsupported_placeholders(tpl) == {"database"}
assert spec.unsupported_placeholders(
"python {code_dir}/a.py --out {output_dir} --cpu {param_cpu}"
) == set()
# ---- analysis_type ----
def test_normalize_analysis_type():
assert spec.normalize_analysis_type("table_filter") == "table_filter"
assert spec.normalize_analysis_type("") == "generic_cli"
assert spec.normalize_analysis_type("bogus") == "generic_cli"
assert spec.normalize_analysis_type(None) == "generic_cli"
def test_infer_analysis_type_rules():
assert spec.infer_analysis_type(template="blastp -query {input_fasta}", display_name="x") == "blast_search"
assert spec.infer_analysis_type(template="hmmsearch {input_hmm} {input_fasta}", display_name="x") == "hmmer_search"
assert spec.infer_analysis_type(template="awk -F'\\t' '{print $1}' {input_file}", display_name="x") == "table_filter"
assert spec.infer_analysis_type(template="python {code_dir}/a.py", display_name="id 清洗 clean_id") == "fasta_header_cleaning"
assert spec.infer_analysis_type(template="python {code_dir}/a.py", display_name="画图工具") == "plot_render"
assert spec.infer_analysis_type(template="python {code_dir}/a.py", display_name="随便") == "generic_cli"
# ---- template_parameters ----
def test_template_parameters_derived():
tpl = "python {code_dir}/a.py --cpu {param_cpu} --out {output_dir}"
params = spec.template_parameters(tpl)
names = {p["name"] for p in params}
assert names == {"cpu"}
assert params[0]["type"] == "string"
def test_template_parameters_keeps_existing():
tpl = "python {code_dir}/a.py --cpu {param_cpu} --mem {param_mem} --out {output_dir}"
params = spec.template_parameters(tpl, [
{"name": "cpu", "type": "int", "required": True, "description": "核数", "default": 4},
])
by_name = {p["name"]: p for p in params}
assert set(by_name) == {"cpu", "mem"}
assert by_name["cpu"]["type"] == "int"
assert by_name["cpu"]["default"] == 4
assert by_name["mem"]["required"] is False
def test_safe_parameter_name():
assert spec.safe_parameter_name("fastq 文件") == "fastq"
assert spec.safe_parameter_name("2x") == "p_2x"
assert spec.safe_parameter_name("param_cpu") == "cpu"
assert spec.safe_parameter_name("") == ""
bioflow_service/tests/test_extensions.py
+404
View File
@@ -0,0 +1,404 @@
"""tests/test_extensions.py —— 扩展生命周期状态机与存储（契约 §1/§2）。
覆盖：创建布局（extension.json/raw_code/raw_test_data/converted_code）/
状态机合法流转 / 非法流转 409 语义 / build 规则版回退（无 LLM key）/
trial 执行（fake executor，成/败）/ validate（无金标+有金标）/
repair（轮次上限/LLM 不可用规则回退）/
confirm（nl_triggers 编辑/parameters 补齐）/ publish / run（仅 published、
run_result 落盘、输出登记）/ 列表 published 优先 / 删除守卫。
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from app.services.bioflow import extension_spec as spec
from app.services.bioflow import extensions as ext
pytestmark = pytest.mark.bioflow_unit
_CODE = "#!/usr/bin/env python\nimport sys\nprint('hello')\n"
@pytest.fixture()
def store(tmp_path, monkeypatch):
monkeypatch.delenv("DOCGEN_LLM_API_KEY", raising=False)
monkeypatch.delenv("BIOFLOW_EXTENSION_API_KEY", raising=False)
return ext.ExtensionStore(base_dir=str(tmp_path / "data" / "extensions"))
def _create(store, *, name="测序质量统计", command="python quality_stats.py --fastq input.fastq --out results/",
code=None, data=None, golden=None):
return store.create_extension(
display_name=name,
command=command,
runtime_image="test4py311", # 旧 conda 环境名 → 映射 bioflow-gf
code_files=code or [("quality_stats.py", _CODE.encode())],
data_files=data or [],
golden=golden,
)
def _fake_executor(exit_code=0, log="run ok\n", outputs=("result.txt",)):
"""sync 可调用 executor：写输出 + 返回 ExtensionRunResult（兼容 _execute 三类形态）。"""
class FakeExec:
def __init__(self):
self.calls = []
def execute(self, spec_in):
self.calls.append(spec_in)
out_dir = Path(spec_in.output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
for name in outputs:
(out_dir / name).write_text("data", encoding="utf-8")
if spec_in.log_path:
Path(spec_in.log_path).parent.mkdir(parents=True, exist_ok=True)
Path(spec_in.log_path).write_text(log, encoding="utf-8")
return ext.ExtensionRunResult(
run_id="", exit_code=exit_code, log_tail=log,
outputs=list(outputs),
error="" if exit_code == 0 else "命令失败",
)
return FakeExec()
# ---- 创建与布局（契约 §1）----
async def test_create_layout(tmp_path):
s = ext.ExtensionStore(base_dir=str(tmp_path / "ext"))
e = _create(s)
assert e["status"] == "draft"
assert e["ext_id"].startswith("ext_")
assert e["runtime_image"] == "bioflow-gf" # test4py311 → 映射
assert e["schema_version"] == spec.SPEC_VERSION
root = s.ext_dir(e["ext_id"])
assert (root / "extension.json").is_file()
assert (root / "raw_code" / "quality_stats.py").read_text(encoding="utf-8") == _CODE
assert (root / "converted_code" / "quality_stats.py").is_file()
assert (root / "raw_test_data").is_dir()
assert (root / "validation").is_dir()
s.delete(e["ext_id"])
def test_create_requires_command_and_code(tmp_path):
s = ext.ExtensionStore(base_dir=str(tmp_path / "ext_req"))
with pytest.raises(ext.ExtensionError, match="display_name"):
s.create_extension(display_name=" ", command="python a.py", code_files=[("a.py", b"x")])
with pytest.raises(ext.ExtensionError, match="command"):
s.create_extension(display_name="n", command="", code_files=[("a.py", b"x")])
with pytest.raises(ext.ExtensionError, match="代码包"):
s.create_extension(display_name="n", command="python a.py", code_files=[])
def test_create_zip_path_traversal_sanitized(store):
e = _create(store, code=[("../../evil.py", b"x"), ("ok.py", b"y")])
root = store.ext_dir(e["ext_id"])
assert (root / "raw_code" / "evil.py").is_file()
assert not (root.parent / "evil.py").exists()
def test_load_missing_404(store):
with pytest.raises(ext.ExtensionNotFoundError):
store.load_extension("ext_nope")
# ---- 状态机 ----
async def test_status_transitions_valid(store):
e = _create(store)
assert (await store.build_extension(e["ext_id"]))["status"] == "built"
async def test_invalid_transition(store):
e = _create(store)
with pytest.raises(ext.ExtensionStatusError):
await store.trial_run(e["ext_id"]) # draft → trial 非法
await store.build_extension(e["ext_id"])
with pytest.raises(ext.ExtensionStatusError):
store.publish(e["ext_id"]) # built → publish 非法（须 validate/confirm）
async def test_build_failed_converges(store):
e = _create(store)
# 命令无 .py 文件可引用 + 无 LLM → 规则版仍能给出模板（不会 build_failed）；
# 构造一个规则版也会失败的场景：空原始命令被创建层拦截。这里直接验证
# 规则版回退路径本身可用（无 LLM key 时 build 成功且 source=rules）。
out = await store.build_extension(e["ext_id"])
assert out["status"] == "built"
assert out["last_build"]["source"] == "rules"
assert "{output_dir}" in out["command_template"]
spec.validate_command_template(out["command_template"])
async def test_llm_invalid_output_falls_back_rules(store, monkeypatch):
"""LLM 返回白名单外占位符 → 视为转换失败 → 规则版回退（build 仍成功）。"""
e = _create(store)
monkeypatch.setattr("app.services.bioflow.extension_llm.llm_config",
lambda: {"base_url": "http://x", "api_key": "k", "model": "m", "timeout": 1.0})
async def fake_complete_json(system, user, **kw):
return {"command_template": "python {code_dir}/a.py --out {not_allowed}",
"requirements": {}, "nl_triggers": ["t"]}
monkeypatch.setattr("app.services.bioflow.llm.complete_json", fake_complete_json)
out = await store.build_extension(e["ext_id"])
assert out["status"] == "built"
assert out["last_build"]["source"] == "rules"
spec.validate_command_template(out["command_template"])
# ---- trial ----
async def test_trial_pass(store):
e = _create(store, data=[("input.fastq", b">s1\nACGT\n")])
await store.build_extension(e["ext_id"])
store.executor = _fake_executor(0, "ok\n", ("result.txt",))
out = await store.trial_run(e["ext_id"])
assert out["status"] == "trial_passed"
assert out["trial"]["ok"] is True
assert out["trial"]["outputs"] == ["result.txt"]
vdir = store.validation_dir(e["ext_id"])
assert (vdir / "trial_1.log").is_file()
assert (vdir / "trial_1.outputs" / "result.txt").is_file()
async def test_trial_fail(store):
e = _create(store)
await store.build_extension(e["ext_id"])
store.executor = _fake_executor(2, "boom\n", ())
out = await store.trial_run(e["ext_id"])
assert out["status"] == "trial_failed"
assert out["trial"]["ok"] is False
assert "命令失败" in out["trial"]["error"]
async def test_trial_retry_increments(store):
e = _create(store)
await store.build_extension(e["ext_id"])
store.executor = _fake_executor(1, "fail\n", ())
await store.trial_run(e["ext_id"])
store.executor = _fake_executor(0, "ok\n", ("r.txt",))
out = await store.trial_run(e["ext_id"])
assert out["trial"]["n"] == 2
assert out["status"] == "trial_passed"
async def test_trial_without_executor_503_semantics(store):
e = _create(store)
await store.build_extension(e["ext_id"])
with pytest.raises(ext.ExtensionRunError):
await store.trial_run(e["ext_id"]) # 占位执行器 → 明确错误（路由层 503）
# ---- validate ----
async def test_validate_without_golden(store):
e = _create(store)
await store.build_extension(e["ext_id"])
store.executor = _fake_executor(0, "ok\n")
await store.trial_run(e["ext_id"])
out = store.validate(e["ext_id"])
assert out["status"] == "validated"
assert out["validation"]["compared"] is False
async def test_validate_golden_pass(store):
e = _create(store, data=[("sample.fastq", b">s\nACGT\n")],
golden={"test_input": "sample.fastq", "expected_outputs": ["result.txt"]})
await store.build_extension(e["ext_id"])
store.executor = _fake_executor(0, "ok\n", ("result.txt",))
await store.trial_run(e["ext_id"])
out = store.validate(e["ext_id"])
assert out["status"] == "validated"
assert out["validation"]["compared"] is True
assert out["validation"]["ok"] is True
async def test_validate_golden_missing_output(store):
e = _create(store, golden={"test_input": "sample.fastq", "expected_outputs": ["gold.tsv"]})
await store.build_extension(e["ext_id"])
store.executor = _fake_executor(0, "ok\n", ("other.txt",))
await store.trial_run(e["ext_id"])
out = store.validate(e["ext_id"])
assert out["status"] == "trial_failed"
assert out["validation"]["missing_outputs"] == ["gold.tsv"]
# ---- repair ----
async def test_repair_limit(store):
e = _create(store)
await store.build_extension(e["ext_id"])
store.executor = _fake_executor(1, "No module named 'pandas'\n", ())
await store.trial_run(e["ext_id"])
store.max_repair_rounds = 1
out = await store.repair(e["ext_id"]) # 第 1 轮成功（规则版并入 pandas）
assert out["repair_count"] == 1
assert "pandas" in out["requirements"]["pip"]
store.executor = _fake_executor(1, "still fail\n", ())
await store.trial_run(e["ext_id"])
with pytest.raises(ext.ExtensionStatusError, match="上限"):
await store.repair(e["ext_id"])
async def test_repair_requires_failure(store):
e = _create(store)
await store.build_extension(e["ext_id"])
with pytest.raises(ext.ExtensionStatusError):
await store.repair(e["ext_id"]) # built（未失败）不可修复
# ---- confirm / publish ----
async def test_full_flow_to_publish(store):
e = _create(store)
ext_id = e["ext_id"]
await store.build_extension(ext_id)
store.executor = _fake_executor(0, "ok\n")
await store.trial_run(ext_id)
store.validate(ext_id)
out = store.confirm(ext_id, nl_triggers=["质量统计", "fastq 统计"])
assert out["status"] == "confirming"
assert out["nl_triggers"] == ["质量统计", "fastq 统计"]
pub = store.publish(ext_id)
assert pub["status"] == "published"
# 已发布不可再发布流转（幂等返回）
assert store.publish(ext_id)["status"] == "published"
async def test_confirm_parameters_reconcile(store):
e = _create(store, command="python quality_stats.py --fastq input.fastq --cpu {param_cpu} --out results/")
await store.build_extension(e["ext_id"])
store.executor = _fake_executor(0)
await store.trial_run(e["ext_id"])
store.validate(e["ext_id"])
# 用户编辑后的 parameters 丢掉 {param_cpu} → 自动补齐
out = store.confirm(e["ext_id"], parameters=[])
names = {p["name"] for p in out["parameters"]}
assert "cpu" in names
# ---- run ----
async def test_run_requires_published(store):
e = _create(store)
await store.build_extension(e["ext_id"])
with pytest.raises(ext.ExtensionStatusError, match="仅已发布"):
await store.run(e["ext_id"], params={}, files=[])
async def test_run_full(store):
e = _create(store)
ext_id = e["ext_id"]
await store.build_extension(ext_id)
store.executor = _fake_executor(0, "run ok\n", ("out.tsv",))
await store.trial_run(ext_id)
store.validate(ext_id)
store.confirm(ext_id)
store.publish(ext_id)
# 输入文件登记（须在白名单根内：data_root 或扩展目录）
data_root = Path(store.base_dir).parent
input_file = data_root / "input.fastq"
input_file.write_text(">s\nACGT\n", encoding="utf-8")
result = await store.run(ext_id, params={"cpu": "4"}, files=[str(input_file)])
assert result["status"] == "done"
assert result["exit_code"] == 0
assert "out.tsv" in result["outputs"]
run_dir = store.runs_dir(ext_id) / result["run_id"]
assert (run_dir / "run_result.json").is_file()
assert (run_dir / "inputs" / "input.fastq").is_file()
assert (run_dir / "outputs" / "out.tsv").is_file()
runs = store.list_runs(ext_id)
assert len(runs) == 1
assert runs[0]["run_id"] == result["run_id"]
async def test_run_rejects_outside_file(store):
e = _create(store)
ext_id = e["ext_id"]
await store.build_extension(ext_id)
store.executor = _fake_executor(0)
await store.trial_run(ext_id)
store.validate(ext_id)
store.publish(ext_id)
with pytest.raises(ext.ExtensionRunError, match="白名单"):
await store.run(ext_id, params={}, files=["/etc/passwd"])
async def test_run_failed_status(store):
e = _create(store)
ext_id = e["ext_id"]
await store.build_extension(ext_id)
store.executor = _fake_executor(3, "output contract failed\n", ())
await store.trial_run(ext_id) # 空输出 → 失败（假执行器 exit=3）
assert store.load_extension(ext_id)["status"] == "trial_failed"
# ---- 列表 / 删除 ----
async def test_list_published_first(store):
a = _create(store, name="扩展 A")
await store.build_extension(a["ext_id"])
b = _create(store, name="扩展 B")
await store.build_extension(b["ext_id"])
store.executor = _fake_executor(0)
await store.trial_run(b["ext_id"])
store.validate(b["ext_id"])
store.publish(b["ext_id"])
items = store.list_extensions()
assert items[0]["ext_id"] == b["ext_id"] # published 优先
assert items[1]["ext_id"] == a["ext_id"]
def test_delete_guard_trial_running(store):
e = _create(store)
# 手工置 trial_running（不可删除守卫；无需 build——直接构造状态）
store.save_extension({**store.load_extension(e["ext_id"]), "status": "trial_running"})
with pytest.raises(ext.ExtensionStatusError, match="试运行"):
store.delete(e["ext_id"])
store.save_extension({**store.load_extension(e["ext_id"]), "status": "draft"})
store.delete(e["ext_id"])
assert store.load_extension(e["ext_id"], required=False) is None
# ---- 并发写合并 ----
async def test_apply_and_write_preserves_fields(store):
e = _create(store)
ext_id = e["ext_id"]
# 两次锁内读-改-写（模拟交错协程写不同字段）：后写不丢先写字段
store._apply_and_write(ext_id, lambda ext: ext.__setitem__(
"command_template", "python {code_dir}/a.py --out {output_dir}"))
store._apply_and_write(ext_id, lambda ext: ext.__setitem__("nl_triggers", ["触发词"]))
merged = store.load_extension(ext_id)
assert merged["command_template"] == "python {code_dir}/a.py --out {output_dir}"
assert merged["nl_triggers"] == ["触发词"]
assert merged["original_command"] # 其余字段保留
# ---- 事件总线发布（契约 §4；run 完成后可回放）----
async def test_run_publishes_events(store, tmp_path):
from app.services.bioflow.events import EventBus
bus = EventBus(tmp_path / "events")
e = _create(store)
ext_id = e["ext_id"]
await store.build_extension(ext_id)
store.executor = _fake_executor(0, "run ok\n", ("o.txt",))
await store.trial_run(ext_id)
store.validate(ext_id)
store.publish(ext_id)
store._bus_override = bus
result = await store.run(ext_id, params={}, files=[])
replay = bus.replay(ext_id)
types = [ev["type"] for ev in replay]
assert "extension_run_started" in types
assert "extension_run_done" in types
assert result["status"] == "done"
docs/EXTENSION_CONTRACT.md
+113
View File
@@ -0,0 +1,113 @@
# 扩展分析契约文档(Dev-A/B/C 与测试 agent 对齐基线)
> 生成 2026-09-12,方案见 docs/DEPLOY_SKILL.md 同源的扩展分析方案。三个开发分支与本契约对齐,测试 agent 按本契约+计划文档验收。
## 1. 扩展资产目录布局(服务端存储)
```
{BIOFLOW_DATA}/extensions/{ext_id}/
├── extension.json # 唯一事实源(schema 见 §2)
├── raw_code/ # 上传 zip 解压的原始代码(含 run.sh 等)
├── raw_test_data/ # 上传的测试数据(可选)
├── converted_code/ # AI 转换后写入的代码(与 raw_code 相同或改写版)
├── validation/ # 试运行/金标/修复循环中间产物
│ ├── trial_{n}.log
│ ├── trial_{n}.outputs/
│ └── golden_report.json
└── runs/{run_id}/ # 每次正式运行
├── inputs/ # 运行输入(路径登记或复制的文件)
├── outputs/ # 运行输出(输出契约校验对象)
├── run.log
└── run_result.json # {run_id, ext_id, exit_code, started_at, finished_at, outputs: [...], error}
```
## 2. extension.json schema(唯一事实源)
```json
{
"schema_version": "extension_stage_v1",
"ext_id": "ext_xxxxxxxx",
"display_name": "测序质量统计",
"status": "draft|building|built|trial_running|trial_passed|trial_failed|validating|validated|repairing|confirming|published|build_failed",
"command_template": "python {code_dir}/quality_stats.py --fastq {input_file} --out {output_dir}",
"original_command": "python quality_stats.py --fastq input.fastq --out results/",
"runtime_image": "bioflow-gf",
"requirements": { "pip": [], "apt": [] },
"parameters": [ { "name": "fastq", "type": "file", "required": true, "description": "..." } ],
"input_roles": { "input_file": "fastq" },
"nl_triggers": ["质量统计", "fastq 统计"],
"analysis_type": "table_filter",
"golden": { "test_input": "sample.fastq", "expected_outputs": ["..."] } | null,
"repair_count": 0,
"created_at": "...", "updated_at": "..."
}
```
**占位符白名单(spec_v2 移植)**:`{code_dir} {input_dir} {input_file} {input_hmm} {input_fasta} {input_query_fasta} {input_reference_fasta} {input_pep_fasta} {input_cds_fasta} {input_id_file} {output_dir} {work_dir}` + `{param_<name>}`。硬约束:`{output_dir}` 必含;白名单外占位符拒绝。
## 3. docker 执行契约(Dev-B 负责实现)
宿主挂载入口 `bioflow_kernel/cli/run_extension.py`,命令形态:
```
docker run --rm --user <uid> \
-v <ext_dir>:/job/ext:ro \
-v <input_dir>:/job/input \
-v <output_dir>:/job/output \
-v <宿主 bioflow_kernel/cli>:/opt/bioflow-entry:ro \
--cpus <白名单> --memory <白名单> \
<image>:dev python -B /opt/bioflow-entry/run_extension.py \
--ext-dir /job/ext --input-dir /job/input --output-dir /job/output [--param name=value ...]
```
run_extension.py 职责:读 /job/ext/extension.json → 渲染 command_template(code_dir=/job/ext/converted_code)→ subprocess 执行 → 日志/退出码返回 → **输出契约校验(输出目录非空,否则退出码 3)**。8 个 base 镜像与自动生成镜像零改动。
## 4. 事件帧契约(chat/stream 扩展分支下发,Dev-C 消费)
帧 = `{event, data}`,事件集:
| event | data 关键字段 | 触发 |
|---|---|---|
| `extension_upload` | {session_ext_id} | 创建意图命中 → 上传引导卡 |
| `extension_created` | {ext_id, display_name, status:"draft"} | 上传完成 |
| `extension_build` | {ext_id, command_template, requirements, runtime_image, nl_triggers, analysis_type, parameters, status} | AI 转换完成 → 转换卡 |
| `extension_trial` | {ext_id, status:"running"\|"done"\|"failed", log_tail, exit_code, error} | 试运行状态/结果 |
| `extension_repair` | {ext_id, repair_count, suggestion, status} | 修复循环 |
| `extension_validate` | {ext_id, golden: {...}\|null, form_defaults} | 金标/确认卡 |
| `extension_published` | {ext_id, display_name, nl_triggers} | 发布完成 |
| `extension_candidates` | {candidates: [{ext_id, display_name, purpose}]} | 多命中候选卡 |
| `extension_confirm` | {ext_id, params_schema, extracted_params, files_required} | 运行前参数确认卡 |
| `extension_run_started` | {run_id, ext_id} | 运行开始 |
| `extension_run_log` | {run_id, log} | 运行日志追加 |
| `extension_run_done` | {run_id, outputs} | 运行完成 |
| `extension_run_failed` | {run_id, exit_code, error, log_tail} | 运行失败 |
## 5. 意图两层判定(Dev-A 负责,intent.py 扩展)
- **A 层创建**:动作词{创建,建,新建,上传,自定义,写,弄,做一个} × 目标词{扩展,拓展分析,分析模板,分析脚本,分析工具},同句命中即进创建流程。
- **B 层运行**:已发布扩展打分 = display_name 模糊 + nl_triggers 关键词 + analysis_type;动作词{帮我跑,用,做一下,分析一下}加分。仅 published 可被触发。
- **歧义消解**:目标词命中但无动作词 → 有 B 层命中走运行;无命中回「没有匹配的已发布扩展,是否创建?」引导卡。
- **漏判安全**:无命中且非创建 → 回落正常生信阶段规划,现网零影响。
## 6. 端点清单(Dev-A 实现,Dev-C 调用)
```
POST /api/bioflow/extensions (multipart zip + display_name + command + runtime_image? + golden?)
GET /api/bioflow/extensions (列表,published 优先)
GET /api/bioflow/extensions/{id}
DELETE /api/bioflow/extensions/{id}
POST /api/bioflow/extensions/{id}/build (AI 转换)
POST /api/bioflow/extensions/{id}/trial-run (test_inputs 文件)
POST /api/bioflow/extensions/{id}/repair (LLM 修复,上限 20 轮可配)
POST /api/bioflow/extensions/{id}/publish
POST /api/bioflow/extensions/{id}/run (params + files → run_id)
GET /api/bioflow/extensions/{id}/runs
GET /api/bioflow/extensions/{id}/runs/{run_id}/events (SSE,首帧快照)
```
## 7. 约束(全部 agent)
- 代码仅限 175 仓库 `/home/xzq/5090/teamWorkspace/zh/PlantAI-rag-new`(worktree 副本)。
- 不 push、不部署 176、不动生产部署树;dev 栈不得重启生产服务。
- 依赖安装:服务端零新增依赖(纯 stdlib+现有 httpx/pydantic/fastapi);镜像内依赖走自动镜像生成(Dev-B)。
- 现有 bioflow 路径零改动:kernel.py/runner.py/planner.py/catalog.py/stage_image_map.json 一律不碰(扩展是并行链路)。