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"