PlantAI-dev/tests/unit/test_ppi_provenance_task6.py

from __future__ import annotations

import hashlib
import json
from pathlib import Path
from types import SimpleNamespace

import pytest
from openpyxl import Workbook

from apps.web import pipeline_backend as backend
from gfa_core.mas.agents.diagnosis import DiagnosisAgent
from gfa_core.mas.tools.stage_status import get_stage_status
from skills.ppi.run import PpiSkill


def _job(tmp_path: Path, job_id: str = "task6") -> backend.JobInfo:
    input_dir = tmp_path / "input"
    workspace = tmp_path / "workspace"
    input_dir.mkdir(parents=True)
    (workspace / "final").mkdir(parents=True)
    (workspace / "logs").mkdir(parents=True)
    return backend.JobInfo(
        job_id=job_id,
        input_dir=str(input_dir),
        workspace_dir=str(workspace),
        logs_dir=str(workspace / "logs"),
        created_at="",
    )


def _sv_preflight(job: backend.JobInfo, *, confirmed: bool):
    return backend.resolve_ppi_species_resolution(
        job=job,
        prefix="Sv.HKL",
        target_species=["Sv"],
        family="HKL",
        target_scientific_name="Setaria viridis",
        ppi_target_tree_taxon_id="4556",
        ppi_reference_taxon_id="4555",
        ppi_reference_storage_key="string_taxon_4555",
        ppi_reference_selection_mode="tree_nearest",
        ppi_reference_user_confirmed=confirmed,
        dependency_owner="ppi",
        requested_stages=["ppi"],
    )


def _assert_complete_identity(identity: dict) -> None:
    assert identity["schema_version"] == "ppi_provenance_v1"
    assert identity["raw_target_species"]
    assert "canonical_target_name" in identity
    assert "target_tree_taxon_id" in identity
    assert "selected_string_taxon_id" in identity
    assert identity["selection_mode"]
    snapshot = identity["string_snapshot"]
    assert snapshot["version"]
    assert snapshot["snapshot_id"]
    assert len(snapshot["sha256"]) == 64
    assert "confirmed" in identity["user_confirmation"]
    assert "reference_species" in identity["ortholog_mapping"]
    assert "path" in identity["ortholog_mapping"]
    assert "normalization_species_label" in identity
    assert identity["diagnosis_code"]


def _prepare_owner_execution_fixture(job: backend.JobInfo) -> Path:
    (Path(job.workspace_dir) / "final" / "Sv.HKL.final_members.id").write_text(
        "SvGene1\n",
        encoding="utf-8",
    )
    scope = Path(job.input_dir) / "ppi-owner.scope.json"
    scope.write_text(
        json.dumps(
            {
                "scope_id": "owner-target-only",
                "scope_mode": "target_only",
                "family": "HKL",
                "target_species": ["Sv"],
                "reference_species": [],
                "selected_species": ["Sv"],
            }
        ),
        encoding="utf-8",
    )
    mapping = Path(job.workspace_dir) / "final" / "Sv.HKL.ppi.orthologs.txt"
    mapping.write_text(
        "SvGene1\tSeita.1\t91\tSv\tstring_taxon_4555\n",
        encoding="utf-8",
    )
    storage = Path(job.input_dir) / "string_taxon_4555"
    storage.mkdir()
    (storage / "string_taxon_4555.pep_cds_idmap.tsv").write_text(
        "species_code\tprotein_id\tpep_id\tcds_id\tgene_id\tcds_header\n"
        "string_taxon_4555\tSeita.1\tSeita.1\tSeita.1\tSeita1\tSeita.1 [gene=Seita1]\n",
        encoding="utf-8",
    )
    return scope


