from __future__ import annotations

import asyncio
import json
import subprocess
from pathlib import Path
from unittest.mock import Mock

import pytest
from fastapi import HTTPException
from openpyxl import Workbook

from apps.web import pipeline_backend as backend
from apps.web.vue3 import main as web_main
from gfa_core.mas.tools.stage_tools import StageToolExecutor, StageToolRegistry


def _source_block(source: str, start: str, end: str) -> str:
    start_index = source.index(start)
    return source[start_index:source.index(end, start_index)]


def _run_node(source: str) -> dict:
    completed = subprocess.run(
        ["/usr/bin/node", "-e", source],
        check=True,
        capture_output=True,
        text=True,
    )
    return json.loads(completed.stdout)


def _ppi_scope_js() -> str:
    return """
const uniqueList = value => [...new Set((Array.isArray(value) ? value : []).filter(Boolean))];
const scopeIdForRoles = (family, targets, refs, comps, mode) => `${family}.target-${targets.join('-')}.${mode}`;
const memberSetForScope = (scopeId, route) => `${scopeId}:${route}:all_members`;
"""


@pytest.mark.parametrize(
    ("stage", "mapping_owner", "confirmed", "expected"),
    [
        ("ppi", "", True, "string_taxon_3562"),
        ("kaks", "ppi", True, "string_taxon_3562"),
        ("kaks", "kaks", True, "At"),
        ("ppi", "", False, "soe"),
    ],
)
def test_technical_reference_bundle_lookup_uses_storage_identity_only_for_ppi_owned_reference(
    stage: str,
    mapping_owner: str,
    confirmed: bool,
    expected: str,
) -> None:
    request = web_main.ExecuteRequest(
        stage=stage,
        ppiReferenceSpecies="soe",
        ppiReferenceStorageKey="string_taxon_3562",
        ppiReferenceUserConfirmed=confirmed,
        orthologMappingOwner=mapping_owner,
    )

    assert web_main._technical_reference_bundle_lookup_code(request, "soe" if stage == "ppi" else "At") == expected


def test_stage_tool_executor_forwards_kaks_ortholog_mapping_owner(
    tmp_path: Path,
    monkeypatch,
) -> None:
    captured: dict = {}

    def fake_run_stage(**kwargs):
        captured.update(kwargs)
        manifest = Path(kwargs["job"].workspace_dir) / "final" / "Ft.SPL.kaks.manifest.json"
        manifest.parent.mkdir(parents=True, exist_ok=True)
        manifest.write_text(
            json.dumps({"manifest_version": "1.0", "stage": "kaks", "status": "success", "prefix": "Ft.SPL"}),
            encoding="utf-8",
        )
        return "KaKs complete", manifest

    monkeypatch.setattr(backend, "run_stage", fake_run_stage)
    tool = StageToolRegistry().get("kaks")
    result = asyncio.run(
        StageToolExecutor(project_root=tmp_path).execute(
            tool,
            "job_owner_forwarding",
            "Ft.SPL",
            extra_params={"orthologMappingOwner": "ppi"},
        )
    )

    assert result["status"] == "success"
    assert captured["ortholog_mapping_owner"] == "ppi"


def test_confirmed_numeric_reference_uses_storage_key_for_local_id_map(
    tmp_path: Path,
    monkeypatch,
) -> None:
    input_dir = tmp_path / "input"
    workspace = tmp_path / "workspace"
    final_dir = workspace / "final"
    logs_dir = workspace / "logs"
    input_dir.mkdir()
    final_dir.mkdir(parents=True)
    logs_dir.mkdir(parents=True)
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = "config"
    sheet.append(["Species_Code", "Species_Name", "TaxID"])
    sheet.append(["Sv", "Setaria viridis", "4556"])
    workbook.save(input_dir / "config.xlsx")

    storage_dir = input_dir / "string_taxon_4555"
    storage_dir.mkdir()
    expected_id_map = storage_dir / "string_taxon_4555.pep_cds_idmap.tsv"
    expected_id_map.write_text(
        "species_code\tprotein_id\nstring_taxon_4555\tSeita.1\n",
        encoding="utf-8",
    )
    (storage_dir / "string_taxon_4555.pep.fasta").write_text(
        ">Seita.1\nMA\n",
        encoding="utf-8",
    )
    (final_dir / "Sv.HKL.final_members.id").write_text("SvGene1\n", encoding="utf-8")
    (final_dir / "Sv.HKL.orthologs.txt").write_text(
        "SvGene1\tSeita.1\t91\tSv\tstring_taxon_4555\n",
        encoding="utf-8",
    )
    scope_file = final_dir / "Sv.HKL.target-ref.scope.json"
    scope_file.write_text(
        json.dumps(
            {
                "family": "HKL",
                "prefix": "Sv.HKL",
                "scope_id": "HKL.target-Sv.ref-string_taxon_4555.target_plus_reference",
                "scope_mode": "target_plus_reference",
                "target_species": ["Sv"],
                "reference_species": ["string_taxon_4555"],
                "selected_species": ["Sv", "string_taxon_4555"],
            }
        ),
        encoding="utf-8",
    )
    job = backend.JobInfo(
        job_id="job_task5_storage",
        input_dir=str(input_dir),
        workspace_dir=str(workspace),
        logs_dir=str(logs_dir),
        created_at="",
    )
    launched: list[list[str]] = []
    monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
    monkeypatch.setattr(
        backend,
        "run_cmd_logged",
        lambda cmd, *_args, **_kwargs: launched.append(list(cmd)) or 1,
    )

    message, _manifest = backend.run_stage(
        "ppi",
        job,
        "Sv.HKL",
        scope_file=str(scope_file),
        ppi_species="4555",
        normalization_species_label="sit",
    )

    wrong_id_map = input_dir / "4555" / "4555.pep_cds_idmap.tsv"
    assert str(wrong_id_map) not in message
    assert launched
    command = launched[0]
    assert command[command.index("--species") + 1] == "4555"
    assert command[command.index("--id-map") + 1] == str(expected_id_map)


def test_ppi_owner_fields_survive_execute_and_worker_transport() -> None:
    from apps.web.vue3.main import ExecuteRequest
    from tasks import normalize_stage_params

    request = ExecuteRequest(
        stage="ppi",
        job_id="job_task5_transport",
        prefix="Sv.HKL",
        ppiReferenceTaxonId="4555",
        ppiReferenceStorageKey="string_taxon_4555",
        ppiReferenceSelectionMode="tree_nearest",
        ppiReferenceUserConfirmed=True,
        dependencyOwner="ppi",
    )
    params, _profile = normalize_stage_params("ppi", request.dict())

    assert params["ppiReferenceTaxonId"] == "4555"
    assert params["ppiReferenceStorageKey"] == "string_taxon_4555"
    assert params["ppiReferenceSelectionMode"] == "tree_nearest"
    assert params["ppiReferenceUserConfirmed"] is True
    assert params["dependencyOwner"] == "ppi"


def test_execute_scope_uses_confirmed_ppi_owner_storage_without_biological_scope_pollution() -> None:
    request = web_main.ExecuteRequest(
        stage="ortholog",
        job_id="job_owner_scope",
        prefix="Sv.HKL",
        scopeMode="target_only",
        analysisRoute="multi_species",
        memberSetId="HKL.target-Sv.target_only:multi_species:all_members",
        targetSpecies=["Sv"],
        ppiReferenceTaxonId="4555",
        ppiReferenceStorageKey="string_taxon_4555",
        ppiReferenceUserConfirmed=True,
        dependencyOwner="ppi",
    )
    scope_params = {
        "scopeMode": "target_only", "targetSpecies": ["Sv"],
        "referenceSpecies": [], "refSpeciesList": [],
    }

    technical_refs, ppi_owned_ortholog, needs_fallback = (
        web_main._prepare_execute_technical_reference_scope(request, scope_params)
    )

    assert technical_refs == ["string_taxon_4555"]
    assert ppi_owned_ortholog is True
    assert needs_fallback is False
    assert request.analysisRoute == "single_species"
    assert request.memberSetId == ""
    assert scope_params == {
        "scopeMode": "target_only", "targetSpecies": ["Sv"],
        "referenceSpecies": [], "refSpeciesList": [],
        "comparativeSpecies": [], "scopeFile": "",
    }


