import asyncio
import hashlib
import inspect
import json
import subprocess
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.mas.stage_capabilities import get_stage_capability
from gfa_core.mas.tools.stage_identity import StageIdentity, get_stage_status_for_identity
from gfa_core.project_paths import get_conda_env_bin, get_conda_env_rscript, get_configured_path


GENOTYPE = "genotype:" + "a" * 64
PHENOTYPE = "phenotype:" + "b" * 64
MODEL = "gwas-model:" + "c" * 64
ANALYSIS = "gwas-analysis:" + "d" * 64
PLINK = str(get_conda_env_bin("ASAW") / "plink")
EMMAX = "/home/xzq/Auto_GWAS/gwas_work/emmax-intel64"
OFFICIAL_EMMAX = get_configured_path("GWAS_EMMAX_BIN", "emmax")
RSCRIPT = str(get_conda_env_rscript("ASAW"))


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_manifests(workspace: Path, prefix: str = "cohort") -> tuple[dict, dict]:
    final = workspace / "final"
    final.mkdir(parents=True, exist_ok=True)
    vcf = final / f"{prefix}.gwas_fmt.sorted.vcf.gz"
    trait = final / f"{prefix}.gwas_trait.trait.txt"
    vcf.write_bytes(b"vcf")
    trait.write_text("<Trait>\theading_days\nS1\t1\nS2\t2\n", encoding="utf-8")
    fmt = {
        "manifest_version": "1.0", "run_id": "fmt-run", "stage": "gwas_fmt",
        "prefix": prefix, "status": "success", "genotype_scope_id": GENOTYPE,
        "artifacts": {"sorted_vcf": _artifact(workspace, vcf, "sorted_vcf")},
    }
    trait_manifest = {
        "manifest_version": "1.0", "run_id": "trait-run", "stage": "gwas_trait",
        "prefix": prefix, "status": "success", "phenotype_scope_id": PHENOTYPE,
        "analysis_scope_id": "gwas-analysis:" + "e" * 64,
        "artifacts": {"trait_table": _artifact(workspace, trait, "trait_table")},
    }
    fmt_path = final / f"{prefix}.gwas_fmt.manifest.json"
    trait_path = final / f"{prefix}.gwas_trait.identity.manifest.json"
    fmt_path.write_text(json.dumps(fmt), encoding="utf-8")
    trait_path.write_text(json.dumps(trait_manifest), encoding="utf-8")
    return {
        "path": fmt_path, "artifact": vcf, "manifest": fmt,
    }, {"path": trait_path, "artifact": trait, "manifest": trait_manifest}


def _emmax_manifest(workspace: Path, prefix: str = "cohort", roles=None) -> Path:
    final = workspace / "final"
    final.mkdir(parents=True, exist_ok=True)
    roles = roles or ("emmax_full", "emmax_pvalues", "report", "environment")
    artifacts = {}
    for role in roles:
        suffix = ".md" if role == "report" else ".txt" if role == "environment" else ".tsv"
        path = final / f"{prefix}.gwas_emmax.{role}{suffix}"
        path.write_text(f"{role}\n", encoding="utf-8")
        artifacts[role] = _artifact(workspace, path, role)
    manifest = {
        "manifest_version": "1.0", "stage": "gwas_emmax", "prefix": prefix,
        "status": "success", "genotype_scope_id": GENOTYPE,
        "phenotype_scope_id": PHENOTYPE, "model_scope_id": MODEL,
        "analysis_scope_id": ANALYSIS, "elapsed_sec": 12.34, "artifacts": artifacts,
    }
    path = final / f"{prefix}.gwas_emmax.identity.manifest.json"
    path.write_text(json.dumps(manifest), encoding="utf-8")
    return path


