from __future__ import annotations

import asyncio
import hashlib
import json
from dataclasses import dataclass
from pathlib import Path

from apps.web.pipeline_backend import _population_cli_command
from apps.web.vue3.main import ExecuteRequest
from gfa_core.mas.agents.base import AgentMessage, AgentState
from gfa_core.mas.agents.intent_v2 import IntentAgentV2
from gfa_core.mas.agents.planning_v2 import PlanningAgentV2
from gfa_core.mas.core.fast_intent import fast_intent_recognition
from gfa_core.mas.stage_catalog import EXECUTABLE_STAGES
from gfa_core.mas.tools.stage_tools import StageTool, StageToolExecutor, StageToolRegistry
from gfa_core.mas.utils.intent_parser import parse_intent_from_query, resolve_population_intent
from gfa_core.project_paths import get_conda_env_bin, get_conda_env_rscript, get_configured_path


POP_STAGES = ["pop_pca", "pop_phylogeny", "pop_structure", "pop_ld_decay", "pop_snp_density"]
POPULATION_REQUEST = (
    "使用 input/job_1784815396884_5db7e7/population_11samples.snp.vcf.gz "
    "和分组表 input/job_1784815396884_5db7e7/ population_11samples.groups.tsv，"
    "执行完整的群体遗传分析。"
)
POPULATION_VCF = "input/job_1784815396884_5db7e7/population_11samples.snp.vcf.gz"
POPULATION_GROUPS = "input/job_1784815396884_5db7e7/population_11samples.groups.tsv"


def _write_complete_pca_manifest(workspace: Path, vcf: Path) -> tuple[str, str]:
    final = workspace / "final"
    final.mkdir(parents=True, exist_ok=True)
    artifact = final / "cohort.pop_pca.test.tsv"
    artifact.write_text("sample\tPC1\nS1\t0.0\n", encoding="utf-8")
    digest = hashlib.sha256(vcf.read_bytes()).hexdigest()
    genotype = f"genotype:{digest}"
    analysis = "population-analysis:" + "b" * 64
    proof = {
        "artifact": "pca_scores",
        "role": "pca_scores",
        "path": str(artifact.relative_to(workspace)),
        "full_path": str(artifact.resolve()),
        "type": "text/tab-separated-values",
        "size": artifact.stat().st_size,
        "sha256": hashlib.sha256(artifact.read_bytes()).hexdigest(),
    }
    roles = {
        "pca_scores", "pca_eigenvalues", "pca_variance_table", "pca_pruning_markers",
        "pca_pc1_pc2_plot", "pca_scree_plot", "pca_tassel_covariate", "report", "environment",
    }
    manifest = {
        "manifest_version": "1.0",
        "stage": "pop_pca",
        "prefix": "cohort",
        "status": "success",
        "start_time": "2026-07-24T00:00:00+00:00",
        "run_id": "pop_pca_test",
        "genotype_scope_id": genotype,
        "analysis_scope_id": analysis,
        "inputs": [{"role": "population_vcf", "sha256": digest}],
        "artifacts": {role: {**proof, "artifact": role, "role": role} for role in roles},
    }
    (final / "cohort.pop_pca.test.manifest.json").write_text(
        json.dumps(manifest),
        encoding="utf-8",
    )
    return genotype, analysis


def test_population_stages_are_executable_and_registry_aligned() -> None:
    definitions = {item["id"]: item for item in StageToolRegistry.STAGE_DEFINITIONS}
    for stage in POP_STAGES:
        assert stage in EXECUTABLE_STAGES
        assert definitions[stage]["script"] == f"run_{stage}.py"
        assert definitions[stage]["category"] == "population_genomics"


def test_population_intent_does_not_steal_gene_family_phylogeny() -> None:
    assert resolve_population_intent("进行群体遗传分析")["stages"] == POP_STAGES
    assert resolve_population_intent("构建群体系统树")["stages"] == ["pop_phylogeny"]
    assert resolve_population_intent("构建基因家族系统树") is None