def test_execute_scope_normalizes_ppi_consumer_to_target_only() -> None:
    request = web_main.ExecuteRequest(
        stage="ppi", job_id="job_ppi_consumer", prefix="Ft.SPL",
        scopeFile="/stale/project.scope.json",
        scopeMode="target_plus_reference", targetSpecies=["Ft"],
        analysisRoute="multi_species",
        memberSetId="SPL.target-Ft.target-reference-At:multi_species:all_members",
        referenceSpecies=["At"], refSpeciesList=["At"], comparativeSpecies=["Os"],
        ppiReferenceTaxonId="3562",
        ppiReferenceStorageKey="string_taxon_3562",
        ppiReferenceUserConfirmed=True,
        dependencyOwner="ppi",
    )
    scope_params = {
        "scopeFile": request.scopeFile,
        "scopeMode": request.scopeMode,
        "targetSpecies": request.targetSpecies,
        "referenceSpecies": request.referenceSpecies,
        "refSpeciesList": request.refSpeciesList,
        "comparativeSpecies": request.comparativeSpecies,
    }

    technical_refs, ppi_owned_ortholog, needs_fallback = (
        web_main._prepare_execute_technical_reference_scope(request, scope_params)
    )

    assert technical_refs == []
    assert ppi_owned_ortholog is False
    assert needs_fallback is False
    assert request.analysisRoute == "single_species"
    assert request.memberSetId == ""
    assert scope_params == {
        "scopeFile": "", "scopeMode": "target_only", "targetSpecies": ["Ft"],
        "referenceSpecies": [], "refSpeciesList": [], "comparativeSpecies": [],
    }


def test_execute_scope_keeps_ordinary_ortholog_reference_fallback_enabled() -> None:
    request = web_main.ExecuteRequest(
        stage="ortholog", job_id="job_ordinary_scope", prefix="Sv.HKL",
        scopeMode="target_only", targetSpecies=["Sv"],
    )

    scope_params = {
        "scopeMode": "target_only", "targetSpecies": ["Sv"],
        "referenceSpecies": [], "refSpeciesList": [],
    }
    technical_refs, ppi_owned_ortholog, needs_fallback = (
        web_main._prepare_execute_technical_reference_scope(request, scope_params)
    )

    assert technical_refs == []
    assert ppi_owned_ortholog is False
    assert needs_fallback is True
    assert scope_params["scopeMode"] == "target_only"


def test_execute_scope_kaks_ignores_ppi_owner_only_reference_fields() -> None:
    request = web_main.ExecuteRequest(
        stage="kaks",
        job_id="job_kaks_scope",
        prefix="Sv.HKL",
        ppiReferenceTaxonId="4555",
        ppiReferenceStorageKey="string_taxon_4555",
        ppiReferenceUserConfirmed=True,
        dependencyOwner="ppi",
    )

    scope_params = {
        "scopeMode": "target_plus_reference", "targetSpecies": ["Sv"],
        "referenceSpecies": [], "refSpeciesList": [],
    }
    technical_refs, ppi_owned_ortholog, needs_fallback = (
        web_main._prepare_execute_technical_reference_scope(request, scope_params)
    )

    assert technical_refs == []
    assert ppi_owned_ortholog is False
    assert needs_fallback is True


@pytest.mark.parametrize(
    ("confirmed", "storage_key"),
    [(False, "string_taxon_4555"), (True, "")],
)
def test_execute_rejects_incomplete_ppi_owned_ortholog_before_scope_or_enqueue(
    monkeypatch, confirmed: bool, storage_key: str,
) -> None:
    scope_called = False

    def fail_if_scope_is_built(**_kwargs):
        nonlocal scope_called
        scope_called = True
        raise ValueError("scope builder must not run")

    monkeypatch.setattr(web_main, "_ensure_gene_family_scope_file", fail_if_scope_is_built)
    request = web_main.ExecuteRequest(
        stage="ortholog", job_id="job_incomplete_owner", prefix="Sv.HKL",
        scopeMode="target_plus_reference", targetSpecies=["Sv"],
        referenceSpecies=["At"], refSpeciesList=["At"],
        ppiKaksReferenceSpecies=["At"], ppiReferenceSpecies="At",
        referenceBundleSpecies="At", ppiReferenceTaxonId="4555",
        ppiReferenceStorageKey=storage_key,
        ppiReferenceUserConfirmed=confirmed, dependencyOwner="ppi",
    )

    with pytest.raises(HTTPException) as exc_info:
        asyncio.run(web_main.execute_stage(request))

    assert exc_info.value.status_code == 400
    assert exc_info.value.detail["code"] == "ppi_owner_reference_incomplete"
    assert scope_called is False


def test_execute_normalizes_polluted_confirmed_ppi_owner_before_scope_creation(monkeypatch) -> None:
    captured = {}

    def capture_scope(**kwargs):
        captured.update(kwargs["params"])
        raise ValueError("captured normalized scope")

    monkeypatch.setattr(web_main, "_ensure_gene_family_scope_file", capture_scope)
    request = web_main.ExecuteRequest(
        stage="ortholog", job_id="job_confirmed_owner", prefix="Sv.HKL",
        scopeMode="target_plus_reference", targetSpecies=["Sv"],
        referenceSpecies=["At"], refSpeciesList=["At"],
        ppiKaksReferenceSpecies=["At"], ppiReferenceSpecies="At",
        referenceBundleSpecies="At", ppiReferenceTaxonId="4555",
        ppiReferenceStorageKey="string_taxon_4555",
        ppiReferenceUserConfirmed=True, dependencyOwner="ppi",
    )

    with pytest.raises(HTTPException):
        asyncio.run(web_main.execute_stage(request))

    assert captured["scopeMode"] == "target_only"
    assert captured["targetSpecies"] == ["Sv"]
    assert captured["referenceSpecies"] == []
    assert captured["refSpeciesList"] == []
    assert [*captured["targetSpecies"], *captured["referenceSpecies"]] == ["Sv"]


def test_execute_preserves_valid_target_only_scope_identity_for_confirmed_ppi_owner(
    tmp_path: Path,
    monkeypatch,
) -> None:
    from gfa_core.mas import resource_governor
    from tasks import execute_single_stage_task

    scope_file = tmp_path / "Sv.HKL.biological.scope.json"
    scope_file.write_text(
        json.dumps(
            {
                "family": "HKL",
                "prefix": "Sv.HKL",
                "scope_id": "fixed-biological-target-only-scope",
                "scope_mode": "target_only",
                "target_species": ["Sv"],
                "reference_species": [],
                "comparative_species": ["Os"],
                "selected_species": ["Sv"],
                "analysis": {"Family": "HKL", "context_marker": "preserve-me"},
            }
        ),
        encoding="utf-8",
    )
    scope_called = False

    def fail_if_scope_is_rebuilt(**_kwargs):
        nonlocal scope_called
        scope_called = True
        raise AssertionError("a validated existing owner scope must be reused")

    monkeypatch.setattr(web_main, "_ensure_gene_family_scope_file", fail_if_scope_is_rebuilt)
    isolated_apply_async = Mock()
    isolated_apply_async.return_value.id = "isolated-valid-owner-scope"
    monkeypatch.setattr(execute_single_stage_task, "apply_async", isolated_apply_async)
    monkeypatch.setattr(web_main, "_ensure_celery_queue_health", lambda *_args, **_kwargs: ({"available": True}, None))
    monkeypatch.setattr(resource_governor, "record_task", lambda **_kwargs: {})
    request = web_main.ExecuteRequest(
        stage="ortholog", job_id="job_owner_existing_scope", prefix="Sv.HKL",
        scopeFile=str(scope_file), scopeMode="target_plus_reference",
        targetSpecies=["WrongTarget"], referenceSpecies=["At"],
        refSpeciesList=["At"], comparativeSpecies=["Zm"],
        ppiKaksReferenceSpecies=["At"], ppiReferenceSpecies="At",
        referenceBundleSpecies="At", ppiReferenceTaxonId="4555",
        ppiReferenceStorageKey="string_taxon_4555",
        ppiReferenceSelectionMode="tree_nearest",
        ppiReferenceUserConfirmed=True, dependencyOwner="ppi",
    )

    asyncio.run(web_main.execute_stage(request))

    assert scope_called is False
    submitted_params = isolated_apply_async.call_args.kwargs["kwargs"]["params"]
    assert submitted_params["scopeFile"] == str(scope_file)
    assert submitted_params["stageContext"]["scope_id"] == "fixed-biological-target-only-scope"
    assert submitted_params["scopeMode"] == "target_only"
    assert submitted_params["analysisRoute"] == "single_species"
    assert submitted_params["memberSetId"] == "fixed-biological-target-only-scope:single_species:all_members"
    assert submitted_params["targetSpecies"] == ["Sv"]
    assert submitted_params["referenceSpecies"] == []
    assert submitted_params["refSpeciesList"] == []
    assert submitted_params["comparativeSpecies"] == ["Os"]
    assert submitted_params["ppiReferenceTaxonId"] == "4555"
    assert submitted_params["ppiReferenceStorageKey"] == "string_taxon_4555"
    assert submitted_params["ppiReferenceSelectionMode"] == "tree_nearest"
    assert submitted_params["ppiReferenceUserConfirmed"] is True
    assert submitted_params["dependencyOwner"] == "ppi"
    assert submitted_params["ppiKaksReferenceSpecies"] == []
    assert submitted_params["ppiReferenceSpecies"] == ""
    assert submitted_params["referenceBundleSpecies"] == ""


