"""Phase 11 integration test — invokes run_overnight --dry-run on real EP001 plan. Part A: end-to-end dry-run of the actual EP001 plan against `recoil.pipeline.cli.run_overnight`. Walks scene JSON output (if produced) and asserts post-Phase-2-through-10 invariants: - no `@Image{` placeholders remain (refs hydrated) - all `start_frame` payload entries are JSON-serializable strings (or None) Part B: cleanup-correctness — asserts that Phase 11 deletions / tombstones landed (no `pass_naming` or `video_naming` imports remain, the three legacy files are gone, no `_orphans/` dirs survive under projects/, both legacy writers are tombstoned). R4 Phase 10 (2026-05-21): the expected-failure marker was removed from test_ep001_dry_run_end_to_end — the PosixPath JSON-serialization bug at persistence.save_scene was fixed via default=str encoder. The dry-run is now a hard assertion. A second smoke exercises `generate.py --pass --dry-run` to guard against the B1 NotImplementedError regression. """ from __future__ import annotations import json import os import subprocess import sys from pathlib import Path REPO = Path(__file__).resolve().parents[2] def _run_overnight_dry() -> subprocess.CompletedProcess[str]: """Invoke `python3 -m recoil.pipeline.cli.run_overnight ... --dry-run`. PYTHONPATH includes REPO (recoil package root) AND recoil/pipeline so legacy `from orchestrator.x import ...` style imports inside pipeline.py resolve. This mirrors recoil/pipeline/tests/conftest.py. """ pythonpath = os.pathsep.join( [ str(REPO), str(REPO / "recoil" / "pipeline"), str(REPO / "recoil"), ] ) return subprocess.run( [ sys.executable, "-m", "recoil.pipeline.cli.run_overnight", "--project", "tartarus", "--episode", "ep_001", "--budget-usd", "5", "--dry-run", ], cwd=REPO, env={"PYTHONPATH": pythonpath, **os.environ}, capture_output=True, text=True, timeout=600, ) def test_ep001_dry_run_end_to_end(): """The real EP001 plan dispatches successfully in dry-run mode. R4 Phase 10: expected-failure marker removed. The PosixPath JSON-serialization bug at persistence.save_scene is fixed (default=str). rc != 0 is now a hard fail. """ result = _run_overnight_dry() assert result.returncode == 0, ( f"run_overnight --dry-run exited {result.returncode}\n" f"stderr={result.stderr[-2000:]}\nstdout={result.stdout[-1000:]}" ) scenes_dir = REPO / "projects" / "tartarus" / "state" / "visual" / "scenes" / "ep_001" if not scenes_dir.exists(): # Dry-run may not write scene files in some configurations — acceptable. return for scene_json in scenes_dir.glob("*.json"): text = scene_json.read_text() assert "@Image{" not in text, ( f"unhydrated ref token in {scene_json.name}" ) scene = json.loads(text) for shot in scene.get("shots", []): sf = shot.get("payload", {}).get("start_frame") assert sf is None or isinstance(sf, str), ( f"non-serializable start_frame in " f"{scene_json.name}/{shot.get('shot_id')}" ) def test_no_pass_naming_imports_remain(): """After cleanup, no recoil/ module imports from recoil.core.pass_naming. Scoped to `recoil/` to ignore documentation references in consultations/, build-logs, and other non-code artifacts. The test file itself is excluded — its docstring and grep argument naturally contain the search string. """ result = subprocess.run( ["git", "grep", "-l", "from recoil.core.pass_naming", "--", "recoil/", ":!recoil/tests/test_integration.py"], cwd=REPO, capture_output=True, text=True, ) assert result.returncode != 0 or result.stdout.strip() == "", ( f"pass_naming imports remain in recoil/: {result.stdout}" ) def test_no_video_naming_imports_remain(): """After cleanup, no recoil/ module imports from recoil.core.video_naming. Scoped to `recoil/` to ignore documentation references in consultations/, build-logs, and other non-code artifacts. The test file itself is excluded — its docstring and grep argument naturally contain the search string. """ result = subprocess.run( ["git", "grep", "-l", "from recoil.core.video_naming", "--", "recoil/", ":!recoil/tests/test_integration.py"], cwd=REPO, capture_output=True, text=True, ) assert result.returncode != 0 or result.stdout.strip() == "", ( f"video_naming imports remain in recoil/: {result.stdout}" ) def test_no_orphans_dirs_remain(): """All projects/*/output/*/_orphans/ directories are removed.""" result = subprocess.run( ["find", "projects/", "-name", "_orphans", "-type", "d"], cwd=REPO, capture_output=True, text=True, ) assert result.stdout.strip() == "", ( f"_orphans dirs remain: {result.stdout}" ) def test_migrate_pass_names_deleted(): """recoil/pipeline/tools/migrate_pass_names.py is deleted.""" assert not (REPO / "recoil/pipeline/tools/migrate_pass_names.py").exists() def test_pass_naming_deleted(): """recoil/core/pass_naming.py is deleted.""" assert not (REPO / "recoil/core/pass_naming.py").exists() def test_video_naming_shim_deleted(): """recoil/core/video_naming.py shim is deleted.""" assert not (REPO / "recoil/core/video_naming.py").exists() def test_legacy_writers_tombstoned(): """Both legacy writers in pipeline.py raise NotImplementedError.""" text = (REPO / "recoil/pipeline/orchestrator/pipeline.py").read_text() assert "Legacy I2VPipeline deprecated" in text assert "Legacy T2VPipeline deprecated" in text assert text.count("NotImplementedError") >= 2 def test_generate_cli_dry_run_dispatches(): """B1 regression guard: generate.py --pass --dry-run must NOT raise NotImplementedError. Sanctioned CLI surface for the /generate-video skill.""" result = subprocess.run( [sys.executable, "recoil/pipeline/cli/generate.py", "--project", "tartarus", "--episode", "1", "--pass", "EP001_PASS_008_SH16_17_18_A_WREN", "--budget", "4", "--dry-run"], cwd=REPO, capture_output=True, text=True, timeout=120, env={"PYTHONPATH": str(REPO), **os.environ}, ) assert "NotImplementedError" not in result.stderr, ( f"B1 regression: stderr={result.stderr[-800:]}" ) assert "phase-21-22 deferred" not in result.stderr # rc may be non-zero if the specific pass id isn't in the current plan; # the assertion is ONLY about the NotImplementedError stub not firing.