backlot: fix all five dogfood findings (F-01..F-05)

- F-01: cost bar crit (red) state past 90% of budget
- F-02: normalize() hardens fetched board state against sparse payloads
- F-03: /thumb 404s for videos with no extractable poster frame instead
  of serving raw video bytes
- F-04: checkpoint artifact path refs only resolve inside the project dir
- F-05 (board half): stall detection — in_progress stage with no disk
  activity >10min renders red 'stalled?' + header badge flips to STALLED?;
  verified against the real wedged why-cities-glow project
- eval harness from dogfood session committed (visual regression +
  interaction smoke, capture watcher, server/gate test suites) +
  regression tests for each finding; 46 backlot tests green, visual eval
  green (restage-before-capture note logged)
This commit is contained in:
calesthio
2026-07-02 08:22:03 -07:00
parent 811480d39b
commit 1d60f0da14
11 changed files with 950 additions and 5 deletions

View File

@@ -0,0 +1,99 @@
"""Gate-integrity scenarios for Backlot and checkpoint hardening."""
import json
from pathlib import Path
import pytest
from backlot import state as state_mod
from backlot.state import load_board_state
from lib.checkpoint import CheckpointValidationError, write_checkpoint
def _script_artifact() -> dict:
return {
"version": "1.0",
"title": "Gate Test",
"total_duration_seconds": 5,
"sections": [{"id": "s1", "text": "Hello.", "start_seconds": 0, "end_seconds": 5}],
}
def _manifest_artifact() -> dict:
return {"version": "1.0", "assets": [], "total_cost_usd": 0.0}
def _write(path: Path, data: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data), encoding="utf-8")
def test_completed_gated_stage_without_approval_is_rejected(tmp_path):
with pytest.raises(CheckpointValidationError, match="GATE VIOLATION"):
write_checkpoint(
tmp_path,
"film",
"script",
"completed",
{"script": _script_artifact()},
pipeline_type="cinematic",
)
def test_typo_pipeline_type_fails_closed(tmp_path):
with pytest.raises(CheckpointValidationError, match="Unknown pipeline_type"):
write_checkpoint(
tmp_path,
"film",
"script",
"completed",
{"script": _script_artifact()},
pipeline_type="cinemtaic",
human_approved=True,
)
def test_handwritten_completed_checkpoint_surfaces_gate_skip(tmp_path, monkeypatch):
monkeypatch.setattr(state_mod, "PROJECTS_DIR", tmp_path)
project = tmp_path / "film"
_write(project / "checkpoint_script.json", {
"version": "1.0",
"project_id": "film",
"pipeline_type": "cinematic",
"stage": "script",
"status": "completed",
"timestamp": "2026-07-02T00:00:00Z",
"artifacts": {"script": _script_artifact()},
})
state = load_board_state(project)
script = next(stage for stage in state["stages"] if stage["name"] == "script")
assert script["gate_skipped"] is True
def test_awaiting_then_approved_archives_history_without_gate_skip(tmp_path):
write_checkpoint(
tmp_path,
"film",
"assets",
"awaiting_human",
{"asset_manifest": _manifest_artifact()},
pipeline_type="cinematic",
)
write_checkpoint(
tmp_path,
"film",
"assets",
"completed",
{"asset_manifest": _manifest_artifact()},
pipeline_type="cinematic",
human_approved=True,
)
state = load_board_state(tmp_path / "film")
assets = next(stage for stage in state["stages"] if stage["name"] == "assets")
assert assets.get("gate_skipped") in (None, False)
assert assets["versions"] == 2
assert assets["history_entries"][0]["status"] == "awaiting_human"

View File

