Viggle Remake
21665fa about 14 hours ago
26.5 kB
"""End-to-end test of the MCP over stdio with no GPU and no paid call: tests/fake_worker.py renders, tests/fake_openai.py
answers the image and chat calls. Wipes and fills $REMAKE_HOME (its path must contain "test").
    REMAKE_HOME=/scratch/remake_home_test python tests/e2e.py CLIP.mp4
CLIP: >= 10 s, landscape, with audio, cuts near 0.8 s and 6.1 s (the shots and keyframes below are for that clip).
Flows:
  1 swap, codex image mode, client-written prompts, worker max 345: one window, sheet/paint briefs, a wrong-aspect import
  2 edit, API image mode: epaint, cut_windows/edit_prompt, a draft and a final, the image-call limit, quota -> STOP
  3 swap, API image mode + server LLM prompts, worker max 124: a hop chain, the USD cap, a stuck hop, the too-long refusal
  4 no swap worker live: windows falls back to 345, 720 is refused, wait returns at once
  5 other aspects (crop): 4:3, 9:16 and 3:1 (squeezed to 2.31:1) canvases, padding, finals at the source's aspect
Flows 1, 2 and 5 ingest spans that are not 17k+5 long (padded with the last shot played backwards), flow 3 one that is.
"""
import asyncio
import json
import os
import re
import shutil
import subprocess
import sys
import time
from contextlib import asynccontextmanager
from pathlib import Path

from mcp import ClientSession, StdioServerParameters, stdio_client
import numpy as np
from PIL import Image, ImageDraw, ImageOps

HERE = Path(__file__).resolve().parent
HOME = Path(os.environ["REMAKE_HOME"]).absolute()
CLIP = Path(sys.argv[1]).resolve()
PY = sys.executable
MCP = str(Path(PY).parent / "viggle-remake-mcp")
LOG = HOME / "fake_openai.jsonl"
assert "test" in str(HOME), f"{HOME}: the test wipes REMAKE_HOME; use a path containing 'test'"
shutil.rmtree(HOME, ignore_errors=True)
(HOME / "chars").mkdir(parents=True)
(HOME / "inputs").mkdir()
procs = {}


def start_worker(name, mx, modes="swap,edit", delay=0.0):
    procs[name] = subprocess.Popen([PY, HERE / "fake_worker.py", "--home", HOME, "--name", name, "--max480", str(mx),
                                    "--max736", str(mx), "--modes", modes, "--delay", str(delay)])
    while not (HOME / "workers" / f"fake_{name}.json").exists():
        time.sleep(0.2)


def stop_worker(name):  # and drop its heartbeat (a dead worker's beat stays "live" for 120 s)
    procs.pop(name).kill()
    time.sleep(0.3)
    (HOME / "workers" / f"fake_{name}.json").unlink()


def card(path, size, text, color):
    im = Image.new("RGB", size, color)
    ImageDraw.Draw(im).text((40, 40), text, fill="white", font_size=80)
    im.save(path)


def probe(p):
    """-> (frames, width, height, has audio)"""
    r = json.loads(subprocess.run(["ffprobe", "-v", "error", "-count_packets", "-show_entries",
                                   "stream=codec_type,nb_read_packets,width,height", "-of", "json", str(p)],
                                  capture_output=True, text=True, check=True).stdout)["streams"]
    v = next(s for s in r if s["codec_type"] == "video")
    return int(v["nb_read_packets"]), v["width"], v["height"], any(s["codec_type"] == "audio" for s in r)


def frames(p):
    """-> uint8 (n, h, w, 3)"""
    _, w, h, _ = probe(p)
    raw = subprocess.run(["ffmpeg", "-v", "error", "-i", str(p), "-f", "rawvideo", "-pix_fmt", "rgb24", "-"],
                         capture_output=True, check=True).stdout
    return np.frombuffer(raw, np.uint8).reshape(-1, h, w, 3)


def requests():
    return [json.loads(x) for x in LOG.read_text().splitlines()] if LOG.exists() else []


