"""Tenet 6 compliance regression tests. This file aggregates the regression tests added during Phases E.6/E.7/E.8 as each silent-swallow site was fixed. Each test exercises the new raise / typed-failure-return / observable-fallback path of one specific site from the Phase E.1 inventory at ``recoil/docs/silent-failure-inventory.md``. Test naming convention: test__ Tests are grouped into one ``Test`` class per site so future phases can extend a class with additional invariants without scrambling diff history. When a future site is fixed, the regression test goes here. The file should grow monotonically — tests are not removed when refactors land. If a fixed site is removed by a future refactor, its test moves to a "removed-because-site-was-deleted" comment block at the bottom of the file with the date and PR/commit reference. ## Phase E.6 — Sites #1–#10 (sidecar/verdict corruption + ref dimension) #1 recoil/core/ref_resolver.py:113 RefDimensionUnknownError #2 recoil/workspace/sidecar.py:200 SidecarCorruptError #3 recoil/workspace/verdict.py:239 VerdictCorruptError #4 recoil/workspace/verdict.py:322 EditorialConfigCorruptError #5 recoil/workspace/verdict.py:381 ImportError narrowed (sanctioned) #6 recoil/workspace/verdict.py:386 VerdictAutofillError (init) #7 recoil/workspace/verdict.py:411 VerdictAutofillError (lookup) #8 recoil/workspace/coverage.py:66 ExecutionStoreUnavailableError #9 recoil/workspace/tree.py:226 SidecarCorruptError (logged, skipped) #10 recoil/workspace/tree.py:615 ExecutionStoreUnavailableError ## Phase E.7 — Sites #11–#20 (orchestrator + workspace/server.py + dailies) #11 recoil/workspace/tree.py:252 SidecarCorruptError (take.json) #12 recoil/workspace/sidecar.py:439 CastingFragmentCorruptError #13 recoil/pipeline/orchestrator/pipeline.py Narrowed except clause #14 recoil/pipeline/lib/keyframe_context.py:697 KeyframeContextLookupError #15 recoil/pipeline/lib/run_shot.py:211 ModelProfileLookupError + budget_estimate_unknown_model_default sanctioned-fallback #16 recoil/workspace/server.py:527 MediaProbeError #17 recoil/workspace/server.py:2538 SidecarCorruptError (orphan reclaim) #18 recoil/workspace/server.py:2902 SidecarCorruptError (covered by #2) #19 recoil/workspace/server.py:3020 SidecarCorruptError (reject path) #20 recoil/pipeline/api/routes/dailies.py:908 RecommendationsCorruptError ## Phase E.8 — Sites #21–#30 (config loaders + state quarantine + observability) #21 recoil/workspace/state.py:84 WorkspaceStateCorruptError + quarantine #22 recoil/lib/prompt_validators.py:146 observability-only (log.warning) #23 recoil/lib/config_loader.py:134 ConfigParseError #24 recoil/lib/config_loader.py:186 ConfigParseError (covered by #23) #25 recoil/lib/prompt_compiler.py:347 PromptCompilerOverridesCorruptError #26 recoil/lib/prompt_compiler.py:413 PromptCompilerOverridesCorruptError (covered) #27 recoil/core/prompt_config.py:89 ConfigParseError #28 recoil/pipeline/lib/run_shot.py:309 model_profile_feature_flag_default sanctioned-fallback #29 recoil/pipeline/lib/run_shot.py:723 observability-only (log.exception) #30 recoil/workspace/server.py:2368 observability-only (log.warning) Sites with "covered by #N" in the disposition share a regression test with the named site (the fix shape is identical; one test asserts both). Sites tagged "observability-only" do not have dedicated tests — the inventory checkbox documents the intent. """ from __future__ import annotations import sys from pathlib import Path import pytest # ── Path setup so `lib.exceptions` resolves to recoil/lib/exceptions.py ── # The Phase 7 additions also exercise pipeline-rooted modules # (pipeline._lib.run_shot, pipeline._lib.keyframe_context, # pipeline.api.routes.dailies, pipeline.orchestrator.pipeline) which # import via `from lib.x import ...` — so pipeline/ must also be on # sys.path AFTER recoil/. RECOIL_ROOT must come FIRST so # `from core.x import ...` resolves to recoil/core/ (not # pipeline/core/, which would shadow it). Same order as # pipeline/tests/conftest.py. # # Pytest may pre-inject either RECOIL_ROOT or the tests dir into # sys.path at session start (importmode=prepend). Naive # `insert(0, …)` guarded by `if … not in sys.path` is a no-op when # the path is already present further down the list — leaving the # pipeline shadow in slot 0. Force-evict any existing entries first, # then insert in the canonical order. _RECOIL_ROOT = Path(__file__).resolve().parent.parent _PIPELINE_ROOT = _RECOIL_ROOT / "pipeline" for _candidate in (str(_RECOIL_ROOT), str(_PIPELINE_ROOT)): while _candidate in sys.path: sys.path.remove(_candidate) # Insert reverse-order so final layout = [RECOIL_ROOT, PIPELINE_ROOT, …]. sys.path.insert(0, str(_PIPELINE_ROOT)) sys.path.insert(0, str(_RECOIL_ROOT)) from recoil.core.exceptions import ( # noqa: E402 CastingFragmentCorruptError, ConfigParseError, EditorialConfigCorruptError, ExecutionStoreUnavailableError, KeyframeContextLookupError, MediaProbeError, ModelProfileLookupError, PromptCompilerOverridesCorruptError, RecommendationsCorruptError, RefDimensionUnknownError, SidecarCorruptError, VerdictAutofillError, VerdictCorruptError, ) def _import_recoil_lib_module(module_name: str): """Import a module that historically lived in ``recoil/lib/`` and now lives in ``recoil/core/`` (post-commit 826c8d65 lib→core migration). The legacy version of this helper bypassed regular package resolution via ``importlib.util.spec_from_file_location`` to dodge a shadowing conflict with ``recoil/pipeline/lib/``. After the migration there is no shadowing — ``recoil/core/`` and ``recoil/pipeline/core/`` are distinct packages — so a normal ``from core import …`` works. """ import importlib return importlib.import_module(f"recoil.core.{module_name}") # ============================================================================= # Section: ref_resolver — RefDimensionUnknownError (Site #1) # ============================================================================= class TestRefResolverDimensionRaise: """Site #1: recoil/core/ref_resolver.py:113. Was: ``except Exception: return None`` after Image.open() — corrupt / unreadable images silently produced no dimensions, so downstream prompt builder used defaults and aspect-ratio computation went undefined. """ def test_ref_resolver_raises_on_unprobeable_image(self, tmp_path): from recoil.core.ref_resolver import get_dimensions bad = tmp_path / "not_an_image.png" bad.write_bytes(b"definitely not an image") with pytest.raises(RefDimensionUnknownError) as exc_info: get_dimensions(bad) assert str(bad) in str(exc_info.value) # ============================================================================= # Section: workspace/sidecar — SidecarCorruptError (Site #2) # ============================================================================= class TestReadSidecarCorruptRaise: """Site #2: recoil/workspace/sidecar.py:200. Was: ``except (json.JSONDecodeError, IOError): return None`` collapsed "no sidecar exists" with "sidecar exists but is corrupt." Now those two conditions are distinguished — missing → None, corrupt → raise. """ def test_read_sidecar_raises_on_corrupt_json(self, tmp_path): from recoil.workspace.sidecar import read_sidecar media = tmp_path / "img.png" media.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 32) bad_sidecar = tmp_path / "img.png.json" bad_sidecar.write_text("{not valid json}") with pytest.raises(SidecarCorruptError) as exc_info: read_sidecar(media) assert str(bad_sidecar) in str(exc_info.value) def test_read_sidecar_returns_none_when_missing(self, tmp_path): # Regression-protect the "no sidecar" branch: it MUST stay None, # not raise. Site #2 fix split FileNotFound vs corrupt. from recoil.workspace.sidecar import read_sidecar media = tmp_path / "img.png" media.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 32) # No sidecar written → returns None, no raise. assert read_sidecar(media) is None # ============================================================================= # Section: workspace/verdict.read_verdict — VerdictCorruptError (Site #3) # ============================================================================= class TestReadVerdictCorruptRaise: """Site #3: recoil/workspace/verdict.py:239. Was: ``except (FileNotFoundError, json.JSONDecodeError): return None``. Now: missing → None (preserved), corrupt JSON → raise VerdictCorruptError. """ def test_read_verdict_raises_on_corrupt_json(self, tmp_path): from recoil.workspace.verdict import read_verdict bad = tmp_path / "shot_take1_verdict.json" bad.write_text("{not valid json") with pytest.raises(VerdictCorruptError) as exc_info: read_verdict(bad) assert str(bad) in str(exc_info.value) def test_read_verdict_returns_none_when_missing(self, tmp_path): from recoil.workspace.verdict import read_verdict missing = tmp_path / "no_such_verdict.json" assert read_verdict(missing) is None # ============================================================================= # Section: workspace/verdict._load_meta_yaml — EditorialConfigCorruptError (Site #4) # ============================================================================= class TestLoadMetaYamlCorruptRaise: """Site #4: recoil/workspace/verdict.py:322. Was: ``except (yaml.YAMLError, FileNotFoundError, OSError): return None`` silently masked editorial-config corruption as "no config." Now: missing → None, malformed YAML → raise EditorialConfigCorruptError. """ def test_verdict_yaml_loader_raises_on_corrupt(self, tmp_path): from recoil.workspace.verdict import _load_meta_yaml bad = tmp_path / "SH01_meta.yaml" # Indentation/tab error → YAMLError bad.write_text("generation:\n\tprompt: 'unclosed\n bad: indent") with pytest.raises(EditorialConfigCorruptError) as exc_info: _load_meta_yaml(bad) assert str(bad) in str(exc_info.value) def test_verdict_yaml_loader_returns_none_when_missing(self, tmp_path): from recoil.workspace.verdict import _load_meta_yaml missing = tmp_path / "no_such_meta.yaml" assert _load_meta_yaml(missing) is None # ============================================================================= # Section: verdict autofill — VerdictAutofillError (Sites #5, #6, #7) # ============================================================================= class TestVerdictAutofillRaise: """Sites #5/#6/#7: recoil/workspace/verdict.py:381/386/411. Site #5 narrowed the ExecutionStore IMPORT swallow to ImportError only — runtime failures during use must propagate. Site #6 raises VerdictAutofillError on store-init failure (was: ``return {}`` silently). Site #7 raises VerdictAutofillError on store.get_shot()/take lookup failure (was: ``return {}`` silently). """ def test_verdict_autofill_raises_on_store_init_failure(self, tmp_path, monkeypatch): # Patch ExecutionStore so __init__ raises a runtime error (not # ImportError). The new code path raises VerdictAutofillError. import recoil.execution.execution_store as es class BoomStoreInit: def __init__(self, project): raise RuntimeError("simulated DB locked") monkeypatch.setattr(es, "ExecutionStore", BoomStoreInit) from recoil.workspace.verdict import _try_execution_store_lookup with pytest.raises(VerdictAutofillError) as exc_info: _try_execution_store_lookup("any_project", "EP001_SH01", 1) assert "executionstore_init" in str(exc_info.value) def test_verdict_autofill_raises_on_lookup_failure(self, tmp_path, monkeypatch): # Patch ExecutionStore so init succeeds but get_shot raises — new # path raises VerdictAutofillError(source=executionstore_lookup). import recoil.execution.execution_store as es class BoomLookupStore: def __init__(self, project): self.project = project def get_shot(self, shot_id): raise RuntimeError("simulated schema mismatch") def close(self): pass monkeypatch.setattr(es, "ExecutionStore", BoomLookupStore) from recoil.workspace.verdict import _try_execution_store_lookup with pytest.raises(VerdictAutofillError) as exc_info: _try_execution_store_lookup("any_project", "EP001_SH01", 1) assert "executionstore_lookup" in str(exc_info.value) def test_verdict_autofill_propagates_real_errors(self, tmp_path, monkeypatch): # Site #5 narrowed: the import swallow used to absorb ANY exception # (return {}). Now only ImportError is sanctioned. We can't easily # mock an import failure inside a function, so this test asserts # the runtime-error path (Site #6 / #7) propagates rather than # being silently swallowed — same property Site #5 protects. import recoil.execution.execution_store as es class BoomStore: def __init__(self, project): raise TypeError("schema-mismatch (not ImportError)") monkeypatch.setattr(es, "ExecutionStore", BoomStore) from recoil.workspace.verdict import _try_execution_store_lookup # Must NOT silently return {} — must raise the canonical type. with pytest.raises(VerdictAutofillError): _try_execution_store_lookup("any_project", "EP001_SH01", 1) # ============================================================================= # Section: workspace/coverage — ExecutionStoreUnavailableError (Site #8) # ============================================================================= class TestCoverageSummaryStoreUnavailable: """Site #8: recoil/workspace/coverage.py:66. Was: ``except Exception: total = 0`` (TODO-PHASE-E removed). Silent store-failure produced 0/0, which was rendered to JT as 100% coverage — false signal. Now: raises ExecutionStoreUnavailableError; HTTP route surfaces as 503. """ def test_coverage_summary_raises_on_store_unavailable(self, monkeypatch): from workspace import coverage def boom(_project): raise RuntimeError("simulated store crash") monkeypatch.setattr(coverage, "_get_store", boom) with pytest.raises(ExecutionStoreUnavailableError) as exc_info: coverage.coverage_summary_for_episode("any_project", "EP001") assert "any_project" in str(exc_info.value) # ============================================================================= # Section: workspace/tree — SidecarCorruptError observable + ExecutionStoreUnavailableError (Sites #9, #10) # ============================================================================= class TestTreeBuilderSidecarObservable: """Site #9: recoil/workspace/tree.py:226. Was: ``except Exception: continue`` swallowed every error in the canonical-ref _meta JSON loop, hiding corrupt sidecars from the tree silently. Now: corrupt JSON / OSError logged-and-skipped (visible); real errors propagate. """ def test_tree_builder_skips_corrupt_meta_sidecars_loudly( self, tmp_path, monkeypatch, caplog ): # Build a project layout with a corrupt _meta canonical ref sidecar. from workspace import tree as ws_tree project = "test_project_phase_e6" project_dir = tmp_path / project # v2 layout: ref sidecars live at assets/{kind}/{subject}/_meta/*.json # (the v1 output/refs/_canonical/ tree was deleted in the paths refactor). meta_dir = project_dir / "assets" / "identity" / "test" / "_meta" meta_dir.mkdir(parents=True) # Write a corrupt _meta sidecar file bad_meta = meta_dir / "hero.png.json" bad_meta.write_text("{not valid json{") (tmp_path / ".recoil-data-root").write_text("recoil-data-root\n") monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) # build_metadata_index also touches the ExecutionStore — short-circuit. class EmptyStore: def get_all_shots(self): return [] def close(self): pass monkeypatch.setattr(ws_tree, "_get_store", lambda _p: EmptyStore()) # Must not crash — the corrupt sidecar is skipped with a WARNING log. import logging with caplog.at_level(logging.WARNING, logger="recoil.workspace.tree"): index = ws_tree.build_metadata_index(project) # Index built successfully (no entry from the corrupt sidecar). assert isinstance(index, dict) assert any( "corrupt asset-ref sidecar" in rec.message for rec in caplog.records ), "expected WARNING log for corrupt _meta sidecar" class TestTreeUncoveredStoreUnavailable: """Site #10: recoil/workspace/tree.py:615. Was: ``except Exception: all_shot_ids = []`` (TODO-PHASE-E removed) silently emptied the uncovered-shot list, hiding all unrendered shots from JT's review queue. Now: raises ExecutionStoreUnavailableError. """ def test_tree_uncovered_raises_on_store_failure(self, monkeypatch, tmp_path): from workspace import tree as ws_tree # group_by_pass_anchors hits _get_store inside the uncovered-shot # block. Patch it to raise. monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) def boom(_project): raise RuntimeError("simulated store failure") monkeypatch.setattr(ws_tree, "_get_store", boom) # Patch PassStore so list_passes returns no passes (the function # reaches the uncovered block only after PassStore traversal). class FakePassStore: def __init__(self, project): pass def list_passes(self, episode_id): return [] def close(self): pass # group_by_pass_anchors does `from recoil.execution.pass_store import PassStore` # at function scope, so we patch the module attribute it'll resolve. import recoil.execution.pass_store as ps_mod monkeypatch.setattr(ps_mod, "PassStore", FakePassStore) # workspace.coverage also resolves PassStore at function scope. # The uncovered-shot try/except runs BEFORE coverage_summary_for_episode # in our code path? Actually it runs AFTER. We need coverage_summary to # not blow up on the patched store, so we patch coverage._get_store too # — but wait, _get_store is the very thing we want to raise. To isolate # the uncovered-block path, patch coverage_summary_for_episode itself. from workspace import coverage as ws_coverage monkeypatch.setattr( ws_coverage, "coverage_summary_for_episode", lambda project, episode_id: { "type": "coverage_summary", "name": "stub", "covered": 0, "total": 0, "awaiting": 0, "review": 0, }, ) monkeypatch.setattr( ws_coverage, "recent_activity_for_episode", lambda project, episode_id: None, ) with pytest.raises(ExecutionStoreUnavailableError) as exc_info: ws_tree.group_by_pass_anchors([], "Episode 001", project="any_project") assert "any_project" in str(exc_info.value) # ============================================================================= # Phase E.7 — Sites #11-#20 # ============================================================================= # # Site #11 recoil/workspace/tree.py:252 SidecarCorruptError (logged + skipped) # Site #12 recoil/workspace/sidecar.py:439 CastingFragmentCorruptError # Site #13 recoil/pipeline/orchestrator/pipeline.py:1863 narrowed except clause # Site #14 recoil/pipeline/lib/keyframe_context.py:697 KeyframeContextLookupError # Site #15 recoil/pipeline/lib/run_shot.py:211 ModelProfileLookupError # Site #16 recoil/workspace/server.py:527 MediaProbeError vs FileNotFoundError # Site #17 recoil/workspace/server.py:2538 SidecarCorruptError (orphan reclaim) # Site #18 recoil/workspace/server.py:2902 covered by Site #2 boundary # Site #19 recoil/workspace/server.py:3020 SidecarCorruptError (reject path) # Site #20 recoil/pipeline/api/routes/dailies.py:908 RecommendationsCorruptError # ============================================================================= # Section: workspace/tree take/universal sidecar — Site #11 # ============================================================================= class TestTreeTakeLoaderSurfacesCorrupt: """Site #11: recoil/workspace/tree.py:252. Was: ``except (json.JSONDecodeError, IOError): continue`` — corrupt take.json silently dropped from the workspace tree (the row simply disappeared, with no diagnostic for the operator). Now: WARNING log is emitted before the skip so the failure is observable. """ def test_tree_take_loader_surfaces_corrupt(self, tmp_path, monkeypatch, caplog): from workspace import tree as ws_tree project = "test_project_phase_e7_site11" project_dir = tmp_path / project # Write a media file + corrupt universal sidecar under a v2 media root # (renders/); the v1 output/video/ tree was deleted in the paths refactor. ep_dir = project_dir / "renders" / "ep_001" ep_dir.mkdir(parents=True) media = ep_dir / "EP001_PASS_001_take1.mp4" media.write_bytes(b"\x00\x00\x00\x18ftypmp42") bad_sc = ep_dir / "EP001_PASS_001_take1.mp4.json" bad_sc.write_text("{not valid json{") (tmp_path / ".recoil-data-root").write_text("recoil-data-root\n") monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) class EmptyStore: def get_all_shots(self): return [] def close(self): pass monkeypatch.setattr(ws_tree, "_get_store", lambda _p: EmptyStore()) import logging with caplog.at_level(logging.WARNING, logger="recoil.workspace.tree"): index = ws_tree.build_metadata_index(project) # Index built — no row added from the corrupt sidecar. assert isinstance(index, dict) assert any( "skipping corrupt take metadata" in rec.message for rec in caplog.records ), "expected WARNING log for corrupt universal sidecar" # ============================================================================= # Section: workspace/sidecar casting fragment — Site #12 # ============================================================================= class TestSidecarWriteRaisesOnCorruptCasting: """Site #12: recoil/workspace/sidecar.py:439. Was: ``except (json.JSONDecodeError, IOError): casting = {}`` — corrupt casting_state.json silently emptied character refs and the promote proceeded with no provenance. Now: ``CastingFragmentCorruptError`` raised so the caller can refuse rather than continue with empty casting. """ def test_sidecar_write_raises_on_corrupt_casting(self, tmp_path): from workspace import sidecar as ws_sidecar project_dir = tmp_path / "test_project_phase_e7_site12" # Make a media file + initial sidecar so promote has something to read. chars_dir = project_dir / "output" / "refs" / "candidates" / "characters" chars_dir.mkdir(parents=True) media = chars_dir / "char_a.png" media.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 32) # Pre-stamp a valid candidate sidecar so read_sidecar() succeeds. ws_sidecar.ensure_sidecar(media) # Pre-corrupt casting_state.json at the path promote_to_canonical # actually reads (ProjectPaths v3: _pipeline/state/visual/...). from recoil.core.paths import ProjectPaths casting_state = ProjectPaths.from_root(project_dir).casting_state_path casting_state.parent.mkdir(parents=True, exist_ok=True) casting_state.write_text("{not json") with pytest.raises(CastingFragmentCorruptError) as exc_info: ws_sidecar.promote_to_canonical( media_path=media, project_dir=project_dir, asset_type="characters", entity_id="char_a", ) assert str(casting_state) in str(exc_info.value) # ============================================================================= # Section: pipeline.py character handoff narrowed — Site #13 # ============================================================================= class TestCharacterHandoffPropagatesSchemaErrors: """Site #13: recoil/pipeline/orchestrator/pipeline.py:1863. Was: outer ``except Exception:`` caught corrupt-bible / schema-mismatch failures (JSONDecodeError, ValueError, TypeError) and silently fell back to the legacy resolver — which itself swallowed FileNotFound / KeyError into an empty-wardrobe character. Now: only the documented "missing data" failure modes (FileNotFoundError, KeyError, AttributeError) trigger the fallback chain; schema errors propagate. """ def test_character_handoff_propagates_schema_errors(self, monkeypatch): # The narrowed except clause lives on MultiShotPipeline._resolve_char_data # at pipeline/orchestrator/pipeline.py:1846. Patch validate_handoff to # raise a schema-style error (JSONDecodeError) and assert it propagates # past the narrowed except (which now only catches FileNotFoundError / # KeyError / AttributeError). import recoil.pipeline.orchestrator.pipeline as pipeline_mod import json as _json try: import recoil.pipeline._lib.render_schema # noqa: F401 except ImportError: pytest.skip( "recoil.pipeline._lib.render_schema not importable in this environment" ) class StubPipeline: breakdown = {} def boom_handoff(*args, **kwargs): raise _json.JSONDecodeError("simulated bible parse error", "", 0) monkeypatch.setattr( "recoil.pipeline._lib.render_schema.validate_handoff", boom_handoff ) # Method lives on ShotAssembler (per inventory Site #13 line 1863 # = body of _resolve_char_data, defined on the assembler). method = pipeline_mod.ShotAssembler._resolve_char_data with pytest.raises(_json.JSONDecodeError): method(StubPipeline(), "wren", 1, {}) # ============================================================================= # Section: keyframe_context lookup — Site #14 # ============================================================================= class TestKeyframeContextPropagatesLookupErrors: """Site #14: recoil/pipeline/lib/keyframe_context.py:697. Was: ``except Exception: return None`` masked corrupt execution_state.json AND missing-data alike. Now: missing data still returns None, but a corrupt state file raises ``KeyframeContextLookupError`` so the caller doesn't silently build a drifted video without the keyframe anchor. """ def test_keyframe_context_propagates_lookup_errors(self, tmp_path, monkeypatch): from recoil.pipeline._lib import keyframe_context as kc_mod from recoil.core.paths import ProjectPaths project = "test_project_phase_e7_site14" # Stage a corrupt execution_state.json under the path ProjectPaths # produces (_pipeline/state//execution_state.json). (tmp_path / ".recoil-data-root").write_text("recoil-data-root\n") (tmp_path / project).mkdir(parents=True) monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) state_dir = ProjectPaths.for_project(project).visual_state_dir state_dir.mkdir(parents=True) (state_dir / "execution_state.json").write_text("{not json") with pytest.raises(KeyframeContextLookupError) as exc_info: kc_mod._find_keyframe_prompt("EP001_SH01", project) assert "EP001_SH01" in str(exc_info.value) def test_keyframe_context_returns_none_when_state_missing( self, tmp_path, monkeypatch ): # Genuinely-missing state file is the sanctioned None path; new # code must NOT raise here. from recoil.pipeline._lib import keyframe_context as kc_mod project = "test_project_phase_e7_site14_missing" (tmp_path / ".recoil-data-root").write_text("recoil-data-root\n") (tmp_path / project).mkdir(parents=True) monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) result = kc_mod._find_keyframe_prompt("EP001_SH01", project) assert result is None # ============================================================================= # Section: run_shot estimated cost — Site #15 # ============================================================================= class TestEstimatedCostPropagatesLookupErrors: """Site #15: recoil/pipeline/lib/run_shot.py:211. Was: ``except Exception: return ESTIMATED_COST_PER_ATTEMPT`` — model profile lookup failure silently substituted a fixed estimate, masking both budget-cap surprises AND silent overruns. Now: lookup failure raises ``ModelProfileLookupError`` so the caller decides whether to fall back explicitly with annotated provenance. """ def test_estimated_cost_propagates_lookup_errors(self, monkeypatch): from recoil.core import model_profiles from recoil.pipeline._lib import run_shot as run_shot_mod def boom_get_profile(_model): raise KeyError("simulated missing model profile") monkeypatch.setattr(model_profiles, "get_profile", boom_get_profile) with pytest.raises(ModelProfileLookupError) as exc_info: run_shot_mod._get_estimated_cost("imaginary-model-xyz") assert "imaginary-model-xyz" in str(exc_info.value) # ============================================================================= # Section: workspace/server probe — Site #16 # ============================================================================= class TestDurationProbeDistinguishesMissingFromFailed: """Site #16: recoil/workspace/server.py:527. Was: ``except Exception: pass; return None`` — collapsed "ffprobe binary missing" with "ffprobe ran and choked on this media." Now: binary-missing still returns None (sanctioned fallback); subprocess failure raises ``MediaProbeError``. """ def test_duration_probe_returns_none_when_binary_missing( self, monkeypatch, tmp_path ): from workspace import server as ws_server import subprocess def boom_run(*args, **kwargs): raise FileNotFoundError("ffprobe not on PATH") monkeypatch.setattr(subprocess, "run", boom_run) media = tmp_path / "vid.mp4" media.write_bytes(b"\x00") assert ws_server._probe_duration_seconds(media) is None def test_duration_probe_raises_on_subprocess_failure(self, monkeypatch, tmp_path): from workspace import server as ws_server import subprocess def boom_run(*args, **kwargs): raise subprocess.SubprocessError("simulated ffprobe crash") monkeypatch.setattr(subprocess, "run", boom_run) media = tmp_path / "vid.mp4" media.write_bytes(b"\x00") with pytest.raises(MediaProbeError) as exc_info: ws_server._probe_duration_seconds(media) assert str(media) in str(exc_info.value) # ============================================================================= # Section: orphan reclaim sidecar — Site #17 # ============================================================================= class TestOrphanReclaimPreservesProvenance: """Site #17: recoil/workspace/server.py:2538. Was: ``except Exception: sc_data = {}`` — corrupt source sidecar silently re-created as empty ``{}`` during orphan reclaim, destroying provenance. Now: the read helper raises ``SidecarCorruptError`` and the route surfaces a 500 instead of obliterating metadata. """ def test_orphan_reclaim_preserves_provenance_or_raises(self, tmp_path): from recoil.workspace.server import _read_orphan_sidecar_for_reclaim # Missing sidecar → empty dict, did_exist=False, no raise. missing = tmp_path / "no_sidecar.mp4.json" sc, existed = _read_orphan_sidecar_for_reclaim(missing) assert sc == {} assert existed is False # Corrupt sidecar → raises SidecarCorruptError. bad = tmp_path / "bad_sidecar.mp4.json" bad.write_text("{not valid json") with pytest.raises(SidecarCorruptError) as exc_info: _read_orphan_sidecar_for_reclaim(bad) assert str(bad) in str(exc_info.value) # ============================================================================= # Section: reject_segment sidecar overwrite — Site #19 # ============================================================================= class TestRejectPathDoesNotOverwriteCorruptSidecar: """Site #19: recoil/workspace/server.py:3020. Was: ``except Exception: sc = {}`` followed by ``sc["status"] = "rejected"`` — a corrupt sidecar on the reject path was silently mutated to ``{status: "rejected"}`` and re-saved, overwriting whatever provenance the corruption was hiding. Now: the read goes through ``ws_sidecar.read_sidecar()`` which propagates ``SidecarCorruptError``; the reject endpoint converts that to 500 instead of overwriting. """ def test_reject_path_does_not_overwrite_corrupt_sidecar(self, tmp_path): # Exercise the boundary: workspace.sidecar.read_sidecar must raise # when the sidecar is corrupt, which is what the reject_segment # route now relies on. The end-to-end HTTP path requires a # full projects_root() and PassStore; the contract we protect here # is that the read raises rather than returning {} (which is what # the route catches). from workspace import sidecar as ws_sidecar media = tmp_path / "shot_001_FROM_PASS_001_seg.mp4" media.write_bytes(b"\x00\x00\x00\x18ftypmp42") bad_sc = tmp_path / "shot_001_FROM_PASS_001_seg.mp4.json" original = bad_sc.read_text() if bad_sc.exists() else None bad_sc.write_text("{corrupt content") with pytest.raises(SidecarCorruptError): ws_sidecar.read_sidecar(media) # The on-disk file must be byte-stable — read MUST NOT mutate. assert bad_sc.read_text() == "{corrupt content" # And the regression-protect: an explicit reject helper would NOT # have written status=rejected on top of corruption (route now # short-circuits to 500 instead). assert original is None or bad_sc.read_text() != original # ============================================================================= # Section: dailies recommendations — Site #20 # ============================================================================= class TestDailiesRecommendationsPreservesCorrupt: """Site #20: recoil/pipeline/api/routes/dailies.py:908. Was: ``except Exception: existing = {}`` — corrupt mark-seen / recommendations file silently overwritten on the next append, losing prior accept/reject context. Now: the read helper raises ``RecommendationsCorruptError`` and the route surfaces a 500. """ def test_dailies_recommendations_preserves_corrupt(self, tmp_path): # routes/__init__ -> console.py does a bare `from api import state`, # which only resolves to pipeline/api (has state.py) when PIPELINE_ROOT # leads sys.path. Force it for the duration of the import, then restore # so `from core.x` resolution for later tests is unaffected. from recoil.core.paths import PIPELINE_ROOT _saved_path = list(sys.path) sys.path.insert(0, str(PIPELINE_ROOT)) try: from recoil.pipeline.api.routes.dailies import ( _read_recommendations_or_raise, ) finally: sys.path[:] = _saved_path # Missing file → empty dict, no raise. missing = tmp_path / "no_seen.json" assert _read_recommendations_or_raise(missing) == {} # Corrupt file → raises RecommendationsCorruptError. bad = tmp_path / "mobile_seen.json" bad.write_text("{not json") with pytest.raises(RecommendationsCorruptError) as exc_info: _read_recommendations_or_raise(bad) assert str(bad) in str(exc_info.value) # And the on-disk file is byte-stable — corruption is preserved # for the operator to inspect, not silently overwritten. assert bad.read_text() == "{not json" # ============================================================================= # Phase E.8 — Sites #21-#30 # ============================================================================= # # Site #21 recoil/workspace/state.py:84 quarantine + WARNING log # Site #22 recoil/lib/prompt_validators.py:146 WARNING log only # Site #23 recoil/lib/config_loader.py:134 ConfigParseError # Site #24 recoil/lib/config_loader.py:186 covered by Site #23 pattern # Site #25 recoil/lib/prompt_compiler.py:347 PromptCompilerOverridesCorruptError # Site #26 recoil/lib/prompt_compiler.py:413 covered by Site #25 pattern # Site #27 recoil/core/prompt_config.py:89 ConfigParseError # Site #28 recoil/pipeline/lib/run_shot.py:309 sanctioned-fallback firing # Site #29 recoil/pipeline/lib/run_shot.py:723 observability-only (no test) # Site #30 recoil/workspace/server.py:2368 observability-only (no test) # ============================================================================= # Section: workspace/state quarantine — Site #21 # ============================================================================= class TestWorkspaceStateDoesNotSilentlyReset: """Site #21: recoil/workspace/state.py:84. Was: ``except (json.JSONDecodeError, IOError): return _fresh_default()`` silently reset the workspace state on corrupt JSON, destroying user selections / viewer position. Now: corrupt state.json is quarantined to ``state.corrupt..json`` and a WARNING is logged before the fresh-default return — operator can recover the prior state from the quarantined file rather than losing it forever. """ def test_workspace_state_does_not_silently_reset( self, tmp_path, monkeypatch, caplog ): from workspace import state as ws_state # Redirect state to tmp_path so we don't touch the user's real state. bad_state = tmp_path / "state.json" bad_state.write_text("{not valid json") monkeypatch.setattr(ws_state, "_STATE_DIR", tmp_path) monkeypatch.setattr(ws_state, "_STATE_PATH", bad_state) import logging with caplog.at_level(logging.WARNING, logger="recoil.workspace.state"): result = ws_state.read_state() # Returns fresh defaults rather than raising. assert isinstance(result, dict) assert result.get("project") is None # Corrupt file was quarantined, not silently overwritten. assert not bad_state.exists(), ( "corrupt state.json should be quarantined (renamed), not left" " in place to be overwritten on next write" ) quarantined = list(tmp_path.glob("state.corrupt.*.json")) assert len(quarantined) == 1, ( f"expected one quarantined file; got {[p.name for p in quarantined]}" ) assert quarantined[0].read_text() == "{not valid json" # WARNING log emitted — operator can grep "quarantined" to find it. assert any("quarantined" in rec.message for rec in caplog.records), ( "expected WARNING log mentioning quarantine" ) # ============================================================================= # Section: prompt_validators overrides — Site #22 (observability) # ============================================================================= class TestPromptValidatorLogsOnCorrupt: """Site #22: recoil/lib/prompt_validators.py:146. Was: ``except (json.JSONDecodeError, OSError): pass`` — corrupt verb_patterns.json silently swallowed; loader fell back to built-ins with no diagnostic. Now: WARNING log before the fall-through so the operator knows the override file failed to parse. """ def test_prompt_validator_logs_on_corrupt(self, tmp_path, monkeypatch, caplog): # `lib` resolves to pipeline/lib (regular package shadows namespace). # Force resolution to recoil/lib via temporary sys.path manipulation. pv = _import_recoil_lib_module("prompt_validators") # Reset cached patterns to force a reload. monkeypatch.setattr(pv, "_loaded_patterns", None) # Stage a corrupt verb_patterns.json next to the prompt_validators # module by patching __file__-derived path. _load_patterns() reads # `Path(__file__).parent / "verb_patterns.json"`. Patch the module's # __file__ to point inside tmp_path. fake_module_dir = tmp_path / "fake_lib" fake_module_dir.mkdir() fake_module_file = fake_module_dir / "prompt_validators.py" fake_module_file.write_text("# stub") bad_overrides = fake_module_dir / "verb_patterns.json" bad_overrides.write_text("{not valid json") monkeypatch.setattr(pv, "__file__", str(fake_module_file)) import logging # caplog with no logger arg sets root level so the canonical # sanctioned-fallbacks logger emits regardless of root-vs-pipeline # import path. with caplog.at_level(logging.WARNING): micro, bare = pv._load_patterns() # Built-in patterns returned (fall-through). assert micro == pv._BUILTIN_MICRO_DETAIL_PATTERNS assert bare == pv._BUILTIN_BARE_PATTERNS # FALLBACK_FIRED log emitted (Tenet 6 prong-1: NAMED + observable). # Phase E debug R3 routed Site #22 through the sanctioned-fallback # registry instead of a plain logger.warning. assert any( "FALLBACK_FIRED" in rec.message and "prompt_validator_pattern_default_builtins" in rec.message for rec in caplog.records ), "expected FALLBACK_FIRED log for corrupt overrides" # ============================================================================= # Section: config_loader — Sites #23, #24 # ============================================================================= class TestConfigLoaderRaisesOnParseError: """Site #23: recoil/lib/config_loader.py:134. Was: ``except (json.JSONDecodeError, OSError): file_config = {}`` — a corrupt project_config.json silently fell back to defaults; user's customizations vanished without diagnostic. Now: missing file is OK (sanctioned), corrupt JSON raises ``ConfigParseError``. Site #24 (line 186, breakdown.json read in load_rendering_directives) is covered by the same pattern + this test class. """ def test_config_loader_raises_on_parse_error(self, tmp_path): config_loader = _import_recoil_lib_module("config_loader") # Build a project layout with a corrupt project_config.json project_dir = tmp_path / "test_project_phase_e8_site23" visual_dir = project_dir / "visual" visual_dir.mkdir(parents=True) bad_cfg = visual_dir / "project_config.json" bad_cfg.write_text("{not valid json") with pytest.raises(ConfigParseError) as exc_info: config_loader.load_project_config(project_dir) assert str(bad_cfg) in str(exc_info.value) def test_config_loader_returns_defaults_when_missing(self, tmp_path): config_loader = _import_recoil_lib_module("config_loader") # Project dir with no project_config.json → defaults returned, no raise. project_dir = tmp_path / "test_project_phase_e8_site23_missing" (project_dir / "visual").mkdir(parents=True) result = config_loader.load_project_config(project_dir) assert isinstance(result, dict) assert "camera_body" in result # default key present # ============================================================================= # Section: prompt_compiler overrides — Sites #25, #26 # ============================================================================= class TestPromptCompilerRaisesOnCorruptOverrides: """Site #25: recoil/lib/prompt_compiler.py:347. Was: ``except (json.JSONDecodeError, OSError): self.overrides = []`` — corrupt prompt_overrides.json silently emptied the override store, so the prompt was compiled WITHOUT the user's customizations. Now: missing file → empty list (sanctioned), corrupt JSON → ``PromptCompilerOverridesCorruptError``. Site #26 (line 413, NoteStore for prompt_notes.json) shares the same pattern + raises the same exception class — covered by this test. """ def test_prompt_compiler_raises_on_corrupt_overrides(self, tmp_path): pc = _import_recoil_lib_module("prompt_compiler") project_dir = tmp_path / "test_project_phase_e8_site25" visual_dir = project_dir / "visual" visual_dir.mkdir(parents=True) bad_overrides = visual_dir / "prompt_overrides.json" bad_overrides.write_text("{not valid json") with pytest.raises(PromptCompilerOverridesCorruptError) as exc_info: pc.OverrideStore(project_dir) assert str(bad_overrides) in str(exc_info.value) def test_prompt_compiler_returns_empty_when_missing(self, tmp_path): pc = _import_recoil_lib_module("prompt_compiler") project_dir = tmp_path / "test_project_phase_e8_site25_missing" (project_dir / "visual").mkdir(parents=True) # No prompt_overrides.json → empty list, no raise. store = pc.OverrideStore(project_dir) assert store.overrides == [] # ============================================================================= # Section: prompt_config corrupt — Site #27 # ============================================================================= class TestPromptConfigRaisesOnCorrupt: """Site #27: recoil/core/prompt_config.py:89. Was: ``except (json.JSONDecodeError, OSError): pass`` — a corrupt per-project prompt_constants.json silently fell through to the global constants, so the project override was ignored without diagnostic. Now: missing → pass-through to global (sanctioned), corrupt JSON → ``ConfigParseError``. """ def test_prompt_config_raises_on_corrupt(self, tmp_path): from recoil.core import prompt_config project_dir = tmp_path / "test_project_phase_e8_site27" project_dir.mkdir() bad_cfg = project_dir / "prompt_constants.json" bad_cfg.write_text("{not valid json") with pytest.raises(ConfigParseError) as exc_info: prompt_config.get_constant( "production", "camera_body", default="", project_dir=str(project_dir), ) assert str(bad_cfg) in str(exc_info.value) # ============================================================================= # Section: run_shot sibling-refs profile failure — Site #28 # ============================================================================= class TestSiblingRefsFiresSanctionedFallbackOnProfileFailure: """Site #28: recoil/pipeline/lib/run_shot.py:309. Was: ``except Exception: pass`` after model_profiles.get_profile(model) — silent fall-through to enable_sibling_refs=True with no diagnostic. Now: registered as sanctioned fallback ``model_profile_feature_flag_default`` that fires a FALLBACK_FIRED log emission AND substitutes the default. """ def test_sibling_refs_fires_sanctioned_fallback_on_profile_failure( self, monkeypatch, caplog ): # Importing run_shot triggers the module-level # register_sanctioned_fallback("model_profile_feature_flag_default") call. import recoil.pipeline._lib.run_shot # noqa: F401 from recoil.pipeline._lib.sanctioned_fallbacks import get_sanctioned_fallback # Verify the registration exists + is correctly named. record = get_sanctioned_fallback("model_profile_feature_flag_default") assert record.introduced_in == "Phase E.8" assert "enable_sibling_refs" in record.justification # Patch model_profiles.get_profile to raise — this is the path the # except block in run_shot covers. from recoil.core import model_profiles from recoil.pipeline._lib.sanctioned_fallbacks import fire_sanctioned_fallback def boom_get_profile(_model): raise KeyError("simulated missing model profile") monkeypatch.setattr(model_profiles, "get_profile", boom_get_profile) # Simulate the exact catch-block behavior in run_shot.py:309 — the # try/except is inline (not a callable on its own), so we exercise # the registered fallback's firing path directly. import logging with caplog.at_level(logging.WARNING): try: model_profiles.get_profile("nonexistent-model-xyz") sibling_refs_enabled = True # unreachable except Exception as e: fire_sanctioned_fallback( "model_profile_feature_flag_default", model="nonexistent-model-xyz", error=str(e), ) sibling_refs_enabled = True # Default substitution applied. assert sibling_refs_enabled is True # FALLBACK_FIRED log emitted, observable to the operator. assert any( "FALLBACK_FIRED" in rec.message and "model_profile_feature_flag_default" in rec.message for rec in caplog.records ), "expected FALLBACK_FIRED log for model_profile_feature_flag_default"