import asyncio import hashlib import json from pathlib import Path import pytest from gfa_core.mas.agents.planning_v2 import PlanningAgentV2, StageStatus from gfa_core.mas.artifact_resolver import ArtifactResolver from gfa_core.mas.context_index import build_project_context from gfa_core.mas.stage_capabilities import get_stage_capability import gfa_core.wgs_upstream_manifest as wgs_manifest WGS_SAMPLE_SET_ID = "wgs-samples:" + "a" * 64 WGS_REFERENCE_ID = "wgs-reference:" + "b" * 64 WGS_QC_ANALYSIS_ID = "wgs-analysis:" + "c" * 64 WGS_INDEX_ANALYSIS_ID = "wgs-analysis:" + "d" * 64 WGS_RUN_ID = "1" * 32 def _wgs_artifact( workspace: Path, bundle: Path, name: str, role: str, *, sample_id: str = "", component: str = "", ) -> dict: path = bundle / name path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(f"{role}:{sample_id or component or 'reference'}".encode()) artifact = { "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_id: artifact["sample_id"] = sample_id if component: artifact["component"] = component return artifact def _write_wgs_align_authorities( workspace: Path, prefix: str = "Cohort" ) -> dict[str, Path]: final_dir = workspace / "final" qc_bundle = final_dir / ".wgs_qc_runs" / prefix / WGS_RUN_ID index_bundle = final_dir / ".wgs_index_runs" / prefix / WGS_RUN_ID qc_artifacts = {} for sample_id in ("S1", "S2"): qc_artifacts[f"clean_reads_r1.{sample_id}"] = _wgs_artifact( workspace, qc_bundle, f"{prefix}.wgs_qc.{sample_id}.R1.fastq.gz", "clean_reads_r1", sample_id=sample_id, ) qc_artifacts[f"clean_reads_r2.{sample_id}"] = _wgs_artifact( workspace, qc_bundle, f"{prefix}.wgs_qc.{sample_id}.R2.fastq.gz", "clean_reads_r2", sample_id=sample_id, ) qc_manifest = { "manifest_version": "1.0", "stage": "wgs_qc", "status": "success", "prefix": prefix, "job_id": "job-1", "run_id": WGS_RUN_ID, "workspace": str(workspace.resolve()), "wgs_sample_set_id": WGS_SAMPLE_SET_ID, "wgs_analysis_id": WGS_QC_ANALYSIS_ID, "artifacts": qc_artifacts, } qc_path = final_dir / f"{prefix}.wgs_qc.manifest.json" qc_path.parent.mkdir(parents=True, exist_ok=True) qc_path.write_text(json.dumps(qc_manifest), encoding="utf-8") index_artifacts = { "reference_fasta": _wgs_artifact( workspace, index_bundle, "reference.fasta", "reference_fasta" ), "reference_fai": _wgs_artifact( workspace, index_bundle, "reference.fasta.fai", "reference_fai" ), "reference_dict": _wgs_artifact( workspace, index_bundle, "reference.dict", "reference_dict" ), } for component in (".0123", ".amb", ".ann", ".bwt.2bit.64", ".pac"): index_artifacts[f"bwa_mem2_index_component.{component}"] = _wgs_artifact( workspace, index_bundle, f"reference.fasta{component}", "bwa_mem2_index_component", component=component, ) index_manifest = { "manifest_version": "1.0", "stage": "wgs_index", "status": "success", "prefix": prefix, "job_id": "job-1", "run_id": WGS_RUN_ID, "workspace": str(workspace.resolve()), "wgs_reference_id": WGS_REFERENCE_ID, "wgs_analysis_id": WGS_INDEX_ANALYSIS_ID, "artifacts": index_artifacts, } index_path = final_dir / f"{prefix}.wgs_index.manifest.json" index_path.write_text(json.dumps(index_manifest), encoding="utf-8") return { "qc_manifest": qc_path, "index_manifest": index_path, "missing_component": Path( index_artifacts["bwa_mem2_index_component..ann"]["full_path"] ), } def test_wgs_qc_resolves_only_canonical_sample_yaml(tmp_path: Path) -> None: input_dir = tmp_path / "input" / "job-1" input_dir.mkdir(parents=True) canonical = input_dir / "wgs_samples.yaml" canonical.write_text("samples: {}\n", encoding="utf-8") (input_dir / "samples.yaml").write_text("samples: {}\n", encoding="utf-8") context = build_project_context( job_id="job-1", prefix="At", project_root=tmp_path, ) capability = get_stage_capability("wgs_qc") assert capability is not None decision = ArtifactResolver(context).resolve_need( "wgs_qc", capability.consumes[0] ) assert decision.resolved assert decision.selected is not None assert Path(decision.selected.path) == canonical def test_wgs_index_preflight_materializes_resolved_reference( tmp_path: Path, monkeypatch ) -> None: from apps.web.vue3 import main as web_main input_dir = tmp_path / "input" / "job-1" input_dir.mkdir(parents=True) reference = input_dir / "reference.9.1m.fasta" reference.write_text(">chr1\nACGT\n", encoding="utf-8") monkeypatch.setattr(web_main, "PROJECT_ROOT", tmp_path) result = web_main._api_semantic_preflight( "job-1", "wgs_index", "Cohort", {"wgsReferenceFasta": ""}, ) assert result["valid"] assert result["materialized_params"]["wgsReferenceFasta"] == str(reference) def test_wgs_index_preflight_does_not_override_explicit_reference( tmp_path: Path, monkeypatch ) -> None: from apps.web.vue3 import main as web_main input_dir = tmp_path / "input" / "job-1" input_dir.mkdir(parents=True) (input_dir / "auto-reference.fasta").write_text( ">chr1\nACGT\n", encoding="utf-8" ) explicit = tmp_path / "explicit-reference.fasta" explicit.write_text(">chr1\nTGCA\n", encoding="utf-8") monkeypatch.setattr(web_main, "PROJECT_ROOT", tmp_path) result = web_main._api_semantic_preflight( "job-1", "wgs_index", "Cohort", {"wgsReferenceFasta": str(explicit)}, ) assert result["valid"] assert "wgsReferenceFasta" not in result.get("materialized_params", {}) def test_wgs_align_planning_accepts_complete_authoritative_inputs( tmp_path: Path, ) -> None: workspace = tmp_path / "workspace" / "job-1" _write_wgs_align_authorities(workspace) input_dir = tmp_path / "input" / "job-1" input_dir.mkdir(parents=True) context = build_project_context( job_id="job-1", prefix="Cohort", project_root=tmp_path ) issues = PlanningAgentV2(None, project_root=str(tmp_path))._check_input_files( "wgs_align", input_dir, workspace, "Cohort", context_index=context ) assert not [issue for issue in issues if issue["severity"] == "error"] assert not [ issue for issue in issues if "缺少 wgs_align 所需输入" in issue["message"] and any( role in issue["message"] for role in ( "clean_reads_r1", "clean_reads_r2", "reference_fasta", "reference_fai", "reference_dict", "bwa_mem2_index_component", ) ) ] def test_wgs_align_planning_reports_one_invalid_authority_without_fallback( tmp_path: Path, ) -> None: workspace = tmp_path / "workspace" / "job-1" authorities = _write_wgs_align_authorities(workspace) authorities["missing_component"].unlink() input_dir = tmp_path / "input" / "job-1" input_dir.mkdir(parents=True) context = build_project_context( job_id="job-1", prefix="Cohort", project_root=tmp_path ) issues = PlanningAgentV2(None, project_root=str(tmp_path))._check_input_files( "wgs_align", input_dir, workspace, "Cohort", context_index=context ) errors = [issue for issue in issues if issue["severity"] == "error"] assert errors == [ { "severity": "error", "category": "input", "code": "invalid_wgs_manifest_identity", "message": ( "artifact bwa_mem2_index_component..ann file is missing or not regular" ), "suggestion": "修复 wgs_qc 或 wgs_index 权威 Manifest 后重试", "auto_fixable": False, } ] assert not [ issue for issue in issues if "缺少 wgs_align 所需输入" in issue["message"] ] def test_wgs_align_planning_allows_planned_upstream_rebuild( tmp_path: Path, monkeypatch ) -> None: agent = PlanningAgentV2(None, project_root=str(tmp_path)) def fail_if_read(*_args, **_kwargs): raise wgs_manifest.WgsUpstreamManifestError( "wgs_qc authority is invalid: upstream Manifest is unreadable" ) monkeypatch.setattr( "gfa_core.mas.agents.planning_v2.materialize_wgs_identity_params", fail_if_read, ) issues = agent._check_input_files( "wgs_align", tmp_path / "input", tmp_path / "workspace" / "job-1", "Cohort", planned_stages=["wgs_qc", "wgs_index", "wgs_align"], ) assert [issue["severity"] for issue in issues] == ["warning"] assert issues[0]["code"] == "wgs_upstream_will_be_rebuilt" def test_unique_wgs_prefix_ignores_stale_request_hint(tmp_path: Path) -> None: workspace = tmp_path / "workspace" / "job-1" _write_wgs_align_authorities(workspace, "Os") assert wgs_manifest.resolve_unique_wgs_prefix( stage="wgs_align", job_id="job-1", params={}, workspace=workspace, ) == "Os" def test_unique_wgs_prefix_rejects_multiple_complete_prefixes(tmp_path: Path) -> None: workspace = tmp_path / "workspace" / "job-1" _write_wgs_align_authorities(workspace, "At") _write_wgs_align_authorities(workspace, "Os") with pytest.raises( wgs_manifest.WgsUpstreamManifestError, match="multiple complete WGS authority prefixes", ): wgs_manifest.resolve_unique_wgs_prefix( stage="wgs_align", job_id="job-1", params={}, workspace=workspace, ) def test_planning_uses_manifest_prefix_for_wgs_consumer( tmp_path: Path, monkeypatch ) -> None: workspace = tmp_path / "workspace" / "job-1" _write_wgs_align_authorities(workspace, "Os") input_dir = tmp_path / "input" / "job-1" input_dir.mkdir(parents=True) (input_dir / "wgs_samples.yaml").write_text("samples: []\n", encoding="utf-8") (input_dir / "reference.fa").write_text(">chr1\nACGT\n", encoding="utf-8") agent = PlanningAgentV2(None, project_root=str(tmp_path)) checked_prefixes: list[str] = [] monkeypatch.setattr( agent, "_check_stage_status_preflight", lambda stage, prefix, checked_workspace, stage_params=None: ( checked_prefixes.append(prefix) or StageStatus(stage=stage, status="not_found") ), ) intent = {"stages": ["wgs_align"], "prefix": "At", "params": {}} result = asyncio.run( agent._run_preflight_validation(intent, "job-1", "At", str(workspace)) ) assert result["valid"] assert intent["prefix"] == "Os" assert checked_prefixes and set(checked_prefixes) == {"Os"} def test_execute_prefix_uses_same_manifest_resolution( tmp_path: Path, monkeypatch ) -> None: from apps.web.vue3 import main as web_main workspace_root = tmp_path / "workspace" _write_wgs_align_authorities(workspace_root / "job-1", "Os") monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root) assert web_main._resolve_wgs_execute_prefix( job_id="job-1", stage="wgs_align", prefix="At", params={}, ) == "Os"