3dai

Running on Zero

3dai / tests /fallback_e2e.py
Bobby
Claude Opus 5.5
Strip drop-shadow ground slab from the sparse structure for shadowed RGBA inputs
ce565eb 27 days ago
35.8 kB
import base64
import json
import os
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "tests"))

os.environ.setdefault("LABS_FALLBACK_SIGNING_KEY", "harness-signing-key")
os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False")

import fallback_stubs  # noqa: E402

fallback_stubs.install()

import gradio as gr  # noqa: E402
from gradio_client import Client, handle_file  # noqa: E402

import gradio6_support  # noqa: E402
import quota_fallback as qf  # noqa: E402
from fallback_contract_checks import wiring_contract, capacity_guard  # noqa: E402

MODE = "controlled_gradio_runtime"
QUOTA_MESSAGE = "You have exceeded your ZeroGPU runs limit. Try again in 1 hour."
PORT = int(os.environ.get("FALLBACK_HARNESS_PORT", "7861"))
BASE_URL = f"http://127.0.0.1:{PORT}"
EMBED_PARENT = qf.EMBED_PARENTS[0]
EVENTS = []
VIOLATIONS = []
app = None
FALLBACK_TEXTURE_SIZE = 1536
STEPS = {"generate": 0, "generate_multi": 0, "preview": 0, "glb_prepare": 0, "glb_package": 0, "stl": 0}
PREPARED_WITH = []


def record(event):
    EVENTS.append(event)


def violate(message):
    VIOLATIONS.append(message)


def reset(denied=False, clear_events=True):
    fallback_stubs.GPU_GATE["error"] = gr.Error(QUOTA_MESSAGE) if denied else None
    fallback_stubs.GPU_ATTEMPTS.clear()
    fallback_stubs.GPU_CALLS.clear()
    if clear_events:
        EVENTS.clear()
        VIOLATIONS.clear()
        PREPARED_WITH.clear()
        for key in STEPS:
            STEPS[key] = 0


def allocations():
    return list(fallback_stubs.GPU_CALLS)


def attempts():
    return list(fallback_stubs.GPU_ATTEMPTS)


def fake_preprocess_image(image):
    return image


def fake_run_original_pipeline(image, seed, ss_gs, ss_steps, slat_gs, slat_steps, strip_ground_slab=False):
    record("generate")
    STEPS["generate"] += 1
    return {"mesh": [seed], "gaussian": [f"gaussian-{seed}"]}


def fake_preprocess_images(images):
    return list(images)


def fake_run_original_pipeline_multi(images, seed, ss_gs, ss_steps, slat_gs, slat_steps, strip_ground_slab=False):
    record("generate_multi")
    STEPS["generate_multi"] += 1
    return {"mesh": [f"multi-{len(images)}-{seed}"], "gaussian": [f"gaussian-multi-{len(images)}"]}


def fake_pack_original_state(outputs):
    return {"runtime": "original", "marker": f"state-{outputs['mesh'][0]}"}


def fake_render_original_preview_frames(outputs, req):
    record("preview")
    STEPS["preview"] += 1
    return [{"frame": index} for index in range(3)]


def fake_prepare_glb(state, texture_size):
    record("glb_prepare")
    STEPS["glb_prepare"] += 1
    PREPARED_WITH.append(texture_size)
    if not fallback_stubs.in_gpu_allocation():
        violate("glb_prepare_outside_allocation")
    return {"glb": f"{state.get('marker')}-{texture_size}"}


def fake_save_export_payload(glb, progress):
    record("glb_package")
    STEPS["glb_package"] += 1
    if fallback_stubs.in_gpu_allocation():
        violate("glb_package_inside_allocation")
    return {"glb_name": "model.glb", "glb_b64": base64.b64encode(json.dumps(glb).encode()).decode()}


def fake_write_mp4(video_path, frames, fps=15):
    Path(video_path).write_bytes(b"harness-mp4")
    return True


def fake_export_stl_from_state(state, export_dir):
    record("stl")
    STEPS["stl"] += 1
    if not isinstance(state, dict) or not str(state.get("marker", "")).startswith("state-"):
        violate("stl_received_no_raw_state")
    stl_path = Path(export_dir) / "harness.stl"
    stl_path.write_bytes(b"harness-stl")
    return str(stl_path), True