@pytest.mark.parametrize(
    ("scope_payload", "expected_code"),
    [
        ("{not-json", "ppi_owner_scope_invalid"),
        (
            json.dumps(
                {
                    "family": "HKL",
                    "scope_id": "wrong-mode-scope",
                    "scope_mode": "target_plus_reference",
                    "target_species": ["Sv"],
                    "reference_species": ["At"],
                    "selected_species": ["Sv", "At"],
                }
            ),
            "ppi_owner_scope_mode_incompatible",
        ),
    ],
)
def test_execute_fails_closed_for_invalid_ppi_owner_scope_file(
    tmp_path: Path,
    monkeypatch,
    scope_payload: str,
    expected_code: str,
) -> None:
    from tasks import execute_single_stage_task

    scope_file = tmp_path / "owner.scope.json"
    scope_file.write_text(scope_payload, encoding="utf-8")
    scope_called = False

    def fail_if_scope_is_built(**_kwargs):
        nonlocal scope_called
        scope_called = True
        raise AssertionError("invalid owner scope must fail before scope construction")

    monkeypatch.setattr(web_main, "_ensure_gene_family_scope_file", fail_if_scope_is_built)
    isolated_apply_async = Mock()
    isolated_apply_async.return_value.id = "must-not-submit-invalid-owner-scope"
    monkeypatch.setattr(execute_single_stage_task, "apply_async", isolated_apply_async)
    request = web_main.ExecuteRequest(
        stage="ortholog", job_id="job_owner_invalid_scope", prefix="Sv.HKL",
        scopeFile=str(scope_file), targetSpecies=["Sv"],
        ppiReferenceTaxonId="4555", ppiReferenceStorageKey="string_taxon_4555",
        ppiReferenceUserConfirmed=True, dependencyOwner="ppi",
    )

    with pytest.raises(HTTPException) as exc_info:
        asyncio.run(web_main.execute_stage(request))

    assert exc_info.value.status_code == 400
    assert exc_info.value.detail["code"] == expected_code
    assert scope_called is False
    isolated_apply_async.assert_not_called()


def test_ppi_owner_fields_are_forwarded_without_becoming_global_reference_scope() -> None:
    main_source = Path("apps/web/vue3/main.py").read_text(encoding="utf-8")
    stage_tools_source = Path("gfa_core/mas/tools/stage_tools.py").read_text(encoding="utf-8")
    technical_start = main_source.index("technical_reference_species: list[str] = []")
    technical_end = main_source.index("gene_family_identity_stages =", technical_start)
    technical_block = main_source[technical_start:technical_end]

    for field in (
        "ppiReferenceTaxonId",
        "ppiReferenceStorageKey",
        "ppiReferenceSelectionMode",
        "ppiReferenceUserConfirmed",
        "dependencyOwner",
    ):
        assert f'"{field}": getattr(req' in main_source or f'"{field}": bool(getattr(req' in main_source
        assert field in stage_tools_source
        assert field not in technical_block


def test_frontend_requires_confirmation_and_keeps_ppi_owner_state_isolated() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    assert "const confirmNearestPpiReferenceCandidate = async" in html
    assert "resolution.candidates" in html
    assert "const normalizePpiReferenceCandidates =" in html
    assert 'v-for="(candidate, index) in confirmParams.ppiKaksReferenceCandidates"' in html
    assert "ppiReferenceTaxonId" in html
    assert "ppiReferenceStorageKey" in html
    assert "ppiReferenceSelectionMode" in html
    assert "ppiReferenceUserConfirmed" in html
    assert "dependencyOwner" in html
    assert "preflight.resolution?.status === 'confirmation_required'" in html
    assert "preflight = await runPpiSpeciesResolutionPreflight" in html
    assert "const applyPpiOwnerSelectionToBatch =" in html


def test_confirmation_applies_owner_then_repeats_preflight() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    owner_block = _source_block(
        html,
        "const ensurePpiKaksDependencyOwner = async",
        "const ensureOrthologProducerForPpiKaks = async",
    )

    assert "preflight.resolution?.status === 'confirmation_required'" in owner_block
    assert "await confirmNearestPpiReferenceCandidate(preflight.resolution)" in owner_block
    assert "applyPpiOwnerSelectionToBatch(ownerSelection" in owner_block
    assert owner_block.count("preflight = await runPpiSpeciesResolutionPreflight") >= 2


def test_ppi_owner_selection_never_mutates_shared_reference_scope() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    selection_block = _source_block(
        html,
        "const applyPpiOwnerSelectionToBatch =",
        "const releaseDirectTargetPpiFromOrtholog =",
    )

    for field in (
        "ppiReferenceTaxonId",
        "ppiReferenceStorageKey",
        "ppiReferenceSelectionMode",
        "ppiReferenceUserConfirmed",
        "dependencyOwner",
    ):
        assert field in selection_block
    assert "getStageId(stage) !== 'ppi'" in selection_block
    assert "ppiKaksReferenceSpecies" not in selection_block
    assert "referenceSpecies:" not in selection_block
    assert "refSpeciesList:" not in selection_block
    assert "scopeMode:" not in selection_block


def test_owner_request_reads_selection_and_allows_kaks_to_reuse_ppi_owner() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    request_block = _source_block(
        html,
        "const buildPpiKaksDependencyRequest =",
        "const attachOrthologDependencyToOwners =",
    )
    owner_block = _source_block(
        html,
        "const ensurePpiKaksDependencyOwner = async",
        "const ensureOrthologProducerForPpiKaks = async",
    )

    for field in (
        "ppiReferenceTaxonId",
        "ppiReferenceStorageKey",
        "ppiReferenceSelectionMode",
        "ppiReferenceUserConfirmed",
        "dependencyOwner",
    ):
        assert field in request_block
    assert "const requestedStages = [ownerStageId]" in owner_block
    assert "const mappingOwner = stageParams.orthologMappingOwner || ownerStageId" in request_block
    assert "dependencyOwner: mappingOwner" in request_block
    assert "ownerStageId === 'ppi'" in request_block


def test_owner_resolution_and_dependency_request_return_live_values() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    selection_block = _source_block(
        html,
        "const ppiOwnerSelectionFromResolution =",
        "const applyPpiOwnerSelectionToBatch =",
    )
    request_block = _source_block(
        html,
        "const buildPpiKaksDependencyRequest =",
        "const ppiKaksInternalOrthologProducers =",
    )

    assert "return {" in selection_block
    assert "const requestBody = {" in request_block
    assert request_block.index("const requestBody = {") < request_block.index(
        "return isPpiOwner ? canonicalizePpiOwnerBiologicalScope(requestBody) : requestBody"
    )
    assert "return {" not in request_block


def test_download_completion_reapplies_owner_without_shared_merge() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    monitor_block = _source_block(
        html,
        "const monitorPpiReferenceBundleDownload =",
        "const ppiHasNearReferenceBeforeArabidopsis =",
    )
    owner_block = _source_block(
        html,
        "const ensurePpiKaksDependencyOwner = async",
        "const ensureOrthologProducerForPpiKaks = async",
    )

    assert "applyPpiOwnerSelectionToBatch(ownerSelection" in monitor_block
    assert "recoveryStages.includes('kaks')" in monitor_block
    assert "mergePpiReferenceSpeciesIntoBatch(storageKey, ['kaks'])" in monitor_block
    assert "mergePpiReferenceSpeciesIntoBatch(storageKey, recoveryStages)" not in monitor_block
    assert "applyPpiOwnerSelectionToBatch(ownerSelection" in owner_block


def test_kaks_download_success_uses_storage_key_and_recovers_shared_ppi_pause() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    recovery_helper = _source_block(
        html,
        "const ppiKaksFallbackRecoveryOwners =",
        "const handlePpiReferenceDownloadFailure =",
    )
    monitor = _source_block(
        html,
        "const monitorPpiReferenceBundleDownload =",
        "const ppiHasNearReferenceBeforeArabidopsis =",
    )
    result = _run_node(
        recovery_helper
        + monitor
        + """
const calls = {applied: [], merged: [], orchestrated: [], resumed: []};
const batchExecuteStages = {value: [
  {id: 'ortholog'},
  {id: 'kaks', dependencies: ['ortholog']},
  {id: 'ppi', dependencies: ['ortholog']}
]};
const stageExecutionStatus = {value: {ortholog: 'paused', kaks: 'paused', ppi: 'paused'}};
const stagePauseReasons = {value: {
  ppi: {type: 'dependency_paused', reason: 'upstream_stage_paused'}
}};
const getStageId = stage => typeof stage === 'object' ? stage.id : stage;
const updateReferenceBundleDownloadProgress = () => {};
const pollReferenceBundleDownloadTask = async () => ({
  success: true,
  species_code: 'string_taxon_3562',
  species_name: 'Spinacia oleracea',
  taxon_id: '3562',
  requested_stages: ['kaks'],
  files: {
    pep: '/job/string_taxon_3562/string_taxon_3562.pep.fasta',
    cds: '/job/string_taxon_3562/string_taxon_3562.cds.fasta'
  }
});
const ppiOwnerSelectionFromResolution = resolution => ({
  taxonId: resolution.selected_taxon_id,
  storageKey: resolution.ppi_reference_storage_key,
  selectionMode: 'tree_nearest',
  userConfirmed: resolution.ppi_reference_user_confirmed
});
const applyPpiOwnerSelectionToBatch = selection => calls.applied.push(selection);
const mergePpiReferenceSpeciesIntoBatch = (code, stages) => calls.merged.push({code, stages});
const clearReferenceBundleDownloadProgress = () => {};
const addAgentMessage = () => {};
const buildPpiKaksDependencyRequest = stages => ({requestedStages: stages});
const runPpiSpeciesResolutionPreflight = async () => ({allowed: true});
const orchestratePpiOwnerAfterReferenceReady = async options => {
  calls.orchestrated.push(options.requestedStages);
  return true;
};
const resumePpiKaksPausedStages = async stages => calls.resumed.push(stages);
const waitForPpiProducerDependency = async () => true;
const ensureOrthologStageForPpiKaks = () => true;
const executePpiOwnedOrthologProducer = async () => true;
const handlePpiReferenceDownloadFailure = async () => false;

monitorPpiReferenceBundleDownload(
  'task-3562',
  {species_code: 'soe', species_name: 'Spinacia oleracea', selected_string_taxon_id: '3562'},
  ['kaks'],
  {ppi_reference_storage_key: 'string_taxon_3562'}
);
setTimeout(() => console.log(JSON.stringify(calls)), 20);
"""
    )

    assert result["merged"] == [{"code": "string_taxon_3562", "stages": ["kaks"]}]
    assert result["applied"] == [{
        "taxonId": "3562",
        "storageKey": "string_taxon_3562",
        "selectionMode": "tree_nearest",
        "userConfirmed": True,
    }]
    assert result["orchestrated"] == [["ppi"]]
    assert result["resumed"] == [["kaks"]]


