from __future__ import annotations

import gzip
import hashlib
import importlib
import json
import asyncio
from pathlib import Path

from gfa_core import gwas_fmt_context as gwas


def _write_vcf(path: Path, sample: str = "S1", pos: int = 10) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    content = (
        "##fileformat=VCFv4.2\n"
        f"#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\t{sample}\n"
        f"1\t{pos}\t.\tA\tG\t.\tPASS\t.\tGT\t0/1\n"
    )
    if path.name.lower().endswith(".gz"):
        with gzip.open(path, "wt", encoding="utf-8") as handle:
            handle.write(content)
    else:
        path.write_text(content, encoding="utf-8")
    return path


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _resolve(project_root: Path, job_id: str = "job_test", selected: str = "") -> dict:
    resolver = getattr(gwas, "resolve_gwas_fmt_input", None)
    assert callable(resolver), "resolve_gwas_fmt_input must be implemented"
    return resolver(
        job_id=job_id,
        prefix="cohort",
        params={"gwasVcf": selected} if selected else {},
        project_root=project_root,
    )


def _write_manifest(final_dir: Path, vcf: Path, *, status: str = "success", digest: str | None = None) -> Path:
    final_dir.mkdir(parents=True, exist_ok=True)
    manifest = {
        "schema_version": "1.0",
        "stage": "wgs_impute",
        "prefix": "cohort",
        "status": status,
        "artifacts": [{
            "role": "imputed_snp_vcf",
            "path": str(vcf),
            "sha256": digest if digest is not None else (_sha256(vcf) if vcf.exists() else ""),
        }],
    }
    path = final_dir / "cohort.wgs_impute.manifest.json"
    path.write_text(json.dumps(manifest), encoding="utf-8")
    return path


def _write_empty_vcf(path: Path, sample: str = "S1") -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    content = (
        "##fileformat=VCFv4.2\n"
        f"#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\t{sample}\n"
    )
    if path.name.lower().endswith(".gz"):
        with gzip.open(path, "wt", encoding="utf-8") as handle:
            handle.write(content)
    else:
        path.write_text(content, encoding="utf-8")
    return path


def _write_wgs_manifest(
    final_dir: Path,
    stage: str,
    artifacts: dict[str, Path],
    *,
    status: str = "success",
    digests: dict[str, str] | None = None,
) -> Path:
    final_dir.mkdir(parents=True, exist_ok=True)
    manifest = {
        "manifest_version": "1.0",
        "stage": stage,
        "prefix": "cohort",
        "status": status,
        "artifacts": {
            role: {
                "artifact": role,
                "role": role,
                "full_path": str(path),
                "sha256": (digests or {}).get(role, _sha256(path) if path.exists() else ""),
            }
            for role, path in artifacts.items()
        },
    }
    path = final_dir / f"cohort.{stage}.manifest.json"
    path.write_text(json.dumps(manifest), encoding="utf-8")
    return path


def test_no_manifest_selects_the_only_input_vcf(tmp_path: Path) -> None:
    vcf = _write_vcf(tmp_path / "input/job_test/upload.vcf.gz")

    result = _resolve(tmp_path)

    assert result["valid"] is True
    assert result["needs_confirmation"] is False
    assert result["decisions"]["external_vcf"]["path"] == str(vcf.resolve())
    assert result["decisions"]["external_vcf"]["source"] == "input"


def test_no_manifest_also_scans_current_job_final(tmp_path: Path) -> None:
    vcf = _write_vcf(tmp_path / "workspace/job_test/final/upstream.vcf")

    result = _resolve(tmp_path)

    assert result["valid"] is True
    assert result["decisions"]["external_vcf"]["path"] == str(vcf.resolve())
    assert result["decisions"]["external_vcf"]["source"] == "final"


def test_no_manifest_requires_confirmation_for_distinct_input_and_final_vcfs(tmp_path: Path) -> None:
    first = _write_vcf(tmp_path / "input/job_test/a.vcf", sample="A", pos=10)
    second = _write_vcf(tmp_path / "workspace/job_test/final/b.vcf.gz", sample="B", pos=20)

    result = _resolve(tmp_path)

    assert result["valid"] is False
    assert result["needs_confirmation"] is True
    assert result["issues"][0]["code"] == "ambiguous_external_vcf"
    assert {item["path"] for item in result["candidates"]} == {str(first.resolve()), str(second.resolve())}


def test_no_manifest_deduplicates_identical_vcf_bytes_across_roots(tmp_path: Path) -> None:
    first = _write_vcf(tmp_path / "input/job_test/a.vcf")
    second = tmp_path / "workspace/job_test/final/copy.vcf"
    second.parent.mkdir(parents=True, exist_ok=True)
    second.write_bytes(first.read_bytes())

    result = _resolve(tmp_path)

    assert result["valid"] is True
    assert len(result["candidates"]) == 1
    assert result["decisions"]["external_vcf"]["sha256"] == _sha256(first)