FAKES = {
    "preprocess_image": fake_preprocess_image,
    "preprocess_images": fake_preprocess_images,
    "_run_original_pipeline": fake_run_original_pipeline,
    "_run_original_pipeline_multi": fake_run_original_pipeline_multi,
    "pack_original_state": fake_pack_original_state,
    "render_original_preview_frames": fake_render_original_preview_frames,
    "_prepare_glb": fake_prepare_glb,
    "save_export_payload": fake_save_export_payload,
    "write_mp4": fake_write_mp4,
    "export_stl_from_state": fake_export_stl_from_state,
}
ORIGINALS = {}


def build_app():
    global app
    import app as application

    app = application
    for name, fake in FAKES.items():
        ORIGINALS.setdefault(name, getattr(app, name, None))
        setattr(app, name, fake)
    return app


def restore_app():
    if app is None:
        return
    for name, original in ORIGINALS.items():
        setattr(app, name, original)
    ORIGINALS.clear()


def launch(app, port=PORT):
    app.demo.queue(max_size=20, default_concurrency_limit=20, api_open=False)
    app.demo.launch(
        server_name="127.0.0.1",
        server_port=port,
        share=False,
        prevent_thread_lock=True,
        quiet=True,
        **gradio6_support.launch_kwargs(app),
    )
    return embed_client(port)


def embed_client(port=PORT, headers=None):
    referer = headers if headers is not None else {"Referer": f"http://127.0.0.1:{port}/?__labs_parent={EMBED_PARENT}"}
    return Client(f"http://127.0.0.1:{port}", verbose=False, headers=referer)


def check(name, condition, detail):
    return {"check": name, "ok": bool(condition), "detail": str(detail)[:200]}


def produced(output):
    if not isinstance(output, dict):
        return output
    return output.get("video") or output.get("value")


def restored_status(result):
    if not isinstance(result, (list, tuple)) or not result:
        return None
    last = result[-1]
    return last if isinstance(last, dict) and ("value" in last or "visible" in last) else None


def issue_permit(ticket):
    receipt = qf.verify_receipt(ticket)
    assert receipt is not None, "harness could not open its own receipt"
    return qf.sign_document({
        "v": qf.DOCUMENT_VERSION,
        "purpose": qf.EXECUTE_PURPOSE,
        "space": receipt["space"],
        "nonce": receipt["nonce"],
        "exp": int(time.time()) + 180,
    })


def permit_for(ticket, **overrides):
    receipt = qf.verify_receipt(ticket)
    payload = {
        "v": qf.DOCUMENT_VERSION,
        "purpose": qf.EXECUTE_PURPOSE,
        "space": receipt["space"],
        "nonce": receipt["nonce"],
        "exp": int(time.time()) + 180,
    }
    payload.update(overrides)
    return qf.sign_document(payload)


def sample_image(path):
    from PIL import Image

    Path(path).parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", (8, 8), (12, 34, 56)).save(path, format="PNG")
    return path


def cached_outputs(app, ticket):
    entry = app.FALLBACK._cache.find(qf.verify_receipt(ticket)["nonce"])
    return None if entry is None else (entry.status, entry.outputs)


def generate(client, image_path, seed=0, texture_size=None):
    size = app.GLB_EXPORT_TEXTURE_SIZE if texture_size is None else texture_size
    return client.predict(handle_file(image_path), seed, 6.0, 12, 1.0, 12, size,
                          api_name="/generate_and_extract")


def materialized_glb(client):
    return produced(client.predict(api_name="/materialize_export_payload"))




