from __future__ import annotations

import asyncio
import json

import pytest

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, StageStatus
from gfa_core.mas.stage_catalog import complete_dependencies
from gfa_core.mas.utils.intent_parser import parse_intent_from_query


WGS_STAGES = [
    "wgs_qc", "wgs_index", "wgs_align", "wgs_markdup", "wgs_bam_qc",
    "wgs_haplotype_calling", "wgs_gvcf_combine", "wgs_joint_genotyping",
    "wgs_variant_select", "wgs_variant_filter", "wgs_impute",
    "wgs_maf_missing_filter",
]
EXPLICIT_STAGE_PHRASES = [
    ("WGS质控", "wgs_qc"), ("WGS quality control", "wgs_qc"),
    ("WGS参考索引", "wgs_index"), ("WGS reference index", "wgs_index"),
    ("WGS比对", "wgs_align"), ("WGS alignment", "wgs_align"),
    ("WGS去重复", "wgs_markdup"), ("WGS mark duplicates", "wgs_markdup"),
    ("WGS 去重复", "wgs_markdup"),
    ("WGS BAM质控", "wgs_bam_qc"), ("WGS BAM QC", "wgs_bam_qc"),
    ("WGS单样本变异检测", "wgs_haplotype_calling"), ("WGS haplotype calling", "wgs_haplotype_calling"),
    ("WGS GVCF合并", "wgs_gvcf_combine"), ("WGS GVCF combine", "wgs_gvcf_combine"),
    ("WGS联合分型", "wgs_joint_genotyping"), ("WGS joint genotyping", "wgs_joint_genotyping"),
    ("WGS变异选择", "wgs_variant_select"), ("WGS variant selection", "wgs_variant_select"),
    ("WGS变异过滤", "wgs_variant_filter"), ("WGS variant filtering", "wgs_variant_filter"),
    ("WGS基因型填充", "wgs_impute"), ("WGS genotype imputation", "wgs_impute"),
    ("WGS MAF缺失率过滤", "wgs_maf_missing_filter"), ("WGS MAF missingness filter", "wgs_maf_missing_filter"),
]


@pytest.mark.parametrize(("phrase", "stage"), EXPLICIT_STAGE_PHRASES)
def test_explicit_wgs_stage_phrase_is_deterministic(phrase: str, stage: str) -> None:
    intent = parse_intent_from_query(f"请执行{phrase}")
    assert intent["stages"] == [stage]
    assert intent["primary_stage"] == stage


@pytest.mark.parametrize("phrase", ["WGS全流程", "VCF全流程"])
def test_bare_wgs_vcf_full_chain_targets_stage12_with_complete_closure(phrase: str) -> None:
    intent = parse_intent_from_query(phrase)
    assert intent["primary_stage"] == "wgs_maf_missing_filter"
    assert complete_dependencies(intent["stages"]) == WGS_STAGES


def test_wgs_deterministic_routing_precedes_llm_fallback(monkeypatch) -> None:
    agent = IntentAgentV2(None)

    async def forbidden_fallback(*_args, **_kwargs):
        raise AssertionError("LLM fallback must not run for deterministic WGS intent")

    monkeypatch.setattr(agent, "_recognize_intent_with_llm", forbidden_fallback)
    state = AgentState(session_id="session")
    message = AgentMessage(
        sender="user", receiver="intent_v2", message_type="query",
        payload={"query": "VCF全流程"},
    )
    result = asyncio.run(agent.execute(state, message))
    intent = result.results["results"]["intent"]
    assert intent["primary_stage"] == "wgs_maf_missing_filter"
    assert intent["stages"] == WGS_STAGES


def test_non_wgs_intent_is_unchanged() -> None:
    intent = parse_intent_from_query("构建系统发育树")
    assert intent["primary_stage"] == "phylogeny"
    assert "wgs_qc" not in intent["stages"]