def test_immediate_download_success_uses_storage_key_and_shared_pause_recovery() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    owner_block = _source_block(
        html,
        "const ensurePpiKaksDependencyOwner = async",
        "const ensureOrthologProducerForPpiKaks = async",
    )

    assert "const storageKey = result.species_code || preflight.resolution.ppi_reference_storage_key" in owner_block
    assert "const recoveryStages = ppiKaksFallbackRecoveryOwners(" in owner_block
    assert "mergePpiReferenceSpeciesIntoBatch(storageKey, ['kaks'])" in owner_block
    assert "if (recoveryStages.includes('ppi'))" in owner_block
    assert "if (recoveryStages.includes('kaks'))" in owner_block


def test_arabidopsis_fallback_is_unselected_until_explicit_confirmation() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    fallback_request = _source_block(
        html,
        "const requestPpiDownloadFallbackDecision =",
        "const clearPpiKaksReferencePause =",
    )

    assert "defaultDecision: 'skip'" in fallback_request
    assert "confirmParams.ppiKaksFallbackDecision: 'skip'" not in html
    assert "ppiKaksFallbackDecision: 'skip'" in html
    assert "confirmParams.ppiKaksFallbackDecision = 'skip'" in html


def test_arabidopsis_fallback_does_not_preselect_kaks_reference_route() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    fallback = _source_block(
        html,
        "const applyArabidopsisFallbackForPpiKaks =",
        "const resumePpiKaksPausedStages =",
    )

    assert "kaksReferenceSource: 'project'" not in fallback
    assert "referenceSpecies: ['At']" not in fallback
    assert "orthologMappingOwner: 'kaks'" not in fallback


def test_kaks_download_failure_recovers_only_ppi_paused_on_same_ortholog() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper_marker = "const ppiKaksFallbackRecoveryOwners ="
    helper = (
        _source_block(html, helper_marker, "const handlePpiReferenceDownloadFailure =")
        if helper_marker in html
        else ""
    )
    handler = _source_block(
        html,
        "const handlePpiReferenceDownloadFailure =",
        "const downloadRecommendedPpiReferenceBundle =",
    )
    result = _run_node(
        _ppi_scope_js()
        + helper
        + handler
        + """
const runCase = async ({stages, statuses, pauses}) => {
  const calls = {dialog: null, fallback: null, orchestrated: [], resumed: []};
  batchExecuteStages.value = stages;
  stageExecutionStatus.value = statuses;
  stagePauseReasons.value = pauses;
  requestPpiDownloadFallbackDecision = async (_err, _resolution, _recommendation, owners) => {
    calls.dialog = owners;
    return 'ath';
  };
  applyArabidopsisFallbackForPpiKaks = owners => { calls.fallback = owners; };
  orchestratePpiOwnerAfterReferenceReady = async options => {
    calls.orchestrated.push(options.requestedStages);
    return true;
  };
  resumePpiKaksPausedStages = async owners => { calls.resumed.push(owners); };
  await handlePpiReferenceDownloadFailure(new Error('download failed'), {}, {}, ['kaks']);
  return calls;
};

let batchExecuteStages = {value: []};
let stageExecutionStatus = {value: {}};
let stagePauseReasons = {value: {}};
let requestPpiDownloadFallbackDecision;
let applyArabidopsisFallbackForPpiKaks;
let orchestratePpiOwnerAfterReferenceReady;
let resumePpiKaksPausedStages;
const addAgentMessage = () => {};
const buildPpiKaksDependencyRequest = () => ({});
const runPpiSpeciesResolutionPreflight = async () => ({allowed: true});
const waitForPpiProducerDependency = async () => true;
const ensureOrthologStageForPpiKaks = () => true;
const getStageId = stage => typeof stage === 'object' ? stage.id : stage;
const executePpiOwnedOrthologProducer = async () => true;
const storePpiKaksReferencePause = () => {};

(async () => {
  const jointPaused = await runCase({
    stages: [
      {id: 'ortholog'},
      {id: 'kaks', dependencies: ['ortholog']},
      {id: 'ppi', dependencies: ['ortholog']}
    ],
    statuses: {ortholog: 'paused', kaks: 'paused', ppi: 'paused'},
    pauses: {ppi: {type: 'dependency_paused', reason: 'upstream_stage_paused'}}
  });
  const completedPpi = await runCase({
    stages: [{id: 'ortholog'}, {id: 'kaks'}, {id: 'ppi', dependencies: ['ortholog']}],
    statuses: {ortholog: 'paused', kaks: 'paused', ppi: 'completed'},
    pauses: {ppi: {type: 'dependency_paused', reason: 'upstream_stage_paused'}}
  });
  const kaksOnly = await runCase({
    stages: [{id: 'ortholog'}, {id: 'kaks', dependencies: ['ortholog']}],
    statuses: {ortholog: 'paused', kaks: 'paused'},
    pauses: {}
  });
  console.log(JSON.stringify({jointPaused, completedPpi, kaksOnly}));
})();
"""
    )

    assert result["jointPaused"] == {
        "dialog": ["kaks", "ppi"],
        "fallback": ["ppi"],
        "orchestrated": [["ppi"]],
        "resumed": [],
    }
    for guarded_case in (result["completedPpi"], result["kaksOnly"]):
        assert guarded_case == {
            "dialog": ["kaks"],
            "fallback": None,
            "orchestrated": [],
            "resumed": [],
        }


def test_final_ppi_request_dynamically_uses_target_only_biological_scope() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(html, "const canonicalizePpiOwnerBiologicalScope =", "const applyGeneFamilyContextToRequest =")
    result = _run_node(
        _ppi_scope_js()
        + helper
        + """
const request = {
  geneFamily: 'SPL', targetSpecies: ['Ft'], comparativeSpecies: ['Os'],
  analysisRoute: 'multi_species', memberSetId: 'stale-member', scopeId: 'stale-scope', scopeFile: '/stale/scope.json',
  referenceSpecies: ['ProjectRef'],
  refSpeciesList: ['ProjectRef'],
  scopeMode: 'target_plus_reference',
  ppiKaksReferenceSpecies: ['KaKsRef'],
  ppiReferenceSpecies: 'KaKsRef',
  referenceBundleSpecies: 'KaKsRef'
};
const batch = {
  ppiReferenceTaxonId: '4555',
  ppiReferenceStorageKey: 'string_taxon_4555',
  ppiReferenceSelectionMode: 'tree_nearest',
  ppiReferenceUserConfirmed: true,
  dependencyOwner: 'ppi'
};
console.log(JSON.stringify(applyPpiOwnerFieldsToRequest(request, 'ppi', {}, batch, {})));
"""
    )

    assert result["ppiReferenceTaxonId"] == "4555"
    assert result["ppiReferenceStorageKey"] == "string_taxon_4555"
    assert result["ppiReferenceSelectionMode"] == "tree_nearest"
    assert result["ppiReferenceUserConfirmed"] is True
    assert result["dependencyOwner"] == "ppi"
    assert result["targetSpecies"] == ["Ft"]
    assert result["referenceSpecies"] == []
    assert result["refSpeciesList"] == []
    assert result["comparativeSpecies"] == []
    assert result["scopeMode"] == "target_only"
    assert result["analysisRoute"] == "single_species"
    assert result["scopeId"] == "SPL.target-Ft.target_only"
    assert result["memberSetId"] == "SPL.target-Ft.target_only:single_species:all_members"
    assert result["scopeFile"] == ""
    for forbidden in ("ppiKaksReferenceSpecies", "ppiReferenceSpecies", "referenceBundleSpecies"):
        assert forbidden not in result