def one_allocation_generation(app, client, image_path):
    reset()
    video = generate(client, image_path)
    results = [
        check("one_gpu_allocation_per_generation", allocations() == ["gpu_task:preview"], allocations()),
        check("generation_attempted_once", attempts() == ["gpu_task:preview"], attempts()),
        check("cached_video_returned", str(produced(video)).endswith(".mp4"), produced(video)),
        check("generation_runs_generate_then_preview",
              EVENTS == ["generate", "preview"], EVENTS),
        check("generation_bakes_no_texture",
              STEPS["glb_prepare"] == 0 and STEPS["glb_package"] == 0 and not PREPARED_WITH,
              (STEPS, PREPARED_WITH)),
    ]
    status = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY,
                            texture_size=app.GLB_EXPORT_TEXTURE_SIZE, api_name="/extract_model")
    results.append(check("first_export_bakes_and_reveals_the_glb",
                         "finalizing" in str(produced(status)).lower()
                         and PREPARED_WITH == [app.GLB_EXPORT_TEXTURE_SIZE],
                         (produced(status), PREPARED_WITH)))
    results.append(check("first_export_allocates_the_texture_bake",
                         allocations() == ["gpu_task:preview", "gpu_task:bake"],
                         allocations()))
    results.append(check("glb_prepared_inside_the_bake_allocation",
                         STEPS["glb_prepare"] == 1 and "glb_prepare_outside_allocation" not in VIOLATIONS,
                         (STEPS, VIOLATIONS)))
    results.append(check("glb_packaged_outside_the_bake_allocation",
                         STEPS["glb_package"] == 1 and "glb_package_inside_allocation" not in VIOLATIONS,
                         (STEPS, VIOLATIONS)))
    glb_path = materialized_glb(client)
    results.append(check("cached_glb_materialized_for_viewer",
                         isinstance(glb_path, str) and glb_path.endswith(".glb"), glb_path))
    results.append(check("materialize_allocates_no_gpu",
                         allocations() == ["gpu_task:preview", "gpu_task:bake"],
                         allocations()))
    baked = len(allocations())
    repeat = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY,
                            texture_size=app.GLB_EXPORT_TEXTURE_SIZE, api_name="/extract_model")
    results.append(check("repeated_matching_export_stays_cached",
                         "finalizing" in str(produced(repeat)).lower()
                         and len(allocations()) == baked and len(PREPARED_WITH) == 1,
                         (produced(repeat), len(allocations()) - baked, PREPARED_WITH)))
    stl = client.predict(api_name="/generate_stl_download")
    results.append(check("stl_export_from_raw_mesh_state",
                         str(produced(stl)).endswith(".stl") and STEPS["stl"] == 1 and not VIOLATIONS,
                         (produced(stl), VIOLATIONS)))
    results.append(check("stl_export_allocates_no_gpu", len(allocations()) == baked, allocations()))
    rebaked = client.predict(mesh_simplify=0.98, texture_size=app.GLB_EXPORT_TEXTURE_SIZE,
                             api_name="/extract_model")
    results.append(check("different_settings_trigger_a_second_bake",
                         "finalizing" in str(produced(rebaked)).lower()
                         and len(allocations()) == baked + 1
                         and PREPARED_WITH[-1] == app.GLB_EXPORT_TEXTURE_SIZE,
                         (produced(rebaked), len(allocations()) - baked, PREPARED_WITH)))
    return results


def texture_size_selection(app, client, image_path):
    results = []
    sizes = list(app.GLB_TEXTURE_SIZES)
    for index, size in enumerate(sizes):
        reset()
        generate(client, image_path, texture_size=size)
        results.append(check(f"generation_requested_the_{size}px_texture",
                             not PREPARED_WITH and allocations() == ["gpu_task:preview"],
                             (PREPARED_WITH, allocations())))
        allocated = len(allocations())
        status = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY, texture_size=size,
                                api_name="/extract_model")
        glb_path = materialized_glb(client)
        results.append(check(f"{size}px_export_bakes_once_and_reveals",
                             "finalizing" in str(produced(status)).lower()
                             and str(glb_path).endswith(".glb")
                             and len(allocations()) == allocated + 1
                             and PREPARED_WITH == [size],
                             (produced(status), glb_path, len(allocations()) - allocated, PREPARED_WITH)))
        baked = len(allocations())
        again = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY, texture_size=size,
                               api_name="/extract_model")
        results.append(check(f"{size}px_export_is_cached_on_repeat",
                             "finalizing" in str(produced(again)).lower()
                             and len(allocations()) == baked,
                             (produced(again), len(allocations()) - baked)))
        later = sizes[(index + 1) % len(sizes)]
        changed = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY, texture_size=later,
                                 api_name="/extract_model")
        results.append(check(f"{size}px_export_rebakes_for_a_later_{later}px_choice",
                             "finalizing" in str(produced(changed)).lower()
                             and len(allocations()) == baked + 1
                             and PREPARED_WITH[-1] == later,
                             (produced(changed), len(allocations()) - baked, PREPARED_WITH[-1:])))
    reset()
    attempts_before = len(attempts())
    raised = None
    try:
        generate(client, image_path, texture_size=768)
    except Exception as error:
        raised = error
    results.append(check("unlisted_texture_size_refused_before_the_gpu_call",
                         raised is not None and len(attempts()) == attempts_before
                         and not allocations() and not PREPARED_WITH,
                         (raised, len(attempts()) - attempts_before, allocations(), PREPARED_WITH)))
    status = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY, texture_size=sizes[-1],
                            api_name="/extract_model")
    glb_path = materialized_glb(client)
    results.append(check("refused_generation_preserved_the_last_model",
                         "finalizing" in str(produced(status)).lower()
                         and str(glb_path).endswith(".glb")
                         and allocations() == ["gpu_task:bake"]
                         and PREPARED_WITH == [sizes[-1]],
                         (produced(status), glb_path, allocations(), PREPARED_WITH)))
    return results