def test_wgs_full_workflow_is_not_gene_family_full_workflow() -> None:
    from apps.web.vue3.main import _looks_like_gene_family_workflow_query

    assert _looks_like_gene_family_workflow_query("执行 WGS 全流程") is False
    assert _looks_like_gene_family_workflow_query("执行基因家族全流程") is True


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

    final_dir = tmp_path / "job" / "final"
    final_dir.mkdir(parents=True)
    (final_dir / "cohort.wgs_qc.manifest.json").write_text(
        json.dumps({
            "stage": "wgs_qc",
            "status": "success",
            "prefix": "cohort",
            "wgs_sample_set_id": "samples-1",
            "wgs_analysis_id": "analysis-1",
        }),
        encoding="utf-8",
    )
    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path)

    assert web_main._wgs_manifest_identity_for_plan("job", "wgs_qc", "cohort") == {
        "wgsSampleSetId": "samples-1",
        "wgsReferenceId": "",
        "wgsAnalysisId": "analysis-1",
    }


def test_wgs_plan_drops_bindings_for_upstreams_that_will_rerun() -> None:
    from apps.web.vue3 import main as web_main

    params = {
        "wgsQcManifest": "/old/cohort.wgs_qc.manifest.json",
        "wgsQcAnalysisId": "old-qc",
        "wgsIndexManifest": "/old/cohort.wgs_index.manifest.json",
        "wgsReferenceId": "old-reference",
        "wgsAlignManifest": "/old/cohort.wgs_align.manifest.json",
        "wgsMethod": "gatk",
    }
    details = [{"id": "wgs_maf_missing_filter", "params": dict(params)}]

    web_main._clear_planned_wgs_upstream_bindings(
        ["wgs_qc", "wgs_index", "wgs_align", "wgs_maf_missing_filter"], params, details
    )

    assert params == {"wgsMethod": "gatk"}
    assert details[0]["params"] == {"wgsMethod": "gatk"}


def test_agent_plan_uses_current_wgs_message_over_stale_intent(monkeypatch) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools import stage_status

    class Status:
        def __init__(self, completed: bool):
            self.status = "success" if completed else "not_found"
            self.completed = completed
            self.manifest_path = None
            self.missing_artifacts = []
            self.errors = []

    monkeypatch.setattr(
        web_main,
        "_load_methodology_plan_for_agent_request",
        lambda _request: (None, None, None),
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_manifest_identity_for_plan",
        lambda *_args: {
            "wgsSampleSetId": "samples",
            "wgsReferenceId": "reference",
            "wgsAnalysisId": "analysis",
        },
    )
    monkeypatch.setattr(
        stage_status,
        "get_stage_status",
        lambda **kwargs: Status(kwargs["stage"] != "wgs_maf_missing_filter"),
    )

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(
                message="执行 WGS MAF缺失率过滤",
                job_id="job-1",
                intent={"stages": ["wgs_variant_select"]},
            )
        )
    )

    assert response["success"] is True
    assert response["mode"] == "deterministic_wgs_plan"
    assert response["plan"]["stages"] == ["wgs_maf_missing_filter"]


def test_agent_plan_keeps_only_missing_wgs_stages(monkeypatch) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools import stage_status

    class Status:
        def __init__(self, completed: bool):
            self.status = "success" if completed else "not_found"
            self.completed = completed
            self.manifest_path = None
            self.missing_artifacts = []
            self.errors = []

    monkeypatch.setattr(
        web_main,
        "_load_methodology_plan_for_agent_request",
        lambda _request: (None, None, None),
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_manifest_identity_for_plan",
        lambda *_args: {
            "wgsSampleSetId": "samples",
            "wgsReferenceId": "reference",
            "wgsAnalysisId": "analysis",
        },
    )
    missing = {"wgs_qc", "wgs_index", "wgs_align", "wgs_markdup"}
    monkeypatch.setattr(
        stage_status,
        "get_stage_status",
        lambda **kwargs: Status(kwargs["stage"] not in missing),
    )

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(
                message="执行 WGS 全流程分析",
                job_id="job-1",
                intent={"stages": ["wgs_variant_select"]},
            )
        )
    )

    assert response["success"] is True
    assert response["plan"]["stages"] == [
        "wgs_qc", "wgs_index", "wgs_align", "wgs_markdup"
    ]