def test_population_request_binds_inputs_through_public_parser() -> None:
    parsed = parse_intent_from_query(POPULATION_REQUEST)

    assert parsed["stages"] == POP_STAGES
    assert parsed["primary_stage"] == "pop_snp_density"
    assert parsed["prefix"] == "cohort"
    assert parsed["params"]["populationVcf"] == POPULATION_VCF
    assert parsed["params"]["populationSampleGroup"] == POPULATION_GROUPS


def test_population_request_binds_inputs_through_fast_parser() -> None:
    parsed = fast_intent_recognition(POPULATION_REQUEST)

    assert parsed is not None
    assert parsed["stages"] == POP_STAGES
    assert parsed["prefix"] == "cohort"
    assert parsed["params"]["populationVcf"] == POPULATION_VCF
    assert parsed["params"]["populationSampleGroup"] == POPULATION_GROUPS


def test_population_request_binds_inputs_through_public_agent() -> None:
    state = AgentState(session_id="population-intent-test", job_id="job_1784815396884_5db7e7")
    message = AgentMessage(
        sender="router",
        receiver="intent",
        message_type="recognize",
        payload={"query": POPULATION_REQUEST, "workspace": "/tmp/population-intent-test"},
    )

    result = asyncio.run(IntentAgentV2(object()).execute(state, message))
    parsed = result.results["results"]["intent"]

    assert parsed["stages"] == POP_STAGES
    assert parsed["prefix"] == "cohort"
    assert parsed["params"]["populationVcf"] == POPULATION_VCF
    assert parsed["params"]["populationSampleGroup"] == POPULATION_GROUPS


def test_population_pca_request_cannot_route_to_rnaseq() -> None:
    parsed = parse_intent_from_query(
        "请使用 input/job_test/population.snp.vcf.gz 进行群体 PCA 主成分分析。"
    )

    assert parsed["stages"] == ["pop_pca"]
    assert parsed["primary_stage"] == "pop_pca"
    assert parsed["params"]["populationVcf"] == "input/job_test/population.snp.vcf.gz"


def test_population_group_table_is_optional() -> None:
    parsed = parse_intent_from_query(
        "使用 input/job_test/population.snp.vcf.gz 执行群体 SNP 密度分析。"
    )

    assert parsed["params"]["populationVcf"] == "input/job_test/population.snp.vcf.gz"
    assert "populationSampleGroup" not in parsed["params"]


def test_population_quoted_paths_keep_intentional_spaces() -> None:
    parsed = parse_intent_from_query(
        '使用 "input/job test/population cohort.vcf.gz" 和分组表 '
        '"input/job test/population groups.tsv"，执行群体 PCA 分析。'
    )

    assert parsed["params"]["populationVcf"] == "input/job test/population cohort.vcf.gz"
    assert parsed["params"]["populationSampleGroup"] == "input/job test/population groups.tsv"


def test_population_request_does_not_guess_between_unanchored_vcfs() -> None:
    parsed = parse_intent_from_query(
        "比较 input/job_test/a.vcf.gz 和 input/job_test/b.vcf.gz，执行群体 PCA 分析。"
    )

    assert parsed["stages"] == ["pop_pca"]
    assert "populationVcf" not in parsed["params"]


def test_population_plan_carries_inputs_to_every_stage() -> None:
    parsed = parse_intent_from_query(POPULATION_REQUEST)
    planner = PlanningAgentV2.__new__(PlanningAgentV2)

    stages = planner._generate_default_stages(
        parsed["stages"],
        parsed["prefix"],
        intent_params=parsed["params"],
    )

    assert [stage["id"] for stage in stages] == POP_STAGES
    for stage in stages:
        assert stage["params"]["populationVcf"] == POPULATION_VCF
        assert stage["params"]["populationSampleGroup"] == POPULATION_GROUPS


def test_population_backend_command_propagates_stage_parameters(tmp_path: Path) -> None:
    command = _population_cli_command(
        stage="pop_snp_density", workspace=tmp_path, job_id="job", prefix="cohort",
        vcf=tmp_path / "input.vcf", sample_group="", cpu=4,
        params={"chromosome_policy": "rice_1_12", "coarse_window_bp": 1_000_000, "fine_window_bp": 500_000, "rscript": "/opt/Rscript"},
    )
    assert command[:3] == [command[0], "-m", "apps.cli.run_pop_snp_density"]
    assert command[command.index("--chromosome-policy") + 1] == "rice_1_12"
    assert command[command.index("--rscript") + 1] == "/opt/Rscript"