def stateless_export(app, image_path):
    client = embed_client()
    reset()
    status = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY,
                            texture_size=app.GLB_EXPORT_TEXTURE_SIZE, api_name="/extract_model")
    return [
        check("stateless_export_returns_no_status", not produced(status), produced(status)),
        check("stateless_export_allocates_no_gpu", not attempts() and not allocations(),
              (attempts(), allocations())),
    ]


def inactive_generation_acks(client, ticket):
    results = []
    quiet_start = time.time()
    for label, ack in (
        ("missing", None),
        ("declined", {"no_fallback": True}),
        ("empty", {}),
        ("cancelled", {"ticket": ticket, "cancelled": True}),
        ("foreign_ticket", {"ticket": ticket[:-3] + "aaa", "ready": True}),
    ):
        quiet = client.predict(ack, {"ticket": ticket}, api_name="/restore_generation_outputs")
        results.append(check(f"inactive_ack_{label}_is_a_noop",
                             produced(quiet[0]) is None and not produced(restored_status(quiet)), quiet))
    results.append(check("inactive_acks_never_block_restore", time.time() - quiet_start < 10,
                         round(time.time() - quiet_start, 1)))
    results.append(check("cancelled_ack_left_no_gpu_work",
                         not allocations() and STEPS["glb_prepare"] == 0, (allocations(), STEPS)))
    errored = client.predict({"ticket": ticket, "error": "Pro allowance is busy right now."},
                             {"ticket": ticket}, api_name="/restore_generation_outputs")
    results.append(check("error_ack_surfaces_message",
                         "busy" in str(produced(restored_status(errored))).lower(), restored_status(errored)))
    provider_error = client.predict(
        {"ticket": ticket, "error": "The shared Labs allowance is unavailable. Please try again after the HF allowance resets."},
        {"ticket": ticket}, api_name="/restore_generation_outputs",
    )
    results.append(check("provider_error_is_displayed_without_provider_names",
                         produced(restored_status(provider_error)) == qf.ALLOWANCE_ERROR,
                         restored_status(provider_error)))
    return results


def pro_texture_check(app, state):
    requested = (state.get(app.GLB_REQUESTED_KEY) or {}).get("texture_size")
    return check(
        "pro_replay_prebaked_the_selected_size",
        PREPARED_WITH == [FALLBACK_TEXTURE_SIZE] and requested == FALLBACK_TEXTURE_SIZE,
        (PREPARED_WITH, state.get(app.GLB_REQUESTED_KEY)),
    )


def wait_for_pro_prebake(app, outputs):
    state = outputs[1] if isinstance(outputs, tuple) and len(outputs) > 1 else {}
    deadline = time.time() + 10
    while app.GLB_PAYLOAD_KEY not in (state or {}) and time.time() < deadline:
        time.sleep(0.02)
    return state or {}


