Blame for tests/fallback_e2e.py

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())