def test_population_ld_defaults_use_verified_project_wrappers() -> None:
    request = ExecuteRequest(stage="pop_ld_decay")
    assert request.popBcftools == "scripts/pysam_bcftools.py"
    assert request.popPlotOne == "scripts/plot_pop_ld_decay.py"
    assert request.popPlotMulti == "scripts/plot_pop_ld_decay.py"


def test_execute_request_exposes_population_confirmation_fields() -> None:
    request = ExecuteRequest(stage="pop_structure", populationVcf="/data/cohort.vcf.gz", popKMin=2, popKMax=10, popGwasK=2)
    assert request.populationVcf == "/data/cohort.vcf.gz"
    assert (request.popKMin, request.popKMax, request.popGwasK) == (2, 10, 2)


def test_execute_api_submits_all_population_fields(monkeypatch, tmp_path: Path) -> None:
    from apps.web.vue3 import main as api

    class DummyExecutor:
        def __init__(self) -> None:
            self.calls = []

        def submit(self, *args):
            self.calls.append(args)

    vcf = tmp_path / "cohort.vcf"
    vcf.write_text(
        "##fileformat=VCFv4.2\n#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\tS1\n"
        "1\t1\trs1\tA\tG\t.\tPASS\t.\tGT\t0/1\n",
        encoding="utf-8",
    )
    groups = tmp_path / "groups.tsv"
    groups.write_text("S1\tGroup A\n", encoding="utf-8")
    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="pop_ld_decay", job_id="job", prefix="cohort",
        populationVcf=str(vcf), populationSampleGroup=str(groups),
        popBcftools="/tools/bcftools", popLddecay="/tools/PopLDdecay",
        popPlotOne="/tools/plot-one", popPlotMulti="/tools/plot-multi",
    )))
    assert result["async"] is True, result
    assert len(dummy.calls) == 1
    submitted = dummy.calls[0][4]
    assert submitted["populationVcf"] == str(vcf)
    assert submitted["populationSampleGroup"] == str(groups)
    assert submitted["popBcftools"] == "/tools/bcftools"
    assert submitted["popLddecay"] == "/tools/PopLDdecay"
    assert submitted["popPlotOne"] == "/tools/plot-one"
    assert submitted["popPlotMulti"] == "/tools/plot-multi"


def test_stage_tool_executor_forwards_population_fields_to_backend(monkeypatch, tmp_path: Path) -> None:
    from apps.web import pipeline_backend

    captured = {}

    @dataclass
    class FakeJobInfo:
        job_id: str
        input_dir: str
        workspace_dir: str
        logs_dir: str
        created_at: str

    def fake_run_stage(**kwargs):
        captured.update(kwargs)
        manifest_path = Path(kwargs["job"].workspace_dir) / "final" / "cohort.pop_structure.manifest.json"
        manifest_path.parent.mkdir(parents=True, exist_ok=True)
        manifest_path.write_text(
            json.dumps({"stage": "pop_structure", "status": "success"}),
            encoding="utf-8",
        )
        return "✅ pop_structure completed", manifest_path

    monkeypatch.setattr(pipeline_backend, "JobInfo", FakeJobInfo)
    monkeypatch.setattr(pipeline_backend, "run_stage", fake_run_stage)

    params = {
        "populationVcf": "/data/cohort.vcf.gz",
        "populationSampleGroup": "/data/groups.tsv",
        "popPlink": "/tools/plink",
        "popAdmixture": "/tools/admixture",
        "popKMin": 3,
        "popKMax": 7,
        "popCvFolds": 4,
        "popRandomSeed": 23,
        "popGwasK": 5,
    }
    tool = StageTool(
        "pop_structure", "Population structure", "run_pop_structure.py",
        "population_genomics", "test", timeout=5,
    )
    result = asyncio.run(
        StageToolExecutor(project_root=tmp_path).execute(tool, "job", "cohort", params)
    )

    assert result["status"] == "success"
    assert captured["population_vcf"] == "/data/cohort.vcf.gz"
    assert captured["population_sample_group"] == "/data/groups.tsv"
    assert captured["population_params"] == {
        "plink": "/tools/plink",
        "tassel": get_configured_path("TASSEL_PIPELINE", "run_pipeline.pl"),
        "fasttree": str(get_conda_env_bin("ASAW") / "FastTree"),
        "admixture": "/tools/admixture",
        "bcftools": "scripts/pysam_bcftools.py",
        "poplddecay": str(get_conda_env_bin("ASAW") / "PopLDdecay"),
        "plot_one": "scripts/plot_pop_ld_decay.py",
        "plot_multi": "scripts/plot_pop_ld_decay.py",
        "rscript": str(get_conda_env_rscript("ASAW")),
        "k_min": 3,
        "k_max": 7,
        "cv_folds": 4,
        "random_seed": 23,
        "gwas_k": 5,
        "chromosome_policy": "all_vcf_contigs",
        "chromosomes": "",
    }