def generation_fallback(app, client, image_path):
    reset(denied=True)
    client.predict(api_name="/reset_fallback_state")
    video = generate(client, image_path, texture_size=FALLBACK_TEXTURE_SIZE)
    receipt = client.predict(api_name="/issue_generation_receipt")
    ticket = (receipt or {}).get("ticket")
    entry = app.FALLBACK._cache.find(qf.verify_receipt(ticket)["nonce"]) if ticket else None
    results = [
        check("denied_generation_attempted_once", attempts() == ["gpu_task:preview"], attempts()),
        check("denied_generation_allocated_nothing", not allocations(), allocations()),
        check("denied_generation_preview_empty", video is None, video),
        check("denied_generation_prepared_no_glb", STEPS["glb_prepare"] == 0 and STEPS["glb_package"] == 0, STEPS),
        check("receipt_issued_to_ui", isinstance(ticket, str) and len(ticket) > 60, ticket),
        check("denied_generation_deferred_the_selected_size",
              entry is not None and entry.args[-1] == FALLBACK_TEXTURE_SIZE, entry and entry.args[-1:]),
    ]
    if not ticket:
        return results
    results.extend(inactive_generation_acks(client, ticket))
    fallback_stubs.GPU_GATE["error"] = None
    permit = issue_permit(ticket)
    allowed = client.predict(ticket, permit, api_name="/run_pro_fallback")
    results.append(check("permit_executes_pro_generate", allowed == {"ok": True, "operation": "generate"}, allowed))
    status, outputs = cached_outputs(app, ticket)
    results.append(check("pro_outputs_cached_server_side", status == "complete" and isinstance(outputs, tuple),
                         (status, outputs)))
    cached_state = wait_for_pro_prebake(app, outputs)
    results.append(check("pro_replay_prebaked_under_one_extra_bake",
                         allocations() == ["gpu_task:preview", "gpu_task:bake"], allocations()))
    results.append(check("pro_replay_cached_the_prebaked_glb_payload",
                         app.GLB_PAYLOAD_KEY in cached_state, sorted(cached_state)))
    results.append(pro_texture_check(app, cached_state))
    replayed = client.predict(ticket, permit, api_name="/run_pro_fallback")
    results.append(check("replayed_permit_idempotent", replayed.get("ok") is True, replayed))
    results.append(check("replay_did_not_rerun_gpu", allocations() == ["gpu_task:preview", "gpu_task:bake"], allocations()))
    restored = client.predict({"ticket": ticket, "ready": True}, {"ticket": ticket},
                              api_name="/restore_generation_outputs")
    video_out, status_out = restored
    results.append(check("preview_restored_to_ui", str(produced(video_out)).endswith(".mp4"), produced(video_out)))
    results.append(check("status_clear_on_success", not produced(status_out), produced(status_out)))
    before = len(allocations())
    status_out = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY,
                                texture_size=FALLBACK_TEXTURE_SIZE, api_name="/extract_model")
    glb_path = materialized_glb(client)
    results.append(check("glb_revealed_after_pro_replay_from_the_prebake",
                         "finalizing" in str(produced(status_out)).lower()
                         and str(glb_path).endswith(".glb") and len(allocations()) == before,
                         (produced(status_out), glb_path, allocations()[before:])))
    cached_before = len(allocations())
    repeat = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY,
                            texture_size=FALLBACK_TEXTURE_SIZE, api_name="/extract_model")
    results.append(check("replayed_export_stays_cached_without_gpu",
                         "finalizing" in str(produced(repeat)).lower()
                         and len(allocations()) == cached_before,
                         (produced(repeat), len(allocations()) - cached_before)))
    return results


def export_fallback_legacy(app, client, image_path):
    reset()
    client.predict(api_name="/reset_fallback_state")
    generate(client, image_path)
    reset(clear_events=False)
    mesh_simplify, texture_size = app.GLB_EXPORT_SIMPLIFY, app.GLB_EXPORT_TEXTURE_SIZE
    receipt, waiting = client.predict(mesh_simplify=mesh_simplify, texture_size=texture_size,
                                      status=QUOTA_MESSAGE, api_name="/issue_export_receipt")
    ticket = (receipt or {}).get("ticket")
    results = [
        check("legacy_export_receipt_issued", isinstance(ticket, str) and len(ticket) > 60, ticket),
        check("legacy_export_waiting_status_shown", qf.WAITING_STATUS in str(waiting), waiting),
    ]
    if not ticket:
        return results
    permitted = client.predict(ticket, issue_permit(ticket), api_name="/run_pro_fallback")
    results.append(check("legacy_permit_executes_pro_export", permitted == {"ok": True, "operation": "export"},
                         permitted))
    entry_status, payload_out = cached_outputs(app, ticket)
    results.append(check("legacy_export_payload_cached",
                         entry_status == "complete" and "glb_b64" in (payload_out or {}), entry_status))
    results.append(check("legacy_export_baked_under_the_permit",
                         allocations() == ["gpu_task:bake"]
                         and attempts() == ["gpu_task:bake"],
                         (allocations(), attempts())))
    restored = client.predict({"ticket": ticket, "ready": True}, {"ticket": ticket},
                              api_name="/restore_export_outputs")
    results.append(check("legacy_export_status_cleared", not produced(restored[0]), produced(restored[0])))
    glb_path = materialized_glb(client)
    results.append(check("legacy_export_materialized_for_viewer",
                         isinstance(glb_path, str) and glb_path.endswith(".glb"), glb_path))
    return results