def test_run_stage_auto_ppi_uses_confirmed_owner_taxon_without_legacy_inference(
    tmp_path: Path,
    monkeypatch,
) -> None:
    job = _job(tmp_path, job_id="owner-auto-execution")
    scope = _prepare_owner_execution_fixture(job)
    launched = []
    monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
    monkeypatch.setattr(
        backend,
        "infer_string_species_from_ortholog_mapping",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("legacy inference must not run")),
    )
    monkeypatch.setattr(
        backend,
        "run_cmd_logged",
        lambda cmd, *_args, **_kwargs: launched.append(list(cmd)) or 1,
    )

    message, manifest_path = backend.run_stage(
        "ppi", job, "Sv.HKL", scope_file=str(scope), ppi_species="auto",
        ppi_reference_taxon_id="4555",
        ppi_reference_storage_key="string_taxon_4555",
        ppi_reference_selection_mode="tree_nearest",
        ppi_reference_user_confirmed=True,
        dependency_owner="ppi",
    )

    assert message.startswith("❌ PPI analysis failed (exit code: 1)")
    assert launched
    command = launched[0]
    assert command[command.index("--species") + 1] == "4555"
    payload = json.loads(Path(manifest_path).read_text(encoding="utf-8"))
    assert payload["parameters"]["string_taxon_id"] == "4555"
    assert payload["ppi_provenance"]["selected_string_taxon_id"] == "4555"


@pytest.mark.parametrize(
    ("owner_taxon", "confirmed", "message_fragment", "expected_failed_taxon"),
    [
        ("not-a-taxon", True, "numeric", ""),
        ("4555", False, "confirmed", "4555"),
    ],
)
def test_run_stage_auto_ppi_owner_fails_closed_without_valid_confirmation(
    tmp_path: Path,
    monkeypatch,
    owner_taxon: str,
    confirmed: bool,
    message_fragment: str,
    expected_failed_taxon: str,
) -> None:
    job = _job(tmp_path, job_id=f"invalid-owner-{confirmed}")
    executed = False

    def fail_if_executed(*_args, **_kwargs):
        nonlocal executed
        executed = True
        raise AssertionError("invalid owner selection must fail before CLI execution")

    monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
    monkeypatch.setattr(backend, "run_cmd_logged", fail_if_executed)
    message, manifest_path = backend.run_stage(
        "ppi", job, "Sv.HKL", ppi_species="auto",
        ppi_reference_taxon_id=owner_taxon,
        ppi_reference_storage_key="string_taxon_4555",
        ppi_reference_user_confirmed=confirmed,
        dependency_owner="ppi",
    )

    assert message.startswith("❌")
    assert message_fragment in message
    assert executed is False
    payload = json.loads(Path(manifest_path).read_text(encoding="utf-8"))
    assert payload["ppi_provenance"]["selected_string_taxon_id"] == expected_failed_taxon


def test_run_stage_explicit_validated_ppi_species_precedes_owner_fallback(
    tmp_path: Path,
    monkeypatch,
) -> None:
    job = _job(tmp_path, job_id="explicit-precedes-owner")
    scope = _prepare_owner_execution_fixture(job)
    launched = []
    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,
    )

    backend.run_stage(
        "ppi", job, "Sv.HKL", scope_file=str(scope), ppi_species="4555",
        ppi_reference_taxon_id="3708",
        ppi_reference_storage_key="string_taxon_4555",
        ppi_reference_user_confirmed=False,
        dependency_owner="ppi",
    )

    assert launched
    command = launched[0]
    assert command[command.index("--species") + 1] == "4555"