def test_population_snp_density_uses_portable_rscript_default() -> None:
    project_root = Path(__file__).resolve().parents[2]
    main_source = (project_root / "apps/web/vue3/main.py").read_text(encoding="utf-8")
    frontend_source = (project_root / "apps/web/vue3/index.html").read_text(encoding="utf-8")

    assert "/data/miniconda3/envs/ASAW/bin/Rscript" not in main_source
    assert "/data/miniconda3/envs/ASAW/bin/Rscript" not in frontend_source
    assert "/home/xzq/miniconda3/envs/ASAW/bin/Rscript" not in main_source
    assert "/home/xzq/miniconda3/envs/ASAW/bin/Rscript" not in frontend_source


def test_celery_artifact_validation_uses_population_manifest_identity(monkeypatch) -> None:
    import tasks
    from gfa_core.mas.tools import stage_identity

    captured = {}

    def fake_get_stage_status_for_identity(*, identity, **_kwargs):
        captured["identity"] = identity
        return type("Status", (), {"completed": True})()

    monkeypatch.setattr(
        stage_identity,
        "get_stage_status_for_identity",
        fake_get_stage_status_for_identity,
    )
    result = tasks.validate_single_stage_artifacts(
        "job",
        "cohort",
        "pop_pca",
        {
            "final_output": {
                "manifest": {
                    "stage": "pop_pca",
                    "prefix": "cohort",
                    "genotype_scope_id": "genotype:abc",
                    "analysis_scope_id": "population-analysis:def",
                    "run_id": "pop_pca_run",
                }
            }
        },
    )

    assert result is None
    assert captured["identity"].genotype_scope_id == "genotype:abc"
    assert captured["identity"].analysis_scope_id == "population-analysis:def"
    assert captured["identity"].run_id == "pop_pca_run"


def test_population_preflight_resolves_relative_paths_from_project_root(monkeypatch, tmp_path: Path) -> None:
    from apps.web.vue3 import main as api

    input_dir = tmp_path / "input" / "job_test"
    input_dir.mkdir(parents=True)
    vcf = input_dir / "cohort.vcf.gz"
    vcf.write_bytes(b"vcf")
    groups = input_dir / "groups.tsv"
    groups.write_text("S1\tA\n", encoding="utf-8")
    elsewhere = tmp_path / "elsewhere"
    elsewhere.mkdir()
    monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
    monkeypatch.chdir(elsewhere)

    result = api._api_semantic_preflight(
        "job_test",
        "pop_pca",
        "cohort",
        {
            "populationVcf": "input/job_test/cohort.vcf.gz",
            "populationSampleGroup": "input/job_test/groups.tsv",
        },
    )

    assert result["valid"] is True
    assert result["materialized_params"] == {
        "populationVcf": str(vcf.resolve()),
        "populationSampleGroup": str(groups.resolve()),
    }


def test_population_preflight_reports_missing_vcf_path(monkeypatch, tmp_path: Path) -> None:
    from apps.web.vue3 import main as api

    monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
    result = api._api_semantic_preflight(
        "job_test",
        "pop_pca",
        "cohort",
        {"populationVcf": "input/job_test/missing.vcf.gz"},
    )

    assert result["valid"] is False
    assert result["issues"] == [{
        "severity": "error",
        "code": "missing_population_vcf",
        "message": "Population VCF does not exist: input/job_test/missing.vcf.gz",
    }]