def security_and_state(app, client, second_client, image_path):
    results = []
    reset(denied=True)
    client.predict(api_name="/reset_fallback_state")
    generate(client, image_path)
    ticket = (client.predict(api_name="/issue_generation_receipt") or {}).get("ticket")
    if not ticket:
        return [check("security_receipt_issued", False, ticket)]
    nonce = qf.verify_receipt(ticket)["nonce"]
    before = len(attempts())
    tampered = ticket[:-3] + ("aaa" if not ticket.endswith("aaa") else "bbb")
    forged = client.predict(tampered, issue_permit(ticket), api_name="/run_pro_fallback")
    results.append(check("forged_ticket_refused", forged.get("ok") is False, forged))
    results.append(check("forged_ticket_no_gpu_call", len(attempts()) == before, len(attempts()) - before))
    cross_space = client.predict(ticket, permit_for(ticket, space="broyang/other"), api_name="/run_pro_fallback")
    results.append(check("cross_space_permit_refused", cross_space.get("ok") is False, cross_space))
    expired = client.predict(ticket, permit_for(ticket, exp=int(time.time()) - 5), api_name="/run_pro_fallback")
    results.append(check("expired_permit_refused", expired.get("ok") is False, expired))
    no_ticket = client.predict("not-a-ticket", permit_for(ticket), api_name="/run_pro_fallback")
    results.append(check("unsigned_ticket_refused", no_ticket.get("ok") is False, no_ticket))
    cross = second_client.predict({"ticket": ticket, "ready": True}, {"ticket": ticket},
                                  api_name="/restore_generation_outputs")
    results.append(check("cross_session_restore_blocked", produced(cross[0]) is None, cross))
    results.append(check("cross_session_no_gpu_rerun", len(attempts()) == before, len(attempts()) - before))
    fallback_stubs.GPU_GATE["error"] = None
    allowed = client.predict(ticket, issue_permit(ticket), api_name="/run_pro_fallback")
    results.append(check("owner_session_still_executes", allowed == {"ok": True, "operation": "generate"}, allowed))
    client.predict(api_name="/reset_fallback_state")
    results.append(check("new_upload_purges_pending", not app.FALLBACK.contains(nonce), nonce))
    late = client.predict(ticket, issue_permit(ticket), api_name="/run_pro_fallback")
    results.append(check("late_result_after_new_upload_refused", late.get("ok") is False, late))
    clean = client.predict({"ticket": ticket, "ready": True}, {"ticket": ticket},
                           api_name="/restore_generation_outputs")
    results.append(check("late_restore_is_noop", produced(clean[0]) is None, clean))
    stale = client.predict({"ticket": ticket, "error": "Pro allowance is busy right now."},
                           {"ticket": ticket}, api_name="/restore_generation_outputs")
    results.append(check("stale_error_after_invalidation_skips_every_output",
                         produced(stale[0]) is None and not produced(restored_status(stale)), stale))
    return results


def ineligible_callers(app, image_path):
    results = []
    spoofed = embed_client(headers={"Referer": f"https://evil.example/?__labs_parent={qf.EMBED_PARENTS[0]}"})
    for label, headers in (("no_referer", {}),
                           ("disallowed_parent", {"Referer": f"{BASE_URL}/?__labs_parent=https://evil.example"})):
        results.extend(ineligible_generation(app, embed_client(headers=headers), image_path, label))
    results.extend(ineligible_generation(app, spoofed, image_path, "foreign_referer_host"))
    results.extend(ineligible_export(app, embed_client(headers={}), image_path))
    return results