def test_joint_batch_ppi_request_dynamically_ignores_kaks_reference() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(html, "const canonicalizePpiOwnerBiologicalScope =", "const applyGeneFamilyContextToRequest =")
    result = _run_node(
        _ppi_scope_js()
        + helper
        + """
const request = { geneFamily: 'SPL', targetSpecies: ['Ft'], referenceSpecies: ['ProjectRef'], refSpeciesList: ['ProjectRef'] };
const batch = {
  ppiKaksReferenceSpecies: ['KaKsRef'],
  ppiReferenceSpecies: 'KaKsRef',
  referenceBundleSpecies: 'KaKsRef',
  ppiReferenceTaxonId: '4555',
  ppiReferenceStorageKey: 'string_taxon_4555',
  ppiReferenceSelectionMode: 'tree_nearest',
  ppiReferenceUserConfirmed: true,
  dependencyOwner: 'ppi'
};
console.log(JSON.stringify(applyPpiOwnerFieldsToRequest(request, 'ppi', {}, batch, {})));
"""
    )

    assert result["referenceSpecies"] == []
    assert "KaKsRef" not in json.dumps(result)
    assert result["ppiReferenceTaxonId"] == "4555"


def test_reference_ready_orchestration_dynamically_inserts_ortholog_before_resume() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const orchestratePpiOwnerAfterReferenceReady =",
        "const monitorPpiReferenceBundleDownload =",
    )
    result = _run_node(
        helper
        + """
(async () => {
  const calls = [];
  let preflightCount = 0;
  const ok = await orchestratePpiOwnerAfterReferenceReady({
    requestBody: { stage: 'ppi' },
    requestedStages: ['ppi'],
    runPreflight: async () => {
      calls.push('preflight');
      preflightCount += 1;
      return { allowed: true, resolution: { status: preflightCount === 1 ? 'needs_ortholog' : 'ready' } };
    },
    ensureOrtholog: async () => { calls.push('ortholog'); return true; },
    resume: async () => { calls.push('resume'); }
  });
  console.log(JSON.stringify({ ok, calls }));
})();
"""
    )

    assert result == {"ok": True, "calls": ["preflight", "ortholog", "preflight", "resume"]}


def test_ppi_owned_ortholog_uses_owner_storage_without_polluting_final_ppi_request() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    producer_helper = _source_block(
        html,
        "const buildPpiOwnedOrthologStage =",
        "const attachOrthologDependencyToOwners =",
    )
    request_helper = _source_block(html, "const canonicalizePpiOwnerBiologicalScope =", "const applyGeneFamilyContextToRequest =")
    ortholog_request_helper = _source_block(
        html,
        "const applyPpiOwnedOrthologFieldsToRequest =",
        "const canonicalizePpiOwnerBiologicalScope =",
    )
    direct_builder = _source_block(
        html,
        "const finalizeExecuteStageDirectRequest =",
        "const executeStageDirect = async",
    )
    result = _run_node(
        _ppi_scope_js()
        + producer_helper
        + ortholog_request_helper
        + request_helper
        + direct_builder
        + """
const request = {
  geneFamily: 'SPL', targetSpecies: ['Ft'],
  referenceSpecies: ['At'], refSpeciesList: ['At'], scopeMode: 'target_plus_reference',
  ppiKaksReferenceSpecies: ['KaKsRef'], ppiReferenceSpecies: 'KaKsRef', referenceBundleSpecies: 'KaKsRef'
};
const owner = {
  ppiReferenceTaxonId: '4555', ppiReferenceStorageKey: 'string_taxon_4555',
  ppiReferenceSelectionMode: 'tree_nearest', ppiReferenceUserConfirmed: true, dependencyOwner: 'ppi'
};
const ortholog = buildPpiOwnedOrthologStage(request, owner, ['ppi']);
const finalOrtholog = applyPpiOwnedOrthologFieldsToRequest(
  {...ortholog.params, ppiKaksReferenceSpecies: ['KaKsRef'], referenceSpecies: ['KaKsRef']},
  'ortholog', ortholog.params
);
const directOrtholog = finalizeExecuteStageDirectRequest(
  {stage: 'ortholog', referenceSpecies: ['KaKsRef'], ppiKaksReferenceSpecies: ['KaKsRef']},
  'ortholog', ortholog.params
);
const ppi = applyPpiOwnerFieldsToRequest({...request}, 'ppi', {}, owner, {});
console.log(JSON.stringify({ortholog, finalOrtholog, directOrtholog, ppi}));
"""
    )

    assert result["ortholog"]["params"]["ppiReferenceTaxonId"] == "4555"
    expected_owner = {
        "ppiReferenceTaxonId": "4555",
        "ppiReferenceStorageKey": "string_taxon_4555",
        "ppiReferenceSelectionMode": "tree_nearest",
        "ppiReferenceUserConfirmed": True,
        "dependencyOwner": "ppi",
    }
    for payload in (result["ortholog"]["params"], result["finalOrtholog"], result["directOrtholog"]):
        for field, expected in expected_owner.items():
            assert payload[field] == expected
        for forbidden in (
            "referenceSpecies", "refSpeciesList", "scopeMode",
            "ppiKaksReferenceSpecies", "ppiReferenceSpecies", "referenceBundleSpecies",
        ):
            assert forbidden not in payload
    assert result["directOrtholog"]["requestedStages"] == ["ppi"]
    assert result["directOrtholog"]["autoInsertedFor"] == ["ppi"]
    assert result["ppi"]["referenceSpecies"] == []
    assert result["ppi"]["refSpeciesList"] == []
    assert result["ppi"]["scopeMode"] == "target_only"
    assert result["ppi"]["scopeId"] == "SPL.target-Ft.target_only"
    for forbidden in ("ppiKaksReferenceSpecies", "ppiReferenceSpecies", "referenceBundleSpecies"):
        assert forbidden not in result["ppi"]


def test_active_scheduler_executes_inserted_ppi_owned_ortholog_before_resume() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const orchestratePpiOwnerAfterReferenceReady =",
        "const monitorPpiReferenceBundleDownload =",
    )
    result = _run_node(
        helper
        + """
(async () => {
  const calls = [];
  let preflightCount = 0;
  const producer = { id: 'ortholog', params: { ppiReferenceStorageKey: 'string_taxon_4555' } };
  const ok = await orchestratePpiOwnerAfterReferenceReady({
    requestBody: { stage: 'ppi' }, requestedStages: ['ppi'], isBatchExecuting: true,
    runPreflight: async () => {
      calls.push('preflight');
      preflightCount += 1;
      return {allowed: true, resolution: {status: preflightCount === 1 ? 'needs_ortholog' : 'ready'}};
    },
    ensureOrtholog: async () => { calls.push('ensure'); return {ready: true, producerStage: producer}; },
    executeProducer: async stage => { calls.push(`execute:${stage.id}`); return true; },
    resume: async () => { calls.push('resume'); }
  });
  console.log(JSON.stringify({ok, calls}));
})();
"""
    )

    assert result == {
        "ok": True,
        "calls": ["preflight", "ensure", "execute:ortholog", "preflight", "resume"],
    }


def test_kaks_soft_gate_releases_only_after_ppi_settles() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const ppiSettledForKaks =",
        "const executeStagesWithDependencies = async",
    )
    result = _run_node(
        helper
        + """
const stages = [{id: 'ppi'}, {id: 'kaks'}];
const getStageId = stage => typeof stage === 'object' ? stage.id : stage;
console.log(JSON.stringify({
  pending: ppiSettledForKaks(stages, {ppi: 'pending'}),
  running: ppiSettledForKaks(stages, {ppi: 'running'}),
  paused: ppiSettledForKaks(stages, {ppi: 'paused'}),
  completed: ppiSettledForKaks(stages, {ppi: 'completed'}),
  failed: ppiSettledForKaks(stages, {ppi: 'failed'}),
  noPpi: ppiSettledForKaks([{id: 'kaks'}], {})
}));
"""
    )

    assert result == {
        "pending": False,
        "running": False,
        "paused": False,
        "completed": True,
        "failed": True,
        "noPpi": True,
    }


def test_kaks_reference_choice_keeps_ppi_and_project_namespaces_separate() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const kaksReferenceChoiceParams =",
        "const confirmedStageParamsForExecution =",
    )
    result = _run_node(
        helper
        + """
console.log(JSON.stringify({
  ppi: kaksReferenceChoiceParams('ppi', 'string_taxon_3562', 'At'),
  project: kaksReferenceChoiceParams('project', 'string_taxon_3562', 'At')
}));
"""
    )

    assert result["ppi"] == {
        "kaksReferenceSource": "ppi",
        "referenceSpecies": ["string_taxon_3562"],
        "refSpeciesList": ["string_taxon_3562"],
        "ppiKaksReferenceSpecies": ["string_taxon_3562"],
        "ppiReferenceSpecies": "string_taxon_3562",
        "referenceBundleSpecies": "string_taxon_3562",
        "orthologMappingOwner": "ppi",
    }
    assert result["project"] == {
        "kaksReferenceSource": "project",
        "referenceSpecies": ["At"],
        "refSpeciesList": ["At"],
        "ppiKaksReferenceSpecies": [],
        "ppiReferenceSpecies": "",
        "referenceBundleSpecies": "",
        "orthologMappingOwner": "kaks",
    }