def test_agent_plan_seeds_completed_qc_only_into_pending_align(monkeypatch) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core import wgs_upstream_manifest
    from gfa_core.mas.tools import stage_status

    class Status:
        def __init__(self, stage: str):
            self.status = "success" if stage == "wgs_qc" else "not_found"
            self.completed = stage == "wgs_qc"
            self.manifest_path = (
                "/workspace/job-1/final/Os.wgs_qc.manifest.json"
                if self.completed else None
            )
            self.missing_artifacts = []
            self.errors = []

    monkeypatch.setattr(
        web_main,
        "_load_methodology_plan_for_agent_request",
        lambda _request: (None, None, None),
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_manifest_identity_for_plan",
        lambda *_args: {
            "wgsSampleSetId": "samples",
            "wgsReferenceId": "reference",
            "wgsAnalysisId": "wgs-analysis:" + "c" * 64,
        },
    )
    monkeypatch.setattr(
        stage_status, "get_stage_status", lambda **kwargs: Status(kwargs["stage"])
    )
    monkeypatch.setattr(
        wgs_upstream_manifest,
        "materialize_completed_wgs_upstream_params",
        lambda **kwargs: (
            {
                "wgsQcManifest": "/workspace/job-1/final/Os.wgs_qc.manifest.json",
                "wgsSampleSetId": "wgs-samples:" + "a" * 64,
            }
            if kwargs["stage"] == "wgs_align"
            and set(kwargs["completed_stages"]) == {"wgs_qc"}
            else {}
        ),
    )

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(
                message="执行 WGS比对",
                job_id="job-1",
                prefix="Os",
            )
        )
    )

    assert response["plan"]["stages"] == ["wgs_index", "wgs_align"]
    details = {item["id"]: item["params"] for item in response["plan"]["stages_detail"]}
    assert "wgsQcManifest" not in details["wgs_index"]
    assert details["wgs_align"]["wgsQcManifest"].endswith(
        "/Os.wgs_qc.manifest.json"
    )
    assert details["wgs_align"]["wgsSampleSetId"].startswith("wgs-samples:")
    assert "wgsIndexManifest" not in details["wgs_align"]
    assert "wgsQcManifest" not in response["plan"]["params"]


def test_agent_plan_keeps_completed_downstream_stages_when_upstreams_were_deleted(
    monkeypatch,
) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools import stage_status

    class Status:
        def __init__(self, completed: bool):
            self.status = "success" if completed else "not_found"
            self.completed = completed
            self.manifest_path = None
            self.missing_artifacts = []
            self.errors = []

    monkeypatch.setattr(
        web_main,
        "_load_methodology_plan_for_agent_request",
        lambda _request: (None, None, None),
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_manifest_identity_for_plan",
        lambda *_args: {
            "wgsSampleSetId": "samples",
            "wgsReferenceId": "reference",
            "wgsAnalysisId": "analysis",
        },
    )
    missing = {
        "wgs_qc", "wgs_index", "wgs_align", "wgs_markdup",
        "wgs_maf_missing_filter",
    }

    def fake_status(**kwargs):
        completed = (
            kwargs["stage"] not in missing
            and kwargs.get("validate_wgs_upstream_authority") is False
        )
        return Status(completed)

    monkeypatch.setattr(stage_status, "get_stage_status", fake_status)

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(
                message="执行 WGS 全流程分析",
                job_id="job-1",
                prefix="Os",
            )
        )
    )

    assert response["success"] is True
    assert response["plan"]["stages"] == [
        "wgs_qc", "wgs_index", "wgs_align", "wgs_markdup",
        "wgs_maf_missing_filter",
    ]


