#!/usr/bin/env python3 """Acceptance tests for Recoil Workspace POC. Tests are designed to run without a live server — they test the Python modules directly. Frontend tests are structural (file existence, content checks) rather than browser-based. Run: python3 -m pytest workspace/tests/test_workspace.py -v """ import json import asyncio import os import re import sys import tempfile from pathlib import Path import pytest # ── Path setup ────────────────────────────────────────────────── _RECOIL_ROOT = Path(__file__).resolve().parent.parent.parent if str(_RECOIL_ROOT) not in sys.path: sys.path.insert(0, str(_RECOIL_ROOT)) # ── Test: state.py ────────────────────────────────────────────── class TestWorkspaceState: """Test workspace state read/write.""" def setup_method(self): """Use a temp directory for state.""" self._original_dir = None self._original_path = None import recoil.workspace.state as ws_state self._original_dir = ws_state._STATE_DIR self._original_path = ws_state._STATE_PATH self._tmp = Path(tempfile.mkdtemp()) ws_state._STATE_DIR = self._tmp ws_state._STATE_PATH = self._tmp / "state.json" def teardown_method(self): import recoil.workspace.state as ws_state ws_state._STATE_DIR = self._original_dir ws_state._STATE_PATH = self._original_path import shutil shutil.rmtree(self._tmp, ignore_errors=True) def test_default_state(self): from recoil.workspace.state import read_state state = read_state() assert state["project"] is None assert state["selection"] == [] assert state["viewer"]["shot_id"] is None def test_set_project(self): from recoil.workspace.state import set_project, get_project set_project("tartarus") assert get_project() == "tartarus" def test_selection(self): from recoil.workspace.state import set_selection, get_selection set_selection(["EP001_SH01", "EP001_SH02"]) assert get_selection() == ["EP001_SH01", "EP001_SH02"] def test_viewer_state(self): from recoil.workspace.state import set_viewer_state, get_viewer_state set_viewer_state( shot_id="EP001_SH03", take_index=2, file_path="output/previs/ep_001/shot_003_take2.png", ) viewer = get_viewer_state() assert viewer["shot_id"] == "EP001_SH03" assert viewer["take_index"] == 2 assert viewer["media_type"] == "image" # Auto-detected from .png def test_video_media_type_detection(self): from recoil.workspace.state import set_viewer_state, get_viewer_state set_viewer_state(file_path="output/video/shot.mp4") assert get_viewer_state()["media_type"] == "video" def test_set_project_clears_selection(self): from recoil.workspace.state import set_project, set_selection, get_selection set_selection(["EP001_SH01"]) set_project("new_project") assert get_selection() == [] def test_dropbox_conflict_detection(self): from recoil.workspace.state import detect_dropbox_conflicts # Create a fake conflict file conflict_dir = self._tmp / "shots" conflict_dir.mkdir() (conflict_dir / "EP001_SH01.json").touch() (conflict_dir / "EP001_SH01 (Joe's conflicted copy).json").touch() conflicts = detect_dropbox_conflicts(conflict_dir) assert len(conflicts) == 1 assert "conflicted copy" in conflicts[0].lower() # ── Test: session_log.py ──────────────────────────────────────── class TestSessionLog: """Test session log append and read.""" def setup_method(self): self._tmp = Path(tempfile.mkdtemp()) def teardown_method(self): import shutil shutil.rmtree(self._tmp, ignore_errors=True) def test_append_and_read(self): from recoil.workspace.session_log import append_entry, read_entries entry = append_entry( "test_project", self._tmp, "feedback", shot_id="EP001_SH03", take_id="T001", data={"text": "looks great", "category": "observation"}, ) assert entry["type"] == "feedback" assert entry["shot_id"] == "EP001_SH03" entries = read_entries("test_project", self._tmp) assert len(entries) == 1 assert entries[0]["type"] == "feedback" def test_since_filter(self): from recoil.workspace.session_log import append_entry, read_entries append_entry("test_project", self._tmp, "action", data={"a": 1}) # Read with a future timestamp should return nothing entries = read_entries("test_project", self._tmp, since="9999-12-31T23:59:59Z") assert len(entries) == 0 def test_multiple_entries(self): from recoil.workspace.session_log import append_entry, read_entries for i in range(5): append_entry("test_project", self._tmp, "action", data={"i": i}) entries = read_entries("test_project", self._tmp) assert len(entries) == 5 # ── Test: mcp_server.py (tool registry) ──────────────────────── class TestSubmitGenerationDispatch: """submit_generation must build a valid dispatch_cli argv and dispatch in a thread. Replaces the prior POC stub behavior.""" def test_argv_basic_no_override(self): from recoil.workspace.mcp_server import _build_dispatch_argv, _DISPATCH_SCRIPT argv = _build_dispatch_argv( project="driver-beware", shot_id="shot_DEER_V01", model="seeddance-2.0", override_prompt=None, ) assert argv[0] == sys.executable assert argv[1] == str(_DISPATCH_SCRIPT) assert ( "--project" in argv and argv[argv.index("--project") + 1] == "driver-beware" ) assert "--shot" in argv and argv[argv.index("--shot") + 1] == "shot_DEER_V01" assert "--model" in argv and argv[argv.index("--model") + 1] == "seeddance-2.0" assert "--prompt" not in argv def test_argv_with_override_prompt(self): from recoil.workspace.mcp_server import _build_dispatch_argv argv = _build_dispatch_argv( project="tartarus", shot_id="EP001_SH02", model="kling-o3", override_prompt="[FRAMING] tighter | [LIGHTING] golden hour", ) assert "--prompt" in argv assert ( argv[argv.index("--prompt") + 1] == "[FRAMING] tighter | [LIGHTING] golden hour" ) def test_argv_omits_prompt_when_empty_string(self): from recoil.workspace.mcp_server import _build_dispatch_argv argv = _build_dispatch_argv( project="tartarus", shot_id="EP001_SH02", model="kling-o3", override_prompt="", ) # Empty string is falsy — same as None — must NOT add --prompt flag assert "--prompt" not in argv def test_dispatch_async_logs_completion_on_success(self, tmp_path, monkeypatch): from recoil.workspace.mcp_server import _run_dispatch_async ops_log = tmp_path / "ops.log.jsonl" class FakeProc: returncode = 0 stdout = "ok" stderr = "" monkeypatch.setattr( "recoil.workspace.mcp_server.subprocess.run", lambda *a, **kw: FakeProc(), ) _run_dispatch_async( op_id="op_test123", ops_log_path=ops_log, argv=["echo", "hi"], cwd=tmp_path, ) records = [ json.loads(line) for line in ops_log.read_text().splitlines() if line ] assert len(records) == 1 assert records[0]["id"] == "op_test123" assert records[0]["status"] == "completed" def test_dispatch_async_logs_failure_on_nonzero(self, tmp_path, monkeypatch): from recoil.workspace.mcp_server import _run_dispatch_async ops_log = tmp_path / "ops.log.jsonl" class FakeProc: returncode = 1 stdout = "" stderr = "boom" monkeypatch.setattr( "recoil.workspace.mcp_server.subprocess.run", lambda *a, **kw: FakeProc(), ) _run_dispatch_async( op_id="op_failz", ops_log_path=ops_log, argv=["false"], cwd=tmp_path, ) records = [ json.loads(line) for line in ops_log.read_text().splitlines() if line ] assert records[0]["status"] == "failed" assert "exited 1" in records[0]["error"] class TestMCPTools: """Test that all MCP tools are registered.""" def test_all_tools_registered(self): from recoil.workspace.mcp_server import _TOOLS expected = [ "prime_project", "get_selection", "get_viewer_state", "show_in_viewer", "get_shot_detail", "get_shot_neighbors", "approve_shot", "reject_shot", "submit_generation", "log_feedback", "get_session_log", "get_activity", "get_file_provenance", "propose_action", "get_recent_clicks", "get_episode_board", ] for name in expected: assert name in _TOOLS, f"Missing tool: {name}" assert len(_TOOLS) == 16 def test_tools_have_handlers(self): from recoil.workspace.mcp_server import _TOOLS for name, tool in _TOOLS.items(): assert "handler" in tool, f"Tool {name} missing handler" assert callable(tool["handler"]), f"Tool {name} handler not callable" def test_tools_have_schemas(self): from recoil.workspace.mcp_server import _TOOLS for name, tool in _TOOLS.items(): assert "inputSchema" in tool, f"Tool {name} missing inputSchema" assert "type" in tool["inputSchema"], f"Tool {name} schema missing 'type'" def test_build_tools_list(self): from recoil.workspace.mcp_server import _build_tools_list tools_list = _build_tools_list() assert len(tools_list) == 16 for t in tools_list: assert "name" in t assert "description" in t assert "inputSchema" in t assert "handler" not in t # Handler should not be in the list response # ── Test: Frontend files ──────────────────────────────────────── class TestFrontend: """Structural checks on frontend files.""" def test_index_html_exists(self): path = _RECOIL_ROOT / "workspace" / "static" / "index.html" assert path.is_file(), f"index.html not found at {path}" def test_workspace_css_exists(self): path = _RECOIL_ROOT / "workspace" / "static" / "workspace.css" assert path.is_file(), f"workspace.css not found at {path}" def test_workspace_js_exists(self): path = _RECOIL_ROOT / "workspace" / "static" / "workspace.js" assert path.is_file(), f"workspace.js not found at {path}" def test_html_links_css(self): html = (_RECOIL_ROOT / "workspace" / "static" / "index.html").read_text() assert "workspace.css" in html def test_html_links_js(self): html = (_RECOIL_ROOT / "workspace" / "static" / "index.html").read_text() assert "workspace.js" in html def test_html_has_viewer(self): html = (_RECOIL_ROOT / "workspace" / "static" / "index.html").read_text() assert re.search(r'id=["\']viewer["\']', html), "Missing viewer element" def test_html_has_file_tree(self): html = (_RECOIL_ROOT / "workspace" / "static" / "index.html").read_text() assert re.search(r'id=["\']file-tree["\']', html), "Missing file-tree element" def test_html_has_inspector(self): html = (_RECOIL_ROOT / "workspace" / "static" / "index.html").read_text() assert re.search(r'id=["\']inspector["\']', html), "Missing inspector element" def test_css_has_design_tokens(self): css = (_RECOIL_ROOT / "workspace" / "static" / "workspace.css").read_text() assert "--bg-base" in css assert "--accent-blue" in css assert "--text-primary" in css def test_js_has_core_functions(self): js = (_RECOIL_ROOT / "workspace" / "static" / "workspace.js").read_text() assert "function init" in js assert "renderFileTree" in js assert "selectFile" in js assert "displayInViewer" in js assert "renderInspector" in js assert "navigateShot" in js def test_js_has_keyboard_handler(self): js = (_RECOIL_ROOT / "workspace" / "static" / "workspace.js").read_text() assert "keydown" in js assert "Escape" in js assert "ArrowLeft" in js assert "ArrowRight" in js # ── Test: server.py ───────────────────────────────────────────── class TestServer: """Test that server.py imports and defines routes.""" def test_server_imports(self): from recoil.workspace.server import app assert app.title == "Recoil Workspace" def test_server_has_routes(self): from recoil.workspace.server import app routes = [r.path for r in app.routes if hasattr(r, "path")] assert "/api/health" in routes assert "/api/state" in routes assert "/api/tree/{project}" in routes assert "/api/shot/{project}/{shot_id}" in routes assert "/api/activity/{project}" in routes # ── Test: Startup script ─────────────────────────────────────── class TestStartup: """Check startup script exists and is valid.""" def test_script_exists(self): path = _RECOIL_ROOT / "workspace" / "start_workspace.sh" assert path.is_file() def test_script_is_bash(self): content = (_RECOIL_ROOT / "workspace" / "start_workspace.sh").read_text() assert content.startswith("#!/bin/bash") def test_script_references_server(self): content = (_RECOIL_ROOT / "workspace" / "start_workspace.sh").read_text() assert "workspace.server" in content or "server.py" in content # ── Test: Design system ──────────────────────────────────────── class TestDesignSystem: """Check design system doc exists.""" def test_design_system_exists(self): path = _RECOIL_ROOT / "workspace" / "DESIGN_SYSTEM.md" assert path.is_file() def test_design_system_has_colors(self): content = (_RECOIL_ROOT / "workspace" / "DESIGN_SYSTEM.md").read_text() assert "Color Palette" in content assert "#08080f" in content or "#0e0e18" in content # ── Test: Pass Detection ────────────────────────────────────── class TestPassDetection: """Test pass video detection and shot normalization.""" def test_pass_pattern_matches_standard(self): from recoil.workspace.server import _PASS_PATTERN m = _PASS_PATTERN.match("EP001_PASS_017_SH33_A_WREN_take1.mp4") assert m is not None assert m.group(1) == "EP001" assert m.group(2) == "017" assert m.group(3) == "33" assert m.group(4) == "A_WREN" assert m.group(5) == "1" def test_pass_pattern_matches_multi_shot(self): from recoil.workspace.server import _PASS_PATTERN m = _PASS_PATTERN.match("EP001_PASS_008_SH16_17_18_A_WREN_take1.mp4") assert m is not None assert m.group(1) == "EP001" assert m.group(2) == "008" assert m.group(3) == "16_17_18" assert m.group(4) == "A_WREN" assert m.group(5) == "1" def test_pass_pattern_no_match_shot(self): from recoil.workspace.server import _PASS_PATTERN assert _PASS_PATTERN.match("shot_001_take1.mp4") is None def test_pass_pattern_no_match_test(self): from recoil.workspace.server import _PASS_PATTERN assert _PASS_PATTERN.match("TEST_2CHAR_SH33_35_take1.mp4") is None def test_pass_pattern_no_match_hatch(self): # HATCH-style scene-based names (no counter, no SH prefix) must be rejected — routes to orphans from recoil.workspace.server import _PASS_PATTERN assert _PASS_PATTERN.match("EP001_PASS_HATCH_001_take1.mp4") is None def test_normalize_shot_num_full_id(self): from recoil.workspace.server import _normalize_shot_num assert _normalize_shot_num("EP001_SH05A_HATCH_MS") == "005a" def test_normalize_shot_num_simple(self): from recoil.workspace.server import _normalize_shot_num assert _normalize_shot_num("EP001_SH12") == "012" def test_normalize_shot_num_bare(self): from recoil.workspace.server import _normalize_shot_num assert _normalize_shot_num("SH3") == "003" def test_normalize_shot_num_invalid(self): from recoil.workspace.server import _normalize_shot_num assert _normalize_shot_num("random_string") is None def test_pass_pattern_matches_with_hash(self): from recoil.workspace.server import _PASS_PATTERN m = _PASS_PATTERN.match("EP001_PASS_017_SH33_A_WREN_take1_31434.mp4") assert m is not None assert m.group(5) == "1" def test_server_has_pass_route(self): from recoil.workspace.server import app routes = [r.path for r in app.routes if hasattr(r, "path")] assert "/api/pass/{project}/{pass_id}" in routes def test_parse_pass_filename_reads_strategy_solo_and_legacy(self): from recoil.workspace.tree import parse_pass_filename cases = [ ("EP001_CONT_002_SH10_11_take3.mp4", "continuity", 2), ("EP001_COV_011_SH23_24_25_take1.mp4", "coverage", 11), ("EP001_ONER_001_SH33_take1.mp4", "oner", 1), ("EP001_SH33_take7.mp4", "solo", 0), ("EP001_PASS_017_SH33_33a_34_35_take1.mp4", "coverage", 17), ("EP001_PASS_017_SH33_33a_34_35_A_WREN_take1.mp4", "coverage", 17), ] for filename, strategy, ordinal in cases: parsed = parse_pass_filename(filename) assert parsed is not None, filename assert parsed["strategy"] == strategy assert parsed["ordinal"] == ordinal legacy = parse_pass_filename("EP001_PASS_017_SH33_A_WREN_take1.mp4") assert legacy["semantic_tag"] == "A_WREN" def test_find_pass_video_resolves_new_and_current_short_grammar(self, tmp_path, monkeypatch): monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) from recoil.execution.pass_store import PassStore from recoil.workspace import server as ws_server project = "parser_proj" proj = tmp_path / project (proj / "output" / "video" / "ep_001").mkdir(parents=True) (proj / "project_config.json").write_text('{"mode": "microdrama"}') store = PassStore(project) cov_pass_id = "EP001_PASS_008_SH16_17_18_A_WREN" short_pass_id = "EP001_PASS_009_SH20_21_B_ENV" store.create_pass(cov_pass_id, ["EP001_SH16", "EP001_SH17", "EP001_SH18"]) store.create_pass(short_pass_id, ["EP001_SH20", "EP001_SH21"]) store.close() ep_dir = proj / "output" / "video" / "ep_001" new_video = ep_dir / "EP001_COV_008_SH16_17_18_take2.mp4" new_video.write_bytes(b"new") (ep_dir / f"{new_video.name}.json").write_text( json.dumps( { "provenance": { "grouping": { "strategy": "coverage", "ordinal": 8, "shot_ids": ["EP001_SH16", "EP001_SH17", "EP001_SH18"], "source_pass_id": cov_pass_id, } } } ) ) short_video = ep_dir / "EP001_PASS_009_SH20_21_take1.mp4" short_video.write_bytes(b"short") assert ws_server._find_pass_video(project, cov_pass_id) == new_video assert ws_server._find_pass_video(project, short_pass_id) == short_video def test_orphan_sweep_keeps_short_and_strategy_grammar(self, tmp_path, monkeypatch): monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) from recoil.workspace import server as ws_server project = "parser_proj" proj = tmp_path / project ep_dir = proj / "output" / "video" / "ep_001" ep_dir.mkdir(parents=True) (proj / "project_config.json").write_text('{"mode": "microdrama"}') keep_names = [ "EP001_CONT_001_SH10_11_take1.mp4", "EP001_COV_002_SH12_13_take1.mp4", "EP001_ONER_001_SH14_take1.mp4", "EP001_SH15_take1.mp4", ] old_time = 1_700_000_000 for name in keep_names: p = ep_dir / name p.write_bytes(b"ok") os.utime(p, (old_time, old_time)) result = ws_server._sweep_orphans(project, "EP001") assert result["moved"] == 0 assert all((ep_dir / name).exists() for name in keep_names) def test_orphan_reclaim_updates_sidecar_video_path(self, tmp_path, monkeypatch): monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) from recoil.workspace import server as ws_server class _Request: async def json(self): return { "orphan_filename": "manual_take1.mp4", "target_episode": "EP001", "segment_shot_ids": ["EP001_SH33"], "semantic_tag": "A_WREN", "reconstructed_provenance": {"model": "seedance-test"}, } project = "parser_proj" proj = tmp_path / project orphan_dir = proj / "output" / "video" / "ep_001" / "_orphans" orphan_dir.mkdir(parents=True) (proj / "project_config.json").write_text('{"mode": "microdrama"}') orphan = orphan_dir / "manual_take1.mp4" orphan.write_bytes(b"video") (orphan_dir / "manual_take1.mp4.json").write_text( json.dumps({"video_path": "output/video/ep_001/_orphans/manual_take1.mp4"}) ) response = asyncio.run(ws_server.register_orphan(project, _Request())) body = json.loads(response.body) sidecar = proj / f"{body['video_path']}.json" data = json.loads(sidecar.read_text()) assert data["video_path"] == body["video_path"] # ── Test: Pass Shot Grouping ───────────────────────────────── class TestPassShotGrouping: """Test pass-sourced takes group correctly with shots.""" def test_shot_pattern_matches_pass_segment(self): from recoil.workspace.server import _SHOT_PATTERN m = _SHOT_PATTERN.match("shot_005a_FROM_PASS_017_A_WREN_take1.mp4") assert m is not None assert m.group(1) == "005a" assert m.group(2) == "1" def test_shot_pattern_matches_pass_segment_multi_take(self): from recoil.workspace.server import _SHOT_PATTERN m = _SHOT_PATTERN.match("shot_012_FROM_PASS_002_B_CORRIDOR_take3.mp4") assert m is not None assert m.group(1) == "012" assert m.group(2) == "3" def test_parse_shot_filename_pass_segment(self): from recoil.workspace.server import _parse_shot_filename parsed = _parse_shot_filename("shot_005a_FROM_PASS_017_A_WREN_take1.mp4") assert parsed is not None assert parsed["shot_num"] == "005a" assert parsed["take_num"] == 1 def test_grouping_orders_direct_before_pass(self): from recoil.workspace.server import _group_episode_files_by_shot files = [ { "name": "shot_005a_FROM_PASS_017_A_WREN_take1.mp4", "type": "file", "path": "output/video/ep_001/shot_005a_FROM_PASS_017_A_WREN_take1.mp4", "media_url": "/media/test/output/video/ep_001/shot_005a_FROM_PASS_017_A_WREN_take1.mp4", "status": "candidate", "status_color": "gray", }, { "name": "shot_005a_take1.mp4", "type": "file", "path": "output/video/ep_001/shot_005a_take1.mp4", "media_url": "/media/test/output/video/ep_001/shot_005a_take1.mp4", "status": "candidate", "status_color": "gray", }, { "name": "shot_005a_take2.mp4", "type": "file", "path": "output/video/ep_001/shot_005a_take2.mp4", "media_url": "/media/test/output/video/ep_001/shot_005a_take2.mp4", "status": "candidate", "status_color": "gray", }, ] result = _group_episode_files_by_shot(files, "Episode 001", "") shots = [n for n in result if n.get("type") == "shot"] assert len(shots) == 1 shot = shots[0] assert shot["take_count"] == 3 # Direct takes first, then pass-sourced assert "_FROM_" not in shot["takes"][0]["name"] assert "_FROM_" not in shot["takes"][1]["name"] assert "_FROM_" in shot["takes"][2]["name"] def test_grouping_counts_all_takes(self): from recoil.workspace.server import _group_episode_files_by_shot files = [ { "name": "shot_003_take1.mp4", "type": "file", "path": "p1", "media_url": "m1", "status": "candidate", "status_color": "gray", }, { "name": "shot_003_FROM_PASS_002_B_TEST_take1.mp4", "type": "file", "path": "p2", "media_url": "m2", "status": "candidate", "status_color": "gray", }, ] result = _group_episode_files_by_shot(files, "Episode 001", "") shots = [n for n in result if n.get("type") == "shot"] assert len(shots) == 1 assert shots[0]["take_count"] == 2 class TestPassProvenance: """Test pass provenance API and inspector support.""" def test_pass_api_route_exists(self): from recoil.workspace.server import app routes = [r.path for r in app.routes if hasattr(r, "path")] assert "/api/pass/{project}/{pass_id}" in routes def test_js_has_pass_provenance_function(self): js = (_RECOIL_ROOT / "workspace" / "static" / "workspace.js").read_text() assert "viewPassSource" in js assert "_findPassVideoPath" in js assert "FROM COVERAGE PASS" in js assert "source_type" in js # ── Test: Recent Feed ──────────────────────────────────────── class TestRecentFeed: """Test recent feed API and frontend.""" def test_recent_api_route_exists(self): from recoil.workspace.server import app routes = [r.path for r in app.routes if hasattr(r, "path")] assert "/api/recent/{project}" in routes def test_js_has_recent_functions(self): js = (_RECOIL_ROOT / "workspace" / "static" / "workspace.js").read_text() assert "switchNavTab" in js assert "pollRecent" in js assert "renderRecentFeed" in js assert "_formatRelativeTime" in js def test_html_has_nav_tabs(self): html = (_RECOIL_ROOT / "workspace" / "static" / "index.html").read_text() assert "nav-tabs" in html assert "tab-shots" in html assert "tab-recent" in html def test_css_has_recent_styles(self): css = (_RECOIL_ROOT / "workspace" / "static" / "workspace.css").read_text() assert ".recent-item" in css assert ".nav-tab" in css assert ".badge-video" in css # ── Test: Pass ↔ Shot Cross-Linking ───────────────────────── class TestCrossLinking: """Test pass ↔ shot cross-linking.""" def test_js_has_cross_link_functions(self): js = (_RECOIL_ROOT / "workspace" / "static" / "workspace.js").read_text() assert "viewPassById" in js assert "_findPassVideoByPassId" in js assert "coverage_passes" in js assert "COVERAGE" in js def test_shot_api_returns_coverage_passes_field(self): """The shot API should include coverage_passes field in response.""" from recoil.workspace.server import get_shot import inspect source = inspect.getsource(get_shot) assert "coverage_passes" in source # ── Test: Filter and CSS Compatibility ────────────────────── class TestFilterAndCSS: """Test filter compatibility and CSS for pass-sourced takes.""" def test_css_has_pass_indicator_style(self): css = (_RECOIL_ROOT / "workspace" / "static" / "workspace.css").read_text() # Should have pass-related styling assert ".viewer-take-pass" in css or "_FROM_" in css or ".badge-" in css def test_js_pass_sourced_take_indicator(self): js = (_RECOIL_ROOT / "workspace" / "static" / "workspace.js").read_text() assert "_FROM_" in js # Pass-sourced detection assert "(P)" in js or "pass" in js.lower() # Pass indicator in take nav def test_filter_works_with_pass_takes(self): """Verify shot grouping includes pass take names for text search.""" from recoil.workspace.server import _group_episode_files_by_shot files = [ { "name": "shot_005a_take1.mp4", "type": "file", "path": "p1", "media_url": "m1", "status": "candidate", "status_color": "gray", }, { "name": "shot_005a_FROM_PASS_017_A_WREN_take1.mp4", "type": "file", "path": "p2", "media_url": "m2", "status": "candidate", "status_color": "gray", }, ] result = _group_episode_files_by_shot(files, "Episode 001", "") shots = [n for n in result if n.get("type") == "shot"] assert len(shots) == 1 shot = shots[0] # Shot node contains all takes — frontend filter searches take names take_names = [t["name"] for t in shot["takes"]] # Verify pass take name is in the takes list assert any("FROM_PASS_017_A_WREN" in name for name in take_names) def test_archive_of_individual_pass_segment(self): """Structural test: archive endpoint exists and handles single files.""" from recoil.workspace.server import app routes = [r.path for r in app.routes if hasattr(r, "path")] assert "/api/archive/{project}" in routes class TestModeGuardSweepOrphans: """Phase 2: _sweep_orphans is a no-op for non-microdrama projects.""" def test_client_mode_skips_sweep(self, tmp_path, monkeypatch): from workspace import server as ws_server # Set up a client_deliverable project. # core.project does not export projects_root() (Phase 4.4 special case); # patching paths + env var is sufficient — project.py reads via projects_root(). monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) proj = tmp_path / "client_proj" proj.mkdir() (proj / "project_config.json").write_text('{"mode": "client_deliverable"}') # Place a video that would normally be orphaned (no PassStore record) ep_dir = proj / "output" / "video" / "ep_001" ep_dir.mkdir(parents=True) bogus = ep_dir / "random_unmatched.mp4" bogus.write_bytes(b"") # Patch projects_root() in server.py module too (eagerly imported) monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) result = ws_server._sweep_orphans("client_proj") # Existing return shape uses 'moved' + 'orphan_dirs', not 'swept' assert result["moved"] == 0 assert result["orphan_dirs"] == [] assert result.get("skipped_reason") == "mode_disabled" # Verify the file was NOT moved assert bogus.exists() assert not (ep_dir / "_orphans" / "random_unmatched.mp4").exists() class TestModeGuardAutoExtract: """Phase 3: _maybe_extract_passes is a no-op for non-microdrama projects. Note: spec named the function `_auto_extract_coverage_segments` but the actual codebase name is `_maybe_extract_passes`. Same intent, different name. """ def test_client_mode_skips_extraction(self, tmp_path, monkeypatch): from workspace import server as ws_server # core.project does not export projects_root() (Phase 4.4 special case). monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) proj = tmp_path / "client_proj" proj.mkdir() (proj / "project_config.json").write_text('{"mode": "client_deliverable"}') # Place a video that would normally trigger extraction ep_dir = proj / "output" / "video" / "ep_001" ep_dir.mkdir(parents=True) pass_video = ep_dir / "EP001_PASS_017_SH33_A_WREN_take1.mp4" pass_video.write_bytes(b"") marker = ep_dir / f".{pass_video.name}.extracted" # Real signature is `(project: str)`. No episode_id parameter. ws_server._maybe_extract_passes("client_proj") assert not marker.exists(), ( "marker should not exist for client_deliverable mode" ) class TestDispatcherClientMode: """Phase 4: client_deliverable mode returns flat deliverables.""" def test_client_mode_returns_flat_list(self, tmp_path, monkeypatch): from workspace import server as ws_server # core.project does not export projects_root() (Phase 4.4 special case). monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) proj_dir = tmp_path / "client_proj" proj_dir.mkdir() (proj_dir / "project_config.json").write_text('{"mode": "client_deliverable"}') files = [ { "name": "REGEN_P02_multishot_take1.mp4", "type": "file", "path": "p1", "media_url": "m1", "status": "candidate", "status_color": "gray", }, { "name": "V1_STAY_IN_CAR_SH01.mp4", "type": "file", "path": "p2", "media_url": "m2", "status": "candidate", "status_color": "gray", }, { "name": "deer_fisheye.png", "type": "file", "path": "p3", "media_url": "m3", "status": "candidate", "status_color": "gray", }, ] result = ws_server._group_episode_files_by_shot( files, "Episode 001", "client_proj" ) assert len(result) == 3 assert all(node.get("type") == "deliverable" for node in result) shot_ids = [n["shot_id"] for n in result] assert "REGEN_P02_multishot_take1" in shot_ids assert "V1_STAY_IN_CAR_SH01" in shot_ids assert "deer_fisheye" in shot_ids def test_client_mode_sorts_deliverables_naturally(self, tmp_path, monkeypatch): """client_deliverable mode must return deliverables in stable, natural order. Filesystem scans return arbitrary order; UI needs a predictable shot list.""" from workspace import server as ws_server # core.project does not export projects_root() (Phase 4.4 special case). monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) proj_dir = tmp_path / "client_proj" proj_dir.mkdir() (proj_dir / "project_config.json").write_text('{"mode": "client_deliverable"}') files = [ { "name": "shot_V10_END.mp4", "type": "file", "path": "p10", "media_url": "m10", "status": "candidate", "status_color": "gray", }, { "name": "shot_V2_BACKWARD.mp4", "type": "file", "path": "p2", "media_url": "m2", "status": "candidate", "status_color": "gray", }, { "name": "shot_V01_OPEN.mp4", "type": "file", "path": "p1", "media_url": "m1", "status": "candidate", "status_color": "gray", }, { "name": "shot_DEER_V01.mp4", "type": "file", "path": "pd", "media_url": "md", "status": "candidate", "status_color": "gray", }, ] result = ws_server._group_episode_files_by_shot( files, "Episode 001", "client_proj" ) shot_ids = [n["shot_id"] for n in result] assert shot_ids == [ "shot_DEER_V01", "shot_V01_OPEN", "shot_V2_BACKWARD", "shot_V10_END", ], f"got: {shot_ids}" def test_microdrama_unchanged(self, tmp_path, monkeypatch): """Existing tartarus-style grouping must continue to work.""" from workspace import server as ws_server # core.project does not export projects_root() (Phase 4.4 special case). monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) proj_dir = tmp_path / "tartarus" proj_dir.mkdir() files = [ { "name": "shot_005a_take1.mp4", "type": "file", "path": "p1", "media_url": "m1", "status": "candidate", "status_color": "gray", }, { "name": "shot_005a_take2.mp4", "type": "file", "path": "p2", "media_url": "m2", "status": "candidate", "status_color": "gray", }, ] result = ws_server._group_episode_files_by_shot( files, "Episode 001", "tartarus" ) types = {n.get("type") for n in result} assert "deliverable" not in types def test_legacy_caller_no_project_arg(self, tmp_path, monkeypatch): """Legacy callers that pass empty/no project argument must fall back to microdrama grouping (no NameError, no crash, no client mode).""" from workspace import server as ws_server files = [ { "name": "shot_005a_take1.mp4", "type": "file", "path": "p1", "media_url": "m1", "status": "candidate", "status_color": "gray", }, ] result = ws_server._group_episode_files_by_shot(files, "Episode 001", "") types = {n.get("type") for n in result} assert "deliverable" not in types def test_tartarus_pass_anchor_surfacing(self, tmp_path, monkeypatch): """Phase 4 must NOT break tartarus pass_anchor surfacing.""" from workspace import server as ws_server # core.project does not export projects_root() (Phase 4.4 special case). monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) proj_dir = tmp_path / "tartarus" proj_dir.mkdir() files = [ { "name": "EP001_PASS_008_SH16_17_18_A_WREN_take1.mp4", "type": "file", "path": "p1", "media_url": "m1", "status": "candidate", "status_color": "gray", }, ] result = ws_server._group_episode_files_by_shot( files, "Episode 001", "tartarus" ) types = {n.get("type") for n in result} assert "pass_anchor" in types or "shot" in types, f"unexpected types: {types}" # ── Phase 3: Verdict sidecar emission tests ───────────────────────── @pytest.fixture def tmp_workspace_project(tmp_path, monkeypatch): """Prime an isolated tmp_path-rooted project ready for approve/reject calls. - Creates `projects/test_project/` with microdrama project_config.json so `Project.captures_verdicts` is True. - Seeds an ExecutionStore with one shot (`EP001_SH01`) containing one take that has a file path + provenance hash. - Patches projects_root() in core.paths and workspace.mcp_server, and sets RECOIL_PROJECTS_ROOT so projects_root() and verdict.py::_projects_root() agree. core.project does not export projects_root() (Phase 4.4 special case); project.py reads via projects_root() which honors the env var. - Clears the workspace.helpers store cache so the patched root takes effect. - Primes ws_state to point at the project (avoids "No project active"). """ from workspace import state as ws_state from workspace import helpers as ws_helpers from recoil.execution.execution_store import ExecutionStore # Patch projects_root() everywhere it's been imported monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) # projects_root() and verdict.py::_projects_root() consult this env var first monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) # Reset the singleton store cache so the new projects_root() is used ws_helpers._stores.clear() project = "test_project" shot_id = "EP001_SH01" proj_dir = tmp_path / project proj_dir.mkdir() # Microdrama mode → captures_verdicts is True (proj_dir / "project_config.json").write_text('{"mode": "microdrama"}') # Use ws_state to register the active project (need isolated state path too, # since ws_state writes to a global state dir) ws_state_dir = tmp_path / "_ws_state" ws_state_dir.mkdir() monkeypatch.setattr(ws_state, "_STATE_DIR", ws_state_dir) monkeypatch.setattr(ws_state, "_STATE_PATH", ws_state_dir / "state.json") ws_state.set_project(project) # Seed an ExecutionStore with one shot + one take store = ExecutionStore(project=project) store.insert_shot( { "shot_id": shot_id, "episode_id": "ep_001", "status": "video_complete", "takes": [ { "take_id": "T0001", "file_path": f"output/video/ep_001/{shot_id}_take1.mp4", "provenance_hash": "deadbeef", } ], } ) store.close() yield { "projects_root": tmp_path, "project": project, "shot_id": shot_id, } # Cleanup the cached store post-test ws_helpers._stores.clear() def _find_verdict_file(projects_root: Path, project: str, shot_id: str) -> Path | None: d = projects_root / project / "output" / "video" / "ep_001" / ".verdicts" matches = list(d.glob(f"{shot_id}_take*_verdict.json")) return matches[0] if matches else None def test_approve_shot_emits_verdict_sidecar(tmp_workspace_project): """After approve_shot fires, a verdict sidecar with verdict='approve' exists.""" from workspace import mcp_server project, shot_id = ( tmp_workspace_project["project"], tmp_workspace_project["shot_id"], ) result = mcp_server.tool_approve_shot( { "shot_id": shot_id, "taxonomy": "taste-shaped", "sub_tags": ["good_lighting"], "claude_proposed_reason": "Landed the beat.", "jt_action": "confirm", } ) assert result.get("shot_id") == shot_id sidecar = _find_verdict_file( tmp_workspace_project["projects_root"], project, shot_id ) assert sidecar is not None, "verdict sidecar missing" data = json.loads(sidecar.read_text()) assert data["verdict"] == "approve" assert data["taxonomy"] == "taste-shaped" assert data["sub_tags"] == ["good_lighting"] assert data["confirmation"]["jt_action"] == "confirm" def test_reject_shot_emits_verdict_sidecar(tmp_workspace_project): from workspace import mcp_server project, shot_id = ( tmp_workspace_project["project"], tmp_workspace_project["shot_id"], ) result = mcp_server.tool_reject_shot( { "shot_id": shot_id, "reason": "flat lighting", "failure_mode": "composition_wrong", "taxonomy": "taste-shaped", "sub_tags": ["flat_lighting"], "claude_proposed_reason": "Lighting soft in seg 2.", "jt_action": "confirm", } ) assert result.get("shot_id") == shot_id sidecar = _find_verdict_file( tmp_workspace_project["projects_root"], project, shot_id ) assert sidecar is not None data = json.loads(sidecar.read_text()) assert data["verdict"] == "reject" assert data["taxonomy"] == "taste-shaped" assert data["sub_tags"] == ["flat_lighting"] assert data["reason_text"] == "flat lighting" def test_approve_without_taxonomy_still_writes_sidecar(tmp_workspace_project): """Backward-compat: old approve_shot calls without the new fields still work.""" from workspace import mcp_server project, shot_id = ( tmp_workspace_project["project"], tmp_workspace_project["shot_id"], ) result = mcp_server.tool_approve_shot({"shot_id": shot_id}) assert "error" not in result sidecar = _find_verdict_file( tmp_workspace_project["projects_root"], project, shot_id ) assert sidecar is not None data = json.loads(sidecar.read_text()) assert data["verdict"] == "approve" assert data["taxonomy"] == "taste-shaped" # default assert data["confirmation"]["jt_action"] == "skip" def test_verdict_write_failure_does_not_break_approve( tmp_workspace_project, monkeypatch ): """If verdict.write_verdict raises, tool_approve_shot still returns success.""" from workspace import mcp_server, verdict as V shot_id = tmp_workspace_project["shot_id"] def boom(**_kwargs): raise RuntimeError("simulated verdict failure") monkeypatch.setattr(V, "write_verdict", boom) result = mcp_server.tool_approve_shot({"shot_id": shot_id}) assert "error" not in result assert result.get("shot_id") == shot_id # ────────────────────────────────────────────────────────────────── # Reclaim badge multi-take lookup (Debug R4 — Bug 1) # ────────────────────────────────────────────────────────────────── def _seed_reclaim_video( projects_root: Path, project: str, rel_dir: str, filename: str ) -> Path: """Create an empty .mp4 file at projects_root/project/rel_dir/filename.""" target_dir = projects_root / project / rel_dir target_dir.mkdir(parents=True, exist_ok=True) video = target_dir / filename video.write_bytes(b"") return video def _write_reclaim_meta( meta_path: Path, *, confidence: str = "high", source: str = "ffprobe" ) -> None: import yaml meta_path.parent.mkdir(parents=True, exist_ok=True) payload = { "shot_id": meta_path.stem.split("_take")[0].rstrip("_meta").rstrip("_"), "reclaim": { "synthetic": True, "confidence": confidence, "inference_sources": [source], "unknown_fields": [], "reclaimed_at": "2026-04-25T00:00:00Z", }, } meta_path.write_text(yaml.safe_dump(payload), encoding="utf-8") def test_reclaim_badge_take1_uses_canonical_meta(tmp_workspace_project): """take==1 → look up `{shot_id}_meta.yaml` (canonical name).""" from workspace import mcp_server project = tmp_workspace_project["project"] root = tmp_workspace_project["projects_root"] rel_dir = "output/video/ep_001" video = _seed_reclaim_video( root, project, rel_dir, "shot_EP001_SH99_take1_kling.mp4" ) _write_reclaim_meta(root / project / rel_dir / "EP001_SH99_meta.yaml") res = mcp_server.tool_get_file_provenance({"path": f"{rel_dir}/{video.name}"}) assert res.get("reclaim_badge"), f"expected reclaim_badge, got {res}" assert res["reclaim_badge"]["synthetic"] is True def test_reclaim_badge_take_gt_1_uses_take_suffixed_meta(tmp_workspace_project): """take>1 → look up `{shot_id}_take{N}_meta.yaml` (R3 collision-safe name).""" from workspace import mcp_server project = tmp_workspace_project["project"] root = tmp_workspace_project["projects_root"] rel_dir = "output/video/ep_001" video = _seed_reclaim_video( root, project, rel_dir, "shot_EP001_SH99_take3_kling.mp4" ) # ONLY the take-suffixed meta exists — canonical {shot_id}_meta.yaml is absent _write_reclaim_meta(root / project / rel_dir / "EP001_SH99_take3_meta.yaml") res = mcp_server.tool_get_file_provenance({"path": f"{rel_dir}/{video.name}"}) assert res.get("reclaim_badge"), ( f"expected reclaim_badge for take>1 (was the take-suffixed meta lookup wired?). " f"got: {res}" ) assert res["reclaim_badge"]["synthetic"] is True assert "_take3_meta.yaml" in res["reclaim_badge"]["meta_yaml"] def test_reclaim_badge_take_gt_1_falls_back_to_canonical(tmp_workspace_project): """take>1 with only canonical meta present → still surfaces the badge.""" from workspace import mcp_server project = tmp_workspace_project["project"] root = tmp_workspace_project["projects_root"] rel_dir = "output/video/ep_001" video = _seed_reclaim_video( root, project, rel_dir, "shot_EP001_SH98_take2_seedance.mp4" ) # Only the canonical name exists (take-suffixed name absent) _write_reclaim_meta(root / project / rel_dir / "EP001_SH98_meta.yaml") res = mcp_server.tool_get_file_provenance({"path": f"{rel_dir}/{video.name}"}) assert res.get("reclaim_badge"), f"expected fallback to canonical meta, got {res}" assert res["reclaim_badge"]["synthetic"] is True assert res["reclaim_badge"]["meta_yaml"].endswith("EP001_SH98_meta.yaml")