def test_kaks_preflight_uses_selected_ppi_mapping_owner() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const buildPpiKaksDependencyRequest =",
        "const ppiKaksInternalOrthologProducers =",
    )
    result = _run_node(
        helper
        + """
const batchExecuteParams = {value: {
  prefix: 'Ft.SPL',
  ppiReferenceTaxonId: '3562',
  ppiReferenceStorageKey: 'string_taxon_3562',
  ppiReferenceSelectionMode: 'tree_nearest',
  ppiReferenceUserConfirmed: true
}};
const batchExecuteStages = {value: [{
  id: 'kaks',
  params: {
    kaksReferenceSource: 'project',
    orthologMappingOwner: 'kaks',
    referenceSpecies: ['At'],
    refSpeciesList: ['At']
  }
}]};
const getStageId = stage => typeof stage === 'object' ? stage.id : stage;
const firstNonEmptyList = (...values) => {
  for (const value of values) {
    if (Array.isArray(value) && value.length) return [...value];
    if (typeof value === 'string' && value.trim()) return [value.trim()];
  }
  return [];
};
const params = {prefix: 'Ft.SPL', geneFamily: 'SPL'};
const confirmParams = {};
const executionPrefix = () => 'Ft.SPL';
const canonicalizePpiOwnerBiologicalScope = value => value;
const confirmedStageParams = {
  kaksReferenceSource: 'ppi',
  orthologMappingOwner: 'ppi',
  referenceSpecies: ['string_taxon_3562'],
  refSpeciesList: ['string_taxon_3562'],
  ppiKaksReferenceSpecies: ['string_taxon_3562'],
  ppiReferenceSpecies: 'string_taxon_3562',
  referenceBundleSpecies: 'string_taxon_3562'
};
console.log(JSON.stringify(
  buildPpiKaksDependencyRequest(['kaks'], confirmedStageParams)
));
"""
    )

    assert {
        "dependencyOwner": result["dependencyOwner"],
        "requestedStages": result["requestedStages"],
        "referenceSpecies": result["referenceSpecies"],
        "ppiReferenceTaxonId": result["ppiReferenceTaxonId"],
        "ppiReferenceStorageKey": result["ppiReferenceStorageKey"],
        "ppiReferenceSelectionMode": result["ppiReferenceSelectionMode"],
        "ppiReferenceUserConfirmed": result["ppiReferenceUserConfirmed"],
    } == {
        "dependencyOwner": "ppi",
        "requestedStages": ["kaks"],
        "referenceSpecies": ["string_taxon_3562"],
        "ppiReferenceTaxonId": "3562",
        "ppiReferenceStorageKey": "string_taxon_3562",
        "ppiReferenceSelectionMode": "tree_nearest",
        "ppiReferenceUserConfirmed": True,
    }


def test_kaks_project_at_route_is_not_blocked_by_ppi_near_reference_policy() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const ppiHasNearReferenceBeforeArabidopsis =",
        "const runPpiSpeciesResolutionPreflight =",
    )
    result = _run_node(
        helper
        + """
const resolution = {
  status: 'needs_ortholog',
  selected_string_species: 'ath',
  diagnostics: {
    reference_recommendations: [{species_code: 'string_taxon_3562'}]
  }
};
console.log(JSON.stringify({
  ppi: ppiNeedsNearReferenceDecision(resolution, ['ppi']),
  kaks: ppiNeedsNearReferenceDecision(resolution, ['kaks'])
}));
"""
    )

    assert result == {"ppi": True, "kaks": False}


def test_prepare_kaks_preflights_the_exact_confirmed_stage_params() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const prepareKaksSelectedReference =",
        "const submitExecuteRequestWithAutoAnnotation =",
    )
    result = _run_node(
        helper
        + """
const confirmedStageParams = {
  kaksReferenceSource: 'ppi',
  orthologMappingOwner: 'ppi',
  referenceSpecies: ['string_taxon_3562'],
  ppiKaksReferenceSpecies: ['string_taxon_3562']
};
let captured = null;
const ensurePpiKaksDependencyOwner = async (...args) => {
  captured = args;
  return true;
};
const ppiKaksDependencyPauses = () => ({});

(async () => {
  const ready = await prepareKaksSelectedReference(confirmedStageParams);
  console.log(JSON.stringify({ready, captured}));
})();
"""
    )

    assert result == {
        "ready": True,
        "captured": [
            "kaks",
            {
                "kaksReferenceSource": "ppi",
                "orthologMappingOwner": "ppi",
                "referenceSpecies": ["string_taxon_3562"],
                "ppiKaksReferenceSpecies": ["string_taxon_3562"],
            },
        ],
    }


def test_submit_preflight_maps_kaks_selected_ortholog_owner_to_dependency_owner() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    retry_helper = _source_block(
        html,
        "const fetchWithTransientRetry = async",
        "const pollReferenceBundleDownloadTask = async",
    )
    helper = _source_block(
        html,
        "const runPpiSpeciesResolutionPreflight =",
        "const ppiKaksDependencyStages =",
    )
    result = _run_node(
        retry_helper
        + helper
        + """
const jobInfo = {job_id: 'job_1786243816439_04955b'};
const executionPrefix = () => 'Ft.SPL';
let capturedPayload = null;
const fetch = async (_url, options) => {
  capturedPayload = JSON.parse(options.body);
  return {
    ok: true,
    json: async () => ({success: true, status: 'ready'})
  };
};

(async () => {
  const outcome = await runPpiSpeciesResolutionPreflight({
    stage: 'kaks',
    prefix: 'Ft.SPL',
    orthologMappingOwner: 'ppi',
    targetSpecies: ['Ft'],
    referenceSpecies: ['string_taxon_3562']
  }, {requestedStages: ['kaks']});
  console.log(JSON.stringify({
    allowed: outcome.allowed,
    dependencyOwner: capturedPayload.dependencyOwner
  }));
})();
"""
    )

    assert result == {"allowed": True, "dependencyOwner": "ppi"}


def test_at_fallback_owner_orchestration_failure_stays_paused_and_returns_false() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    handler = _source_block(
        html,
        "const handlePpiReferenceDownloadFailure =",
        "const downloadRecommendedPpiReferenceBundle =",
    )
    result = _run_node(
        handler
        + """
const batchExecuteStages = {value: [{id: 'ppi', params: {}}]};
const stageExecutionStatus = {value: {ppi: 'paused'}};
const stagePauseReasons = {value: {}};
const pauses = [];
const ppiKaksFallbackRecoveryOwners = () => ['ppi'];
const requestPpiDownloadFallbackDecision = async () => 'ath';
const applyArabidopsisFallbackForPpiKaks = () => {};
const buildPpiKaksDependencyRequest = () => ({stage: 'ppi'});
const orchestratePpiOwnerAfterReferenceReady = async () => false;
const runPpiSpeciesResolutionPreflight = async () => ({allowed: true});
const waitForPpiProducerDependency = async () => true;
const ensureOrthologStageForPpiKaks = () => true;
const getPpiKaksInternalOrthologProducer = () => null;
const executePpiOwnedOrthologProducer = async () => false;
const resumePpiKaksPausedStages = async () => {};
const addAgentMessage = () => {};
const storePpiKaksReferencePause = (owners, resolution, message) => {
  pauses.push({owners, resolution, message});
};

(async () => {
  const accepted = await handlePpiReferenceDownloadFailure(
    new Error('download failed'),
    {},
    {},
    ['ppi']
  );
  console.log(JSON.stringify({accepted, pauses}));
})();
"""
    )

    assert result["accepted"] is False
    assert result["pauses"] == [
        {
            "owners": ["ppi"],
            "resolution": {},
            "message": "拟南芥兜底参考尚未形成可用的 PPI Ortholog 映射；PPI 保持暂停。",
        }
    ]


def test_kaks_missing_reference_data_does_not_open_download_decision() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const ensurePpiKaksDependencyOwner = async",
        "const preflightPpiKaksDependencyOwners = async",
    )
    result = _run_node(
        helper
        + """
const batchExecuteStages = {value: [{id: 'kaks', params: {orthologMappingOwner: 'ppi'}}]};
const ppiKaksDependencyStages = new Set(['ppi', 'kaks']);
const getStageId = stage => typeof stage === 'object' ? stage.id : stage;
const buildPpiKaksDependencyRequest = () => ({stage: 'kaks'});
const runPpiSpeciesResolutionPreflight = async () => ({
  allowed: false,
  resolution: {status: 'needs_reference_data', issue_code: 'kaks_reference_cds_missing'}
});
const ppiNeedsNearReferenceDecision = () => false;
let decisionCalls = 0;
const requestPpiReferenceDataDecision = async () => { decisionCalls += 1; return 'skip'; };
const formatPpiSpeciesResolutionBlockMessage = () => 'reference data missing';
const storePpiKaksReferencePause = () => {};
const addAgentMessage = () => {};

(async () => {
  const ready = await ensurePpiKaksDependencyOwner('kaks');
  console.log(JSON.stringify({ready, decisionCalls}));
})();
"""
    )

    assert result == {"ready": False, "decisionCalls": 0}