def test_valid_manifest_wins_over_unmanifested_final_candidates(tmp_path: Path) -> None:
    final_dir = tmp_path / "workspace/job_test/final"
    authoritative = _write_vcf(final_dir / "upstream.imputed.vcf.gz", sample="M")
    _write_manifest(final_dir, authoritative)
    _write_vcf(final_dir / "unmanifested.vcf", sample="F", pos=30)

    result = _resolve(tmp_path)

    assert result["valid"] is True
    assert result["decisions"]["external_vcf"]["path"] == str(authoritative.resolve())
    assert result["decisions"]["external_vcf"]["source"] == "manifest"
    assert {item["path"] for item in result["candidates"]} == {str(authoritative.resolve())}


def test_failed_wgs_manifest_allows_uploaded_vcf(tmp_path: Path) -> None:
    final_dir = tmp_path / "workspace/job_test/final"
    declared = _write_vcf(final_dir / "declared.vcf")
    _write_manifest(final_dir, declared, status="failed")
    _write_vcf(tmp_path / "input/job_test/upload.vcf", sample="I")
    _write_vcf(final_dir / "other.vcf", sample="F")

    upload = tmp_path / "input/job_test/upload.vcf"
    result = _resolve(tmp_path)

    assert result["valid"] is True
    assert result["needs_confirmation"] is False
    assert result["decisions"]["external_vcf"]["path"] == str(upload.resolve())


def test_manifest_hash_mismatch_never_falls_back(tmp_path: Path) -> None:
    final_dir = tmp_path / "workspace/job_test/final"
    declared = _write_vcf(final_dir / "declared.vcf")
    _write_manifest(final_dir, declared, digest="0" * 64)
    _write_vcf(tmp_path / "input/job_test/upload.vcf", sample="I")

    result = _resolve(tmp_path)

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


def test_prefers_maf_snp_manifest_and_ignores_empty_indel_artifacts(tmp_path: Path) -> None:
    final_dir = tmp_path / "workspace/job_test/final"
    pass_snp = _write_vcf(final_dir / "pass.snp.vcf.gz", sample="P", pos=10)
    empty_indel = _write_empty_vcf(final_dir / "pass.indel.vcf.gz", sample="P")
    final_snp = _write_vcf(final_dir / "final.snp.vcf.gz", sample="M", pos=20)
    _write_wgs_manifest(
        final_dir,
        "wgs_variant_filter",
        {"pass_snp_vcf": pass_snp, "pass_indel_vcf": empty_indel},
    )
    maf_manifest = _write_wgs_manifest(
        final_dir,
        "wgs_maf_missing_filter",
        {"final_snp_vcf": final_snp},
    )

    result = _resolve(tmp_path)

    assert result["valid"] is True
    assert result["needs_confirmation"] is False
    decision = result["decisions"]["external_vcf"]
    assert decision["path"] == str(final_snp.resolve())
    assert decision["source_manifest"] == str(maf_manifest)
    assert {item["path"] for item in result["candidates"]} == {str(final_snp.resolve())}


def test_falls_back_through_successful_wgs_snp_stages(tmp_path: Path) -> None:
    final_dir = tmp_path / "workspace/job_test/final"
    pass_snp = _write_vcf(final_dir / "pass.snp.vcf.gz", sample="P", pos=10)
    imputed_snp = _write_vcf(final_dir / "imputed.snp.vcf.gz", sample="I", pos=20)
    _write_wgs_manifest(final_dir, "wgs_variant_filter", {"pass_snp_vcf": pass_snp})
    _write_wgs_manifest(final_dir, "wgs_impute", {"imputed_snp_vcf": imputed_snp})

    result = _resolve(tmp_path)

    assert result["valid"] is True
    assert result["decisions"]["external_vcf"]["path"] == str(imputed_snp.resolve())


def test_distinct_uploaded_and_authoritative_wgs_vcfs_require_confirmation(tmp_path: Path) -> None:
    final_dir = tmp_path / "workspace/job_test/final"
    upload = _write_vcf(tmp_path / "input/job_test/upload.vcf", sample="U", pos=10)
    final_snp = _write_vcf(final_dir / "final.snp.vcf.gz", sample="M", pos=20)
    _write_wgs_manifest(final_dir, "wgs_maf_missing_filter", {"final_snp_vcf": final_snp})

    result = _resolve(tmp_path)

    assert result["valid"] is False
    assert result["needs_confirmation"] is True
    assert result["issues"][0]["code"] == "ambiguous_external_vcf"
    assert {item["path"] for item in result["candidates"]} == {
        str(upload.resolve()),
        str(final_snp.resolve()),
    }