@pytest.mark.parametrize(
    ("message", "target"),
    [
        ("执行 WGS BAM质控", "wgs_bam_qc"),
        ("执行 WGS变异过滤", "wgs_variant_filter"),
    ],
)
def test_agent_plan_limits_every_status_set_to_requested_wgs_closure(
    monkeypatch, message: str, target: str
) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools import stage_status

    class Status:
        def __init__(self, stage: str):
            self.status = "success" if stage == "wgs_markdup" else "not_found"
            self.completed = stage == "wgs_markdup"
            self.manifest_path = "/final/Os.wgs_markdup.manifest.json" if self.completed else None
            self.missing_artifacts = []
            self.errors = []

    monkeypatch.setattr(
        web_main, "_load_methodology_plan_for_agent_request", lambda _request: (None, None, None)
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_manifest_identity_for_plan",
        lambda *_args: {
            "wgsSampleSetId": "samples",
            "wgsReferenceId": "reference",
            "wgsAnalysisId": "analysis",
        },
    )
    monkeypatch.setattr(
        stage_status, "get_stage_status", lambda **kwargs: Status(kwargs["stage"])
    )

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(message=message, job_id="job-1", prefix="Os")
        )
    )
    closure = complete_dependencies([target])

    assert response["plan"]["stages"] == [
        stage for stage in closure if stage != "wgs_markdup"
    ]
    assert [item["id"] for item in response["plan"]["stages_detail"]] == response["plan"]["stages"]
    assert response["completed_stages"] == ["wgs_markdup"]
    assert list(response["plan"]["stage_status_summary"]) == closure


def test_agent_plan_rejects_multiple_manifest_prefixes_instead_of_using_cohort(
    tmp_path, monkeypatch
) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools import stage_status

    final_dir = tmp_path / "job-1" / "final"
    final_dir.mkdir(parents=True)
    for prefix in ("Os", "Rice"):
        (final_dir / f"{prefix}.wgs_qc.manifest.json").write_text(
            json.dumps({"stage": "wgs_qc", "status": "success", "prefix": prefix}),
            encoding="utf-8",
        )

    class Status:
        status = "not_found"
        completed = False
        manifest_path = None
        missing_artifacts = []
        errors = []

    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path)
    monkeypatch.setattr(
        web_main, "_load_methodology_plan_for_agent_request", lambda _request: (None, None, None)
    )
    monkeypatch.setattr(stage_status, "get_stage_status", lambda **_kwargs: Status())

    response = asyncio.run(
        web_main.agent_plan(web_main.AgentPlanRequest(message="执行 WGS质控", job_id="job-1"))
    )

    assert response["success"] is False
    assert "prefix" in response.get("message", "").lower()


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

    final_dir = tmp_path / "job-1" / "final"
    final_dir.mkdir(parents=True)
    (final_dir / "Os.wgs_qc.manifest.json").write_text(
        json.dumps({"stage": "wgs_qc", "status": "success", "prefix": "Os"}),
        encoding="utf-8",
    )
    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path)

    assert web_main._wgs_prefix_for_plan(
        "job-1", ["wgs_qc", "wgs_index", "wgs_align"], "At"
    ) == "Os"


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

    final_dir = tmp_path / "job-1" / "final"
    final_dir.mkdir(parents=True)
    for suffix, identity in (("", "analysis-1"), (".copy", "analysis-2")):
        (final_dir / f"Os.wgs_qc{suffix}.manifest.json").write_text(
            json.dumps({
                "stage": "wgs_qc",
                "status": "success",
                "prefix": "Os",
                "wgs_sample_set_id": "samples",
                "wgs_analysis_id": identity,
            }),
            encoding="utf-8",
        )

    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path)
    monkeypatch.setattr(
        web_main, "_load_methodology_plan_for_agent_request", lambda _request: (None, None, None)
    )

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(message="执行 WGS质控", job_id="job-1", prefix="Os")
        )
    )

    assert response["success"] is False
    assert "identity" in response.get("message", "").lower()