@@ -0,0 +1,205 @@
"""Server/API tests for Backlot.
These cover the deterministic eval surface in internal/evals/BACKLOT_EVAL_PLAN.md:
API shape, path safety, media/thumb serving, range requests, and loose
performance budgets.
"""
from __future__ import annotations
import io
import json
import time
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from PIL import Image
from backlot import server as server_mod
from backlot import state as state_mod
@pytest.fixture
def projects_root(tmp_path, monkeypatch):
root = tmp_path / "projects"
root.mkdir()
monkeypatch.setattr(state_mod, "PROJECTS_DIR", root)
monkeypatch.setattr(server_mod, "PROJECTS_DIR", root)
monkeypatch.setattr(server_mod, "_summary_cache", {})
monkeypatch.setattr(server_mod, "_PROJECTS_ROOT_STR", __import__("os").path.normcase(str(root.resolve())))
monkeypatch.setattr(server_mod, "THUMB_CACHE_DIR", tmp_path / "thumbs")
return root
@pytest.fixture
def client(projects_root, monkeypatch):
async def no_watch():
return None
monkeypatch.setattr(server_mod, "_watch_projects", no_watch)
with TestClient(server_mod.create_app()) as c:
yield c
def _write_json(path: Path, data: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data), encoding="utf-8")
def _make_project(root: Path, project_id: str = "film") -> Path:
project = root / project_id
(project / "artifacts").mkdir(parents=True)
(project / "assets" / "images").mkdir(parents=True)
(project / "assets" / "video").mkdir(parents=True)
(project / "renders").mkdir(parents=True)
_write_json(
project / "project.json",
{
"project_id": project_id,
"title": "Film",
"pipeline_type": "cinematic",
"created_at": "2026-07-02T00:00:00Z",
},
)
_write_json(
project / "checkpoint_script.json",
{
"version": "1.0",
"project_id": project_id,
"pipeline_type": "cinematic",
"stage": "script",
"status": "awaiting_human",
"timestamp": "2026-07-02T00:01:00Z",
"artifacts": {},
},
)
return project
def _write_png(path: Path, color: tuple[int, int, int] = (200, 40, 80)) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
img = Image.new("RGB", (24, 16), color)
buf = io.BytesIO()
img.save(buf, format="PNG")
path.write_bytes(buf.getvalue())
class TestBacklotServerApi:
def test_health(self, client):
response = client.get("/api/health")
assert response.status_code == 200
assert response.json() == {"ok": True, "app": "backlot"}
def test_projects_shape_and_state(self, client, projects_root):
_make_project(projects_root, "film")
projects = client.get("/api/projects")
assert projects.status_code == 200
body = projects.json()
assert len(body) == 1
assert body[0]["project_id"] == "film"
assert body[0]["awaiting_human"] is True
assert "stage_states" in body[0]
state = client.get("/api/project/film/state")
assert state.status_code == 200
state_body = state.json()
assert state_body["project_id"] == "film"
assert state_body["title"] == "Film"
assert state_body["stages"]
@pytest.mark.parametrize(
("url", "status"),
[
("/api/project/../state", 404),
("/api/project/C:/state", 400),
("/api/project/nope/state", 404),
],
)
def test_project_id_rejects_bad_or_unknown_ids(self, client, url, status):
response = client.get(url)
assert response.status_code == status
def test_media_rejects_path_traversal(self, client, projects_root):
_make_project(projects_root, "film")
response = client.get("/media/film/%2E%2E/project.json")
assert response.status_code == 403
def test_media_serves_range_requests(self, client, projects_root):
project = _make_project(projects_root, "film")
media = project / "renders" / "final.mp4"
media.write_bytes(b"0123456789")
response = client.get("/media/film/renders/final.mp4", headers={"Range": "bytes=2-5"})
assert response.status_code == 206
assert response.content == b"2345"
assert response.headers["content-range"].startswith("bytes 2-5/10")
def test_thumb_downscales_image_and_passes_through_non_media(self, client, projects_root):
project = _make_project(projects_root, "film")
_write_png(project / "assets" / "images" / "sc1.png")
text = project / "artifacts" / "note.txt"
text.write_text("hello", encoding="utf-8")
image = client.get("/thumb/film/assets/images/sc1.png?w=320")
assert image.status_code == 200
assert image.headers["content-type"] == "image/jpeg"
assert image.content.startswith(b"\xff\xd8")
passthrough = client.get("/thumb/film/artifacts/note.txt")
assert passthrough.status_code == 200
assert passthrough.content == b"hello"
class TestBacklotPerformanceBudgets:
def test_projects_and_state_stay_within_loose_budgets(self, client, projects_root):
for i in range(25):
project = _make_project(projects_root, f"film-{i:02d}")
_write_json(
project / "artifacts" / "scene_plan.json",
{"version": "1.0", "scenes": [{"id": "sc1", "start_seconds": 0, "end_seconds": 1}]},
)
t0 = time.perf_counter()
cold = client.get("/api/projects")
cold_s = time.perf_counter() - t0
assert cold.status_code == 200
assert cold_s < 2.0
t1 = time.perf_counter()
warm = client.get("/api/projects")
warm_s = time.perf_counter() - t1
assert warm.status_code == 200
assert warm_s < 0.150
t2 = time.perf_counter()
state = client.get("/api/project/film-00/state")
state_s = time.perf_counter() - t2
assert state.status_code == 200
assert state_s < 0.400
def test_image_thumb_generation_stays_within_budget(self, client, projects_root):
project = _make_project(projects_root, "film")
_write_png(project / "assets" / "images" / "sc1.png")
t0 = time.perf_counter()
response = client.get("/thumb/film/assets/images/sc1.png?w=640")
elapsed = time.perf_counter() - t0
assert response.status_code == 200
assert elapsed < 1.5
class TestFindingsFixes:
"""Regression tests for dogfood findings F-03 (thumb video fallback)."""
def test_thumb_never_serves_raw_video_bytes(self, client, projects_root):
p = _make_project(projects_root, "vid")
fake_video = p / "renders" / "final.mp4"
fake_video.parent.mkdir(parents=True, exist_ok=True)
# Not a real video: ffmpeg poster extraction will fail.
fake_video.write_bytes(b"\x00" * 4096)
res = client.get("/thumb/vid/renders/final.mp4")
assert res.status_code == 404 # never the raw video bytes (F-03)