def test_invalid_authoritative_wgs_snp_never_falls_back(tmp_path: Path) -> None:
    final_dir = tmp_path / "workspace/job_test/final"
    final_snp = _write_vcf(final_dir / "final.snp.vcf.gz", sample="M", pos=20)
    _write_wgs_manifest(
        final_dir,
        "wgs_maf_missing_filter",
        {"final_snp_vcf": final_snp},
        digests={"final_snp_vcf": "0" * 64},
    )
    _write_vcf(tmp_path / "input/job_test/upload.vcf", sample="U", pos=10)

    result = _resolve(tmp_path)

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


def test_scanner_excludes_gvcf_gwas_outputs_and_symlink_escape(tmp_path: Path) -> None:
    input_dir = tmp_path / "input/job_test"
    final_dir = tmp_path / "workspace/job_test/final"
    _write_vcf(input_dir / "sample.g.vcf.gz")
    _write_vcf(final_dir / "cohort.gwas_fmt.sorted.vcf.gz")
    outside = _write_vcf(tmp_path / "outside.vcf")
    input_dir.mkdir(parents=True, exist_ok=True)
    (input_dir / "escape.vcf").symlink_to(outside)

    result = _resolve(tmp_path)

    assert result["valid"] is False
    assert result["needs_confirmation"] is False
    assert result["issues"][0]["code"] == "missing_external_vcf"
    assert result["candidates"] == []


def test_selected_vcf_must_be_one_of_the_current_candidates(tmp_path: Path) -> None:
    allowed = _write_vcf(tmp_path / "input/job_test/allowed.vcf")
    outside = _write_vcf(tmp_path / "outside.vcf", sample="X")

    rejected = _resolve(tmp_path, selected=str(outside))
    accepted = _resolve(tmp_path, selected=str(allowed))

    assert rejected["valid"] is False
    assert rejected["issues"][0]["code"] == "invalid_external_vcf_selection"
    assert accepted["valid"] is True
    assert accepted["decisions"]["external_vcf"]["path"] == str(allowed.resolve())


def test_planning_input_check_resolves_single_job_vcf_without_path_prompt(tmp_path: Path) -> None:
    from gfa_core.mas.agents.planning_v2 import PlanningAgentV2

    vcf = _write_vcf(tmp_path / "input/job_test/cohort.vcf.gz")
    workspace = tmp_path / "workspace/job_test"
    workspace.mkdir(parents=True)
    agent = PlanningAgentV2.__new__(PlanningAgentV2)
    agent.project_root = tmp_path

    issues = agent._check_input_files(
        "gwas_fmt",
        tmp_path / "input/job_test",
        workspace,
        "cohort",
        context_index=None,
        planned_stages=["gwas_fmt"],
    )

    assert not any(issue.get("fix_action") == "confirm_external_vcf" for issue in issues)
    resolved = next(issue for issue in issues if issue.get("category") == "input_resolved")
    assert resolved["resolver"]["decisions"]["external_vcf"]["path"] == str(vcf.resolve())


def test_planning_stage_params_reuse_resolved_gwas_fields() -> None:
    from gfa_core.mas.agents.planning_v2 import PlanningAgentV2

    agent = PlanningAgentV2.__new__(PlanningAgentV2)
    stages = agent._generate_default_stages(
        ["gwas_fmt"],
        "cohort",
        intent_params={
            "gwasVcf": "/srv/job/input/cohort.vcf.gz",
            "gwasVcfCandidates": [{"path": "/srv/job/input/cohort.vcf.gz", "name": "cohort.vcf.gz"}],
        },
    )

    assert stages[0]["params"]["gwasVcf"] == "/srv/job/input/cohort.vcf.gz"
    assert stages[0]["params"]["gwasVcfCandidates"][0]["name"] == "cohort.vcf.gz"