def test_population_preflight_reports_missing_group_path(monkeypatch, tmp_path: Path) -> None:
    from apps.web.vue3 import main as api

    vcf = tmp_path / "cohort.vcf.gz"
    vcf.write_bytes(b"vcf")
    monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
    result = api._api_semantic_preflight(
        "job_test",
        "pop_pca",
        "cohort",
        {
            "populationVcf": str(vcf),
            "populationSampleGroup": "input/job_test/missing.groups.tsv",
        },
    )

    assert result["valid"] is False
    assert result["issues"] == [{
        "severity": "error",
        "code": "missing_population_sample_group",
        "message": "Population sample-group table does not exist: input/job_test/missing.groups.tsv",
    }]


def test_population_preflight_materializes_empty_scope_from_manifest(monkeypatch, tmp_path: Path) -> None:
    from apps.web.vue3 import main as api

    vcf = tmp_path / "cohort.vcf.gz"
    vcf.write_bytes(b"vcf")
    workspace_root = tmp_path / "workspace"
    genotype, analysis = _write_complete_pca_manifest(workspace_root / "job", vcf)
    monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
    monkeypatch.setattr(api, "WORKSPACE_ROOT", workspace_root)

    result = api._api_semantic_preflight(
        "job",
        "pop_pca",
        "cohort",
        {"populationVcf": str(vcf)},
    )

    assert result["valid"] is True
    assert result["materialized_params"]["genotypeScopeId"] == genotype
    assert result["materialized_params"]["analysisScopeId"] == analysis
    assert result["materialized_params"]["runId"] == "pop_pca_test"


def test_population_preflight_rejects_partial_scope(monkeypatch, tmp_path: Path) -> None:
    from apps.web.vue3 import main as api

    vcf = tmp_path / "cohort.vcf.gz"
    vcf.write_bytes(b"vcf")
    monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
    monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")

    result = api._api_semantic_preflight(
        "job",
        "pop_pca",
        "cohort",
        {"populationVcf": str(vcf), "genotypeScopeId": "genotype:" + "a" * 64},
    )

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


def test_population_plan_recognizes_completed_manifest_without_request_scope(monkeypatch, tmp_path: Path) -> None:
    from apps.web.vue3 import main as api

    vcf = tmp_path / "cohort.vcf.gz"
    vcf.write_bytes(b"vcf")
    workspace_root = tmp_path / "workspace"
    _write_complete_pca_manifest(workspace_root / "job", vcf)
    monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
    monkeypatch.setattr(api, "WORKSPACE_ROOT", workspace_root)
    details = [{
        "id": "pop_pca",
        "params": {"prefix": "cohort", "populationVcf": str(vcf)},
    }]

    completed = api._identity_completed_stages_for_plan(
        "job",
        ["pop_pca"],
        {"populationVcf": str(vcf)},
        details,
        effective_prefix="cohort",
    )

    assert completed == ["pop_pca"]
    assert details[0]["params"]["genotypeScopeId"].startswith("genotype:")
    assert details[0]["params"]["analysisScopeId"].startswith("population-analysis:")


def test_population_results_materialize_unique_identity_when_scopes_are_empty(monkeypatch, tmp_path: Path) -> None:
    from apps.web.vue3 import main as api

    vcf = tmp_path / "cohort.vcf.gz"
    vcf.write_bytes(b"vcf")
    workspace_root = tmp_path / "workspace"
    _write_complete_pca_manifest(workspace_root / "job", vcf)
    monkeypatch.setattr(api, "WORKSPACE_ROOT", workspace_root)

    result = asyncio.run(api.get_results(
        job_id="job",
        stage="pop_pca",
        result_prefix="cohort",
        rnaseq_scope_id="",
        scope_id="",
        genotype_scope_id="",
        phenotype_scope_id="",
        covariate_scope_id="",
        model_scope_id="",
        source_model_scope_id="",
        analysis_scope_id="",
    ))

    assert result["pop_pca"]["files"]
    assert result["pop_pca"]["status"] == "success"