View File

@@ -188,3 +188,59 @@ class TestLibrary:
summary = summarize_project(p)
assert summary["awaiting_human"] is True
assert summary["active_stage"] == "script"
class TestFindingsFixes:
"""Regression tests for dogfood findings F-04/F-05."""
def test_artifact_refs_outside_project_are_not_followed(self, projects_root, tmp_path):
# F-04: a checkpoint pointing at JSON outside the project tree
# must not surface that file on the board.
secret = tmp_path / "secret.json"
secret.write_text(json.dumps({"version": "1.0", "leaked": True}), encoding="utf-8")
p = _make_project(projects_root, "sneaky-ref")
_write(p / "checkpoint_script.json", {
"stage": "script", "status": "completed",
"timestamp": "2026-01-01T01:00:00Z",
"artifacts": {"script": str(secret)},
})
s = load_board_state(p)
assert "script" not in s["artifacts"]
def test_inside_project_absolute_refs_still_resolve(self, projects_root):
p = _make_project(projects_root, "abs-ref")
_write(p / "artifacts" / "inline_script.json", SCRIPT)
_write(p / "checkpoint_script.json", {
"stage": "script", "status": "completed",
"timestamp": "2026-01-01T01:00:00Z",
"artifacts": {"script": str((p / "artifacts" / "inline_script.json").resolve())},
})
s = load_board_state(p)
assert s["artifacts"]["script"]["title"] == "Test Film"
def test_stalled_in_progress_stage_flagged(self, projects_root):
# F-05: an in_progress stage with no recent activity reads stalled.
import os
p = _make_project(projects_root, "wedged")
_write(p / "checkpoint_research.json", {
"stage": "research", "status": "in_progress",
"timestamp": "2026-01-01T01:00:00Z", "artifacts": {},
})
past = time.time() - 30 * 60
for f in p.rglob("*"):
if f.is_file():
os.utime(f, (past, past))
s = load_board_state(p)
research = next(x for x in s["stages"] if x["name"] == "research")
assert research["stalled"] is True
assert research["stalled_minutes"] >= 29
def test_fresh_in_progress_not_stalled(self, projects_root):
p = _make_project(projects_root, "busy")
_write(p / "checkpoint_research.json", {
"stage": "research", "status": "in_progress",
"timestamp": "2026-01-01T01:00:00Z", "artifacts": {},
})
s = load_board_state(p)
research = next(x for x in s["stages"] if x["name"] == "research")
assert "stalled" not in research