def test_run_stage_owner_mapping_defaults_normalization_to_storage_identity(
    tmp_path: Path,
    monkeypatch,
) -> None:
    from gfa_core.id_resolution import ReferenceAliasIndex

    job = _job(tmp_path, job_id="owner-storage-normalization")
    scope = _prepare_owner_execution_fixture(job)
    launched = []
    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,
    )

    backend.run_stage(
        "ppi", job, "Sv.HKL", scope_file=str(scope), ppi_species="auto",
        ppi_reference_taxon_id="4555",
        ppi_reference_storage_key="string_taxon_4555",
        ppi_reference_user_confirmed=True,
        dependency_owner="ppi",
    )

    command = launched[0]
    assert command[command.index("--species") + 1] == "4555"
    normalization_label = command[command.index("--normalization-species-label") + 1]
    assert normalization_label == "string_taxon_4555"
    id_map = Path(job.input_dir) / "string_taxon_4555" / "string_taxon_4555.pep_cds_idmap.tsv"
    ReferenceAliasIndex.from_tsv(id_map, normalization_label)


def test_run_stage_explicit_normalization_label_precedes_owner_storage_default(
    tmp_path: Path,
    monkeypatch,
) -> None:
    job = _job(tmp_path, job_id="owner-explicit-normalization")
    scope = _prepare_owner_execution_fixture(job)
    launched = []
    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,
    )

    backend.run_stage(
        "ppi", job, "Sv.HKL", scope_file=str(scope), ppi_species="auto",
        normalization_species_label="explicit-reference-label",
        ppi_reference_taxon_id="4555",
        ppi_reference_storage_key="string_taxon_4555",
        ppi_reference_user_confirmed=True,
        dependency_owner="ppi",
    )

    command = launched[0]
    assert command[command.index("--normalization-species-label") + 1] == "explicit-reference-label"


def test_run_stage_direct_exact_os_keeps_osa_normalization_default(
    tmp_path: Path,
    monkeypatch,
) -> None:
    job = _job(tmp_path, job_id="direct-os-normalization")
    final_dir = Path(job.workspace_dir) / "final"
    (final_dir / "Os.HKL.final_members.id").write_text("Os01g01010\n", encoding="utf-8")
    scope = Path(job.input_dir) / "os.scope.json"
    scope.write_text(
        json.dumps(
            {
                "scope_id": "HKL.target-Os.target_only",
                "scope_mode": "target_only",
                "family": "HKL",
                "target_species": ["Os"],
                "reference_species": [],
                "selected_species": ["Os"],
            }
        ),
        encoding="utf-8",
    )
    launched = []
    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,
    )

    backend.run_stage("ppi", job, "Os.HKL", scope_file=str(scope), ppi_species="auto")

    command = launched[0]
    assert command[command.index("--species") + 1] == "osa"
    assert command[command.index("--normalization-species-label") + 1] == "osa"


def test_waiting_preflights_write_decision_records_without_failed_stage_manifest(tmp_path: Path) -> None:
    job = _job(tmp_path)
    confirmation = _sv_preflight(job, confirmed=False)
    missing_pep = _sv_preflight(job, confirmed=True)
    storage = Path(job.input_dir) / "string_taxon_4555"
    storage.mkdir()
    (storage / "string_taxon_4555.pep.fasta").write_text(">Seita1\nMA\n", encoding="utf-8")
    (storage / "string_taxon_4555.cds.fasta").write_text(">Seita1\nATGGCT\n", encoding="utf-8")
    needs_ortholog = _sv_preflight(job, confirmed=True)

    assert [confirmation["status"], missing_pep["status"], needs_ortholog["status"]] == [
        "confirmation_required", "needs_reference_data", "needs_ortholog"
    ]
    _assert_complete_identity(missing_pep["ppi_provenance"])
    decision_path = Path(job.workspace_dir) / "reproducibility" / "Sv.HKL.ppi.preflight_decisions.jsonl"
    records = [json.loads(line) for line in decision_path.read_text(encoding="utf-8").splitlines()]
    assert [record["status"] for record in records[-3:]] == [
        "confirmation_required", "needs_reference_data", "needs_ortholog"
    ]
    assert all(record["record_type"] == "ppi_preflight_decision" for record in records[-3:])
    assert all("ppi_provenance" in record for record in records[-3:])
    assert not (Path(job.workspace_dir) / "final" / "Sv.HKL.ppi.manifest.json").exists()