def test_execute_api_returns_concrete_population_preflight_error(monkeypatch, tmp_path: Path) -> None:
    from apps.web.vue3 import main as api

    monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
    monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
    monkeypatch.setattr(api, "INPUT_ROOT", tmp_path / "input")
    result = asyncio.run(api.execute_stage(api.ExecuteRequest(
        stage="pop_pca",
        job_id="job_test",
        prefix="cohort",
        populationVcf="input/job_test/missing.vcf.gz",
    )))

    assert result["status"] == "preflight_failed"
    assert result["message"] == "Population VCF does not exist: input/job_test/missing.vcf.gz"


def test_frontend_contains_all_population_stage_ids() -> None:
    html = (Path(__file__).resolve().parents[2] / "apps" / "web" / "vue3" / "index.html").read_text(encoding="utf-8")
    for stage in POP_STAGES:
        assert stage in html
    assert "applyPopulationExecutionParams" in html
    assert "appendPopulationIdentityQuery" in html
    assert "populationVcf" in html


def test_frontend_population_binding_uses_stage_before_confirmation_params() -> None:
    html = (Path(__file__).resolve().parents[2] / "apps" / "web" / "vue3" / "index.html").read_text(encoding="utf-8")
    function_start = html.index("const applyPopulationExecutionParams")
    function_end = html.index("const executeSingleStageWithParams", function_start)
    function_source = html[function_start:function_end]

    assert function_source.index("stageParams[camelKey]") < function_source.index("batchParams[camelKey]")
    assert function_source.index("batchParams[camelKey]") < function_source.index("planParams[camelKey]")
    assert function_source.index("planParams[camelKey]") < function_source.index("confirmParams[camelKey]")


def test_population_plink_commands_use_plink19_missing_id_template() -> None:
    project = Path(__file__).resolve().parents[2]
    for script in ("run_pop_pca.py", "run_pop_structure.py"):
        source = (project / "apps" / "cli" / script).read_text(encoding="utf-8")
        assert "@:#:$1:$2" in source
        assert "@:#:$r:$a" not in source