@pytest.mark.parametrize(
    ("producer", "message", "target"),
    [
        ("wgs_variant_select", "执行 WGS变异过滤", "wgs_variant_filter"),
        ("wgs_variant_filter", "执行 WGS基因型填充", "wgs_impute"),
    ],
)
def test_agent_plan_propagates_bcftools_method_from_completed_direct_authority(
    tmp_path, monkeypatch, producer: str, message: str, target: str
) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools import stage_status

    manifest = tmp_path / "job-1" / "final" / f"Cohort.{producer}.bcftools.manifest.json"
    manifest.parent.mkdir(parents=True)
    manifest.write_text(
        json.dumps({
            "stage": producer,
            "status": "success",
            "prefix": "Cohort",
            "method": "bcftools",
            "parameters": {"method": "bcftools"},
        }),
        encoding="utf-8",
    )

    class Status:
        def __init__(self, stage: str):
            self.status = "not_found" if stage == target else "success"
            self.completed = stage != target
            self.manifest_path = manifest if stage == producer else None
            self.missing_artifacts = []
            self.errors = []

    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path)
    monkeypatch.setattr(
        web_main, "_load_methodology_plan_for_agent_request", lambda _request: (None, None, None)
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_manifest_identity_for_plan",
        lambda *_args: {
            "wgsSampleSetId": "samples",
            "wgsReferenceId": "reference",
            "wgsAnalysisId": "analysis",
        },
    )
    monkeypatch.setattr(
        stage_status, "get_stage_status", lambda **kwargs: Status(kwargs["stage"])
    )

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(message=message, job_id="job-1", prefix="Cohort")
        )
    )

    assert response["plan"]["stages"] == [target]
    assert response["plan"]["params"]["wgsMethod"] == "bcftools"
    assert response["plan"]["stages_detail"][0]["params"]["wgsMethod"] == "bcftools"


def test_agent_plan_rejects_explicit_method_conflicting_with_completed_authority(
    tmp_path, monkeypatch
) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools import stage_status

    manifest = tmp_path / "job-1" / "final" / "Cohort.wgs_variant_select.bcftools.manifest.json"
    manifest.parent.mkdir(parents=True)
    manifest.write_text(
        json.dumps({
            "stage": "wgs_variant_select",
            "status": "success",
            "prefix": "Cohort",
            "method": "bcftools",
        }),
        encoding="utf-8",
    )

    class Status:
        def __init__(self, stage: str):
            self.status = "not_found" if stage == "wgs_variant_filter" else "success"
            self.completed = stage != "wgs_variant_filter"
            self.manifest_path = manifest if stage == "wgs_variant_select" else None
            self.missing_artifacts = []
            self.errors = []

    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", tmp_path)
    monkeypatch.setattr(
        web_main, "_load_methodology_plan_for_agent_request", lambda _request: (None, None, None)
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_manifest_identity_for_plan",
        lambda *_args: {
            "wgsSampleSetId": "samples",
            "wgsReferenceId": "reference",
            "wgsAnalysisId": "analysis",
        },
    )
    monkeypatch.setattr(
        stage_status, "get_stage_status", lambda **kwargs: Status(kwargs["stage"])
    )

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(
                message="执行 WGS变异过滤",
                job_id="job-1",
                prefix="Cohort",
                intent={"params": {"wgsMethod": "gatk"}},
            )
        )
    )

    assert response["success"] is False
    assert "method conflict" in response["message"].lower()