def test_planning_preflight_carries_resolved_vcf_into_intent_params(tmp_path: Path, monkeypatch) -> None:
    from gfa_core.mas.agents import planning_v2

    vcf = _write_vcf(tmp_path / "input/job_test/cohort.vcf")
    agent = planning_v2.PlanningAgentV2.__new__(planning_v2.PlanningAgentV2)
    agent.project_root = tmp_path
    agent.workspace_root = tmp_path / "workspace"
    agent.input_root = tmp_path / "input"
    monkeypatch.setattr(planning_v2, "build_project_context", lambda **_kwargs: object())
    monkeypatch.setattr(
        agent,
        "_check_stage_status_preflight",
        lambda *_args, **_kwargs: planning_v2.StageStatus(stage="gwas_fmt", status="not_found"),
    )
    monkeypatch.setattr(agent, "_check_dependencies", lambda *_args, **_kwargs: [])
    monkeypatch.setattr(agent, "_check_resources", lambda: [])
    intent = {"stages": ["gwas_fmt"], "params": {}}

    result = asyncio.run(agent._run_preflight_validation(intent, "job_test", "cohort", ""))

    assert result["valid"] is True
    assert intent["params"]["gwasVcf"] == str(vcf.resolve())
    assert intent["params"]["gwasVcfCandidates"][0]["path"] == str(vcf.resolve())


def test_api_semantic_preflight_binds_single_job_vcf_server_side(tmp_path: Path, monkeypatch) -> None:
    api = importlib.import_module("apps.web.vue3.main")
    vcf = _write_vcf(tmp_path / "input/job_test/cohort.vcf.gz")
    tassel = tmp_path / "run_pipeline.pl"
    tassel.write_text("#!/usr/bin/env bash\nexit 0\n", encoding="utf-8")
    tassel.chmod(0o755)
    monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
    params = {"tasselPipeline": str(tassel)}

    result = api._api_semantic_preflight("job_test", "gwas_fmt", "cohort", params)

    assert result["valid"] is True
    assert result["decisions"]["external_vcf"]["path"] == str(vcf.resolve())
    assert params["gwasVcf"] == str(vcf.resolve())


def test_semantic_preflight_uses_deployment_tassel_for_default_request(tmp_path: Path, monkeypatch) -> None:
    vcf = _write_vcf(tmp_path / "cohort.vcf")
    tassel = tmp_path / "tassel-5-standalone/run_pipeline.pl"
    tassel.parent.mkdir(parents=True)
    tassel.write_text("#!/usr/bin/env bash\nexit 0\n", encoding="utf-8")
    tassel.chmod(0o755)
    monkeypatch.setenv("TASSEL_PIPELINE", str(tassel))

    result = gwas.semantic_external_vcf_preflight({
        "gwasVcf": str(vcf),
        "tasselPipeline": "run_pipeline.pl",
    })

    assert result["valid"] is True
    assert result["decisions"]["tassel"] == {
        "requested": "run_pipeline.pl",
        "path": str(tassel.resolve()),
        "executable": True,
    }


def test_api_preflight_passes_resolved_deployment_tassel_to_execution_params(tmp_path: Path, monkeypatch) -> None:
    api = importlib.import_module("apps.web.vue3.main")
    _write_vcf(tmp_path / "input/job_test/cohort.vcf")
    tassel = tmp_path / "tassel-5-standalone/run_pipeline.pl"
    tassel.parent.mkdir(parents=True)
    tassel.write_text("#!/usr/bin/env bash\nexit 0\n", encoding="utf-8")
    tassel.chmod(0o755)
    monkeypatch.setenv("TASSEL_PIPELINE", str(tassel))
    monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
    params = {"tasselPipeline": "run_pipeline.pl"}

    result = api._api_semantic_preflight("job_test", "gwas_fmt", "cohort", params)

    assert result["valid"] is True
    assert params["tasselPipeline"] == str(tassel.resolve())


def test_api_semantic_preflight_returns_confirmation_for_multiple_job_vcfs(tmp_path: Path, monkeypatch) -> None:
    api = importlib.import_module("apps.web.vue3.main")
    _write_vcf(tmp_path / "input/job_test/a.vcf", sample="A")
    _write_vcf(tmp_path / "workspace/job_test/final/b.vcf", sample="B")
    monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)

    result = api._api_semantic_preflight("job_test", "gwas_fmt", "cohort", {})

    assert result["valid"] is False
    assert result["needs_confirmation"] is True
    assert result["issues"][0]["code"] == "ambiguous_external_vcf"
    assert len(result["candidates"]) == 2


def test_execute_api_surfaces_the_first_gwas_preflight_root_cause(tmp_path: Path, monkeypatch) -> None:
    api = importlib.import_module("apps.web.vue3.main")
    monkeypatch.setattr(api, "PROJECT_ROOT", tmp_path)
    monkeypatch.setattr(api, "WORKSPACE_ROOT", tmp_path / "workspace")
    api.tasks.clear()
    request = api.ExecuteRequest(stage="gwas_fmt", job_id="job_missing", prefix="cohort")

    result = asyncio.run(api.execute_stage(request))

    issue_message = result["semantic_preflight"]["issues"][0]["message"]
    assert result["status"] == "preflight_failed"
    assert result["message"] == issue_message