""" Parity test โ€” after any sequence of asset-ops, ref_resolver (the pipeline view) and _api_assets_list (the UI view) must return the same hero for every asset. This is the structural proof that the Phase 1 refactor eliminated the three-codepath drift that caused the April 6 9-round debug convergence. This is the mechanical implementation of the Phase 1 acceptance gate "UI and pipeline return the same hero." Spec ยง10. """ import random from pathlib import Path import pytest from recoil.pipeline._lib import asset_ops from recoil.core import ref_resolver from recoil.pipeline.editors.review_server import _api_assets_list_sync RANDOM_SEQUENCE_COUNT = 20 ASSET_SLUGS = [ "sadie", "dusty", "widower", "int_dusty_bar", "int_sadie_apt", "ext_street", ] EXTENSIONS = [".png", ".jpg", ".jpeg", ".webp"] ASSET_TYPES = ["character", "location"] OPS = ["set_hero", "add_ref", "delete_ref", "tag_phase"] PHASES = [None, "ph1", "ph2", "ph3"] @pytest.fixture def synthetic_canonical(tmp_path): """Create a minimal _canonical/ tree for testing.""" canonical = tmp_path / "output" / "refs" / "_canonical" for asset_type in ["characters", "locations", "props"]: (canonical / asset_type).mkdir(parents=True) (tmp_path / "state" / "visual").mkdir(parents=True) return canonical def _asset_type_for(slug: str) -> str: return "character" if slug in ("sadie", "dusty", "widower") else "location" def _generate_random_op(rng: random.Random) -> dict: op = rng.choice(OPS) slug = rng.choice(ASSET_SLUGS) asset_type = _asset_type_for(slug) ext = rng.choice(EXTENSIONS) phase = rng.choice(PHASES) return { "op": op, "slug": slug, "asset_type": asset_type, "ext": ext, "phase": phase, } def _make_source(tmp_path: Path, name: str, ext: str) -> Path: src = tmp_path / f"{name}{ext}" src.write_bytes(b"\x89PNG\r\n\x1a\n" + b"0" * 1024) return src def _apply_op(canonical_root: Path, op_spec: dict, tmp_path: Path, step: int) -> None: """Apply an op via asset_ops (same path the Phase 1 refactor routes through). canonical_root is .../output/refs/_canonical. project_root is its great-grandparent. """ project_root = canonical_root.parent.parent.parent op = op_spec["op"] slug = op_spec["slug"] asset_type = op_spec["asset_type"] ext = op_spec["ext"] phase = op_spec["phase"] if op == "set_hero": src = _make_source(tmp_path, f"src_{step}_hero", ext) asset_ops.set_hero( project_root=project_root, asset_type=asset_type, asset_id=slug, source=src, phase=phase, ) elif op == "add_ref": src = _make_source(tmp_path, f"src_{step}_ref", ext) asset_ops.add_ref( project_root=project_root, asset_type=asset_type, asset_id=slug, source=src, ) elif op == "delete_ref": # Pick a real file in the canonical dir to delete (best-effort). type_folder = {"character": "characters", "location": "locations"}[asset_type] canonical_dir = canonical_root / type_folder / slug if not canonical_dir.exists(): return candidates = [f for f in canonical_dir.glob("*.*") if f.is_file()] if not candidates: return # Deterministic pick โ€” first match. Avoids extra rng calls so the # parametrized seed remains the only source of nondeterminism. victim = sorted(candidates)[0] try: asset_ops.delete_ref( project_root=project_root, asset_type=asset_type, asset_id=slug, filename=victim.name, ) except Exception: pass elif op == "tag_phase": # tag_phase needs an existing file to rename. Find any non-hero ref. type_folder = {"character": "characters", "location": "locations"}[asset_type] canonical_dir = canonical_root / type_folder / slug if not canonical_dir.exists(): return candidates = [ f for f in canonical_dir.glob("*.*") if f.is_file() and not f.stem.startswith("hero") ] if not candidates: return victim = sorted(candidates)[0] try: asset_ops.tag_phase( project_root=project_root, asset_type=asset_type, asset_id=slug, filename=victim.name, phase=phase or "ph1", ) except Exception: pass def _assert_parity(project_root: Path, slug: str, asset_type: str): """The core invariant: ref_resolver and _api_assets_list must agree on hero.""" refs_root = project_root / "output" / "refs" if asset_type == "character": pipeline_view = ref_resolver.resolve_character_refs(refs_root, slug) else: pipeline_view = ref_resolver.resolve_location_refs(refs_root, slug) ui_view = _api_assets_list_sync(project_root=project_root) ui_section = ui_view.get(asset_type + "s", {}) ui_asset = ui_section.get(slug, {}) ui_hero = next((r for r in ui_asset.get("refs", []) if r.get("hero")), None) ui_hero_filename = ui_hero["filename"] if ui_hero else None pipeline_hero = ( pipeline_view.get("hero") if isinstance(pipeline_view, dict) else None ) pipeline_hero_filename = pipeline_hero.name if pipeline_hero else None assert pipeline_hero_filename == ui_hero_filename, ( f"PARITY VIOLATION for {asset_type}/{slug}:\n" f" pipeline (ref_resolver): {pipeline_hero_filename}\n" f" UI (_api_assets_list): {ui_hero_filename}\n" f"This is the Phase 1 no-drift gate. One codepath is wrong." ) @pytest.mark.parametrize("seed", [0, 1, 2, 42, 1337]) def test_ui_pipeline_parity_random_sequences(synthetic_canonical, tmp_path, seed): """Deterministic parity test: for each seed, run 20 random asset-ops, assert parity after every step. Phase 1 gate fails if any seed fails. Seed 0 is the canonical "must always pass" case (hardcoded into CI); other seeds are sanity spot-checks.""" rng = random.Random(seed) project_root = synthetic_canonical.parent.parent.parent for step in range(RANDOM_SEQUENCE_COUNT): op_spec = _generate_random_op(rng) try: _apply_op(synthetic_canonical, op_spec, tmp_path, step) except (FileNotFoundError, ValueError): # Some random ops are invalid (e.g. delete_ref on missing file); # that's fine โ€” the point is that WHATEVER state we end up in, # pipeline and UI views must agree. pass # After every step, verify parity for every known asset for slug in ASSET_SLUGS: asset_type = _asset_type_for(slug) _assert_parity( project_root=project_root, slug=slug, asset_type=asset_type, )