fix(bioflow): rnaseq/wgs 阶段参数契约——CLI 参数注入+样本配置自动生成+产物路径对齐

- stage_image_map 加 cli_contract(13 个 rnaseq 阶段):-i/-o/-c 等
  标志按契约注入;无契约阶段保持 --input-dir 默认形态
- rnaseq 系 -o 语义=产物目录,契约注入 workspace/final(CLI 内部以
  output.parent 为 workspace 并建 output.parent/final——此前注入
  workspace 导致 /job/final 权限错误)
- 登记目录自动生成 samples.yaml(按 f1/r1+f2/r2 配对,不覆盖手工配置;
  此前 CLI -c 必填无来源,rnaseq 全链从未真实跑通)
- catalog 补 hisat2_index/gtf/prefix 参数 schema(确认表单收集)
- 338 用例全绿;176 真实任务验证:fastp 在真实 fastq 上产出 clean
  产物+报告(首个真实跑通的 rnaseq 阶段)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-16 15:23:50 +08:00
co-authored by Claude Fable 5
parent 04fb306624
commit 3b512fffd0
6 changed files with 300 additions and 16 deletions
+163 -13
View File
@@ -388,68 +388,218 @@
"image": "bioflow-rnaR",
"module": "cli.run_expression",
"executable": true,
"note": "跨环境: 主 rnaR(R 绘图); 内部 rnaseq_quant 矩阵触发走 rnaseq 镜像子阶段(编排层拆分)"
"note": "跨环境: 主 rnaR(R 绘图); 内部 rnaseq_quant 矩阵触发走 rnaseq 镜像子阶段(编排层拆分)",
"cli_contract": [
[
"--workspace",
"workspace"
],
[
"--prefix",
"prefix"
]
]
},
"rnaseq_qc": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_qc",
"executable": true
"executable": true,
"cli_contract": [
[
"--input",
"input_dir"
],
[
"--config",
"samples_yaml"
],
[
"--output",
"output_final"
]
]
},
"rnaseq_align": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_align",
"executable": true
"executable": true,
"cli_contract": [
[
"--input",
"input_dir"
],
[
"--hisat2-index",
"param:hisat2_index"
],
[
"--output",
"output_final"
]
]
},
"rnaseq_quant": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_quant",
"executable": true
"executable": true,
"cli_contract": [
[
"--input",
"input_dir"
],
[
"--gtf",
"param:gtf"
],
[
"--output",
"output_final"
]
]
},
"rnaseq_sample_qc": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_sample_qc",
"executable": true
"executable": true,
"cli_contract": [
[
"--workspace",
"workspace"
]
]
},
"rnaseq_wgcna": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_wgcna",
"executable": true
"executable": true,
"cli_contract": [
[
"--workspace",
"workspace"
]
]
},
"rnaseq_de": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_de",
"executable": true,
"note": "跨环境: DE 表在 rnaseq; DESeq2 R 统计/绘图可拆分 r_de_analysis 镜像子阶段(编排层)"
"note": "跨环境: DE 表在 rnaseq; DESeq2 R 统计/绘图可拆分 r_de_analysis 镜像子阶段(编排层)",
"cli_contract": [
[
"--input",
"input_dir"
],
[
"--config",
"samples_yaml"
],
[
"--output",
"output_final"
]
]
},
"rnaseq_go_enrich": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_enrich",
"executable": true
"executable": true,
"cli_contract": [
[
"--input",
"input_dir"
],
[
"--output",
"output_final"
],
[
"--prefix",
"prefix"
]
]
},
"rnaseq_kegg_enrich": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_enrich",
"executable": true
"executable": true,
"cli_contract": [
[
"--input",
"input_dir"
],
[
"--output",
"output_final"
],
[
"--prefix",
"prefix"
]
]
},
"rnaseq_fgsea": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_fgsea",
"executable": true
"executable": true,
"cli_contract": [
[
"--workspace",
"workspace"
]
]
},
"rnaseq_candidate_family": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_candidate_family",
"executable": true
"executable": true,
"cli_contract": [
[
"--workspace",
"workspace"
],
[
"--prefix",
"prefix"
]
]
},
"rnaseq_report": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_report",
"executable": true
"executable": true,
"cli_contract": [
[
"--workspace",
"workspace"
],
[
"--prefix",
"prefix"
]
]
},
"rnaseq_enrich": {
"image": "bioflow-rnaseq",
"module": "cli.run_rnaseq_enrich_v2",
"executable": true
"executable": true,
"cli_contract": [
[
"--input",
"input_dir"
],
[
"--output",
"output_final"
],
[
"--prefix",
"prefix"
],
[
"--job-id",
"job_id"
]
]
},
"enrichment": {
"image": "bioflow-rnaseq",
+7
View File
@@ -281,6 +281,13 @@ async def _register_path(record: Any, server_path: str, settings: Any) -> dict[s
link.symlink_to(p)
except OSError as exc: # noqa: BLE001
logger.warning("登记 symlink 失败(不阻断): %s -> %s (%s)", p, link, exc)
# rnaseq 系阶段需要 samples.yaml(CLI -c 必填):从 fastq 自动配对生成(不覆盖手工配置)
try:
from ..services.bioflow.runner import ensure_samples_yaml
if ensure_samples_yaml(input_dir):
logger.info("已自动生成 samples.yaml(fastq 配对): %s", input_dir)
except Exception as exc: # noqa: BLE001
logger.warning("samples.yaml 自动生成失败(不阻断登记): %s", exc)
registered = [str(p) for p in files]
try:
_persist_registered_paths(record, registered)
@@ -343,12 +343,16 @@ _RNASEQ_ALIGN_PARAMS = [
"help": "样本列表(逗号分隔)"},
{"name": "ref_genome", "type": "string", "default": "", "source": "user",
"help": "参考基因组路径/索引名"},
{"name": "hisat2_index", "type": "string", "default": "", "source": "user",
"help": "HISAT2 索引前缀(容器内可访问路径,如 /bioflow-data/hmm/hisat2/xxx)"},
{"name": "cpu", "type": "choice", "choices": ["4", "8", "16", "32"], "default": "8",
"source": "runtime", "help": "容器 CPU 数"},
]
_RNASEQ_QUANT_PARAMS = [
{"name": "samples", "type": "string", "default": "", "source": "user",
"help": "样本列表(逗号分隔)"},
{"name": "gtf", "type": "string", "default": "", "source": "user",
"help": "参考基因组 GTF 注释(容器内可访问路径)"},
{"name": "cpu", "type": "choice", "choices": ["4", "8", "16", "32"], "default": "6",
"source": "runtime", "help": "容器 CPU 数"},
]
@@ -361,6 +365,8 @@ _RNASEQ_SAMPLE_QC_PARAMS = [
_RNASEQ_DE_PARAMS = [
{"name": "design", "type": "string", "default": "", "source": "user",
"help": "实验设计矩阵文件(对比组声明)"},
{"name": "prefix", "type": "string", "default": "", "source": "contract",
"help": "分析前缀"},
{"name": "cpu", "type": "choice", "choices": ["4", "8", "16", "32"], "default": "4",
"source": "runtime", "help": "容器 CPU 数"},
]
@@ -373,6 +379,8 @@ _RNASEQ_WGCNA_PARAMS = [
_RNASEQ_ENRICH_PARAMS = [
{"name": "gene_list", "type": "string", "default": "", "source": "contract",
"help": "差异基因列表(上游 DE 结果)"},
{"name": "prefix", "type": "string", "default": "", "source": "contract",
"help": "分析前缀"},
{"name": "cpu", "type": "choice", "choices": ["4", "8", "16", "32"], "default": "4",
"source": "runtime", "help": "容器 CPU 数"},
]
+47 -3
View File
@@ -165,9 +165,13 @@ class Kernel:
cmd.append(f"{tag}:{self.cfg.docker_tag_suffix}")
cmd.extend(CONTAINER_CMD_PREFIX)
cmd.extend(["-m", mod])
# 通用容器内参数:input-dir/workspace/cpu(各入口 argparse 兼容)
cmd += ["--input-dir", self.cfg.container_input_dir]
cmd += ["--workspace", self.cfg.container_workspace_dir]
# 参数注入:有 cli_contract 的阶段按契约,否则沿用默认 --input-dir/--workspace
contract = entry.get("cli_contract")
if contract:
cmd += self._contract_args(contract, params, job_input_dir)
else:
cmd += ["--input-dir", self.cfg.container_input_dir]
cmd += ["--workspace", self.cfg.container_workspace_dir]
for key in sorted(params):
value = params[key]
if key in ("env", "cpu", "memory"):
@@ -185,6 +189,46 @@ class Kernel:
cmd.append(str(value))
return cmd
def _contract_args(self, contract: list[Any], params: dict[str, Any],
job_input_dir: str | Path) -> list[str]:
"""cli_contract 解析:[(flag, source), ...] → argv。
source 取值:
- input_dir/workspace:容器内 /job/input、/job/workspace
- output_final:/job/workspace/final(rnaseq 系 -o 语义:产物目录,
CLI 内部以 output.parent 为 workspace 并建 output.parent/final——
故 -o 必须指向 workspace/final,manifest 也落此处与 runner 检索对齐)
- samples_yaml:/job/input/samples.yaml(宿主侧不存在则跳过该 flag,
交由 CLI 报缺失,错误信息比服务端更可读)
- prefix/job_id:从 params 取值,缺失跳过
- param:<name>:从 params[name] 取值,缺失跳过(用户经确认表单提供)
"""
args: list[str] = []
for flag, source in contract:
if source == "input_dir":
args += [flag, self.cfg.container_input_dir]
elif source == "workspace":
args += [flag, self.cfg.container_workspace_dir]
elif source == "output_final":
args += [flag, f"{self.cfg.container_workspace_dir}/final"]
elif source == "samples_yaml":
if (Path(job_input_dir) / "samples.yaml").is_file():
args += [flag, f"{self.cfg.container_input_dir}/samples.yaml"]
elif source == "prefix":
value = str(params.get("prefix") or "").strip()
if value:
args += [flag, value]
elif source == "job_id":
value = str(params.get("job_id") or "").strip()
if value:
args += [flag, value]
elif str(source).startswith("param:"):
key = str(source).split(":", 1)[1]
value = params.get(key)
if value is not None and str(value).strip():
args += [flag, str(value)]
return args
def _safe_cpu(self, cpu: str) -> str:
if str(cpu).strip() in {c.strip() for c in self.cfg.cpu_allow.split(",")}:
return str(cpu).strip()
@@ -92,6 +92,43 @@ def _stage_input_satisfied(stage: str, job_input_dir: Path) -> tuple[bool, list[
return False, list(DEFAULT_INPUT_HINTS)
def ensure_samples_yaml(job_input_dir: Path) -> bool:
"""为 rnaseq 系阶段自动生成 samples.yaml(从 fastq 按样本前缀配对)。
扫描 input/ 下 *.fq.gz/fastq.gz,按 <样本>_f1/_r1 与 _f2/_r2 配对,生成
{samples: {<样本>: {r1: <名>, r2: <名>}}}。已存在则不覆盖(用户手工配置优先)。
返回是否生成了新文件。
"""
import re as _re
import yaml as _yaml
yaml_path = job_input_dir / "samples.yaml"
if yaml_path.exists():
return False
pairs: dict[str, dict[str, str]] = {}
for p in sorted(job_input_dir.rglob("*")):
if not p.is_file() or p.name == "samples.yaml":
continue
m = _re.match(r"^(.+?)[._-](f1|r1|1)(\.(fastq|fq))?(\.gz)?$", p.name, _re.I)
m2 = _re.match(r"^(.+?)[._-](f2|r2|2)(\.(fastq|fq))?(\.gz)?$", p.name, _re.I)
m = m or m2
if not m:
continue
sample, mate = m.group(1), (m.group(2) or "").lower()
rec = pairs.setdefault(sample, {})
if mate in ("f1", "r1", "1"):
rec["r1"] = p.name
else:
rec["r2"] = p.name
samples = {s: v for s, v in sorted(pairs.items()) if v.get("r1") and v.get("r2")}
if not samples:
return False
yaml_path.write_text(
_yaml.safe_dump({"samples": samples}, allow_unicode=True), encoding="utf-8"
)
return True
# ---- runner ----
ExecutorFn = Callable[..., Awaitable[dict[str, Any]]]
+38
View File
@@ -89,6 +89,44 @@ def test_build_command_kernel_dir_mount(tmp_path):
assert "-w" in cmd and "/bioflow-kernel" in cmd
def test_build_command_cli_contract(tmp_path):
"""契约阶段:按 cli_contract 注入 -i/-o/-c;无契约阶段保持 --input-dir。"""
from app.services.bioflow.kernel import Kernel, KernelConfig, ImageMap
cfg = KernelConfig(host_uid=1234, data_root=str(tmp_path / "data"))
img = ImageMap(
images={"rnaseq": {"docker_tag": "bioflow-rnaseq"}},
stages={
"rnaseq_qc": {"image": "rnaseq", "module": "cli.run_rnaseq_qc",
"cli_contract": [["--input", "input_dir"],
["--config", "samples_yaml"],
["--output", "output_final"]]},
"rnaseq_align": {"image": "rnaseq", "module": "cli.run_rnaseq_align",
"cli_contract": [["--input", "input_dir"],
["--hisat2-index", "param:hisat2_index"],
["--output", "output_final"]]},
"plain": {"image": "rnaseq", "module": "cli.run_plain"},
},
)
k = Kernel(img, cfg)
# 有 samples.yaml 时注入 -c
(tmp_path / "in").mkdir(parents=True, exist_ok=True)
(tmp_path / "in" / "samples.yaml").write_text("samples: {}", encoding="utf-8")
cmd = k.build_command("rnaseq_qc", tmp_path / "in", tmp_path / "ws")
assert "--input" in cmd and "/job/input" in cmd
assert "--config" in cmd and "/job/input/samples.yaml" in cmd
assert "--output" in cmd and "/job/workspace/final" in cmd # rnaseq -o 语义对齐
assert "--input-dir" not in cmd # 契约阶段不再注入默认形态
# param: 有值注入、无值跳过
cmd2 = k.build_command("rnaseq_align", tmp_path / "in", tmp_path / "ws",
params={"hisat2_index": "/bioflow-data/hmm/hisat2/osa"})
assert "--hisat2-index" in cmd2 and "/bioflow-data/hmm/hisat2/osa" in cmd2
cmd3 = k.build_command("rnaseq_align", tmp_path / "in", tmp_path / "ws")
assert "--hisat2-index" not in cmd3
# 无契约阶段回归默认
cmd4 = k.build_command("plain", tmp_path / "in", tmp_path / "ws")
assert "--input-dir" in cmd4 and "--workspace" in cmd4
def test_build_command_params(kernel, tmp_path):
cmd = kernel.build_command("identify", tmp_path / "in", tmp_path / "ws",
{"prefix": "Ft.SPL", "required_pfams": "PF03110",