def ineligible_generation(app, client, image_path, label):
    reset(denied=True)
    client.predict(api_name="/reset_fallback_state")
    before = app.FALLBACK._cache.size()
    raised = None
    try:
        generate(client, image_path)
    except Exception as error:
        raised = error
    receipt = client.predict(api_name="/issue_generation_receipt")
    return [
        check(f"{label}_quota_error_surfaces", raised is not None and qf.ALLOWANCE_ERROR in str(raised), raised),
        check(f"{label}_issues_no_receipt", receipt is None, receipt),
        check(f"{label}_defers_no_work", app.FALLBACK._cache.size() == before, app.FALLBACK._cache.size()),
        check(f"{label}_allocated_nothing", not allocations(), allocations()),
    ]


def ineligible_export(app, client, image_path):
    reset()
    generate(client, image_path)
    reset(clear_events=False)
    receipt, waiting = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY,
                                      texture_size=app.GLB_EXPORT_TEXTURE_SIZE,
                                      status=QUOTA_MESSAGE, api_name="/issue_export_receipt")
    status = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY,
                            texture_size=app.GLB_EXPORT_TEXTURE_SIZE, api_name="/extract_model")
    glb_path = materialized_glb(client)
    return [
        check("ineligible_export_issues_no_receipt", receipt is None, receipt),
        check("ineligible_export_does_not_show_waiting", "requesting" not in str(waiting).lower(), waiting),
        check("ineligible_export_still_reveals_cached_glb",
              "finalizing" in str(produced(status)).lower() and str(glb_path).endswith(".glb"),
              (produced(status), glb_path)),
        check("ineligible_export_bakes_without_a_permit",
              allocations() == ["gpu_task:bake"]
              and attempts() == ["gpu_task:bake"],
              (allocations(), attempts())),
    ]




def generation_api_for(app, first_input_type):
    config = app.demo.get_config_file()
    components = {component["id"]: component for component in config["components"]}
    for dep in config["dependencies"]:
        if str(dep.get("api_name", "")).startswith("generate_and_extract") and dep.get("inputs"):
            if components[dep["inputs"][0]]["type"] == first_input_type:
                return f"/{dep['api_name']}"
    raise AssertionError(f"no generation endpoint starts with a {first_input_type} input")


def multi_image_generation(app, client, image_paths):
    reset()
    multi_api = generation_api_for(app, "gallery")
    size = app.GLB_EXPORT_TEXTURE_SIZE
    video = client.predict([{"image": handle_file(path)} for path in image_paths], 0, 6.0, 12, 1.0, 12, size,
                           api_name=multi_api)
    results = [
        check("multi_generation_uses_one_gpu_allocation",
              allocations() == ["gpu_task:preview"], allocations()),
        check("multi_generation_runs_generate_then_preview",
              EVENTS == ["generate_multi", "preview"], EVENTS),
        check("multi_generation_uses_the_multi_runner",
              STEPS["generate_multi"] == 1 and STEPS["generate"] == 0, STEPS),
        check("multi_generation_returns_a_cached_video", str(produced(video)).endswith(".mp4"), produced(video)),
        check("multi_generation_bakes_no_texture",
              STEPS["glb_prepare"] == 0 and STEPS["glb_package"] == 0 and not PREPARED_WITH,
              (STEPS, PREPARED_WITH)),
    ]
    allocated = len(allocations())
    status = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY, texture_size=size, api_name="/extract_model")
    glb_path = materialized_glb(client)
    results.append(check("multi_generation_glb_revealed_under_one_bake",
                         "finalizing" in str(produced(status)).lower()
                         and str(glb_path).endswith(".glb") and len(allocations()) == allocated + 1
                         and PREPARED_WITH == [size],
                         (produced(status), glb_path, allocations()[allocated:], PREPARED_WITH)))
    single_view = refused_generation(client, image_paths[:1], multi_api, size)
    results.append(check("multi_generation_refuses_one_view_before_the_gpu",
                         single_view is not None and "at least" in str(single_view).lower() and not attempts(),
                         (single_view, attempts())))
    too_many = refused_generation(client, image_paths * 4, multi_api, size)
    results.append(check("multi_generation_refuses_too_many_views_before_the_gpu",
                         too_many is not None and "at most" in str(too_many).lower() and not attempts(),
                         (too_many, attempts())))
    return results