def test_orchestrator_waits_for_identify_and_does_not_resume_on_producer_failure() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const orchestratePpiOwnerAfterReferenceReady =",
        "const monitorPpiReferenceBundleDownload =",
    )
    result = _run_node(
        helper
        + """
(async () => {
  const calls = [];
  let identify = 'running';
  const base = {
    requestBody: {stage: 'ppi'}, requestedStages: ['ppi'],
    runPreflight: async () => {
      calls.push('preflight');
      return {allowed: true, resolution: {status: calls.filter(item => item === 'preflight').length <= 2 ? 'needs_ortholog' : 'ready'}};
    },
    dependencyStatus: () => identify,
    waitForDependency: async () => { calls.push('wait:identify'); identify = 'completed'; return true; },
    ensureOrtholog: async () => ({ready: true, producerStage: {id: 'ortholog'}}),
    resume: async () => { calls.push('resume'); }
  };
  const failed = await orchestratePpiOwnerAfterReferenceReady({
    ...base, executeProducer: async () => { calls.push('execute:failed'); return false; }
  });
  const succeeded = await orchestratePpiOwnerAfterReferenceReady({
    ...base, executeProducer: async () => { calls.push('execute:success'); return true; }
  });
  console.log(JSON.stringify({failed, succeeded, calls}));
})();
"""
    )

    assert result == {
        "failed": False,
        "succeeded": True,
        "calls": ["preflight", "wait:identify", "execute:failed", "preflight", "execute:success", "preflight", "resume"],
    }


def test_ppi_stage_error_path_keeps_confirmed_params_in_catch_scope_and_batch_has_final_cleanup() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    stage_start = html.index("const executeStageWithConfirmation = async")
    stage_end = html.index("const NCBI_SPECIES_DEFAULTS =", stage_start)
    stage_block = html[stage_start:stage_end]
    try_index = stage_block.index("try {")
    declaration_index = stage_block.index("let confirmedStageParams")
    catch_index = stage_block.index("} catch (err)")

    assert declaration_index < try_index < catch_index
    assert "confirmedStageParams = confirmedStageParamsForExecution" in stage_block
    assert "stageExecutionStatus.value[stageId] = 'failed'" in stage_block[catch_index:]

    batch_start = html.index("const startBatchExecute = async")
    batch_end = html.index("const ppiSettledForKaks =", batch_start)
    batch_block = html[batch_start:batch_end]
    execute_index = batch_block.index("await executeStagesWithDependencies")
    finally_index = batch_block.index("finally", execute_index)
    cleanup_index = batch_block.index("isBatchExecuting.value = false", finally_index)
    assert execute_index < finally_index < cleanup_index


def test_hidden_ppi_ortholog_reports_success_to_the_user() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    block = _source_block(
        html,
        "const executePpiOwnedOrthologProducer = async",
        "const waitForPpiProducerDependency = async",
    )
    assert "专属同源映射生成完成" in block


def test_arabidopsis_fallback_keeps_ppi_paused_until_owner_ortholog_is_ready() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const applyArabidopsisFallbackForPpiKaks =",
        "const resumePpiKaksPausedStages =",
    )
    result = _run_node(
        helper
        + """
const calls = {selection: null, clear: []};
const batchExecuteStages = {value: [{id: 'ppi', params: {}}]};
const stageExecutionStatus = {value: {ppi: 'paused'}};
const getStageId = stage => typeof stage === 'object' ? stage.id : stage;
const getStageLabel = stage => stage;
const stageDependencies = {};
const applyPpiOwnerSelectionToBatch = selection => { calls.selection = selection; };
const mergePpiReferenceSpeciesIntoBatch = () => {};
const clearPpiKaksReferencePause = stages => { calls.clear.push(stages); };

applyArabidopsisFallbackForPpiKaks(['ppi'], 'download failed');
console.log(JSON.stringify({calls, status: stageExecutionStatus.value.ppi}));
"""
    )

    assert result == {
        "calls": {
            "selection": {
                "taxonId": "3702",
                "storageKey": "At",
                "selectionMode": "model_fallback",
                "userConfirmed": True,
            },
            "clear": [],
        },
        "status": "paused",
    }


def test_at_fallback_batch_owner_overrides_stale_spinach_stage_for_ppi_request() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const applyPpiOwnerFieldsToRequest =",
        "const applyGeneFamilyContextToRequest =",
    )
    result = _run_node(
        helper
        + """
const canonicalizePpiOwnerBiologicalScope = request => request;
const request = {};
applyPpiOwnerFieldsToRequest(request, 'ppi', {
  ppiReferenceTaxonId: '3562',
  ppiReferenceStorageKey: 'string_taxon_3562',
  ppiReferenceSelectionMode: 'tree_nearest',
  ppiReferenceUserConfirmed: true
}, {
  ppiReferenceTaxonId: '3702',
  ppiReferenceStorageKey: 'At',
  ppiReferenceSelectionMode: 'model_fallback',
  ppiReferenceUserConfirmed: true
}, {});
console.log(JSON.stringify(request));
"""
    )

    assert result["ppiReferenceTaxonId"] == "3702"
    assert result["ppiReferenceStorageKey"] == "At"
    assert result["ppiReferenceSelectionMode"] == "model_fallback"
    assert result["ppiReferenceUserConfirmed"] is True


def test_normal_spinach_ppi_owner_request_is_unchanged() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const applyPpiOwnerFieldsToRequest =",
        "const applyGeneFamilyContextToRequest =",
    )
    result = _run_node(
        helper
        + """
const canonicalizePpiOwnerBiologicalScope = request => request;
const request = {};
const spinach = {
  ppiReferenceTaxonId: '3562',
  ppiReferenceStorageKey: 'string_taxon_3562',
  ppiReferenceSelectionMode: 'tree_nearest',
  ppiReferenceUserConfirmed: true
};
applyPpiOwnerFieldsToRequest(request, 'ppi', spinach, spinach, {});
console.log(JSON.stringify(request));
"""
    )

    assert result["ppiReferenceTaxonId"] == "3562"
    assert result["ppiReferenceStorageKey"] == "string_taxon_3562"
    assert result["ppiReferenceSelectionMode"] == "tree_nearest"
    assert result["ppiReferenceUserConfirmed"] is True


def test_ppi_owned_ortholog_producer_is_single_flight_for_same_owner_and_reference_identity() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const executePpiOwnedOrthologProducer = async",
        "const waitForPpiProducerDependency = async",
    )
    result = _run_node(
        helper
        + """
const ppiKaksInternalOrthologStatus = new Map();
const ppiKaksInternalOrthologPromises = new Map();
const ppiKaksInternalOrthologIdentityKey = producerStage => {
  const producerParams = producerStage.params || {};
  return [
    String(producerParams.dependencyOwner || 'ppi').trim().toLowerCase(),
    String(producerParams.ppiReferenceTaxonId || '').trim(),
    String(producerParams.ppiReferenceStorageKey || '').trim()
  ].join('::');
};
const getStageId = stage => stage.id;
const getStageLabel = stage => stage;
const addAgentMessage = () => {};
let executeCalls = 0;
const executeStageDirect = async () => {
  executeCalls += 1;
  await new Promise(resolve => setTimeout(resolve, 10));
};
const producer = {id: 'ortholog', params: {
  dependencyOwner: 'ppi',
  ppiReferenceTaxonId: '3562',
  ppiReferenceStorageKey: 'string_taxon_3562'
}};

(async () => {
  const results = await Promise.all([
    executePpiOwnedOrthologProducer(producer),
    executePpiOwnedOrthologProducer(producer)
  ]);
  console.log(JSON.stringify({results, executeCalls}));
})();
"""
    )

    assert result == {"results": [True, True], "executeCalls": 1}