def test_confirmed_arabidopsis_model_fallback_is_accepted_outside_nearest_candidates(
    tmp_path: Path,
) -> None:
    job = _job(tmp_path, job_id="ft-arabidopsis-fallback")
    storage = Path(job.input_dir) / "At"
    storage.mkdir()
    (storage / "At.pep.fasta").write_text(">AT1G01010.1\nMA\n", encoding="utf-8")

    result = backend.resolve_ppi_species_resolution(
        job=job,
        prefix="Ft.SPL",
        target_species=["Ft"],
        family="SPL",
        target_scientific_name="Fagopyrum tataricum",
        ppi_target_tree_taxon_id="62330",
        ppi_reference_taxon_id="3702",
        ppi_reference_storage_key="At",
        ppi_reference_selection_mode="model_fallback",
        ppi_reference_user_confirmed=True,
        dependency_owner="ppi",
        ignore_existing_ortholog=True,
        requested_stages=["ppi"],
    )

    assert result["status"] == "needs_ortholog"
    assert result["selected_taxon_id"] == "3702"
    assert result["reference_species"] == "At"
    assert result["ppi_provenance"]["selection_mode"] == "model_fallback"
    assert result["ppi_provenance"]["user_confirmation"]["confirmed"] is True


@pytest.mark.parametrize(
    ("selection_mode", "status", "expected"),
    [
        ("exact_string", "ready", "ppi_string_exact"),
        ("tree_nearest", "confirmation_required", "ppi_string_tree_nearest"),
        ("model_fallback", "confirmation_required", "ppi_string_model_fallback"),
        ("unresolved", "unresolved", "ppi_string_unresolved"),
        ("resource_error", "resource_error", "ppi_string_resource_error"),
    ],
)
def test_diagnosis_exposes_stable_ppi_resolution_codes(selection_mode: str, status: str, expected: str) -> None:
    result = DiagnosisAgent.ppi_identity_diagnosis({
        "status": status,
        "ppi_provenance": {"selection_mode": selection_mode},
    })
    assert result["code"] == expected


def test_skill_success_and_failure_manifests_share_provenance(tmp_path: Path) -> None:
    provenance = {
        "schema_version": "ppi_provenance_v1",
        "raw_target_species": ["Os"],
        "canonical_target_name": "Oryza sativa Japonica Group",
        "target_tree_taxon_id": "39947",
        "selected_string_taxon_id": "39947",
        "selection_mode": "exact_string",
        "string_snapshot": {"version": "12.0", "snapshot_id": "snapshot", "sha256": "a" * 64},
        "user_confirmation": {"required": False, "confirmed": False},
        "ortholog_mapping": {"reference_species": "", "path": ""},
        "normalization_species_label": "Os",
        "diagnosis_code": "ppi_string_exact",
    }
    empty_ids = tmp_path / "empty.ids"
    empty_ids.write_text("", encoding="utf-8")
    output = tmp_path / "ppi"
    failed = PpiSkill().run(
        empty_ids, "39947", "Os.HKL", output,
        normalization_species_label="Os", ppi_provenance=provenance,
    )
    failed_manifest = json.loads(Path(failed["manifest_path"]).read_text(encoding="utf-8"))
    assert failed["status"] == "failed"
    assert failed_manifest["status"] == "failed"
    assert failed_manifest["ppi_provenance"] == provenance