def test_agent_plan_does_not_inherit_stale_authority_params(monkeypatch) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools import stage_status

    class Status:
        def __init__(self, stage: str):
            self.status = "not_found" if stage == "wgs_variant_filter" else "success"
            self.completed = stage != "wgs_variant_filter"
            self.manifest_path = None
            self.missing_artifacts = []
            self.errors = []

    monkeypatch.setattr(
        web_main, "_load_methodology_plan_for_agent_request", lambda _request: (None, None, None)
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_manifest_identity_for_plan",
        lambda *_args: {
            "wgsSampleSetId": "samples",
            "wgsReferenceId": "reference",
            "wgsAnalysisId": "analysis",
        },
    )
    monkeypatch.setattr(
        stage_status, "get_stage_status", lambda **kwargs: Status(kwargs["stage"])
    )

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(
                message="执行 WGS变异过滤",
                job_id="job-1",
                prefix="Os",
                intent={
                    "params": {
                        "wgsQcManifest": "/stale/Os.wgs_qc.manifest.json",
                        "wgsQcAnalysisId": "stale-analysis",
                    }
                },
            )
        )
    )

    assert "wgsQcManifest" not in response["plan"]["params"]
    assert "wgsQcAnalysisId" not in response["plan"]["params"]


def test_agent_plan_uses_selected_job_and_manifest_prefix_when_request_omits_both(monkeypatch) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools import stage_status

    class Status:
        def __init__(self, completed: bool):
            self.status = "success" if completed else "not_found"
            self.completed = completed
            self.manifest_path = None
            self.missing_artifacts = []
            self.errors = []

    monkeypatch.setattr(web_main, "current_job_id", "job-selected")
    monkeypatch.setattr(
        web_main,
        "_load_methodology_plan_for_agent_request",
        lambda _request: (None, None, None),
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_prefix_for_plan",
        lambda job_id, _stages, _prefix: "Os" if job_id == "job-selected" else "",
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_manifest_identity_for_plan",
        lambda *_args: {
            "wgsSampleSetId": "samples",
            "wgsReferenceId": "reference",
            "wgsAnalysisId": "analysis",
        },
    )
    seen = []

    def fake_status(**kwargs):
        seen.append(kwargs)
        return Status(kwargs["stage"] != "wgs_maf_missing_filter")

    monkeypatch.setattr(stage_status, "get_stage_status", fake_status)

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(message="执行 WGS MAF缺失率过滤")
        )
    )

    assert response["mode"] == "deterministic_wgs_plan"
    assert response["plan"]["prefix"] == "Os"
    assert response["plan"]["stages"] == ["wgs_maf_missing_filter"]
    assert seen and all(item["job_id"] == "job-selected" for item in seen)
    assert all(item["prefix"] == "Os" for item in seen)


def test_agent_plan_prefers_current_request_prefix_over_stale_chat_intent(monkeypatch) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools import stage_status

    class Status:
        status = "not_found"
        completed = False
        manifest_path = None
        missing_artifacts = []
        errors = []

    monkeypatch.setattr(
        web_main,
        "_load_methodology_plan_for_agent_request",
        lambda _request: (None, None, None),
    )
    seen_prefixes = []

    def capture_prefix(_job_id, _stages, requested_prefix):
        seen_prefixes.append(requested_prefix)
        return requested_prefix

    monkeypatch.setattr(web_main, "_wgs_prefix_for_plan", capture_prefix)
    monkeypatch.setattr(web_main, "_wgs_manifest_identity_for_plan", lambda *_args: {})
    monkeypatch.setattr(stage_status, "get_stage_status", lambda **_kwargs: Status())

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(
                message="执行 WGS比对",
                job_id="job-selected",
                prefix="Os",
                intent={
                    "stages": ["wgs_align"],
                    "taskType": "pipeline",
                    "prefix": "At",
                    "params": {"prefix": "At"},
                },
            )
        )
    )

    assert response["mode"] == "deterministic_wgs_plan"
    assert response["plan"]["prefix"] == "Os"
    assert seen_prefixes == ["Os"]


