"""Phase 1 — per-project failure isolation in list_projects() / list_legacy_project_warnings(). Spec: consultations/recoil/bugfix-sprint-post-overhaul/BUILD_SPEC.md, Phase 1. Pre-fix bug: any exception other than LegacyProjectFormatError raised inside the per-slug body (Pydantic ValidationError on Project() construction, KeyError in _project_name, OSError in nested adapter calls, late LegacyProjectFormatError raised from inside list_episodes(slug)) bubbled up and 500-d /api/projects, killing the project picker. Post-fix invariant: one bad slug is dropped, the rest load, and the fallback `project_load_failure_isolated` fires with the slug + error type in the payload. Build A Phase 5 (2026-05-09): migrated from recoil.api.sanctioned_fallbacks (deleted) to recoil.pipeline._lib.sanctioned_fallbacks (canonical). Counter checks replaced with log-line assertions (canonical registry has no counter). """ from __future__ import annotations import logging from recoil.api.adapters import projects as projects_adapter from recoil.api.adapters.projects import ( LegacyProjectFormatError, LegacyWarning, list_legacy_project_warnings, list_projects, ) from recoil.pipeline._lib.sanctioned_fallbacks import ( list_sanctioned_fallbacks, ) def test_phase1_fallback_is_registered(): """Sanity: Law 14 — the fallback we emit must exist in the registry.""" names = {r.name for r in list_sanctioned_fallbacks()} assert "project_load_failure_isolated" in names # ── Helpers ────────────────────────────────────────────────────────────── def _stub_slugs(monkeypatch, slugs: list[str]) -> None: monkeypatch.setattr(projects_adapter, "_slugs_in_root", lambda: list(slugs)) def _make_minimal_project(slug: str): """Construct a real Project shape that satisfies the schema without touching disk. Bypasses _build_project's nested adapter calls. """ from recoil.api.schemas.engine import SCHEMA_VERSION, Project return Project( schema_version=SCHEMA_VERSION, id=slug, name=slug, aspect="9_16", aspect_synthesized=False, score=None, episodes=[], ) def _fallback_count(caplog, name: str) -> int: """Count FALLBACK_FIRED log lines for a given fallback name. Matches the canonical log format: 'FALLBACK_FIRED name= ...' as produced by recoil.pipeline._lib.sanctioned_fallbacks.fire_sanctioned_fallback. """ return sum( 1 for r in caplog.records if "FALLBACK_FIRED" in r.message and name in r.message ) # ── list_projects ──────────────────────────────────────────────────────── def test_list_projects_skips_project_with_pydantic_validation_error( monkeypatch, caplog ): """Bad slug raises ValueError inside _build_project → isolated, fallback fired, other slugs still load.""" caplog.set_level(logging.WARNING) _stub_slugs(monkeypatch, ["good-project", "bad-project", "another-good"]) monkeypatch.setattr(projects_adapter, "_load_bible", lambda slug: {"project": slug}) def _build(slug, bible): if slug == "bad-project": # Stand-in for any unexpected exception inside _build_project, # including a Pydantic ValidationError raised by Project(...). raise ValueError("invalid aspect_ratio: not a real value") return _make_minimal_project(slug) monkeypatch.setattr(projects_adapter, "_build_project", _build) result = list_projects() assert [p.id for p in result] == ["good-project", "another-good"], ( "bad slug must be dropped; good slugs must still load" ) assert _fallback_count(caplog, "project_load_failure_isolated") == 1 assert any( "FALLBACK_FIRED" in r.message and "project_load_failure_isolated" in r.message for r in caplog.records ) # The skip log line records the slug + error class. assert any( "bad-project" in r.message and "ValueError" in r.message for r in caplog.records ) def test_list_projects_skips_project_with_unexpected_oserror(monkeypatch, caplog): """Bad slug raises OSError inside _load_bible → isolated, fallback fired, other slugs still load.""" caplog.set_level(logging.WARNING) _stub_slugs(monkeypatch, ["alpha", "broken-fs", "gamma"]) def _load(slug): if slug == "broken-fs": raise OSError("[Errno 5] Input/output error") return {"project": slug} monkeypatch.setattr(projects_adapter, "_load_bible", _load) monkeypatch.setattr( projects_adapter, "_build_project", lambda slug, bible: _make_minimal_project(slug), ) result = list_projects() assert [p.id for p in result] == ["alpha", "gamma"], ( "OSError on one slug must not take down the others" ) assert _fallback_count(caplog, "project_load_failure_isolated") == 1 assert any( "FALLBACK_FIRED" in r.message and "project_load_failure_isolated" in r.message for r in caplog.records ) assert any( "broken-fs" in r.message and "OSError" in r.message for r in caplog.records ) def test_list_projects_still_silently_skips_legacy_format_error(monkeypatch, caplog): """LegacyProjectFormatError stays a silent skip — pre-fix behavior preserved. No fallback fires for legacy (it's surfaced via list_legacy_project_warnings).""" caplog.set_level(logging.WARNING) _stub_slugs(monkeypatch, ["legacy-one", "loaded-one"]) def _load(slug): if slug == "legacy-one": raise LegacyProjectFormatError(slug, "missing state/visual/global_bible.json") return {"project": slug} monkeypatch.setattr(projects_adapter, "_load_bible", _load) monkeypatch.setattr( projects_adapter, "_build_project", lambda slug, bible: _make_minimal_project(slug), ) result = list_projects() assert [p.id for p in result] == ["loaded-one"] # Legacy is NOT a project_load_failure_isolated case — no fallback fires. assert _fallback_count(caplog, "project_load_failure_isolated") == 0 def test_list_projects_isolates_late_legacy_error_from_list_episodes( monkeypatch, caplog ): """LegacyProjectFormatError raised LATER from inside list_episodes (i.e. inside _build_project) is treated as an unexpected failure and isolated via the fallback — pre-fix it bubbled up and 500'd.""" caplog.set_level(logging.WARNING) _stub_slugs(monkeypatch, ["ok-project", "late-legacy", "also-ok"]) monkeypatch.setattr(projects_adapter, "_load_bible", lambda slug: {"project": slug}) def _build(slug, bible): if slug == "late-legacy": raise LegacyProjectFormatError(slug, "list_episodes detected legacy shape") return _make_minimal_project(slug) monkeypatch.setattr(projects_adapter, "_build_project", _build) result = list_projects() assert [p.id for p in result] == ["ok-project", "also-ok"] assert _fallback_count(caplog, "project_load_failure_isolated") == 1 # ── list_legacy_project_warnings ───────────────────────────────────────── def test_list_legacy_warnings_continues_past_unexpected_exception( monkeypatch, caplog ): """Unexpected exception during _load_bible MUST NOT short-circuit the loop. The bad slug is recorded as LegacyWarning(reason='unexpected: ...') and remaining slugs still get evaluated.""" caplog.set_level(logging.WARNING) _stub_slugs(monkeypatch, ["legacy-1", "exploding-slug", "legacy-2", "loaded"]) def _load(slug): if slug == "exploding-slug": raise OSError("disk vanished") if slug.startswith("legacy"): raise LegacyProjectFormatError(slug, "missing global_bible.json") return {"project": slug} # "loaded" returns a real bible monkeypatch.setattr(projects_adapter, "_load_bible", _load) warnings = list_legacy_project_warnings() by_slug = {w.slug: w for w in warnings} assert "legacy-1" in by_slug assert "legacy-2" in by_slug assert "exploding-slug" in by_slug, ( "exploding-slug must still be reported as a warning, not crash the call" ) assert "loaded" not in by_slug, "non-legacy slugs are not warnings" assert by_slug["exploding-slug"].reason == "unexpected: OSError" assert by_slug["legacy-1"].reason == "missing global_bible.json" # And the fallback fired exactly once for the unexpected slug. assert _fallback_count(caplog, "project_load_failure_isolated") == 1 assert any( "FALLBACK_FIRED" in r.message and "project_load_failure_isolated" in r.message for r in caplog.records ) def test_list_legacy_warnings_returns_legacywarning_instances(monkeypatch): """Type sanity: warnings returned for both legacy and unexpected paths are LegacyWarning instances (not bare dicts).""" _stub_slugs(monkeypatch, ["legacy", "boom"]) def _load(slug): if slug == "boom": raise RuntimeError("anything") raise LegacyProjectFormatError(slug, "missing") monkeypatch.setattr(projects_adapter, "_load_bible", _load) warnings = list_legacy_project_warnings() assert all(isinstance(w, LegacyWarning) for w in warnings) assert {w.slug for w in warnings} == {"legacy", "boom"} class TestEpisodeSynthesisGating: """Build B Phase 7: episode synthesis is gated by Project.supports_episodes.""" def test_microdrama_returns_synthesized_episodes(self): from recoil.api.adapters.projects import get_project as adapter_get_project from recoil.core.project import get_project as core_get_project core_get_project.cache_clear() proj = adapter_get_project("tartarus") assert proj is not None assert len(proj.episodes) > 0 def test_client_video_returns_empty_episodes(self): from recoil.api.adapters.projects import get_project as adapter_get_project from recoil.core.project import get_project as core_get_project core_get_project.cache_clear() proj = adapter_get_project("driver-beware") assert proj is not None assert proj.episodes == [] class TestProjectTypeOnWire: def setup_method(self): from recoil.core.project import get_project as core_get_project core_get_project.cache_clear() def test_driver_beware_projectType_is_client_video(self): from recoil.api.adapters.projects import get_project as adapter_get_project proj = adapter_get_project("driver-beware") assert proj is not None assert proj.project_type == "client_video" def test_tartarus_projectType_is_microdrama(self): from recoil.api.adapters.projects import get_project as adapter_get_project proj = adapter_get_project("tartarus") assert proj is not None assert proj.project_type == "microdrama"