from __future__ import annotations import hashlib import json from pathlib import Path from apps.web import pipeline_backend from apps.web.vue3.main import ExecuteRequest from gfa_core.mas.agents.planning_v2 import PlanningAgentV2 from gfa_core.mas.stage_capabilities import CAPABILITIES from gfa_core import project_paths from gfa_core.mas.stage_catalog import EXECUTABLE_STAGES, STAGE_DEPENDENCIES from gfa_core.mas.tools.stage_tools import StageToolRegistry SAMPLES = "wgs-samples:" + "a" * 64 REFERENCE = "wgs-reference:" + "b" * 64 MARKDUP_ANALYSIS = "wgs-analysis:" + "c" * 64 INDEX_ANALYSIS = "wgs-analysis:" + "d" * 64 def _record(workspace: Path, path: Path, role: str, sample: str = "") -> dict: value = { "artifact": role, "role": role, "path": str(path.relative_to(workspace)), "full_path": str(path.resolve()), "size": path.stat().st_size, "size_bytes": path.stat().st_size, "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), } if sample: value["sample_id"] = sample return value def _authorities(workspace: Path) -> tuple[Path, Path]: final = workspace / "final" markdup_bundle = final / ".wgs_markdup_runs" / "Cohort" / ("4" * 32) markdup_bundle.mkdir(parents=True) artifacts = {} for sample in ("S1", "S2", "S3"): for role, suffix in ( ("markdup_bam", "markdup.bam"), ("markdup_bam_index", "markdup.bam.bai"), ): path = markdup_bundle / f"Cohort.wgs_markdup.{sample}.{suffix}" path.write_bytes(f"{sample}-{role}".encode()) artifacts[f"{role}.{sample}"] = _record( workspace, path, role, sample ) markdup = final / "Cohort.wgs_markdup.manifest.json" markdup.write_text(json.dumps({ "manifest_version": "1.0", "stage": "wgs_markdup", "status": "success", "prefix": "Cohort", "job_id": "job-1", "run_id": "4" * 32, "workspace": str(workspace.resolve()), "wgs_sample_set_id": SAMPLES, "wgs_reference_id": REFERENCE, "wgs_analysis_id": MARKDUP_ANALYSIS, "artifacts": artifacts, })) index_bundle = final / ".wgs_index_runs" / "Cohort" / ("2" * 32) index_bundle.mkdir(parents=True) fai = index_bundle / "Cohort.wgs_index.reference.fasta.fai" fai.write_text("chr1\t10\t0\t0\t0\n") index = final / "Cohort.wgs_index.manifest.json" index.write_text(json.dumps({ "manifest_version": "1.0", "stage": "wgs_index", "status": "success", "prefix": "Cohort", "job_id": "job-1", "run_id": "2" * 32, "workspace": str(workspace.resolve()), "wgs_reference_id": REFERENCE, "wgs_analysis_id": INDEX_ANALYSIS, "artifacts": {"reference_fai": _record(workspace, fai, "reference_fai")}, })) return markdup, index def _publish_result(workspace: Path) -> Path: final = workspace / "final" bundle = final / ".wgs_bam_qc_runs" / "Cohort" / ("5" * 32) bundle.mkdir(parents=True) artifacts = {} for sample in ("S1", "S2", "S3"): for role, suffix in ( ("bam_flagstat", "flagstat.txt"), ("bam_depth_summary", "depth.tsv"), ): path = bundle / f"Cohort.wgs_bam_qc.{sample}.{suffix}" path.write_text(f"{sample}-{role}\n") artifacts[f"{role}.{sample}"] = _record( workspace, path, role, sample ) for role, suffix in ( ("bam_qc_summary", "cohort.summary.tsv"), ("bam_qc_report", "cohort.report.md"), ): path = bundle / f"Cohort.wgs_bam_qc.{suffix}" path.write_text(f"{role}\n") artifacts[role] = _record(workspace, path, role) output = final / "Cohort.wgs_bam_qc.manifest.json" output.write_text(json.dumps({ "manifest_version": "1.0", "stage": "wgs_bam_qc", "status": "success", "prefix": "Cohort", "job_id": "job-1", "run_id": "5" * 32, "workspace": str(workspace.resolve()), "wgs_sample_set_id": SAMPLES, "wgs_reference_id": REFERENCE, "upstream_wgs_analysis_ids": { "wgs_markdup": MARKDUP_ANALYSIS, "wgs_index": INDEX_ANALYSIS, }, "wgs_analysis_id": "wgs-analysis:" + "e" * 64, "parameters": {"metric_name": "coverage_ge_1x"}, "metrics": { "input_sample_ids": ["S1", "S2", "S3"], "output_sample_ids": ["S1", "S2", "S3"], "total_reference_bases": 10, "coverage_ge_1x": 0.2, }, "artifacts": artifacts, })) return output def test_wgs_bam_qc_vertical_registration_is_exact() -> None: assert STAGE_DEPENDENCIES["wgs_bam_qc"] == ["wgs_markdup", "wgs_index"] assert "wgs_bam_qc" in EXECUTABLE_STAGES capability = CAPABILITIES["wgs_bam_qc"] assert {item.role for item in capability.consumes} == { "markdup_bam", "markdup_bam_index", "reference_fai" } assert {item.role for item in capability.provides} == { "bam_flagstat", "bam_depth_summary", "bam_qc_summary", "bam_qc_report" } definition = next( item for item in StageToolRegistry.STAGE_DEFINITIONS if item["id"] == "wgs_bam_qc" ) assert definition["script"] == "run_wgs_bam_qc.py" assert definition["conda_env"] == "rnaseq" assert capability.env == "rnaseq" assert definition["required_inputs"] == [ "wgs_markdup_manifest", "wgs_index_manifest" ] assert PlanningAgentV2.REQUIRED_INPUTS["wgs_bam_qc"] == [] def test_wgs_bam_qc_resolves_samtools_from_declared_rnaseq_environment() -> None: resolver = getattr(project_paths, "get_wgs_samtools", None) assert resolver is not None assert resolver("wgs_bam_qc", "").endswith( "/envs/rnaseq/bin/samtools" ) explicit = "/opt/asaw/bin/samtools" assert resolver("wgs_align", explicit) == explicit def test_backend_uses_both_authorities_and_exact_canonical_command( monkeypatch, tmp_path: Path ) -> None: workspace = tmp_path / "workspace" markdup, index = _authorities(workspace) calls = [] def fake_run(command, *_args, **_kwargs): calls.append(command) _publish_result(workspace) return 0 monkeypatch.setattr(pipeline_backend, "run_cmd_logged", fake_run) message, output = pipeline_backend.run_stage( stage="wgs_bam_qc", job=pipeline_backend.JobInfo( "job-1", "", str(workspace), str(workspace / "logs"), "now" ), prefix="Cohort", wgs_markdup_manifest=str(markdup), wgs_index_manifest=str(index), wgs_sample_set_id=SAMPLES, wgs_reference_id=REFERENCE, wgs_analysis_id=MARKDUP_ANALYSIS, wgs_index_analysis_id=INDEX_ANALYSIS, wgs_samtools="/opt/samtools", wgs_coverage_threshold=1, wgs_threads=2, wgs_max_jobs=3, ) assert "completed" in message.lower() assert output == workspace / "final" / "Cohort.wgs_bam_qc.manifest.json" command = calls[0] assert command[:3] == [ pipeline_backend.sys.executable, "-m", "apps.cli.run_wgs_bam_qc" ] assert command[command.index("--markdup-manifest") + 1] == str(markdup) assert command[command.index("--index-manifest") + 1] == str(index) assert command[command.index("--coverage-threshold") + 1] == "1" assert command[command.index("--threads") + 1] == "2" assert command[command.index("--max-jobs") + 1] == "3" def test_api_contract_carries_stage05_authorities_and_identity_pair() -> None: request = ExecuteRequest( stage="wgs_bam_qc", wgsMarkdupManifest="final/Cohort.wgs_markdup.manifest.json", wgsIndexManifest="final/Cohort.wgs_index.manifest.json", wgsSampleSetId=SAMPLES, wgsReferenceId=REFERENCE, wgsAnalysisId=MARKDUP_ANALYSIS, wgsIndexAnalysisId=INDEX_ANALYSIS, wgsCoverageThreshold=1, ) assert request.wgsMarkdupManifest assert request.wgsIndexAnalysisId == INDEX_ANALYSIS def test_planning_status_keeps_completed_bam_qc_after_upstreams_are_deleted( tmp_path: Path, ) -> None: from gfa_core.mas.tools.stage_status import get_stage_status workspace_root = tmp_path / "workspace" workspace = workspace_root / "job-1" markdup, index = _authorities(workspace) _publish_result(workspace) markdup.unlink() index.unlink() strict_status = get_stage_status( "job-1", "Cohort", "wgs_bam_qc", workspace_root, wgs_sample_set_id=SAMPLES, wgs_reference_id=REFERENCE, wgs_analysis_id="wgs-analysis:" + "e" * 64, ) planning_status = get_stage_status( "job-1", "Cohort", "wgs_bam_qc", workspace_root, wgs_sample_set_id=SAMPLES, wgs_reference_id=REFERENCE, wgs_analysis_id="wgs-analysis:" + "e" * 64, validate_wgs_upstream_authority=False, ) assert strict_status.completed is False assert planning_status.completed is True