@pytest.mark.parametrize(
    "failure_case",
    [
        "unsupported_species",
        "missing_scope",
        "unreadable_scope",
        "mapping_without_source_ids",
        "missing_scoped_member_ids",
        "multi_target_without_mapping",
        "member_resolution_error",
        "member_resolution_without_path",
    ],
)
def test_backend_pre_command_failures_write_current_ppi_manifest(
    tmp_path: Path, monkeypatch, failure_case: str,
) -> None:
    job = _job(tmp_path / failure_case)
    monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
    kwargs = {"ppi_species": "osa"}

    scope_cases = {
        "missing_scope",
        "unreadable_scope",
        "mapping_without_source_ids",
        "missing_scoped_member_ids",
        "multi_target_without_mapping",
    }
    if failure_case == "unsupported_species":
        kwargs["ppi_species"] = "definitely_unsupported"
    elif failure_case in scope_cases:
        scope = Path(job.input_dir) / "ppi.scope.json"
        kwargs["scope_file"] = str(scope)
        if failure_case != "missing_scope":
            scope.write_text("{}", encoding="utf-8")
        if failure_case == "unreadable_scope":
            monkeypatch.setattr(backend, "read_scope", lambda _path: (_ for _ in ()).throw(ValueError("bad scope")))
        else:
            scope_data = {
                "scope_id": "scope1",
                "family": "HKL",
                "target_species": ["Sv", "Zm"] if failure_case == "multi_target_without_mapping" else ["Sv"],
                "reference_species": ["string_taxon_4555"],
            }
            monkeypatch.setattr(backend, "read_scope", lambda _path, data=scope_data: data)
            kwargs["ppi_species"] = "4555"
            if failure_case == "mapping_without_source_ids":
                mapping = Path(job.workspace_dir) / "final" / "Sv.HKL.scope1.orthologs.txt"
                mapping.write_text("SvGene\tSeitaGene\n", encoding="utf-8")
                monkeypatch.setattr(backend, "build_ppi_gene_ids_from_ortholog_mapping", lambda *_args: 0)
            elif failure_case == "missing_scoped_member_ids":
                monkeypatch.setattr(backend, "resolve_single_target_gene_id_file", lambda *_args: None)
    elif failure_case == "member_resolution_error":
        monkeypatch.setattr(
            backend,
            "resolve_effective_members",
            lambda **_kwargs: (_ for _ in ()).throw(backend.MemberResolutionError("missing members")),
        )
    elif failure_case == "member_resolution_without_path":
        monkeypatch.setattr(
            backend,
            "resolve_effective_members",
            lambda **_kwargs: SimpleNamespace(path_id=None, warnings=[], source_kind="test", member_count=0),
        )

    message, manifest_path = backend.run_stage("ppi", job, "Sv.HKL", **kwargs)

    expected = Path(job.workspace_dir) / "final" / "Sv.HKL.ppi.manifest.json"
    assert message.startswith("❌")
    assert manifest_path == expected
    payload = json.loads(expected.read_text(encoding="utf-8"))
    assert payload["stage"] == "ppi"
    assert payload["prefix"] == "Sv.HKL"
    assert payload["status"] == "failed"
    _assert_complete_identity(payload["ppi_provenance"])