def test_population_results_require_exact_identity_and_valid_hashes(monkeypatch, tmp_path: Path) -> None:
    from apps.web.vue3 import main as web

    workspace_root = tmp_path / "workspace"
    final = workspace_root / "job" / "final"
    final.mkdir(parents=True)
    artifact = final / "cohort.pop_pca.123456789abc.pca.tsv"
    artifact.write_text("sample\tPC1\nS1\t0.0\n", encoding="utf-8")
    genotype = "genotype:" + "a" * 64
    analysis = "population-analysis:" + "b" * 64
    pca_roles = {
        "pca_scores", "pca_eigenvalues", "pca_variance_table", "pca_pruning_markers",
        "pca_pc1_pc2_plot", "pca_scree_plot", "pca_tassel_covariate", "report", "environment",
    }
    proof = {
        "path": f"final/{artifact.name}", "size": artifact.stat().st_size,
        "sha256": hashlib.sha256(artifact.read_bytes()).hexdigest(),
    }
    manifest = {
        "stage": "pop_pca", "prefix": "cohort", "status": "success",
        "genotype_scope_id": genotype, "analysis_scope_id": analysis,
        "artifacts": {role: {"role": role, **proof} for role in pca_roles},
    }
    (final / "cohort.pop_pca.123456789abc.manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
    sibling_artifact = final / "sibling.pop_pca.123456789abc.pca.tsv"
    sibling_artifact.write_text("sample\tPC1\nS1\t9.9\n", encoding="utf-8")
    sibling = {
        **manifest, "prefix": "sibling",
        "artifacts": {"pca_scores": {
            "role": "pca_scores", "path": f"final/{sibling_artifact.name}",
            "size": sibling_artifact.stat().st_size,
            "sha256": hashlib.sha256(sibling_artifact.read_bytes()).hexdigest(),
        }},
    }
    (final / "sibling.pop_pca.123456789abc.manifest.json").write_text(json.dumps(sibling), encoding="utf-8")
    monkeypatch.setattr(web, "WORKSPACE_ROOT", workspace_root)

    from gfa_core.mas.tools.stage_identity import StageIdentity, get_stage_status_for_identity
    partial = {**manifest, "prefix": "partial", "artifacts": {"pca_scores": {"role": "pca_scores", **proof}}}
    (final / "partial.pop_pca.123456789abc.manifest.json").write_text(json.dumps(partial), encoding="utf-8")
    partial_status = get_stage_status_for_identity(
        "job", StageIdentity(stage="pop_pca", prefix="partial", genotype_scope_id=genotype, analysis_scope_id=analysis),
        workspace_root=workspace_root,
    )
    assert partial_status.completed is False
    assert any("missing required population artifact role" in error for error in partial_status.errors)
    partial_results = asyncio.run(web.get_results(
        job_id="job", stage="pop_pca", result_prefix="partial", rnaseq_scope_id="", scope_id="",
        genotype_scope_id=genotype, analysis_scope_id=analysis,
        phenotype_scope_id="", covariate_scope_id="", model_scope_id="", source_model_scope_id="",
    ))["pop_pca"]
    assert partial_results["files"] == []

    ld_qc = final / "partialld.group_qc.json"
    ld_qc.write_text(json.dumps({"valid_groups": [{"group": "North", "slug": "north", "sample_count": 3}]}), encoding="utf-8")
    ld_qc_proof = {
        "path": f"final/{ld_qc.name}", "size": ld_qc.stat().st_size,
        "sha256": hashlib.sha256(ld_qc.read_bytes()).hexdigest(),
    }
    ld_roles = {"ld_all_stat", "ld_all_plot_png", "ld_all_plot_pdf", "group_sample_qc", "report", "environment"}
    ld_artifacts = {role: {"role": role, **proof} for role in ld_roles}
    ld_artifacts["group_sample_qc"] = {"role": "group_sample_qc", **ld_qc_proof}
    partial_ld = {
        "stage": "pop_ld_decay", "prefix": "partialld", "status": "success",
        "genotype_scope_id": genotype, "analysis_scope_id": analysis,
        "metrics": {"valid_group_count": 1}, "artifacts": ld_artifacts,
    }
    (final / "partialld.pop_ld_decay.123456789abc.manifest.json").write_text(json.dumps(partial_ld), encoding="utf-8")
    partial_ld_status = get_stage_status_for_identity(
        "job", StageIdentity(stage="pop_ld_decay", prefix="partialld", genotype_scope_id=genotype, analysis_scope_id=analysis),
        workspace_root=workspace_root,
    )
    assert partial_ld_status.completed is False
    assert any("ld_group_stat_north" in error for error in partial_ld_status.errors)

    def results(**identity):
        return asyncio.run(web.get_results(
            job_id="job", stage="pop_pca", result_prefix="cohort", rnaseq_scope_id="", scope_id="",
            genotype_scope_id=identity.get("genotype", ""), analysis_scope_id=identity.get("analysis", ""),
            phenotype_scope_id="", covariate_scope_id="", model_scope_id="", source_model_scope_id="",
        ))

    exact_files = results(genotype=genotype, analysis=analysis)["pop_pca"]["files"]
    assert exact_files
    assert all("sibling" not in item["name"] for item in exact_files)
    assert results(genotype=genotype)["pop_pca"]["files"] == []
    artifact.write_text("tampered\n", encoding="utf-8")
    assert results(genotype=genotype, analysis=analysis)["pop_pca"]["files"] == []
    artifact.write_text("sample\tPC1\nS1\t0.0\n", encoding="utf-8")
    failed = {
        "stage": "pop_pca", "prefix": "cohort", "status": "failed",
        "start_time": "2099-01-01T00:00:00+00:00",
        "genotype_scope_id": genotype, "analysis_scope_id": analysis,
        "artifacts": {}, "error": "controlled retry failure",
    }
    (final / "cohort.pop_pca.failed.manifest.json").write_text(json.dumps(failed), encoding="utf-8")
    latest = results(genotype=genotype, analysis=analysis)["pop_pca"]
    assert latest["files"] == []
    assert latest["status"] == "failed"