def answer(text):
    """Do what a CODEX IMAGE STEP says: 'generate' (a tinted copy of the last referenced picture at the brief's size)
    and copy it to copy_to."""
    refs = json.loads(re.search(r"referenced_image_paths = (\[.*\])", text).group(1))
    copy_to = Path(re.search(r"\bto (/\S+)\n", text).group(1))
    W, H = map(int, json.loads(copy_to.with_suffix(".json").read_text())["size"].split("x"))
    im = Image.open(refs[-1]).convert("L")
    ImageOps.colorize(im, "#103010", "#c0f0c0").resize((W, H)).save(copy_to)
    return copy_to


@asynccontextmanager
async def server(**env):
    p = StdioServerParameters(command=MCP, env={"REMAKE_HOME": str(HOME), "PATH": os.environ["PATH"], **env})
    async with stdio_client(p) as (r, w), ClientSession(r, w) as s:
        await s.initialize()

        async def call(tool, ok=True, **args):
            t0 = time.time()
            res = await s.call_tool(tool, args, read_timeout_seconds=900)
            text = "\n".join(c.text for c in res.content if c.type == "text")
            nimg = sum(c.type == "image" for c in res.content)
            assert res.is_error != ok, f"{tool}({args}) is_error={res.is_error}:\n{text[:3000]}"
            print(f"  {time.time() - t0:5.1f}s {tool}{'' if ok else ' (expected error)'}: {text[:140]!r}{f' +{nimg} img' if nimg else ''}",
                  flush=True)
            return text, nimg
        call.session = s
        yield call


def plan(nf, cuts, x1=None):
    b = [0] + cuts + [nf]
    people = [{"id": "P1", "desc": "the tall man in the green elf costume and yellow tights", "replace_with": "Vela"},
              {"id": "P2", "desc": "the big man in the dark red store vest", "replace_with": "none"}]
    if x1:
        people[0]["replace_with"] = "none"
        people.append({"id": "X1", "desc": "the white clipboard the man in the red vest holds", "replace_with": x1})
    return {"people": people, "shots": [{"start": s, "end": e, "people": [p["id"] for p in people], "desc": "in a toy store"}
                                        for s, e in zip(b, b[1:])]}


def where(fs, x1=False):
    return {str(f): {"P1": "left, facing right", "P2": "right, facing left", **({"X1": "in P2's left hand"} if x1 else {})}
            for f in fs}


def prompt_text(n, extra=""):
    pics = " ".join(f"<Picture {k + 1}> is a frame of the target video." for k in range(n))
    return (f"subject_definitions:\n<Video 1> is the source video. {pics} <Subject 1> is Vela.{extra}\n\n"
            "summary:\nThe target video is <Video 1> with the elf replaced by <Subject 1>.\n\n"
            "retention_analysis:\n<Video 1>: fully_preserved - shots, cuts, camera and timing.\n\n"
            "detailed_description:\n[Shot 1] At 00:00.000 <Subject 1> talks: <d>[unclear]</d>.\n\n"
            "overall_soundscape:\nStore ambience as in the source.\n\nnon_diegetic_music:\nN/A\n")