def _scopes() -> dict:
    return {"genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE}


def test_emmax_unset_default_is_resolved_server_side_across_entry_points(monkeypatch, tmp_path):
    workspace = tmp_path / "workspace" / "job"
    _producer_manifests(workspace)
    monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")

    request = api.ExecuteRequest(stage="gwas_emmax")
    preflight = api._api_semantic_preflight("job", "gwas_emmax", "cohort", _scopes())

    assert request.gwasEmmax == ""
    assert preflight["materialized_params"]["gwasEmmax"] == OFFICIAL_EMMAX
    assert inspect.signature(backend.run_stage).parameters["gwas_emmax"].default == OFFICIAL_EMMAX
    assert "default=GWAS_EMMAX_BIN" in inspect.getsource(
        __import__("gfa_core.mas.tools.stage_tools", fromlist=["StageToolExecutor"]).StageToolExecutor.execute
    )


def test_emmax_semantic_preflight_materializes_two_exact_producer_manifests(monkeypatch, tmp_path):
    workspace = tmp_path / "workspace" / "job"
    fmt, trait = _producer_manifests(workspace)
    monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")

    result = api._api_semantic_preflight("job", "gwas_emmax", "cohort", _scopes())

    assert result["valid"] is True
    assert result["materialized_params"] == {
        "gwasSortedVcf": str(fmt["artifact"]), "gwasTraitFile": str(trait["artifact"]),
        "genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE,
        "gwasPlink": PLINK, "gwasEmmax": OFFICIAL_EMMAX, "gwasRscript": RSCRIPT,
        "gwasRequestedPcs": 3,
    }
    assert result["decisions"]["sorted_vcf"]["source_manifest"] == str(fmt["path"])
    assert result["decisions"]["trait_table"]["source_manifest"] == str(trait["path"])


def test_emmax_semantic_preflight_materializes_scopes_from_unique_producer_manifests(
    monkeypatch, tmp_path
):
    workspace = tmp_path / "workspace" / "job"
    fmt, trait = _producer_manifests(workspace)
    monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")

    result = api._api_semantic_preflight("job", "gwas_emmax", "cohort", {})

    assert result["valid"] is True
    assert result["materialized_params"] == {
        "gwasSortedVcf": str(fmt["artifact"]),
        "gwasTraitFile": str(trait["artifact"]),
        "genotypeScopeId": GENOTYPE,
        "phenotypeScopeId": PHENOTYPE,
        "gwasPlink": PLINK,
        "gwasEmmax": OFFICIAL_EMMAX,
        "gwasRscript": RSCRIPT,
        "gwasRequestedPcs": 3,
    }


@pytest.mark.parametrize("requested,valid", [(0, True), (-1, False), (True, False)])
def test_emmax_preflight_requested_pcs_matches_cli_nonnegative_contract(
    monkeypatch, tmp_path, requested, valid,
):
    _producer_manifests(tmp_path / "workspace" / "job")
    monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
    result = api._api_semantic_preflight(
        "job", "gwas_emmax", "cohort", {**_scopes(), "gwasRequestedPcs": requested},
    )
    assert result["valid"] is valid
    if valid:
        assert result["materialized_params"]["gwasRequestedPcs"] == requested


@pytest.mark.parametrize("case", [
    "missing_scope", "wrong_scope", "failed", "bad_hash", "bad_size",
    "missing_manifest", "caller_vcf", "caller_trait", "bad_version",
    "missing_run_id", "bad_trait_analysis",
])
def test_emmax_semantic_preflight_fails_closed(monkeypatch, tmp_path, case):
    workspace = tmp_path / "workspace" / "job"
    fmt, trait = _producer_manifests(workspace)
    params = _scopes()
    target = fmt if case in {"failed", "bad_hash", "bad_size", "bad_version", "missing_run_id"} else trait
    manifest = target["manifest"]
    if case == "missing_scope": params["phenotypeScopeId"] = ""
    elif case == "wrong_scope": params["phenotypeScopeId"] = "phenotype:" + "9" * 64
    elif case == "failed": manifest["status"] = "failed"
    elif case == "bad_hash": next(iter(manifest["artifacts"].values()))["sha256"] = "0" * 64
    elif case == "bad_size": next(iter(manifest["artifacts"].values()))["size"] += 1
    elif case == "missing_manifest": trait["path"].unlink()
    elif case == "caller_vcf": params["gwasSortedVcf"] = str(trait["artifact"])
    elif case == "caller_trait": params["gwasTraitFile"] = str(fmt["artifact"])
    elif case == "bad_version": manifest["manifest_version"] = "0.9"
    elif case == "missing_run_id": manifest.pop("run_id")
    elif case == "bad_trait_analysis": manifest["analysis_scope_id"] = ""
    if case in {"failed", "bad_hash", "bad_size", "bad_version", "missing_run_id", "bad_trait_analysis"}:
        target["path"].write_text(json.dumps(manifest), encoding="utf-8")
    monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")

    result = api._api_semantic_preflight("job", "gwas_emmax", "cohort", params)

    assert result["valid"] is False
    assert result["issues"][0]["code"] == "invalid_gwas_emmax_upstream"


def test_emmax_command_is_only_canonical_runner_and_preserves_explicit_values(tmp_path):
    command = backend._gwas_emmax_cli_command(
        workspace=tmp_path, job_id="job", prefix="cohort",
        sorted_vcf=tmp_path / "in.vcf.gz", trait=tmp_path / "trait.txt",
        vcf_manifest=tmp_path / "fmt.json", trait_manifest=tmp_path / "trait.json",
        genotype_scope_id=GENOTYPE, phenotype_scope_id=PHENOTYPE,
        plink=PLINK, emmax=EMMAX, rscript=RSCRIPT, requested_pcs=3,
        threads=6, log_file=tmp_path / "logs/emmax.log",
    )
    assert command[:3] == [backend.sys.executable, "-m", "apps.cli.run_gwas_emmax"]
    expected = {
        "--vcf": str(tmp_path / "in.vcf.gz"), "--trait": str(tmp_path / "trait.txt"),
        "--vcf-manifest": str(tmp_path / "fmt.json"),
        "--trait-manifest": str(tmp_path / "trait.json"),
        "--genotype-scope-id": GENOTYPE, "--phenotype-scope-id": PHENOTYPE,
        "--plink": PLINK, "--emmax": EMMAX, "--rscript": RSCRIPT,
        "--requested-pcs": "3", "--threads": "6",
        "--log-file": str(tmp_path / "logs/emmax.log"),
    }
    for flag, value in expected.items():
        assert command.count(flag) == 1
        assert command[command.index(flag) + 1] == value
    assert not any(token.endswith(".sh") or token.endswith(".R") for token in command)


def test_execute_request_and_frontend_dag_forward_emmax_contract(tmp_path):
    request = api.ExecuteRequest(
        stage="gwas_emmax", gwasPlink=PLINK, gwasEmmax=EMMAX,
        gwasRscript=RSCRIPT, gwasRequestedPcs=2, **_scopes(),
    )
    assert (request.gwasPlink, request.gwasEmmax, request.gwasRequestedPcs) == (PLINK, EMMAX, 2)
    frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    start = frontend.index("const applyGwasExecutionParams =")
    execute = frontend.index("const executeSingleStageWithParams =", start)
    end = frontend.index("// 执行单个阶段（内部函数）", execute)
    stage_params = {**_scopes(), "gwasPlink": PLINK, "gwasEmmax": EMMAX,
                    "gwasRscript": RSCRIPT, "gwasRequestedPcs": 2}
    script = tmp_path / "emmax-request.cjs"
    script.write_text(
        "let captured=null;const batchExecuteParams={value:{plan:{params:{}}}};"
        "const batchExecuteStages={value:[{id:'gwas_emmax',params:" + json.dumps(stage_params) + "}]};"
        "const confirmParams=new Proxy({cpu:6,comboPanels:[]},{get:(o,k)=>o[k]});"
        "const sessionId={value:'s'};const jobInfo={job_id:'job'};const getStageId=s=>s.id;"
        "const getStageAction=s=>s.action;const usesRnaseqSampleParallel=()=>false;"
        "const stageCpuForExecution=()=>6;const confirmCpuForStage=()=>6;"
        "const executionPrefix=()=> 'cohort';const applyGeneFamilyContextToRequest=()=>{};"
        "const applyRnaseqExecutionParams=()=>{};const applyWgsExecutionParams=()=>{};const submitExecuteRequestWithAutoAnnotation=async b=>{captured=b;return {success:true,async:false};};"
        "const createStructuredExecuteError=x=>new Error(JSON.stringify(x));const runInteractiveSubfamilyConfirm=async()=>({});"
        "const syncIdentifyExecutionSpecies=()=>{};const addAgentMessage=()=>{};const getStageLabel=x=>x;"
        "const activePipelineTaskIds={value:[]};const pollTaskUntilComplete=async()=>({});const console={log:()=>{}};"
        + frontend[start:execute] + frontend[execute:end]
        + "executeSingleStageWithParams('gwas_emmax').then(()=>process.stdout.write(JSON.stringify(captured))).catch(e=>{process.stderr.write(String(e));process.exit(1);});",
        encoding="utf-8",
    )
    completed = subprocess.run(["node", str(script)], text=True, capture_output=True, check=False)
    assert completed.returncode == 0, completed.stderr
    body = json.loads(completed.stdout)
    for key, value in stage_params.items():
        assert body[key] == value


def _run_frontend_emmax_handoff(tmp_path: Path, producer_results: list[tuple[str, dict]]) -> dict:
    frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    handoff_start = frontend.index("const handoffGwasEmmaxProducerIdentity =")
    apply_start = frontend.index("const applyGwasExecutionParams =", handoff_start)
    execute_start = frontend.index("const executeSingleStageWithParams =", apply_start)
    stages = [
        {"id": "gwas_fmt", "params": {}},
        {"id": "gwas_trait", "params": {}},
        {"id": "gwas_emmax", "params": {}},
    ]
    script = tmp_path / "emmax-handoff.cjs"
    script.write_text(
        "const batchExecuteStages={value:" + json.dumps(stages) + "};"
        "const batchExecuteParams={value:{plan:{params:{}}}};"
        "const stageExecutionStatus={value:{gwas_fmt:'running',gwas_trait:'pending',gwas_emmax:'pending'}};"
        "const confirmParams={};const getStageId=s=>s.id;const executionPrefix=()=> 'cohort';"
        + frontend[handoff_start:apply_start]
        + frontend[apply_start:execute_start]
        + "const producerResults=" + json.dumps(producer_results) + ";"
        + "for(const [stage,result] of producerResults) handoffGwasEmmaxProducerIdentity(stage,result);"
        + "const body={};const emmax=batchExecuteStages.value.find(s=>s.id==='gwas_emmax');"
        + "applyGwasExecutionParams(body,'gwas_emmax',emmax.params);"
        + "process.stdout.write(JSON.stringify({body,params:emmax.params}));",
        encoding="utf-8",
    )
    completed = subprocess.run(["node", str(script)], text=True, capture_output=True, check=False)
    assert completed.returncode == 0, completed.stderr
    return json.loads(completed.stdout)


def test_frontend_runtime_handoff_feeds_exact_producer_scopes_to_emmax_preflight(
    monkeypatch, tmp_path,
):
    _producer_manifests(tmp_path / "workspace/job")
    monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
    producer_results = [
        ["gwas_fmt", {"status": "success", "result": {
            "stage_identity": {"stage": "gwas_fmt", "prefix": "cohort",
                               "genotype_scope_id": GENOTYPE},
            "stage_status": {"stage": "gwas_fmt", "completed": True,
                             "prefix": "cohort", "genotype_scope_id": GENOTYPE},
        }}],
        ["gwas_trait", {"status": "success", "result": {
            "stage_identity": {"stage": "gwas_trait", "prefix": "cohort",
                               "phenotype_scope_id": PHENOTYPE},
            "stage_status": {"stage": "gwas_trait", "completed": True,
                             "prefix": "cohort", "phenotype_scope_id": PHENOTYPE},
        }}],
    ]
    captured = _run_frontend_emmax_handoff(tmp_path, producer_results)
    assert captured["params"] == {
        "genotypeScopeId": GENOTYPE, "phenotypeScopeId": PHENOTYPE,
    }
    preflight = api._api_semantic_preflight(
        "job", "gwas_emmax", "cohort", captured["body"],
    )
    assert preflight["valid"] is True
    assert preflight["materialized_params"]["genotypeScopeId"] == GENOTYPE
    assert preflight["materialized_params"]["phenotypeScopeId"] == PHENOTYPE
    frontend = (Path(__file__).resolve().parents[2] / "apps/web/vue3/index.html").read_text(encoding="utf-8")
    verified = frontend.index("if (!manifestCompleted)")
    handoff_call = frontend.index("handoffGwasEmmaxProducerIdentity(stageAction, stageResult);", verified)
    completed_mark = frontend.index("stageExecutionStatus.value[stageId] = 'completed';", handoff_call)
    assert verified < handoff_call < completed_mark


@pytest.mark.parametrize("bad_result", [
    {"status": "success", "result": {"stage_identity": {
        "stage": "gwas_trait", "prefix": "cohort", "genotype_scope_id": GENOTYPE,
    }, "stage_status": {"stage": "gwas_trait", "prefix": "cohort", "completed": True}}},
    {"status": "success", "result": {"stage_identity": {
        "stage": "gwas_fmt", "prefix": "cohort", "genotype_scope_id": "genotype:bad",
    }, "stage_status": {"stage": "gwas_fmt", "prefix": "cohort", "completed": True}}},
    {"status": "success", "result": {"stage_status": {
        "stage": "gwas_fmt", "prefix": "cohort", "completed": False,
        "genotype_scope_id": GENOTYPE,
    }}},
    {"status": "success", "result": {
        "stage_identity": {"stage": "gwas_fmt", "prefix": "other",
                           "genotype_scope_id": GENOTYPE},
        "stage_status": {"stage": "gwas_fmt", "prefix": "other", "completed": True},
    }},
])
def test_frontend_handoff_rejects_bad_identity_but_backend_uses_authoritative_manifests(
    monkeypatch, tmp_path, bad_result,
):
    _producer_manifests(tmp_path / "workspace/job")
    monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
    captured = _run_frontend_emmax_handoff(tmp_path, [["gwas_fmt", bad_result]])
    assert "genotypeScopeId" not in captured["params"]
    result = api._api_semantic_preflight("job", "gwas_emmax", "cohort", captured["body"])
    assert result["valid"] is True
    assert result["materialized_params"]["genotypeScopeId"] == GENOTYPE
    assert result["materialized_params"]["phenotypeScopeId"] == PHENOTYPE


def test_emmax_declarations_and_exact_four_scope_status(tmp_path):
    from gfa_core.mas.tools.stage_tools import StageToolRegistry

    capability = get_stage_capability("gwas_emmax")
    assert [item.role for item in capability.consumes] == ["sorted_vcf", "trait_table"]
    assert [item.role for item in capability.provides] == [
        "emmax_full", "emmax_pvalues", "manifest", "report", "environment",
    ]
    tool = StageToolRegistry().get("gwas_emmax")
    assert tool.outputs == ["emmax_full", "emmax_pvalues", "manifest", "report", "environment"]
    workspace = tmp_path / "workspace" / "job"
    _emmax_manifest(workspace)
    identity = StageIdentity(
        stage="gwas_emmax", prefix="cohort", genotype_scope_id=GENOTYPE,
        phenotype_scope_id=PHENOTYPE, 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
    identity.model_scope_id = ""
    partial = get_stage_status_for_identity(
        "job", identity, workspace_root=tmp_path / "workspace", strict_identity=True,
    )
    assert partial.completed is False
    assert "all four" in partial.errors[0]


def test_emmax_results_require_exact_scopes_and_elapsed(monkeypatch, tmp_path):
    workspace = tmp_path / "workspace" / "job"
    _emmax_manifest(workspace)
    monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
    exact = asyncio.run(api.get_results(
        job_id="job", stage="gwas_emmax", genotype_scope_id=GENOTYPE,
        phenotype_scope_id=PHENOTYPE, model_scope_id=MODEL, analysis_scope_id=ANALYSIS,
    ))
    assert exact["gwas_emmax"]["images"] == []
    assert {item["role"] for item in exact["gwas_emmax"]["files"]} == {
        "emmax_full", "emmax_pvalues", "report", "environment", "manifest",
    }
    assert exact["gwas_emmax"]["elapsed_time"] == 12.3
    wrong = asyncio.run(api.get_results(
        job_id="job", stage="gwas_emmax", genotype_scope_id=GENOTYPE,
        phenotype_scope_id=PHENOTYPE, model_scope_id="gwas-model:" + "9" * 64,
        analysis_scope_id=ANALYSIS,
    ))
    assert wrong["gwas_emmax"]["files"] == []


def test_backend_emmax_accepts_only_fresh_exact_success_manifest(monkeypatch, tmp_path):
    workspace = tmp_path / "workspace"
    fmt, trait = _producer_manifests(workspace)
    decisions = {
        "sorted_vcf": {"path": str(fmt["artifact"]), "source_manifest": str(fmt["path"])},
        "trait_table": {"path": str(trait["artifact"]), "source_manifest": str(trait["path"])},
    }
    monkeypatch.setattr(backend, "_resolve_gwas_emmax_inputs", lambda **_kwargs: decisions)

    def fake_run(command, cwd=None):
        assert command[command.index("--log-file") + 1].endswith("_gwas_emmax.log")
        _emmax_manifest(workspace)
        return 0

    monkeypatch.setattr(backend, "_run_child_owned_log", fake_run)
    message, manifest = backend.run_stage(
        stage="gwas_emmax",
        job=backend.JobInfo("job", str(tmp_path / "input"), str(workspace), str(workspace / "logs"), "now"),
        prefix="cohort", cpu=6, genotype_scope_id=GENOTYPE,
        phenotype_scope_id=PHENOTYPE, gwas_plink=PLINK, gwas_emmax=EMMAX,
        gwas_rscript=RSCRIPT, gwas_requested_pcs=3,
    )
    assert message.startswith("✅")
    assert manifest == workspace / "final/cohort.gwas_emmax.identity.manifest.json"


def test_backend_emmax_failure_never_replays_prior_success(monkeypatch, tmp_path):
    workspace = tmp_path / "workspace"
    fmt, trait = _producer_manifests(workspace)
    prior = _emmax_manifest(workspace)
    decisions = {
        "sorted_vcf": {"path": str(fmt["artifact"]), "source_manifest": str(fmt["path"])},
        "trait_table": {"path": str(trait["artifact"]), "source_manifest": str(trait["path"])},
    }
    monkeypatch.setattr(backend, "_resolve_gwas_emmax_inputs", lambda **_kwargs: decisions)
    monkeypatch.setattr(backend, "_run_child_owned_log", lambda *_args, **_kwargs: 7)
    message, manifest = backend.run_stage(
        stage="gwas_emmax",
        job=backend.JobInfo("job", str(tmp_path / "input"), str(workspace), str(workspace / "logs"), "now"),
        prefix="cohort", cpu=6, genotype_scope_id=GENOTYPE,
        phenotype_scope_id=PHENOTYPE,
    )
    assert "exit code: 7" in message
    assert manifest is None
    assert prior.is_file()


def _configure_task_test(monkeypatch, tmp_path, fake_run_stage):
    import gfa_core.mas.resource_governor as resource_governor

    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", tmp_path / "workspace")
    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", fake_run_stage)


def test_api_task_forwards_emmax_fields_and_accepts_complete_postflight(monkeypatch, tmp_path):
    captured = {}

    def fake_run_stage(**kwargs):
        captured.update(kwargs)
        return "✅ done", _emmax_manifest(tmp_path / "workspace/job")

    _configure_task_test(monkeypatch, tmp_path, fake_run_stage)
    api.tasks["emmax-task"] = {
        "id": "emmax-task", "stage": "gwas_emmax", "job_id": "job",
        "status": "pending", "message": "", "created_at": "2026-07-21T00:00:00",
    }
    params = {"prefix": "cohort", **_scopes(), "gwasPlink": PLINK,
              "gwasEmmax": EMMAX, "gwasRscript": RSCRIPT, "gwasRequestedPcs": 2}
    api.run_stage_sync("emmax-task", "gwas_emmax", "job", params)
    assert {key: captured[key] for key in (
        "gwas_plink", "gwas_emmax", "gwas_rscript", "gwas_requested_pcs",
    )} == {
        "gwas_plink": PLINK, "gwas_emmax": EMMAX,
        "gwas_rscript": RSCRIPT, "gwas_requested_pcs": 2,
    }
    task = api.tasks["emmax-task"]
    assert task["status"] == "success", task
    files = task["result"]["stage_files"]["gwas_emmax"]["files"]
    assert {item["role"] for item in files} == {
        "emmax_full", "emmax_pvalues", "report", "environment", "manifest",
    }
    assert task["result"]["stage_files"]["gwas_emmax"]["elapsed_time"] == 12.3


def test_api_task_rejects_partial_emmax_manifest_without_publishing_result(monkeypatch, tmp_path):
    def fake_run_stage(**_kwargs):
        path = _emmax_manifest(
            tmp_path / "workspace/job",
            roles=("emmax_full", "report", "environment"),
        )
        return "✅ done", path

    _configure_task_test(monkeypatch, tmp_path, fake_run_stage)
    api.tasks["partial-emmax"] = {
        "id": "partial-emmax", "stage": "gwas_emmax", "job_id": "job",
        "status": "pending", "message": "", "created_at": "2026-07-21T00:00:00",
    }
    api.run_stage_sync("partial-emmax", "gwas_emmax", "job", {"prefix": "cohort", **_scopes()})
    task = api.tasks["partial-emmax"]
    assert task["status"] == "failed"
    assert task["result"]["success"] is False
    assert "files" not in task["result"] and "images" not in task["result"]