def test_ppi_owned_ortholog_producer_does_not_reuse_in_flight_promise_for_new_reference_identity() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const executePpiOwnedOrthologProducer = async",
        "const waitForPpiProducerDependency = async",
    )
    result = _run_node(
        helper
        + """
const ppiKaksInternalOrthologStatus = new Map();
const ppiKaksInternalOrthologPromises = new Map();
const ppiKaksInternalOrthologIdentityKey = producerStage => {
  const producerParams = producerStage.params || {};
  return [
    String(producerParams.dependencyOwner || 'ppi').trim().toLowerCase(),
    String(producerParams.ppiReferenceTaxonId || '').trim(),
    String(producerParams.ppiReferenceStorageKey || '').trim()
  ].join('::');
};
const getStageId = stage => stage.id;
const getStageLabel = stage => stage;
const addAgentMessage = () => {};
const executeCalls = [];
const executeStageDirect = async (_stage, params) => {
  executeCalls.push(params.ppiReferenceStorageKey);
  if (params.ppiReferenceStorageKey === 'string_taxon_3562') {
    await new Promise(resolve => setTimeout(resolve, 20));
    throw new Error('old reference failed');
  }
  await new Promise(resolve => setTimeout(resolve, 5));
};
const oldProducer = {id: 'ortholog', params: {
  dependencyOwner: 'ppi',
  ppiReferenceTaxonId: '3562',
  ppiReferenceStorageKey: 'string_taxon_3562'
}};
const atProducer = {id: 'ortholog', params: {
  dependencyOwner: 'ppi',
  ppiReferenceTaxonId: '3702',
  ppiReferenceStorageKey: 'At'
}};

(async () => {
  const results = await Promise.all([
    executePpiOwnedOrthologProducer(oldProducer),
    executePpiOwnedOrthologProducer(atProducer)
  ]);
  console.log(JSON.stringify({results, executeCalls}));
})();
"""
    )

    assert result == {
        "results": [False, True],
        "executeCalls": ["string_taxon_3562", "At"],
    }


def test_resetting_ppi_producer_clears_completed_identity_for_same_reference_rerun() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    execute_helper = _source_block(
        html,
        "const executePpiOwnedOrthologProducer = async",
        "const waitForPpiProducerDependency = async",
    )
    remember_helper = _source_block(
        html,
        "const ppiKaksInternalOrthologProducers = new Map();",
        "const getPpiKaksInternalOrthologProducer =",
    )
    result = _run_node(
        execute_helper
        + remember_helper
        + """
const ppiKaksInternalOrthologIdentityKey = producerStage => {
  const producerParams = producerStage.params || {};
  return [
    String(producerParams.dependencyOwner || 'ppi').trim().toLowerCase(),
    String(producerParams.ppiReferenceTaxonId || '').trim(),
    String(producerParams.ppiReferenceStorageKey || '').trim()
  ].join('::');
};
const getStageId = stage => stage.id;
const getStageLabel = stage => stage;
const addAgentMessage = () => {};
let executeCalls = 0;
const executeStageDirect = async () => { executeCalls += 1; };
const atProducer = {id: 'ortholog', params: {
  dependencyOwner: 'ppi',
  ppiReferenceTaxonId: '3702',
  ppiReferenceStorageKey: 'At'
}};

(async () => {
  rememberPpiKaksInternalOrthologProducer('ppi', atProducer);
  const first = await executePpiOwnedOrthologProducer(atProducer);
  rememberPpiKaksInternalOrthologProducer('ppi', null);
  rememberPpiKaksInternalOrthologProducer('ppi', atProducer);
  const second = await executePpiOwnedOrthologProducer(atProducer);
  console.log(JSON.stringify({first, second, executeCalls}));
})();
"""
    )

    assert result == {"first": True, "second": True, "executeCalls": 2}


def test_resume_does_not_change_running_ppi_back_to_pending() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const resumePpiKaksPausedStages = async",
        "const ppiKaksFallbackRecoveryOwners =",
    )
    result = _run_node(
        helper
        + """
const batchExecuteStages = {value: [{id: 'ppi'}]};
const stageExecutionStatus = {value: {ppi: 'running'}};
const isBatchExecuting = {value: true};
const shouldStopBatchExecute = {value: false};
const currentBatchStage = {value: ''};
const getStageId = stage => typeof stage === 'object' ? stage.id : stage;
const ppiKaksPauseStageIds = stages => stages;
const clearPpiKaksReferencePause = () => {};
const addAgentMessage = () => {};
const executeStagesWithDependencies = async () => {};

(async () => {
  await resumePpiKaksPausedStages(['ppi'], 'At ready');
  console.log(JSON.stringify({status: stageExecutionStatus.value.ppi}));
})();
"""
    )

    assert result == {"status": "running"}


def test_ppi_manifest_and_background_status_queries_use_canonical_owner_identity() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    manifest_block = _source_block(html, "const checkManifestExists = async", "const pollTaskUntilComplete = async")
    background_block = _source_block(html, "const fetchStageCompleted = async", "const finalizeExecuteStageDirectRequest =")
    expected = "canonicalizePpiOwnerBiologicalScope({ ...batchPlanParams, ...batchParams, ...stageParams })"
    assert expected in manifest_block
    assert expected in background_block


def test_ppi_preflight_retries_one_transient_network_error_before_returning_ready() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    retry_helper = _source_block(
        html,
        "const fetchWithTransientRetry = async",
        "const pollReferenceBundleDownloadTask = async",
    )
    preflight = _source_block(
        html,
        "const runPpiSpeciesResolutionPreflight = async",
        "const ppiKaksDependencyStages =",
    )
    result = _run_node(
        retry_helper
        + preflight
        + """
const jobInfo = {job_id: 'job_network_retry'};
const executionPrefix = () => 'Ft.SPL';
const canonicalizePpiOwnerBiologicalScope = payload => payload;
const ppiNeedsNearReferenceDecision = () => false;
const formatPpiSpeciesResolutionBlockMessage = () => 'not ready';
const showStageConfirmDialog = async () => true;
const addAgentMessage = () => {};
const ElMessageBox = {alert: async () => {}};
const setTimeout = callback => callback();
let attempts = 0;
const fetch = async () => {
  attempts += 1;
  if (attempts === 1) {
    throw new TypeError('NetworkError when attempting to fetch resource.');
  }
  return {
    ok: true,
    status: 200,
    statusText: 'OK',
    json: async () => ({success: true, status: 'ready'})
  };
};

(async () => {
  const outcome = await runPpiSpeciesResolutionPreflight({
    stage: 'ppi', prefix: 'Ft.SPL', targetSpecies: ['Ft']
  }, {requestedStages: ['ppi']});
  console.log(JSON.stringify({attempts, allowed: outcome.allowed}));
})();
"""
    )

    assert result == {"attempts": 2, "allowed": True}


def test_ppi_preflight_does_not_retry_a_valid_business_response() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    retry_helper = _source_block(
        html,
        "const fetchWithTransientRetry = async",
        "const pollReferenceBundleDownloadTask = async",
    )
    preflight = _source_block(
        html,
        "const runPpiSpeciesResolutionPreflight = async",
        "const ppiKaksDependencyStages =",
    )
    result = _run_node(
        retry_helper
        + preflight
        + """
const jobInfo = {job_id: 'job_business_response'};
const executionPrefix = () => 'Ft.SPL';
const canonicalizePpiOwnerBiologicalScope = payload => payload;
const ppiNeedsNearReferenceDecision = () => false;
const formatPpiSpeciesResolutionBlockMessage = () => 'reference data missing';
const showStageConfirmDialog = async () => true;
const addAgentMessage = () => {};
const ElMessageBox = {alert: async () => {}};
const setTimeout = callback => callback();
let attempts = 0;
const fetch = async () => {
  attempts += 1;
  return {
    ok: true,
    status: 200,
    statusText: 'OK',
    json: async () => ({success: true, status: 'needs_reference_data'})
  };
};

(async () => {
  const outcome = await runPpiSpeciesResolutionPreflight({
    stage: 'ppi', prefix: 'Ft.SPL', targetSpecies: ['Ft']
  }, {requestedStages: ['ppi']});
  console.log(JSON.stringify({attempts, allowed: outcome.allowed}));
})();
"""
    )

    assert result == {"attempts": 1, "allowed": False}


def test_reference_bundle_status_poll_recovers_from_one_transient_network_error() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    helper = _source_block(
        html,
        "const fetchWithTransientRetry = async",
        "const ppiKaksOwnedOrthologOwners =",
    )
    result = _run_node(
        helper
        + """
const setTimeout = callback => callback();
const referenceBundleErrorMessage = (_data, fallback) => fallback;
const updateReferenceBundleDownloadProgress = () => {};
let attempts = 0;
const fetch = async () => {
  attempts += 1;
  if (attempts === 1) throw new TypeError('Failed to fetch');
  return {
    ok: true,
    status: 200,
    json: async () => ({status: 'success', result: {status: 'success', species_code: 'string_taxon_3562'}})
  };
};

(async () => {
  const result = await pollReferenceBundleDownloadTask('task-download');
  console.log(JSON.stringify({attempts, status: result.status, species: result.species_code}));
})();
"""
    )

    assert result == {"attempts": 2, "status": "success", "species": "string_taxon_3562"}


def test_transient_retry_is_not_used_for_download_or_analysis_submission() -> None:
    html = Path("apps/web/vue3/index.html").read_text(encoding="utf-8")
    download_submit = _source_block(
        html,
        "const downloadRecommendedPpiReferenceBundle = async",
        "const updateReferenceBundleDownloadProgress =",
    )
    execute_submit = _source_block(
        html,
        "const submitExecuteRequestWithAutoAnnotation = async",
        "const waitForDialogTransition = async",
    )

    assert "fetch('/api/reference-bundle/download'" in download_submit
    assert "fetchWithTransientRetry('/api/reference-bundle/download'" not in download_submit
    assert "fetch('/api/execute'" in execute_submit
    assert "fetchWithTransientRetry('/api/execute'" not in execute_submit