import asyncio import hashlib import json from pathlib import Path import pytest from apps.web import pipeline_backend as backend from apps.web.vue3 import main as api from gfa_core import gwas_upstream_manifest as upstream from gfa_core.gwas_upstream_manifest import GwasUpstreamError from gfa_core.mas.stage_capabilities import get_stage_capability from gfa_core.mas.tools.stage_identity import StageIdentity, get_stage_status_for_identity GENOTYPE = "genotype:" + "a" * 64 PHENOTYPE = "phenotype:" + "b" * 64 COVARIATE = "covariate:" + "c" * 64 MODEL = "gwas-model:" + "d" * 64 ANALYSIS = "gwas-analysis:" + "e" * 64 TASSEL = "/opt/tassel/run_pipeline.pl" def _sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _artifact(workspace: Path, path: Path, role: str) -> dict: return { "role": role, "artifact": role, "path": str(path.relative_to(workspace)), "full_path": str(path.resolve()), "size": path.stat().st_size, "sha256": _sha(path), } def _producer(workspace: Path, stage: str, role: str, scopes: dict, tag: str = "source") -> tuple[Path, Path]: final = workspace / "final" final.mkdir(parents=True, exist_ok=True) suffix = ".vcf.gz" if role == "sorted_vcf" else ".txt" artifact = final / f"cohort.{stage}.{tag}.{role}{suffix}" artifact.write_bytes(b"authoritative\n") manifest = { "manifest_version": "1.0", "run_id": f"{stage}-run", "stage": stage, "prefix": "cohort", "status": "success", **scopes, "artifacts": {role: _artifact(workspace, artifact, role)}, } if stage != "gwas_fmt": manifest["analysis_scope_id"] = "gwas-analysis:" + "f" * 64 path = final / f"cohort.{stage}.{tag}.manifest.json" path.write_text(json.dumps(manifest), encoding="utf-8") return path, artifact def _upstream(workspace: Path) -> dict: fmt_m, vcf = _producer(workspace, "gwas_fmt", "sorted_vcf", {"genotype_scope_id": GENOTYPE}) trait_m, trait = _producer(workspace, "gwas_trait", "trait_table", {"phenotype_scope_id": PHENOTYPE}) cov_m, cov = _producer(workspace, "gwas_cov", "covariate_table", { "genotype_scope_id": GENOTYPE, "covariate_scope_id": COVARIATE, }) return {"vcf_manifest": fmt_m, "sorted_vcf": vcf, "trait_manifest": trait_m, "trait_table": trait, "covariate_manifest": cov_m, "covariate_table": cov} def _scopes() -> dict: return {"genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE, "covariateScopeId": COVARIATE} def _glm_manifest(workspace: Path, roles=None, *, model=MODEL, analysis=ANALYSIS, tag="identity") -> Path: final = workspace / "final" final.mkdir(parents=True, exist_ok=True) roles = roles or ( "glm_full", "glm_structure_full", "glm_pvalues", "glm_structure_pvalues", "log", "report", "environment", ) artifacts = {} for role in roles: suffix = ".md" if role == "report" else ".txt" if role in {"log", "environment", "glm_full", "glm_structure_full"} else ".tsv" path = final / f"cohort.gwas_glm.{tag}.{role}{suffix}" path.write_text(role + "\n", encoding="utf-8") artifacts[role] = _artifact(workspace, path, role) manifest = { "manifest_version": "1.0", "stage": "gwas_glm", "prefix": "cohort", "status": "success", "genotype_scope_id": GENOTYPE, "phenotype_scope_id": PHENOTYPE, "covariate_scope_id": COVARIATE, "model_scope_id": model, "analysis_scope_id": analysis, "elapsed_sec": 4.26, "artifacts": artifacts, } path = final / f"cohort.gwas_glm.{tag}.manifest.json" path.write_text(json.dumps(manifest), encoding="utf-8") return path def test_glm_resolver_returns_three_exact_manifest_artifacts(tmp_path): workspace = tmp_path / "workspace" expected = _upstream(workspace) resolved = upstream.resolve_gwas_glm_upstream( workspace=workspace, prefix="cohort", genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE, ) assert {key: resolved[key]["path"] for key in ("sorted_vcf", "trait_table", "covariate_table")} == { "sorted_vcf": str(expected["sorted_vcf"]), "trait_table": str(expected["trait_table"]), "covariate_table": str(expected["covariate_table"]), } assert resolved["sorted_vcf"]["source_manifest"] == str(expected["vcf_manifest"]) @pytest.mark.parametrize("failure", ["scope", "hash", "status", "multiple"]) def test_glm_resolver_fails_closed_without_scan_latest_or_override(tmp_path, failure): workspace = tmp_path / "workspace" paths = _upstream(workspace) if failure == "scope": payload = json.loads(paths["covariate_manifest"].read_text()) payload["genotype_scope_id"] = "genotype:" + "9" * 64 paths["covariate_manifest"].write_text(json.dumps(payload)) elif failure == "hash": paths["trait_table"].write_text("changed\n") elif failure == "status": payload = json.loads(paths["trait_manifest"].read_text()); payload["status"] = "failed" paths["trait_manifest"].write_text(json.dumps(payload)) else: _producer(workspace, "gwas_trait", "trait_table", {"phenotype_scope_id": PHENOTYPE}, tag="copy") with pytest.raises(GwasUpstreamError): upstream.resolve_gwas_glm_upstream( workspace=workspace, prefix="cohort", genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE, ) def test_glm_preflight_materializes_exact_paths_and_manifests(monkeypatch, tmp_path): workspace = tmp_path / "workspace/job" paths = _upstream(workspace) monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace") result = api._api_semantic_preflight("job", "gwas_glm", "cohort", _scopes()) assert result["valid"] is True assert result["materialized_params"] == { "gwasSortedVcf": str(paths["sorted_vcf"]), "gwasVcfManifest": str(paths["vcf_manifest"]), "gwasTraitFile": str(paths["trait_table"]), "gwasTraitManifest": str(paths["trait_manifest"]), "gwasCovariateFile": str(paths["covariate_table"]), "gwasCovariateManifest": str(paths["covariate_manifest"]), "tasselPipeline": "run_pipeline.pl", "tasselXmx": "10g", **_scopes(), } def test_glm_preflight_materializes_scopes_from_unique_producer_manifests( monkeypatch, tmp_path ): workspace = tmp_path / "workspace/job" paths = _upstream(workspace) monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace") result = api._api_semantic_preflight("job", "gwas_glm", "cohort", {}) assert result["valid"] is True assert result["materialized_params"] == { "gwasSortedVcf": str(paths["sorted_vcf"]), "gwasVcfManifest": str(paths["vcf_manifest"]), "gwasTraitFile": str(paths["trait_table"]), "gwasTraitManifest": str(paths["trait_manifest"]), "gwasCovariateFile": str(paths["covariate_table"]), "gwasCovariateManifest": str(paths["covariate_manifest"]), "tasselPipeline": "run_pipeline.pl", "tasselXmx": "10g", **_scopes(), } def test_glm_command_is_only_canonical_runner(tmp_path): command = backend._gwas_glm_cli_command( workspace=tmp_path, job_id="job", prefix="cohort", sorted_vcf=tmp_path / "x.vcf.gz", trait=tmp_path / "trait.txt", covariate=tmp_path / "cov.txt", vcf_manifest=tmp_path / "fmt.json", trait_manifest=tmp_path / "trait.json", covariate_manifest=tmp_path / "cov.json", genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE, tassel_pipeline=TASSEL, xmx="20g", log_file=tmp_path / "logs/gwas_glm.log", ) assert command[:3] == [backend.sys.executable, "-m", "apps.cli.run_gwas_tassel_glm"] for flag in ("--vcf", "--trait", "--covariate", "--vcf-manifest", "--trait-manifest", "--covariate-manifest", "--genotype-scope-id", "--phenotype-scope-id", "--covariate-scope-id", "--tassel-pipeline", "--xmx", "--log-file"): assert command.count(flag) == 1 assert not any(token.endswith(".sh") for token in command) def test_backend_glm_accepts_only_fresh_exact_success_manifest(monkeypatch, tmp_path): workspace = tmp_path / "workspace" decisions = { role: {"path": str(path), "source_manifest": str(manifest)} for role, manifest, path in ( ("sorted_vcf", *_producer(workspace, "gwas_fmt", "sorted_vcf", {"genotype_scope_id": GENOTYPE})), ("trait_table", *_producer(workspace, "gwas_trait", "trait_table", {"phenotype_scope_id": PHENOTYPE})), ("covariate_table", *_producer(workspace, "gwas_cov", "covariate_table", { "genotype_scope_id": GENOTYPE, "covariate_scope_id": COVARIATE, })), ) } monkeypatch.setattr(backend, "_resolve_gwas_glm_inputs", lambda **_kwargs: decisions) def fake_run(command, cwd=None): assert command[:3] == [backend.sys.executable, "-m", "apps.cli.run_gwas_tassel_glm"] assert command[command.index("--log-file") + 1].endswith("_gwas_glm.log") _glm_manifest(workspace) return 0 monkeypatch.setattr(backend, "_run_child_owned_log", fake_run) message, manifest = backend.run_stage( stage="gwas_glm", job=backend.JobInfo("job", str(tmp_path / "input"), str(workspace), str(workspace / "logs"), "now"), prefix="cohort", genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE, tassel_pipeline=TASSEL, tassel_xmx="20g", ) assert message.startswith("✅") assert manifest == workspace / "final/cohort.gwas_glm.identity.manifest.json" def test_api_task_rejects_partial_glm_postflight(monkeypatch, tmp_path): import gfa_core.mas.resource_governor as resource_governor workspace_root = tmp_path / "workspace" class NoopLease: def __init__(self, **_kwargs): pass def wait(self, update_state=None): return None def release(self, **_kwargs): return None monkeypatch.setattr(resource_governor, "ResourceLease", NoopLease) monkeypatch.setattr(resource_governor, "get_task", lambda _task_id: {}) monkeypatch.setattr(resource_governor, "snapshot", lambda **_kwargs: {"usage": {}}) monkeypatch.setattr(api, "WORKSPACE_ROOT", workspace_root) monkeypatch.setattr(api, "INPUT_ROOT", tmp_path / "input") monkeypatch.setattr(api, "_attach_single_stage_status", lambda _task: None) monkeypatch.setattr(api, "_attach_annotation_progress", lambda _task: None) monkeypatch.setattr(api.time, "sleep", lambda _seconds: None) monkeypatch.setattr(backend, "run_stage", lambda **_kwargs: ( "✅ done", _glm_manifest(workspace_root / "job", roles=( "glm_full", "glm_structure_full", "glm_pvalues", "log", "report", "environment", )), )) api.tasks["partial-glm"] = { "id": "partial-glm", "stage": "gwas_glm", "job_id": "job", "status": "pending", "message": "", "created_at": "2026-07-21T00:00:00", } api.run_stage_sync("partial-glm", "gwas_glm", "job", {"prefix": "cohort", **_scopes()}) task = api.tasks["partial-glm"] assert task["status"] == "failed" assert task["result"]["error"] == "invalid_gwas_glm_outputs" def test_execute_request_capability_tool_and_skill_are_closed(): from gfa_core.mas.tools.stage_tools import StageToolRegistry request = api.ExecuteRequest(stage="gwas_glm", gwasVcfManifest="v", gwasTraitManifest="t", gwasCovariateManifest="c", **_scopes()) assert request.gwasCovariateManifest == "c" capability = get_stage_capability("gwas_glm") assert [item.role for item in capability.consumes] == ["sorted_vcf", "trait_table", "covariate_table"] expected = ["glm_full", "glm_structure_full", "glm_pvalues", "glm_structure_pvalues", "manifest", "report", "environment"] assert [item.role for item in capability.provides] == expected tool = StageToolRegistry().get("gwas_glm") assert tool.required_inputs == ["sorted_vcf", "trait_table", "covariate_table"] assert tool.outputs == expected config = (Path(api.PROJECT_ROOT) / "skills/gwas-tassel-glm/config.yaml").read_text() skill = (Path(api.PROJECT_ROOT) / "skills/gwas-tassel-glm/SKILL.md").read_text() for token in ("apps.cli.run_gwas_tassel_glm", *expected, "sorted_vcf", "trait_table", "covariate_table"): assert token in config and token in skill def test_glm_status_and_results_require_exact_five_scopes(monkeypatch, tmp_path): workspace = tmp_path / "workspace/job" _glm_manifest(workspace) monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace") identity = StageIdentity(stage="gwas_glm", prefix="cohort", genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE, model_scope_id=MODEL, analysis_scope_id=ANALYSIS) assert get_stage_status_for_identity("job", identity, workspace_root=tmp_path / "workspace", strict_identity=True).completed is True exact = asyncio.run(api.get_results(job_id="job", stage="gwas_glm", genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE, covariate_scope_id=COVARIATE, model_scope_id=MODEL, analysis_scope_id=ANALYSIS)) assert exact["gwas_glm"]["images"] == [] assert {item["role"] for item in exact["gwas_glm"]["files"]} == { "glm_full", "glm_structure_full", "glm_pvalues", "glm_structure_pvalues", "log", "report", "environment", "manifest", } assert exact["gwas_glm"]["elapsed_time"] == 4.3 wrong = asyncio.run(api.get_results(job_id="job", stage="gwas_glm", genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE, covariate_scope_id="covariate:" + "9" * 64, model_scope_id=MODEL, analysis_scope_id=ANALYSIS)) assert wrong["gwas_glm"]["files"] == [] def test_glm_partial_identity_and_partial_outputs_fail_closed(monkeypatch, tmp_path): workspace = tmp_path / "workspace/job" _glm_manifest(workspace, roles=("glm_full", "glm_pvalues", "log", "report", "environment")) partial = StageIdentity(stage="gwas_glm", prefix="cohort", genotype_scope_id=GENOTYPE) report = get_stage_status_for_identity("job", partial, workspace_root=tmp_path / "workspace", strict_identity=True) assert report.completed is False and "all five" in report.errors[0] def test_glm_frontend_declares_typed_handoff_five_scope_queries_and_no_dialog(): frontend = (Path(api.PROJECT_ROOT) / "apps/web/vue3/index.html").read_text(encoding="utf-8") assert "handoffGwasGlmProducerIdentity" in frontend assert "stageExecutionStatus.value.gwas_glm !== 'pending'" in frontend for producer in ("gwas_fmt", "gwas_trait", "gwas_cov"): assert producer in frontend assert "gwasGlmExactIdentity" in frontend and "rememberGwasGlmIdentity" in frontend assert "sourceModelScopeId" not in frontend[frontend.index("const gwasGlmExactIdentity ="):frontend.index("const checkManifestExists =", frontend.index("const gwasGlmExactIdentity ="))] for role in ("glm_full", "glm_structure_full", "glm_pvalues", "glm_structure_pvalues", "manifest", "report", "environment"): assert role in frontend assert "showGwasGlm" not in frontend and "gwasGlmDialog" not in frontend def test_glm_natural_language_is_explicit_and_bare_gwas_is_not_alias(): from gfa_core.mas.utils.intent_parser import parse_intent_from_query explicit = parse_intent_from_query("使用 TASSEL 执行固定效应 GLM 分析") bare = parse_intent_from_query("执行 GWAS 分析") assert explicit["primary_stage"] == "gwas_glm" assert bare["primary_stage"] == "gwas_glm" assert len(bare["stages"]) > 1 and "gwas_glm" in bare["stages"] def test_execute_glm_preserves_nondefault_tassel_heap_to_worker_submission(monkeypatch, tmp_path): workspace = tmp_path / "workspace/job" _upstream(workspace) class DummyExecutor: def __init__(self): self.calls = [] def submit(self, *args): self.calls.append(args) dummy = DummyExecutor() monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path) monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace") monkeypatch.setattr(api, "INPUT_ROOT", tmp_path / "input") monkeypatch.setattr(api, "CELERY_AVAILABLE", False) monkeypatch.setattr(api, "executor", dummy) result = asyncio.run(api.execute_stage(api.ExecuteRequest( stage="gwas_glm", job_id="job", prefix="cohort", tasselXmx="20g", **_scopes(), ))) assert result["async"] is True and len(dummy.calls) == 1 submitted_params = dummy.calls[0][4] assert submitted_params["tasselXmx"] == "20g" def test_background_completion_materializes_all_three_glm_scopes(tmp_path): frontend = (Path(api.PROJECT_ROOT) / "apps/web/vue3/index.html").read_text(encoding="utf-8") handoff_start = frontend.index("const handoffGwasGlmProducerIdentity =") apply_start = frontend.index("const applyGwasExecutionParams =", handoff_start) handler_start = frontend.index("const handleStageCompleted =") handler_end = frontend.index("const resolveSubfamilyReview =", handler_start) stages = [{"id": "gwas_glm", "params": {}}] results = [ {"stage": "gwas_fmt", "scope": {"genotype_scope_id": GENOTYPE}}, {"stage": "gwas_trait", "scope": {"phenotype_scope_id": PHENOTYPE}}, {"stage": "gwas_cov", "scope": {"genotype_scope_id": GENOTYPE, "covariate_scope_id": COVARIATE}}, ] script = tmp_path / "background-glm-handoff.cjs" script.write_text( "const batchExecuteStages={value:" + json.dumps(stages) + "};" "const stageExecutionStatus={value:{gwas_glm:'pending'}};" "const getStageId=s=>s.id;const executionPrefix=()=> 'cohort';" "const setResultStageFilter=()=>{};const refreshResults=async()=>{};" "const openCompletedSubfamilyReview=async()=>{};" "const methodologyIdentifyMemberReviewRequired=()=>false;" "const subfamilyPerSpeciesMode=()=>false;const maybeShowIdentifyDomainReview=async()=>false;" "const rememberGwasGlmIdentity=()=>false;" + frontend[handoff_start:apply_start] + frontend[handler_start:handler_end] + "const rows=" + json.dumps(results) + ";" + "Promise.all(rows.map(row=>handleStageCompleted(row.stage,{result:{stage_status:{completed:true,stage:row.stage,prefix:'cohort',...row.scope},stage_identity:{stage:row.stage,prefix:'cohort',...row.scope}}},{})))" + ".then(()=>process.stdout.write(JSON.stringify(batchExecuteStages.value[0].params)))" + ".catch(e=>{process.stderr.write(String(e));process.exit(1)});", encoding="utf-8", ) completed = __import__("subprocess").run(["node", str(script)], text=True, capture_output=True, check=False) assert completed.returncode == 0, completed.stderr assert json.loads(completed.stdout) == _scopes() def test_frontend_limits_each_scheduler_wave_to_one_gwas_model_heavy(tmp_path): frontend = (Path(api.PROJECT_ROOT) / "apps/web/vue3/index.html").read_text(encoding="utf-8") assert "const limitGwasModelHeavyWave =" in frontend assert "limitGwasModelHeavyWave(executableCandidates, stageExecutionStatus.value)" in frontend assert "limitGwasModelHeavyWave(executableCandidates, localStatus)" in frontend start = frontend.index("const gwasModelHeavyStages =") end = frontend.index("const confirmParams =", start) script = tmp_path / "gwas-heavy-wave.cjs" script.write_text( frontend[start:end] + "const candidates=['gwas_mlm','gwas_emmax','gwas_glm','gwas_plot'];" + "const pending=limitGwasModelHeavyWave(candidates,{});" + "const running=limitGwasModelHeavyWave(candidates,{gwas_emmax:'running'});" + "process.stdout.write(JSON.stringify({pending,running}));", encoding="utf-8", ) completed = __import__("subprocess").run(["node", str(script)], text=True, capture_output=True, check=False) assert completed.returncode == 0, completed.stderr assert json.loads(completed.stdout) == { "pending": ["gwas_mlm", "gwas_plot"], "running": ["gwas_plot"], } def test_gwas_model_heavy_lease_is_cross_job_exclusive_and_released(monkeypatch, tmp_path): from gfa_core.mas import resource_governor from gfa_core.mas.stage_catalog import get_stage_resource_profile monkeypatch.setenv("BBAV5_RESOURCE_STATE_DIR", str(tmp_path / "governor")) for stage in ("gwas_mlm", "gwas_emmax", "gwas_glm"): assert get_stage_resource_profile(stage)["concurrency_group"] == "gwas_model_heavy" first = resource_governor.try_acquire( task_id="one", stage="gwas_mlm", job_id="job-a", prefix="cohort", cpu=1, concurrency_group="gwas_model_heavy", ) blocked = resource_governor.try_acquire( task_id="two", stage="gwas_glm", job_id="job-b", prefix="cohort", cpu=1, concurrency_group="gwas_model_heavy", ) assert first["granted"] is True assert blocked["granted"] is False and "gwas_model_heavy" in blocked["reason"] resource_governor.release("one", status="failed") acquired = resource_governor.try_acquire( task_id="two", stage="gwas_glm", job_id="job-b", prefix="cohort", cpu=1, concurrency_group="gwas_model_heavy", ) assert acquired["granted"] is True def test_both_threadpool_and_celery_pass_the_catalog_concurrency_group(): main_source = (Path(api.PROJECT_ROOT) / "apps/web/vue3/main.py").read_text(encoding="utf-8") celery_source = (Path(api.PROJECT_ROOT) / "tasks.py").read_text(encoding="utf-8") assert 'concurrency_group=str(profile.get("concurrency_group") or "")' in main_source assert 'concurrency_group=str(resource_profile.get("concurrency_group") or "")' in celery_source