def multi_image_fallback(app, client, image_paths):
    reset(denied=True)
    client.predict(api_name="/reset_fallback_state")
    multi_api = generation_api_for(app, "gallery")
    size = app.GLB_EXPORT_TEXTURE_SIZE
    client.predict([{"image": handle_file(path)} for path in image_paths], 0, 6.0, 12, 1.0, 12, size,
                   api_name=multi_api)
    ticket = (client.predict(api_name="/issue_generation_receipt") or {}).get("ticket")
    entry = app.FALLBACK._cache.find(qf.verify_receipt(ticket)["nonce"]) if ticket else None
    results = [
        check("multi_denied_generation_allocated_nothing", not allocations(), allocations()),
        check("multi_denied_receipt_issued", isinstance(ticket, str) and len(ticket) > 60, ticket),
        check("multi_denied_deferred_the_view_list",
              entry is not None and isinstance(entry.args[0], list) and len(entry.args[0]) == 2
              and entry.args[-1] == size,
              entry and (type(entry.args[0]).__name__, entry.args[-1:])),
    ]
    if not ticket:
        return results
    fallback_stubs.GPU_GATE["error"] = None
    allowed = client.predict(ticket, issue_permit(ticket), api_name="/run_pro_fallback")
    results.append(check("multi_permit_executes_pro_generate",
                         allowed == {"ok": True, "operation": "generate"}, allowed))
    status, outputs = cached_outputs(app, ticket)
    cached_state = wait_for_pro_prebake(app, outputs)
    results.append(check("multi_pro_replay_uses_the_multi_runner",
                         STEPS["generate_multi"] == 1 and STEPS["generate"] == 0
                         and allocations() == ["gpu_task:preview", "gpu_task:bake"], (STEPS, allocations())))
    results.append(check("multi_pro_replay_cached_the_prebaked_glb_payload",
                         status == "complete" and PREPARED_WITH == [size]
                         and app.GLB_PAYLOAD_KEY in cached_state,
                         (status, sorted(cached_state), PREPARED_WITH)))
    client.predict({"ticket": ticket, "ready": True}, {"ticket": ticket},
                   api_name="/restore_generation_outputs")
    before = len(allocations())
    export_status = client.predict(mesh_simplify=app.GLB_EXPORT_SIMPLIFY, texture_size=size,
                                   api_name="/extract_model")
    glb_path = materialized_glb(client)
    results.append(check("multi_pro_replay_exports_from_the_prebake",
                         "finalizing" in str(produced(export_status)).lower()
                         and str(glb_path).endswith(".glb")
                         and allocations()[before:] == [],
                         (produced(export_status), glb_path, allocations()[before:])))
    return results


def refused_generation(client, image_paths, multi_api, size):
    reset()
    try:
        client.predict([{"image": handle_file(path)} for path in image_paths], 0, 6.0, 12, 1.0, 12, size, api_name=multi_api)
    except Exception as error:
        return error
    return None


def main():
    app = build_app()
    client = launch(app)
    second_client = embed_client()
    image_path = sample_image(str(ROOT / "cache" / "harness-input.png"))
    multi_image_paths = [
        sample_image(str(ROOT / "cache" / f"harness-multi-{index}.png")) for index in range(2)
    ]
    report = {
        "mode": MODE,
        "wiring_contract": wiring_contract(app, check, fallback_stubs),
        "one_allocation_generation": one_allocation_generation(app, client, image_path),
        "multi_image_generation": multi_image_generation(app, client, multi_image_paths),
        "multi_image_fallback": multi_image_fallback(app, client, multi_image_paths),
        "texture_size_selection": texture_size_selection(app, client, image_path),
        "stateless_export": stateless_export(app, image_path),
        "generation_fallback": generation_fallback(app, client, image_path),
        "export_fallback_legacy": export_fallback_legacy(app, client, image_path),
        "security_and_state": security_and_state(app, client, second_client, image_path),
        "ineligible_callers": ineligible_callers(app, image_path),
        "capacity_guard": capacity_guard(app, image_path, check, qf),
    }
    flat = [item for group in report.values() if isinstance(group, list) for item in group]
    report["summary"] = {
        "total": len(flat),
        "passed": sum(1 for item in flat if item["ok"]),
        "failed": [item for item in flat if not item["ok"]],
    }
    Path(os.environ.get("FALLBACK_EVIDENCE", "/tmp/3dai-fallback-e2e.json")).write_text(json.dumps(report, indent=2))
    print(json.dumps(report["summary"], indent=2))
    return 1 if report["summary"]["failed"] else 0


if __name__ == "__main__":
    sys.exit(main())