View File

@@ -0,0 +1,43 @@
"""Tests for Backlot visual eval image comparison helpers."""
from pathlib import Path
from PIL import Image
from scripts.backlot_visual_eval import compare_images
def _img(path: Path, color: tuple[int, int, int]) -> None:
Image.new("RGB", (10, 10), color).save(path)
def test_compare_images_detects_large_drift(tmp_path):
expected = tmp_path / "expected.png"
actual = tmp_path / "actual.png"
diff = tmp_path / "diff.png"
_img(expected, (0, 0, 0))
_img(actual, (255, 255, 255))
result = compare_images(expected, actual, diff, threshold=0.015)
assert result["passed"] is False
assert result["changed_ratio"] == 1.0
assert diff.exists()
def test_compare_images_can_mask_regions(tmp_path):
expected = tmp_path / "expected.png"
actual = tmp_path / "actual.png"
diff = tmp_path / "diff.png"
_img(expected, (0, 0, 0))
_img(actual, (0, 0, 0))
img = Image.open(actual)
for x in range(5):
for y in range(5):
img.putpixel((x, y), (255, 255, 255))
img.save(actual)
result = compare_images(expected, actual, diff, threshold=0.015, masks=[(0, 0, 5, 5)])
assert result["passed"] is True
assert result["changed_ratio"] == 0.0

View File

@@ -0,0 +1,47 @@
"""Tests for the Backlot dogfood screenshot watcher helpers."""
from scripts.backlot_watch_captures import capture_slug, state_fingerprint
def test_capture_slug_keeps_names_filesystem_safe():
assert capture_slug("why-cities-glow", "scene_plan", "awaiting_human") == (
"why-cities-glow-scene_plan-awaiting_human"
)
assert capture_slug("../bad id", "C:\\stage", "in progress!") == "bad-id-C-stage-in-progress"
def test_state_fingerprint_changes_on_board_relevant_state_only():
state = {
"stages": [
{"name": "script", "status": "completed", "partial_progress": None},
{"name": "assets", "status": "in_progress", "partial_progress": {"done": ["sc1"]}},
],
"storyboard": {
"scenes": [
{
"id": "sc1",
"generating": False,
"visual": {"path": "assets/images/sc1.png", "exists": True},
"takes": [{"path": "assets/images/sc1.png"}],
},
{"id": "sc2", "generating": True, "generating_tool": "flux_image", "visual": None},
]
},
"cost": {"total_spent_usd": 0.1},
"media": {"renders": []},
"events": [{"event": "start", "tool": "flux_image"}],
"last_activity": 123,
}
same = dict(state)
same["last_activity"] = 999
changed = dict(state)
changed["storyboard"] = {
"scenes": [
state["storyboard"]["scenes"][0],
{"id": "sc2", "generating": False, "visual": {"path": "assets/images/sc2.png", "exists": True}},
]
}
assert state_fingerprint(state) == state_fingerprint(same)
assert state_fingerprint(state) != state_fingerprint(changed)