def test_tree_nearest_confirmation_history_survives_preflight_to_execution(
    tmp_path: Path, monkeypatch,
) -> None:
    job = _job(tmp_path)
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = "config"
    sheet.append(["Species_Code", "Species_Name", "TaxID"])
    sheet.append(["Sv", "Setaria viridis", "4556"])
    workbook.save(Path(job.input_dir) / "config.xlsx")

    unconfirmed = _sv_preflight(job, confirmed=False)["ppi_provenance"]
    assert unconfirmed["user_confirmation"] == {"required": True, "confirmed": False}
    preflight = _sv_preflight(job, confirmed=True)["ppi_provenance"]
    assert preflight["user_confirmation"] == {"required": True, "confirmed": True}

    scope = Path(job.input_dir) / "ppi.scope.json"
    scope.write_text("{}", encoding="utf-8")
    scope_data = {
        "scope_id": "scope1", "family": "HKL", "target_species": ["Sv"],
        "reference_species": ["string_taxon_4555"],
    }
    monkeypatch.setattr(backend, "read_scope", lambda _path: scope_data)
    mapping = Path(job.workspace_dir) / "final" / "Sv.HKL.scope1.orthologs.txt"
    mapping.write_text("SvGene\tSeitaGene\n", encoding="utf-8")

    def build_ids(_mapping, output, _targets):
        Path(output).parent.mkdir(parents=True, exist_ok=True)
        Path(output).write_text("SvGene\n", encoding="utf-8")
        return 1

    monkeypatch.setattr(backend, "build_ppi_gene_ids_from_ortholog_mapping", build_ids)
    storage = Path(job.input_dir) / "string_taxon_4555"
    storage.mkdir()
    (storage / "string_taxon_4555.pep_cds_idmap.tsv").write_text("a\tb\n", encoding="utf-8")
    captured = {}

    def fail_after_submission(cmd, *_args, **_kwargs):
        captured.update(json.loads(Path(cmd[cmd.index("--provenance-json") + 1]).read_text(encoding="utf-8")))
        return 1

    monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
    monkeypatch.setattr(backend, "run_cmd_logged", fail_after_submission)
    _message, manifest_path = backend.run_stage(
        "ppi", job, "Sv.HKL", scope_file=str(scope), ppi_species="4555",
        ppi_reference_taxon_id="4555",
        ppi_reference_storage_key="string_taxon_4555",
        ppi_reference_selection_mode="tree_nearest",
        ppi_reference_user_confirmed=True,
        normalization_species_label="Sv",
    )
    executed = json.loads(Path(manifest_path).read_text(encoding="utf-8"))["ppi_provenance"]

    for field in (
        "raw_target_species", "canonical_target_name", "target_tree_taxon_id",
        "selected_string_taxon_id", "selection_mode", "string_snapshot",
        "normalization_species_label", "diagnosis_code",
    ):
        assert executed[field] == captured[field]
    assert executed["raw_target_species"] == preflight["raw_target_species"]
    assert executed["canonical_target_name"] == preflight["canonical_target_name"]
    assert executed["target_tree_taxon_id"] == preflight["target_tree_taxon_id"]
    assert executed["selected_string_taxon_id"] == preflight["selected_string_taxon_id"]
    assert executed["selection_mode"] == preflight["selection_mode"] == "tree_nearest"
    assert executed["string_snapshot"] == preflight["string_snapshot"]
    assert executed["user_confirmation"] == {"required": True, "confirmed": True}


def test_backend_submitted_failure_always_writes_identity_manifest(tmp_path: Path, monkeypatch) -> None:
    job = _job(tmp_path)
    members = Path(job.workspace_dir) / "final" / "Os.HKL.final_members.id"
    members.write_text("Os01g0100100\n", encoding="utf-8")
    monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
    monkeypatch.setattr(backend, "run_cmd_logged", lambda *_args, **_kwargs: 1)

    message, manifest_path = backend.run_stage(
        "ppi", job, "Os.HKL", ppi_species="osa",
        normalization_species_label="Os", ppi_score=200,
    )

    assert message.startswith("❌")
    assert manifest_path == Path(job.workspace_dir) / "final" / "Os.HKL.ppi.manifest.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    assert manifest["status"] == "failed"
    _assert_complete_identity(manifest["ppi_provenance"])


