"""Tests for workspace/verdict.py (Phase 1).""" from __future__ import annotations import logging import os from datetime import datetime, timedelta, timezone import pytest from workspace import verdict as V @pytest.fixture def tmp_projects(tmp_path, monkeypatch): root = tmp_path / "projects" (root / "driver-beware" / "renders" / "ep_001").mkdir(parents=True) (root / ".recoil-data-root").write_text("recoil-data-root\n") (root / "driver-beware" / "project_config.json").write_text("{}") monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(root)) # Bust the lru/cache-lite via re-import import importlib importlib.reload(V) return root def test_round_trip_write_read(tmp_projects): out = V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="SEQ_TEST", take_number=1, verdict="approve", taxonomy="taste-shaped", sub_tags=["good_lighting"], reason_text="Landed the beat.", reason_source="jt_confirmed", auto_filled={"model": "seeddance-2.0"}, confirmation={"jt_action": "confirm"}, session_id="test-session", machine_id="TEST", generation_id="SEQ_TEST_12345", ) assert out.name == "SEQ_TEST_take1_verdict.json" assert out.exists() data = V.read_verdict(out) assert data["verdict"] == "approve" assert data["taxonomy"] == "taste-shaped" assert data["sub_tags"] == ["good_lighting"] assert data["writer_version"] == V.WRITER_VERSION assert "_unknown_sub_tags" not in data def test_rejects_invalid_taxonomy(tmp_projects): with pytest.raises(ValueError, match="taxonomy"): V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="X", take_number=1, verdict="reject", taxonomy="nonsense", sub_tags=[], reason_text="", reason_source="claude_only", auto_filled={}, confirmation={"jt_action": "skip"}, ) def test_rejects_invalid_verdict(tmp_projects): with pytest.raises(ValueError, match="verdict"): V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="X", take_number=1, verdict="maybe", taxonomy="taste-shaped", sub_tags=[], reason_text="", reason_source="claude_only", auto_filled={}, confirmation={"jt_action": "skip"}, ) def test_rejects_invalid_jt_action(tmp_projects): with pytest.raises(ValueError, match="jt_action"): V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="X", take_number=1, verdict="approve", taxonomy="taste-shaped", sub_tags=[], reason_text="", reason_source="claude_only", auto_filled={}, confirmation={"jt_action": "shrug"}, ) def test_unknown_sub_tag_flagged_not_rejected(tmp_projects): out = V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="Y", take_number=2, verdict="reject", taxonomy="taste-shaped", sub_tags=["flat_lighting", "chartreuse_vibes"], reason_text="", reason_source="jt_confirmed", auto_filled={}, confirmation={"jt_action": "confirm"}, ) data = V.read_verdict(out) assert data["sub_tags"] == ["flat_lighting", "chartreuse_vibes"] assert data.get("_unknown_sub_tags") == ["chartreuse_vibes"] def test_file_lands_in_correct_directory(tmp_projects): out = V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="SEQ_A", take_number=3, verdict="approve", taxonomy="taste-shaped", sub_tags=[], reason_text="", reason_source="claude_only", auto_filled={}, confirmation={"jt_action": "skip"}, ) expected = tmp_projects / "driver-beware" / "renders" / "ep_001" / ".verdicts" / "SEQ_A_take3_verdict.json" assert out == expected def test_atomic_write_no_partial_on_failure(tmp_projects, monkeypatch): """Simulate a mid-write crash; no partial verdict file should remain.""" target_dir = tmp_projects / "driver-beware" / "renders" / "ep_001" def flaky_replace(src, dst): raise RuntimeError("simulated crash") monkeypatch.setattr(os, "replace", flaky_replace) with pytest.raises(RuntimeError, match="simulated crash"): V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="CRASH", take_number=1, verdict="approve", taxonomy="taste-shaped", sub_tags=[], reason_text="", reason_source="claude_only", auto_filled={}, confirmation={"jt_action": "skip"}, ) # No *_verdict.json file should remain assert list(target_dir.glob("CRASH_*.json")) == [] # No leftover temp file either assert list(target_dir.glob(".verdict_*.json")) == [] def test_iter_and_find_verdict(tmp_projects): V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="A", take_number=1, verdict="approve", taxonomy="taste-shaped", sub_tags=[], reason_text="", reason_source="claude_only", auto_filled={}, confirmation={"jt_action": "skip"}, ) V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="B", take_number=1, verdict="reject", taxonomy="validator-escape", sub_tags=[], reason_text="", reason_source="jt_confirmed", auto_filled={}, confirmation={"jt_action": "confirm"}, ) all_paths = list(V.iter_verdicts("driver-beware")) assert len(all_paths) == 2 assert V.find_verdict("driver-beware", "ep_001", "A", 1) is not None assert V.find_verdict("driver-beware", "ep_001", "MISSING", 1) is None # ── Phase 2: build_auto_filled tests ───────────────────────────── YAML_META_TAKE1 = """\ generation: id: SEQ_TEST_1700000000 timestamp: '2026-04-23T12:00:00Z' model: seeddance-2.0 endpoint: i2v prompt_builder: build_seeddance_i2v_prompt prompt_text: "[0s-2s] Medium shot.\\n[2s-4s] Wide." prompt_word_count: 6 inputs: character_refs: [] start_frame: /path/start.jpg parameters: duration: 6 aspect_ratio: "16:9" cost_usd: 1.0 latency_seconds: 30.0 """ YAML_META_TAKE2 = """\ generation: id: SEQ_TEST_1700000600 timestamp: '2026-04-23T12:10:00Z' model: seeddance-2.0 endpoint: i2v prompt_builder: build_seeddance_i2v_prompt prompt_text: "[0s-2s] Medium shot with harder key light.\\n[2s-4s] Wide, no haze." prompt_word_count: 11 inputs: character_refs: [] start_frame: /path/start.jpg parameters: duration: 6 aspect_ratio: "16:9" cost_usd: 1.0 latency_seconds: 32.0 """ def _seed_meta(projects_root, filename, yaml_text): d = projects_root / "driver-beware" / "renders" / "ep_001" d.mkdir(parents=True, exist_ok=True) (d / filename).write_text(yaml_text, encoding="utf-8") def test_autofill_from_real_meta_yaml(tmp_projects): _seed_meta(tmp_projects, "SEQ_TEST_meta.yaml", YAML_META_TAKE1) af = V.build_auto_filled(project="driver-beware", shot_id="SEQ_TEST", take_number=1) assert af["model"] == "seeddance-2.0" assert af["prompt_word_count"] == 6 assert af["params"]["duration"] == 6 assert af["params"]["aspect_ratio"] == "16:9" assert af["ref_set"]["start_frame"] == "/path/start.jpg" assert af.get("_reference_videos_inferred") is True assert af["dispatch_timestamp"] == "2026-04-23T12:00:00Z" def test_autofill_prompt_delta_take2_vs_take1(tmp_projects): _seed_meta(tmp_projects, "SEQ_TEST_take1_meta.yaml", YAML_META_TAKE1) _seed_meta(tmp_projects, "SEQ_TEST_take2_meta.yaml", YAML_META_TAKE2) af = V.build_auto_filled(project="driver-beware", shot_id="SEQ_TEST", take_number=2) delta = af["prompt_delta_from_prior_take"] assert delta is not None assert delta != "identical" assert "+" in delta or "-" in delta def test_autofill_missing_meta_does_not_crash(tmp_projects): af = V.build_auto_filled(project="driver-beware", shot_id="GHOST", take_number=1) assert af["_meta_yaml_missing"] is True assert af["model"] is None assert af["prompt_word_count"] is None assert af["ref_set"]["character_refs"] == [] def test_guess_reason_from_chat_flat_lighting(tmp_projects): _seed_meta(tmp_projects, "SEQ_TEST_meta.yaml", YAML_META_TAKE1) chat = [ {"role": "user", "content": "This lighting feels flat in seg 2."}, ] af = V.build_auto_filled( project="driver-beware", shot_id="SEQ_TEST", take_number=1, chat_context_window=chat, ) assert af["guessed_reason_from_chat"] is not None assert "flat_lighting" in af["guessed_reason_from_chat"] def test_no_chat_means_no_guess(tmp_projects): _seed_meta(tmp_projects, "SEQ_TEST_meta.yaml", YAML_META_TAKE1) af = V.build_auto_filled(project="driver-beware", shot_id="SEQ_TEST", take_number=1) assert af["guessed_reason_from_chat"] is None # ── Phase 5: summarize_session_verdicts tests ─────────────────── def test_summarize_empty_returns_empty_string(tmp_projects): assert V.summarize_session_verdicts("driver-beware", "2026-04-23T00:00:00Z") == "" def test_summarize_with_verdicts(tmp_projects): V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="A", take_number=1, verdict="approve", taxonomy="taste-shaped", sub_tags=["good_lighting"], reason_text="", reason_source="jt_confirmed", auto_filled={}, confirmation={"jt_action": "confirm"}, ) V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="B", take_number=1, verdict="reject", taxonomy="taste-shaped", sub_tags=["flat_lighting"], reason_text="too soft", reason_source="jt_confirmed", auto_filled={}, confirmation={"jt_action": "confirm"}, ) V.write_verdict( project="driver-beware", episode_id="ep_001", shot_id="C", take_number=1, verdict="reject", taxonomy="strategic", sub_tags=[], reason_text="wrong model — should be Kling not SeedDance", reason_source="jt_added", auto_filled={}, confirmation={"jt_action": "qualify"}, ) one_hour_ago = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat().replace("+00:00", "Z") summary = V.summarize_session_verdicts("driver-beware", one_hour_ago) assert summary, "expected non-empty summary" assert "3 verdicts" in summary assert "1 approve" in summary assert "2 reject" in summary assert "flat_lighting" in summary assert "STRATEGIC NOTE" in summary # ── Phase 6: ExecutionStore join tests ────────────────────────── def test_focus_character_and_location_populate_when_store_returns_data(tmp_projects, monkeypatch): """If ExecutionStore returns a take with inputs_snapshot, fields populate.""" class FakeStore: def __init__(self, project): self.project = project def get_shot(self, shot_id): return { "takes": [{ "take_id": "T1", "inputs_snapshot": { "characters": [{"char_id": "maya", "display_name": "Maya"}], "location_id": "interior_sedan", }, }], } def close(self): pass import recoil.execution.execution_store as es monkeypatch.setattr(es, "ExecutionStore", FakeStore) _seed_meta(tmp_projects, "SEQ_TEST_meta.yaml", YAML_META_TAKE1) af = V.build_auto_filled(project="driver-beware", shot_id="SEQ_TEST", take_number=1) assert af["focus_character"] == "maya" assert af["location_id"] == "interior_sedan" def test_missing_execution_store_inner_raises_outer_degrades(tmp_projects, monkeypatch, caplog): """ExecutionStore init failure raises VerdictAutofillError at the _try_execution_store_lookup layer (Tenet 6 fail-loud) but the outer build_auto_filled catches + logs + degrades to empty autofill (its "never crashes on missing data" docstring contract). See recoil/docs/silent-failure-inventory.md Sites #5–#7. """ from recoil.core.exceptions import VerdictAutofillError import recoil.execution.execution_store as es class BoomStore: def __init__(self, project): raise RuntimeError("no store") monkeypatch.setattr(es, "ExecutionStore", BoomStore) _seed_meta(tmp_projects, "SEQ_TEST_meta.yaml", YAML_META_TAKE1) # Inner layer: raises (Tenet 6 fail-loud at the lookup site). with pytest.raises(VerdictAutofillError): V._try_execution_store_lookup( project="driver-beware", shot_id="SEQ_TEST", take_number=1 ) # Outer layer: build_auto_filled catches + degrades (per its docstring). with caplog.at_level(logging.WARNING): result = V.build_auto_filled( project="driver-beware", shot_id="SEQ_TEST", take_number=1 ) assert result.get("focus_character") is None assert result.get("location_id") is None assert any("execution-store lookup failed" in r.message for r in caplog.records)