async def main():
    fo = procs["openai"] = subprocess.Popen([PY, HERE / "fake_openai.py", LOG], stdout=subprocess.PIPE, text=True)
    port = int(fo.stdout.readline())
    api = {"OPENAI_API_KEY": "sk-fake-test", "OPENAI_BASE_URL": f"http://127.0.0.1:{port}/v1"}
    card(HOME / "chars" / "Vela_sheet.png", (1536, 1024), "Vela sheet", "#305080")
    card(HOME / "chars" / "Vela_photo.png", (768, 1024), "Vela photo", "#803050")
    (HOME / "chars" / "lib_test.json").write_text(json.dumps({
        "chars": {"Vela": "Vela, a woman in her thirties, slim, short silver hair, teal flight jacket"},
        "outfits": {"Vela": "teal flight jacket, grey turtleneck, cargo trousers, brown boots"}}))
    card(HOME / "inputs" / "orin.png", (800, 1000), "Orin", "#607030")
    card(HOME / "inputs" / "nest.png", (800, 800), "nest", "#706050")
    start_worker("A", 345)

    print("== 1 swap, codex image mode, client prompts, worker max 345")
    async with server(REMAKE_PAINT="codex", **api) as c:
        tools = (await c.session.list_tools()).tools
        assert len(tools) == 23 and {"set_prompt", "write_prompts", "cut_windows"} <= {t.name for t in tools}, [t.name for t in tools]
        sid, d = "swapc", HOME / "sessions" / "swapc"
        await c("new_session", sid=sid, kind="swap")
        await c("new_session", ok=False, sid=sid, kind="edit")
        await c("new_session", ok=False, sid="bad_name", kind="swap")
        t, n = await c("peek", sid=sid, src=str(CLIP), dur=10)
        assert n == 1 and json.loads(t)["size"] == [1920, 1080]
        t, n = await c("ingest", sid=sid, src=str(CLIP), dur=10)
        nf, cuts = int((d / "case/nf").read_text()), json.loads((d / "case/cuts.json").read_text())
        meta = json.loads((d / "case/meta.json").read_text())
        assert nf == 243 and meta["frames"] == 240 and cuts == [19, 147] and n == 1, (nf, meta, cuts)
        assert meta["draft"] == [832, 480] and meta["final"] == [1280, 736] and "frames 240-242 are padding" in t, t
        assert probe(d / "case/cond.mp4")[:3] == (243, 832, 480) and probe(d / "case/cond720.mp4")[:3] == (243, 1280, 736)
        # the padding is the last shot played backwards: frame 240 is frame 238, 242 is 236
        v = frames(d / "case/cond.mp4")
        assert all(abs(v[a].astype(int) - v[b]).mean() < 3 for a, b in ((240, 238), (242, 236))), "padding"
        await c("ingest", ok=False, sid=sid, src=str(CLIP), dur=10)  # already ingested, no force
        t, _ = await c("sheet", sid=sid, name="Vela", source="lib")
        assert "filled from lib_test.json" in t
        kf = [9, 83, 186]
        t, _ = await c("save_case", sid=sid, plan_grid=plan(nf, cuts), where=where(kf))
        assert "keyframes not painted yet: [9, 83, 186]" in t, t
        t, _ = await c("save_case", sid=sid, where=where([9, 83, 148]))
        assert "within 3 frames of a cut" in t
        t, _ = await c("save_case", sid=sid, where=where([9, 83, 241]))
        assert "frame 241 is in the padding" in t, t
        await c("save_case", sid=sid, where=where(kf))
        t, n = await c("frames", sid=sid, frames=kf)
        assert n == 1
        t, _ = await c("windows", sid=sid)
        assert "keyframe chosen but not painted yet" in t, t
        await c("write_prompts", ok=False, sid=sid)  # no painted keyframe
        # a second character from a photo outside the session: brief, generate, import
        case = json.loads((d / "case/case.json").read_text())
        case["chars"]["Orin"], case["outfits"]["Orin"] = "Orin, a man of fifty", "brown suit, black shoes"
        await c("save_case", sid=sid, case=case)
        photo = str(HOME / "inputs" / "orin.png")
        t, n = await c("sheet", sid=sid, name="Orin", source="photo", photo=photo)
        assert t.startswith("CODEX IMAGE STEP") and n == 0 and not (d / "case/chars/Orin_sheet.png").exists()
        assert (d / "photos/orin.png").exists()
        answer(t)
        t, n = await c("sheet", sid=sid, name="Orin", source="photo", photo=photo)
        assert n == 1 and Image.open(d / "case/chars/Orin_sheet.png").size == (1536, 1024)
        for fs in ([9], [83, 186]):
            t, n = await c("paint", sid=sid, frames=fs, quality="medium")
            assert t.startswith("CODEX IMAGE STEP") and n == 0
            answer(t)
            t, n = await c("paint", sid=sid, frames=fs, quality="medium")
            assert n == 1 and json.loads(t)["frames"] == fs
        await c("paint", ok=False, sid=sid, frames=[9])  # painted already, no force
        await c("paint", ok=False, sid=sid, frames=[9], quality="high")
        # a repaint whose import has the wrong aspect: nothing imported, the current paint stays
        before = (d / "case/paint/g2/f9.png").read_bytes()
        t, _ = await c("paint", sid=sid, frames=[9], force=True, note="hair colour")
        Image.new("RGB", (1024, 1024), "red").save(answer(t))
        t, _ = await c("paint", ok=False, sid=sid, frames=[9], force=True, note="hair colour")
        assert "aspect" in t and (d / "case/paint/g2/f9.png").read_bytes() == before
        t, _ = await c("paint", sid=sid, frames=[9], force=True, note="hair colour")
        answer(t)
        await c("paint", sid=sid, frames=[9], force=True, note="hair colour")
        assert (d / "case/paint/g2/old/f9_v0.png").read_bytes() == before
        t, _ = await c("budget", sid=sid)
        b = json.loads(t)
        assert b["paint_mode"] == "codex" and b["session_calls"] == 4 and b["session_limit"] == 80 and b["session_usd"] == 0, b
        await c("windows", sid=sid)
        rf = json.loads((d / "case/refs.json").read_text())["windows"]
        assert [(w["render"], w["refs"]) for w in rf] == [([0, 243], kf)], rf
        t, n = await c("write_prompts", sid=sid)
        assert "set_prompt(window=0" in t and n == 4 and "<d>[unclear]</d>" in t, (n, t[:500])
        assert t.count("=== GUIDE ===") == 1 and "/case/paint/" not in t, t[:500]  # inlined: no server paths to read
        await c("submit", ok=False, sid=sid)  # no prompt yet
        t, _ = await c("set_prompt", ok=False, sid=sid, window=0, text=prompt_text(2))
        assert "<Picture [3]>" in t
        await c("set_prompt", ok=False, sid=sid, window=0, text="summary:\n" + prompt_text(3))
        t, _ = await c("set_prompt", sid=sid, window=0, text=prompt_text(3, " The store is dimly lit — warm."))
        assert "warning: 'dim'" in t and "non-ASCII" in t
        await c("set_prompt", sid=sid, window=0, text=prompt_text(3))
        t, _ = await c("submit", sid=sid)
        assert json.loads(t)["queued"] == 1
        t, _ = await c("submit", sid=sid)  # queued or done already: skipped
        assert json.loads(t)["queued"] == 0
        t, _ = await c("wait", sid=sid, max_s=120)
        assert json.loads(t)["done"] == 1 and json.loads(t)["failed"] == 0, t
        t, n = await c("strip", sid=sid)
        assert n == 1
        await c("finish", ok=False, sid=sid, picks={"0": 3})
        t, _ = await c("finish", sid=sid, picks={"0": 0})
        final = d / "case/final/full_720.mp4"
        assert probe(final) == (240, 1280, 720, True), probe(final)  # the padding cut off, at the source's 16:9
        t, _ = await c("status", sid=sid)
        assert "image calls 4/80" in t and "full_720.mp4" in t
        t, n = await c("view", sid=sid, path="case/chars/Orin_sheet.png")
        assert n == 1
        await c("view", ok=False, sid=sid, path=str(HOME / "inputs" / "orin.png"))  # outside session and library
        await c("view", sid=sid, path=str(HOME / "chars" / "Vela_sheet.png"))
        await c("save_result", ok=False, sid=sid, result={"status": "done", "final": "nope.mp4"})
        await c("save_result", sid=sid, result={"status": "done", "final": str(final), "seconds": 9.4, "picks": {"0": 0},
                                                "windows": [{"w": 0, "grade": "OK", "note": "fake"}], "overall": "OK"})
        assert json.loads((d / "result.json").read_text())["kind"] == "swap"
        # case.json changed after the prompt was written: submit refuses the stale prompt
        orig = json.loads(json.dumps(case))
        case["chars"]["Vela"] += ", freckles"
        await c("save_case", sid=sid, case=case)
        t, _ = await c("submit", ok=False, sid=sid, seeds=[1])
        assert "stale prompt" in t
        await c("save_case", sid=sid, case=orig)  # the same bytes again: the prompt is current
        # a symlink planted under case/ blocks every tool
        (d / "case/evil").symlink_to("/tmp")
        await c("status", ok=False, sid=sid)
        (d / "case/evil").unlink()
    assert requests() == [], "codex image mode called the API"

    print("== 2 edit, API image mode")
    async with server(**api) as c:
        sid, d = "edita", HOME / "sessions" / "edita"
        await c("new_session", sid=sid, kind="edit")
        t, _ = await c("ingest", sid=sid, src=str(CLIP), dur=5)
        nf, cuts = int((d / "case/nf").read_text()), json.loads((d / "case/cuts.json").read_text())
        assert nf == 124 and cuts == [19], (nf, cuts)
        assert "frames 120-123 are padding" in t, t
        case = {"chars": {"NoBoard": "removed: no clipboard anywhere"}, "outfits": {"NoBoard": "none"},
                "edit": "The clipboard is removed."}
        t, _ = await c("save_case", sid=sid, plan_grid=plan(nf, cuts, x1="NoBoard"), case=case, where=where([9, 60], x1=True))
        assert "ERRORS" not in t
        await c("sheet", ok=False, sid=sid, name="Vela", source="lib")  # swap only
        await c("paint", ok=False, sid=sid, frames=[9])  # swap only
        t, n = await c("epaint", sid=sid, frame=9, edit="Remove the clipboard: his hand is open and empty.")
        assert json.loads(t)["usd"] == 0.0925 and n == 1
        t, n = await c("epaint", sid=sid, frame=60, edit="Replace the clipboard by a bird nest.", quality="medium",
                       refs=[str(HOME / "inputs" / "nest.png")])
        rq = requests()
        assert len(rq) == 2 and all(r["path"] == "/v1/images/edits" for r in rq), rq
        assert rq[0]["fields"]["model"] == "gpt-image-2" and rq[0]["fields"]["quality"] == "low" and rq[0]["fields"]["size"] == "1664x960"
        assert rq[1]["fields"]["quality"] == "medium" and rq[1]["images"] == [[800, 800], [1664, 960]], rq[1]
        assert rq[0]["auth"].startswith("Bearer sk-")
        assert (d / "photos/nest.png").exists()
        await c("windows", sid=sid)
        rf = json.loads((d / "case/refs.json").read_text())["windows"]
        assert [(w["render"], w["refs"]) for w in rf] == [([0, 124], [9, 60])], rf
        await c("write_prompts", ok=False, sid=sid)  # swap only
        await c("edit_prompt", ok=False, sid=sid, window=0, text=prompt_text(2))  # not cut yet
        t, n = await c("cut_windows", sid=sid)
        assert n == 1 and t.count("=== GUIDE ===") == 1 and "<Video 1> is the source video for the target video edit." in t
        t, _ = await c("edit_prompt", sid=sid, window=0, text=prompt_text(2))
        assert "bound to w0" in t
        t, _ = await c("submit", sid=sid, canvas="480")
        assert json.loads(t)["queued"] == 1
        await c("submit", sid=sid, canvas="720")
        t, _ = await c("wait", sid=sid, max_s=120)
        assert json.loads(t)["done"] == 2, t
        await c("strip", sid=sid, canvas="480")
        await c("finish", sid=sid, picks={"0": 0}, canvas="480")
        assert probe(d / "case/final/full_480.mp4") == (120, 832, 468, True)
        await c("finish", sid=sid, picks={"0": 0})
        assert probe(d / "case/final/full_720.mp4") == (120, 1280, 720, True)
        # the session's image-call limit (a paint_limit file)
        (d / "paint_limit").write_text("2")
        t, _ = await c("epaint", ok=False, sid=sid, frame=9, edit="x", force=True)
        assert "used its 2 image calls" in t
        (d / "paint_limit").unlink()
        # a quota error stops every session
        Path(str(LOG) + ".quota").touch()
        t, _ = await c("epaint", ok=False, sid=sid, frame=9, edit="Remove the clipboard.", force=True)
        assert "quota" in t and (HOME / "STOP").exists()
        n0 = len(requests())
        t, _ = await c("epaint", ok=False, sid=sid, frame=9, edit="Remove the clipboard.", force=True)
        assert "PAINT STOPPED" in t and len(requests()) == n0
        t, _ = await c("budget", sid=sid)
        b = json.loads(t)
        assert b["stopped"] and b["calls_failed"] == 1 and b["calls_ok"] == 2 and b["session_usd"] == 0.185, b
        t, _ = await c("status", sid=sid)
        assert "PAINT STOPPED" in t
        Path(str(LOG) + ".quota").unlink()
        (HOME / "STOP").unlink()
        await c("save_result", sid=sid, result={"status": "done", "final": str(d / "case/final/full_720.mp4")})

    print("== 3 swap, API image mode + server LLM prompts, worker max 124: hops")
    stop_worker("A")
    start_worker("B", 124, delay=2)
    async with server(REMAKE_LLM_MODEL="fake-llm", REMAKE_MAX_USD="0.5", **api) as c:
        t, _ = await c("submit", ok=False, sid="swapc", seeds=[2])  # its 243-frame window is too long now
        assert "at most 124" in t, t
        sid, d = "hop", HOME / "sessions" / "hop"
        await c("new_session", sid=sid, kind="swap")
        t, _ = await c("ingest", sid=sid, src=str(CLIP), dur=9.42)  # 226 = 17k+5 frames: no padding
        assert "padding" not in t and json.loads((d / "case/meta.json").read_text())["frames"] == 226, t
        await c("sheet", sid=sid, name="Vela", source="lib")
        nf, cuts = int((d / "case/nf").read_text()), json.loads((d / "case/cuts.json").read_text())
        await c("save_case", sid=sid, plan_grid=plan(nf, cuts), where=where([9, 83, 186]))
        t, _ = await c("windows", sid=sid, maxlen=200)
        assert "lowered to 124" in t and "hop of w0" in t, t
        await c("paint", sid=sid, frames=[9, 83])
        await c("paint", sid=sid, frames=[186])
        case = json.loads((d / "case/case.json").read_text())
        case["chars"]["Kit"], case["outfits"]["Kit"] = "Kit, a red fox", "blue overalls, yellow boots"
        await c("save_case", sid=sid, case=case)
        await c("sheet", sid=sid, name="Kit", source="text", text="an anthropomorphic red fox")
        rq = requests()
        assert rq[-1]["path"] == "/v1/images/generations" and rq[-1]["fields"]["size"] == "1536x1024", rq[-1]
        assert rq[-3]["images"] == [[1536, 1024], [768, 1024], [832, 960]], rq[-3]  # sheet, library photo, the pair grid
        t, _ = await c("paint", ok=False, sid=sid, frames=[186], force=True)  # 3 x 0.0925 + 0.3 > 0.5
        assert "budget reached" in t
        await c("windows", sid=sid)
        rf = json.loads((d / "case/refs.json").read_text())["windows"]
        assert [(w["render"], w["own"], w["refs"], w.get("prefix_of"), w.get("prefix_at")) for w in rf] == \
            [([0, 124], [0, 119], [9, 83], None, None), ([102, 226], [119, 226], [186], 0, 102)], rf
        n0 = len(requests())
        t, n = await c("write_prompts", sid=sid)
        assert n == 0 and t.count("===== case/win") == 2, t
        chat = requests()[n0:]
        assert sorted(r["images"] for r in chat) == [2, 3] and all(r["model"] == "fake-llm" for r in chat), chat
        for i in (0, 1):
            assert (d / f"case/win{i}/prompt.txt").read_text().startswith("subject_definitions:")
        t, _ = await c("submit", sid=sid)
        sub = json.loads(t)
        assert sub["queued"] == 2 and sub["hops_stuck"] == [], sub
        hop = json.loads(next((HOME / "queue").glob(f"50_s00_*_{sid}_720_w01_*.json")).read_text())
        assert hop["prefix_at"] == 102 and hop["num_frames"] == 124 and hop["pin"] == [186 - 102], hop
        t, _ = await c("status", sid=sid)
        assert "hops waiting for their prefix" in t, t
        t, _ = await c("wait", sid=sid, max_s=120)
        assert json.loads(t)["done"] == 2, t
        got = [json.loads(x) for x in (HOME / "fake_B.jsonl").read_text().splitlines()]
        assert [g["name"].split("_")[-2] for g in got] == ["w00", "w01"] and got[1]["job"]["prefix"] == got[0]["job"]["out"]
        await c("finish", ok=False, sid=sid, picks={"0": 0})  # the hop window has no pick
        t, _ = await c("finish", sid=sid, picks={"0": 0, "1": 0})
        assert probe(d / "case/final/full_720.mp4") == (226, 1280, 720, True)
        # a hop submitted without the window it continues is stuck: wait does not block on it
        t, _ = await c("submit", sid=sid, seeds=[1], windows=[1])
        assert len(json.loads(t)["hops_stuck"]) == 1, t
        t0 = time.time()
        t, _ = await c("wait", sid=sid, max_s=120)
        assert time.time() - t0 < 5 and "submit the window before it" in t, t
        await c("submit", sid=sid, seeds=[1], windows=[0])
        t, _ = await c("wait", sid=sid, max_s=120)
        assert json.loads(t)["done"] == 4 and json.loads(t)["hops_stuck"] == [], t
        await c("finish", ok=False, sid=sid, picks={"0": 0, "1": 1})  # a hop must take its prefix window's seed
        await c("finish", sid=sid, picks={"0": 1, "1": 1})
        t, _ = await c("budget", sid=sid)
        assert abs(json.loads(t)["session_usd"] - 0.2835) < 0.001, t  # 3 image calls at 0.0925 + 2 prompts at 0.003

    print("== 4 no swap worker live")
    stop_worker("B")
    start_worker("C", 345, modes="edit")
    async with server(**api) as c:
        t, _ = await c("windows", sid="hop")  # the only live worker renders edit sessions
        assert "no live worker; windows sized for 345" in t, t
        t, _ = await c("submit", ok=False, sid="hop")
        assert "no live swap worker renders finals (tier 736)" in t, t
        stop_worker("C")
        t0 = time.time()
        t, _ = await c("wait", sid="hop")
        assert time.time() - t0 < 5 and "no live worker" in t, t

    print("== 5 other aspects")
    start_worker("D", 345)
    async with server(**api) as c:
        # crop, draft, final, finished size: 4:3; 9:16 (portrait); 3:1, wider than the model takes, squeezed to 960x416
        for sid, crop, draft, final, out in (("a43", "1440:1080:240:0", [704, 544], [1120, 832], (1120, 840)),
                                             ("a916", "608:1080:656:0", [480, 832], [736, 1280], (720, 1280)),
                                             ("a31", "1920:640:0:220", [960, 416], [1472, 640], (1472, 490))):
            d = HOME / "sessions" / sid
            await c("new_session", sid=sid, kind="swap")
            t, _ = await c("ingest", sid=sid, src=str(CLIP), dur=4, crop=crop)
            meta, nf = json.loads((d / "case/meta.json").read_text()), int((d / "case/nf").read_text())
            assert (meta["draft"], meta["final"], meta["frames"], nf) == (draft, final, 96, 107), (meta, nf)
            assert probe(d / "case/cond.mp4")[:3] == (107, *draft) and probe(d / "case/cond720.mp4")[:3] == (107, *final)
            assert ("wider than the 2.31:1" in t) == (sid == "a31"), t
            await c("sheet", sid=sid, name="Vela", source="lib")
            cuts = json.loads((d / "case/cuts.json").read_text())
            await c("save_case", sid=sid, plan_grid=plan(nf, cuts), where=where([9, 60]))
            t, n = await c("paint", sid=sid, frames=[9, 60])
            assert n == 1 and all(Image.open(d / f"case/paint/g2/f{f}.png").size == tuple(draft) for f in (9, 60)), t
            await c("windows", sid=sid)
            await c("write_prompts", sid=sid)
            await c("set_prompt", sid=sid, window=0, text=prompt_text(2))
            await c("submit", sid=sid, canvas="480")
            await c("submit", sid=sid)
            t, _ = await c("wait", sid=sid, max_s=120)
            assert json.loads(t)["done"] == 2, t
            await c("finish", sid=sid, picks={"0": 0})
            assert probe(d / "case/final/full_720.mp4") == (96, *out, True), probe(d / "case/final/full_720.mp4")
        got = [json.loads(x)["job"] for x in (HOME / "fake_D.jsonl").read_text().splitlines()]
        assert sorted(g["tier"] for g in got) == [480] * 3 + [736] * 3, got
    print("ALL OK")


try:
    asyncio.run(main())
finally:
    for p in procs.values():
        p.kill()