def test_backend_and_skill_success_manifest_use_the_same_submitted_identity(tmp_path: Path, monkeypatch) -> None:
    job = _job(tmp_path)
    members = Path(job.workspace_dir) / "final" / "Os.HKL.final_members.id"
    members.write_text("Os01g0100100\n", encoding="utf-8")
    monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))

    def fake_run(cmd, *_args, **_kwargs):
        output = Path(cmd[cmd.index("--output") + 1])
        prefix = cmd[cmd.index("--prefix") + 1]
        provenance_path = Path(cmd[cmd.index("--provenance-json") + 1])
        provenance = json.loads(provenance_path.read_text(encoding="utf-8"))
        output.mkdir(parents=True, exist_ok=True)
        network = output / f"{prefix}_network.tsv"
        network.write_text("protein1\tprotein2\tscore\n", encoding="utf-8")
        (output / f"{prefix}.manifest.json").write_text(json.dumps({
            "manifest_version": "1.0", "stage": "ppi", "status": "success", "prefix": prefix,
            "parameters": {}, "metrics": {}, "ppi_provenance": provenance,
            "artifacts": {"network": {"full_path": str(network), "path": network.name}},
        }), encoding="utf-8")
        return 0

    monkeypatch.setattr(backend, "run_cmd_logged", fake_run)
    message, manifest_path = backend.run_stage(
        "ppi", job, "Os.HKL", ppi_species="osa",
        normalization_species_label="Os", ppi_score=200,
    )

    assert message.startswith("✅")
    manifest = json.loads(Path(manifest_path).read_text(encoding="utf-8"))
    _assert_complete_identity(manifest["ppi_provenance"])
    assert manifest["ppi_provenance"]["selected_string_taxon_id"] == "39947"


def test_status_returns_manifest_identity_without_rewriting_historical_snapshot(tmp_path: Path) -> None:
    workspace_root = tmp_path / "workspace"
    final_dir = workspace_root / "job" / "final"
    final_dir.mkdir(parents=True)
    artifact = final_dir / "Os.HKL.ppi_network.tsv"
    artifact.write_text("protein1\tprotein2\tscore\n", encoding="utf-8")
    provenance = {
        "schema_version": "ppi_provenance_v1",
        "raw_target_species": ["Os"],
        "canonical_target_name": "Oryza sativa Japonica Group",
        "target_tree_taxon_id": "39947",
        "selected_string_taxon_id": "39947",
        "selection_mode": "exact_string",
        "string_snapshot": {"version": "11.0", "snapshot_id": "historic", "sha256": "b" * 64},
        "user_confirmation": {"required": False, "confirmed": False},
        "ortholog_mapping": {"reference_species": "", "path": ""},
        "normalization_species_label": "Os",
        "diagnosis_code": "ppi_string_exact",
    }
    manifest_path = final_dir / "Os.HKL.ppi.manifest.json"
    manifest_path.write_text(json.dumps({
        "manifest_version": "1.0", "stage": "ppi", "status": "success", "prefix": "Os.HKL",
        "ppi_provenance": provenance,
        "artifacts": {"network": {"full_path": str(artifact)}},
    }), encoding="utf-8")
    before = hashlib.sha256(manifest_path.read_bytes()).hexdigest()

    report = get_stage_status("job", "Os.HKL", "ppi", workspace_root=workspace_root)

    assert report.completed is True
    assert report.to_dict()["ppi_provenance"] == provenance
    assert hashlib.sha256(manifest_path.read_bytes()).hexdigest() == before


def test_job_diagnosis_reads_the_manifest_provenance_identity(tmp_path: Path) -> None:
    final_dir = tmp_path / "workspace" / "job" / "final"
    final_dir.mkdir(parents=True)
    provenance = {
        "selection_mode": "model_fallback",
        "selected_string_taxon_id": "3702",
        "string_snapshot": {"version": "12.0", "snapshot_id": "fixed", "sha256": "c" * 64},
        "diagnosis_code": "ppi_string_model_fallback",
    }
    (final_dir / "Ft.SPL.ppi.manifest.json").write_text(json.dumps({
        "manifest_version": "1.0", "stage": "ppi", "status": "failed", "prefix": "Ft.SPL",
        "ppi_provenance": provenance, "artifacts": {},
    }), encoding="utf-8")

    report = DiagnosisAgent(None).diagnose_job_stage("job", "ppi", "Ft.SPL", project_root=tmp_path)

    assert report["ppi_identity_diagnosis"]["code"] == "ppi_string_model_fallback"
    assert report["ppi_identity_diagnosis"]["snapshot"]["snapshot_id"] == "fixed"