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