def test_agent_plan_recovers_wgs_job_id_from_chat_intent(monkeypatch) -> None:
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools import stage_status

    class Status:
        def __init__(self, completed: bool):
            self.status = "success" if completed else "not_found"
            self.completed = completed
            self.manifest_path = None
            self.missing_artifacts = []
            self.errors = []

    monkeypatch.setattr(web_main, "current_job_id", "default_job")
    monkeypatch.setattr(
        web_main,
        "_load_methodology_plan_for_agent_request",
        lambda _request: (None, None, None),
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_prefix_for_plan",
        lambda job_id, _stages, _prefix: "Os" if job_id == "job-selected" else "",
    )
    monkeypatch.setattr(
        web_main,
        "_wgs_manifest_identity_for_plan",
        lambda *_args: {
            "wgsSampleSetId": "samples",
            "wgsReferenceId": "reference",
            "wgsAnalysisId": "analysis",
        },
    )
    missing = {
        "wgs_qc", "wgs_index", "wgs_align", "wgs_markdup",
        "wgs_maf_missing_filter",
    }
    seen = []

    def fake_status(**kwargs):
        seen.append(kwargs)
        return Status(kwargs["stage"] not in missing)

    monkeypatch.setattr(stage_status, "get_stage_status", fake_status)

    response = asyncio.run(
        web_main.agent_plan(
            web_main.AgentPlanRequest(
                message="执行 WGS全流程",
                intent={
                    "job_id": "job-selected",
                    "stages": WGS_STAGES,
                    "taskType": "pipeline",
                },
            )
        )
    )

    assert response["mode"] == "deterministic_wgs_plan"
    assert response["plan"]["prefix"] == "Os"
    assert response["plan"]["stages"] == [
        "wgs_qc", "wgs_index", "wgs_align", "wgs_markdup",
        "wgs_maf_missing_filter",
    ]
    assert seen and all(item["job_id"] == "job-selected" for item in seen)


@pytest.mark.parametrize(
    ("status", "must_plan"),
    [
        ("success", False),
        ("failed", True),
        ("partial_success", True),
        ("not_found", True),
        ("stale", True),
        ("hash_mismatch", True),
    ],
)
def test_planner_skips_only_revalidated_exact_wgs_success(
    tmp_path, monkeypatch, status: str, must_plan: bool
) -> None:
    agent = PlanningAgentV2(None, project_root=str(tmp_path))
    monkeypatch.setattr(
        agent,
        "_check_stage_status_preflight",
        lambda stage, prefix, workspace, stage_params=None: StageStatus(stage=stage, status=status),
    )
    plan = asyncio.run(
        agent._generate_plan_with_llm(
            query="WGS质控",
            intent={"stages": ["wgs_qc"], "prefix": "Cohort"},
            completed_stages=[],
            job_id="job",
            workspace=str(tmp_path / "job"),
            preflight_result={
                "stage_status": {
                    "wgs_qc": {
                        "status": "success",
                        "manifest_path": "final/Cohort.wgs_qc.manifest.json",
                    }
                }
            },
        )
    )
    assert bool(plan["stages"]) is must_plan


@pytest.mark.parametrize("stage", WGS_STAGES)
def test_planner_passes_stage_specific_exact_wgs_identity_to_status(
    tmp_path, monkeypatch, stage: str
) -> None:
    agent = PlanningAgentV2(None, project_root=str(tmp_path))
    captured = {}

    def capture_status(
        stage_name, prefix, workspace, stage_params=None
    ) -> StageStatus:
        captured[stage_name] = dict(stage_params or {})
        return StageStatus(stage=stage_name, status="failed")

    monkeypatch.setattr(agent, "_check_stage_status_preflight", capture_status)
    asyncio.run(
        agent._generate_plan_with_llm(
            query=f"execute {stage}",
            intent={
                "stages": [stage],
                "prefix": "Cohort",
                "params": {
                    "wgsSampleSetId": "sample-set",
                    "wgsReferenceId": "reference",
                    "wgsAnalysisId": "analysis",
                },
            },
            completed_stages=[],
            job_id="job",
            workspace=str(tmp_path / "job"),
        )
    )
    target_params = captured[stage]
    expected = {"wgsAnalysisId": "analysis"}
    if stage != "wgs_index":
        expected["wgsSampleSetId"] = "sample-set"
    if stage != "wgs_qc":
        expected["wgsReferenceId"] = "reference"
    assert {key: target_params.get(key) for key in expected} == expected