gateway_app.py

"""Public CPU login gateway for the private AutoML GPU Space."""

from __future__ import annotations

import os
import shutil
import tempfile
import time
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any, Mapping

import gradio as gr
import pandas as pd

import auth_storage
from gateway_auth import (
    AuthResult,
    change_password,
    handle_denied_request,
    login,
    request_signup,
    requester_key,
)
from gateway_client import BackendResult, BackendUnavailable, GatewayBackendClient
from gateway_contract import (
    PROTOCOL_VERSION,
    ContractError,
    check_upload_size,
    inspect_package_manifest,
    load_gateway_catalog,
    predictions_frame,
)
from gateway_jobs import ROUTED_JOB_ADMISSION, JobRejected
from gateway_routing import (
    VALID_ROUTING_MODES,
    RouteDecision,
    RoutingError,
    declare_route,
    route_existing,
    route_prediction,
    route_uploaded,
)


GATEWAY_CATALOG_PATH = Path(__file__).with_name("gateway_catalog.json")
AUTO_CHOICE = "🤖 Auto"
AUTO_DEVICE = "auto - GPU if available"
IMAGE_DEVICE_CHOICES = [AUTO_DEVICE, "cpu", "cuda - requires GPU hardware"]
IMAGE_TASK_CHOICES = [
    "classification",
    "object_detection",
    "instance_segmentation",
    "semantic_segmentation",
]
IMAGE_BACKEND_CHOICES = [
    "Auto",
    "Existing classification pipeline",
    "Medical image agent",
    "AutoGluon AutoMM",
    "YOLO",
    "Benchmark backends",
]
IMAGE_FORMAT_CHOICES = ["auto", "folder", "csv", "coco", "yolo", "semantic-mask"]
IMAGE_BACKBONE_CHOICES = [
    AUTO_CHOICE,
    "raw_pixels",
    "mobilenet_v3_small",
    "resnet18",
    "resnet50",
    "densenet121",
    "efficientnet_b0",
]
IMAGE_CLASSIFIER_CHOICES = [
    AUTO_CHOICE,
    "logistic_regression",
    "random_forest",
    "autogluon",
]
IMAGE_SIZE_CHOICES = [AUTO_CHOICE, "32", "48", "64", "96", "128", "224"]
CANCER_PRESET_CHOICES = [
    "Auto - agent decides",
    "lc25000_densenet121",
    "lc25000_resnet50",
    "breakhis_efficientnet_b0",
    "ham10000_efficientnet_b0",
    "ham10000_mobilenet_v3",
    "pcam_resnet18",
]
FEATURE_SELECTION_CHOICES = [
    "🤖 Agent Decides",
    "no_feature_selection_tool",
    "t_test_feature_selection_tool",
]
TRANSFORM_CHOICES = [
    "🤖 Agent Decides",
    "no_transform_tool",
    "pca_transform_tool",
    "subspace_transform_tool",
    "cluster_undersample_transform_tool",
]
ML_CHOICES = ["🤖 Agent Decides", "autogluon_tool"]
QUALITY_PRESETS = ["medium_quality", "high_quality", "best_quality", "interpretable"]


def _catalog() -> dict[str, Any]:
    if GATEWAY_CATALOG_PATH.is_file():
        return load_gateway_catalog(GATEWAY_CATALOG_PATH)
    return {
        "schema_version": 1,
        "datasets": {
            "No datasets configured": {"targets": [], "image": False}
        },
    }


def _dataset_metadata(dataset_name: str) -> dict[str, Any]:
    return dict(_catalog()["datasets"].get(dataset_name) or {})


def routing_mode_from_env() -> str:
    mode = os.getenv("AUTOML_ROUTING_MODE", "dual").strip().lower()
    if mode not in VALID_ROUTING_MODES:
        raise RuntimeError(
            "AUTOML_ROUTING_MODE must be dual or single-gpu."
        )
    return mode


def format_route_preview(decision: RouteDecision) -> str:
    label = "CPU" if decision.profile == "cpu" else "GPU"
    return f"**Compute route: Private {label}** — {decision.explanation}"


def _route_preview_error(exc: Exception) -> str:
    return f"**Compute route unavailable** — {exc}"


def existing_route_preview(
    dataset: Any,
    image_backbone: Any,
    image_device: Any,
    image_task: Any,
    image_backend: Any,
    image_benchmark_backends: Any,
    cancer_preset: Any,
) -> str:
    payload = {
        "image_backbone": image_backbone,
        "image_device": image_device,
        "image_task": image_task,
        "image_backend": image_backend,
        "image_benchmark_backends": image_benchmark_backends,
        "cancer_preset": cancer_preset,
    }
    try:
        decision = route_existing(
            payload,
            dataset_is_image=bool(_dataset_metadata(str(dataset)).get("image")),
            routing_mode=routing_mode_from_env(),
        )
        return format_route_preview(decision)
    except (RoutingError, RuntimeError) as exc:
        return _route_preview_error(exc)


def uploaded_route_preview(
    data_type: Any,
    image_backbone: Any,
    image_device: Any,
    uploaded_image_backend: Any,
    cancer_preset: Any,
) -> str:
    payload = {
        "data_type": data_type,
        "image_backbone": image_backbone,
        "image_device": image_device,
        "uploaded_image_backend": uploaded_image_backend,
        "cancer_preset": cancer_preset,
    }
    try:
        return format_route_preview(
            route_uploaded(payload, routing_mode=routing_mode_from_env())
        )
    except (RoutingError, RuntimeError) as exc:
        return _route_preview_error(exc)


def replay_route_preview(manifest: Mapping[str, Any] | None, image_device: Any) -> str:
    try:
        return format_route_preview(
            route_prediction(
                {"image_device": image_device},
                dict(manifest or {}),
                routing_mode=routing_mode_from_env(),
            )
        )
    except (RoutingError, RuntimeError) as exc:
        return _route_preview_error(exc)


def _normalize_token(value: Any) -> str:
    return str(value or "").strip().lower().replace("-", "_").replace(" ", "_")


def normalize_image_task(value: Any) -> str:
    token = _normalize_token(value)
    aliases = {
        "classification": "classification",
        "object_detection": "object_detection",
        "detection": "object_detection",
        "instance_segmentation": "instance_segmentation",
        "semantic_segmentation": "semantic_segmentation",
    }
    return aliases.get(token, "classification")


def normalize_image_backend(value: Any) -> str:
    aliases = {
        "auto": "auto",
        "existing_classification_pipeline": "existing",
        "existing": "existing",
        "medical_image_agent": "medical_agent",
        "medical_agent": "medical_agent",
        "autogluon_automm": "automm",
        "automm": "automm",
        "yolo": "yolo",
        "benchmark_backends": "benchmark",
        "benchmark": "benchmark",
    }
    return aliases.get(_normalize_token(value), "auto")


def get_image_task_ui_state(image_task: Any, image_backend: Any) -> dict[str, Any]:
    task = normalize_image_task(image_task)
    backend = normalize_image_backend(image_backend)
    effective = backend
    if backend == "auto":
        effective = "existing" if task == "classification" else (
            "automm" if task == "semantic_segmentation" else "yolo"
        )
    notes = {
        "classification": "Classification supports the existing classifier, Medical Image Agent, AutoMM, and benchmark mode.",
        "object_detection": "Object detection uses the managed COCO dataset for an existing image dataset.",
        "instance_segmentation": "Instance segmentation uses managed COCO polygon/RLE annotations.",
        "semantic_segmentation": "Semantic segmentation requires a managed image/mask manifest.",
    }
    return {
        "classification": task == "classification" and effective == "existing",
        "medical": task == "classification" and effective == "medical_agent",
        "detection": task == "object_detection",
        "instance": task == "instance_segmentation",
        "semantic": task == "semantic_segmentation",
        "benchmark": backend == "benchmark",
        "optional": effective in {"automm", "yolo", "benchmark"},
        "note": notes[task],
    }


def image_task_ui_updates(image_task: Any, image_backend: Any):
    state = get_image_task_ui_state(image_task, image_backend)
    return (
        gr.update(visible=state["classification"]),
        gr.update(visible=state["medical"]),
        gr.update(visible=state["detection"]),
        gr.update(visible=state["instance"]),
        gr.update(visible=state["semantic"]),
        gr.update(visible=state["benchmark"]),
        gr.update(visible=state["optional"]),
        gr.update(value=state["note"]),
    )


def uploaded_image_backend_updates(image_backend: Any):
    backend = normalize_image_backend(image_backend)
    return gr.update(visible=backend == "existing"), gr.update(visible=backend == "medical_agent")


def update_existing_dataset_controls(dataset_name: str):
    metadata = _dataset_metadata(dataset_name)
    targets = list(metadata.get("targets") or [])
    image = bool(metadata.get("image"))
    return (
        gr.update(choices=targets, value=targets[0] if targets else None),
        gr.update(visible=not image, value=False if image else False),
        gr.update(label="📊 Validation Size" if image else "📊 Test Size"),
        gr.update(visible=not image),
        gr.update(visible=image),
    )


def update_uploaded_type_controls(data_type: str):
    image = data_type == "Image Dataset"
    return (
        gr.update(
            label="📁 Training Data (Image ZIP or CSV Manifest)" if image else "📁 Training Data (CSV)",
            file_types=[".zip", ".csv"] if image else [".csv"],
        ),
        gr.update(
            label="📁 Test Data (Image ZIP or CSV Manifest) - Optional" if image else "📁 Test Data (CSV) - Optional",
            file_types=[".zip", ".csv"] if image else [".csv"],
        ),
        gr.update(visible=not image),
        gr.update(visible=not image),
        gr.update(label="📊 Validation Size" if image else "📊 Test Size"),
        gr.update(visible=not image),
        gr.update(visible=image),
    )


def _file_path(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, (str, os.PathLike)):
        return str(value)
    for attribute in ("path", "name"):
        candidate = getattr(value, attribute, None)
        if candidate:
            return str(candidate)
    if isinstance(value, Mapping):
        for key in ("path", "name"):
            if value.get(key):
                return str(value[key])
    return ""


def update_uploaded_targets(data_type: str, train_file: Any = None):
    if data_type == "Image Dataset":
        return gr.update(choices=["label"], value="label")
    path = _file_path(train_file)
    if not path:
        return gr.update(choices=[], value=None)
    try:
        columns = pd.read_csv(path, nrows=1).columns.tolist()
    except Exception:
        return gr.update(choices=[], value=None)
    return gr.update(choices=columns, value=columns[0] if columns else None)


def _saved_model_choices(
    session_token: str | None, request: gr.Request | None = None
) -> list[tuple[str, str]]:
    username = auth_storage.current_username(request, session_token)
    if not username:
        return []
    choices = []
    for record in auth_storage.list_saved_models(username):
        task = str(record.get("task_type") or "").strip()
        artifact = str(record.get("artifact_type") or "model")
        detail = f"{artifact}/{task}" if task else artifact
        created = str(record.get("created_at") or "")[:10] or "unknown date"
        target = str(record.get("target_column") or "unknown target")
        label = f"{record.get('display_name') or 'AutoML model'} - {detail} - {target} - {created}"
        choices.append((label, str(record["model_id"])))
    return choices


def refresh_saved_model_choices(
    selected: str | None = None,
    session_token: str | None = None,
    request: gr.Request | None = None,
):
    choices = _saved_model_choices(session_token, request)
    values = {value for _, value in choices}
    return gr.update(
        choices=choices,
        value=selected if selected in values else None,
        interactive=bool(choices),
        label="Saved Cloud Model",
    )


def _auth_controls(signed_in: bool):
    return (
        gr.update(interactive=signed_in),
        gr.update(interactive=signed_in),
        gr.update(interactive=signed_in),
        gr.update(interactive=False),
    )


def _neutral_replay_outputs(signed_in: bool):
    return (
        "Choose a saved cloud model or upload a trained model package."
        if signed_in
        else "Sign in before loading a model package.",
        "",
        "",
        gr.update(visible=False),
        gr.update(visible=False),
        gr.update(visible=False, value=AUTO_DEVICE),
        gr.update(visible=False, value=16),
        {},
        "Choose a model to inspect its compute route."
        if signed_in
        else "Sign in to inspect the model compute route.",
        "Status: _idle_",
        "",
        pd.DataFrame(),
    )


def _pending_account_controls():
    requests = auth_storage.list_pending_account_requests()
    rows = [
        [str(record.get("username") or ""), str(record.get("requested_at") or "")]
        for record in requests
    ]
    choices = [row[0] for row in rows]
    return rows, choices


def _admin_ui_outputs(username: str):
    if not auth_storage.is_admin(username):
        return (
            gr.update(visible=False),
            gr.update(value=[]),
            gr.update(choices=[], value=None),
            "",
        )
    rows, choices = _pending_account_controls()
    return (
        gr.update(visible=True),
        gr.update(value=rows),
        gr.update(choices=choices, value=choices[0] if choices else None),
        f"{len(choices)} pending account request(s).",
    )


def _auth_outputs(
    result: AuthResult,
    request: gr.Request | None = None,
):
    if result.ok:
        return (
            result.message,
            result.session_token,
            gr.update(visible=False),
            gr.update(visible=True),
            gr.update(visible=True),
            f"Signed in as **{result.username}**",
            refresh_saved_model_choices(
                None,
                result.session_token,
                request,
            ),
            "",
            *_auth_controls(True),
            *_neutral_replay_outputs(True),
            gr.update(visible=False),
            "",
            "",
            None,
            *_admin_ui_outputs(result.username),
        )
    denied = result.account_status == auth_storage.ACCOUNT_STATUS_DENIED
    return (
        result.message,
        "",
        gr.update(visible=True),
        gr.update(visible=False),
        gr.update(visible=False),
        "Sign in or create an account to continue.",
        refresh_saved_model_choices(None, "", request),
        "",
        *_auth_controls(False),
        *_neutral_replay_outputs(False),
        gr.update(visible=denied),
        result.username if denied else "",
        "",
        None,
        *_admin_ui_outputs(""),
    )


def restore_auth_ui(
    session_token: str | None = None,
    request: gr.Request | None = None,
):
    token = str(session_token or "").strip()
    username = auth_storage.current_username(request, token)
    if username:
        return _auth_outputs(
            AuthResult(
                True,
                f"Signed in as {username}.",
                session_token=token,
                username=username,
            ),
            request,
        )
    message = (
        "Your session expired. Sign in again."
        if token
        else "Sign in or create an account to continue."
    )
    return _auth_outputs(AuthResult(False, message), request)


def handle_login_ui(
    username: str, password: str, request: gr.Request | None = None
):
    return _auth_outputs(
        login(username, password, requester_key(request)),
        request,
    )


def handle_signup_ui(
    username: str,
    password: str,
    confirm_password: str,
    request: gr.Request | None = None,
):
    return _auth_outputs(
        request_signup(
            username,
            password,
            confirm_password,
            requester_key(request),
        ),
        request,
    )


def handle_denied_request_ui(
    username: str,
    password: str,
    choice: str,
):
    result = handle_denied_request(username, password, choice)
    denied = result.account_status == auth_storage.ACCOUNT_STATUS_DENIED
    return (
        result.message,
        gr.update(visible=denied),
        result.username if denied else "",
        "",
        None,
    )


def handle_logout_ui():
    return _auth_outputs(AuthResult(False, "Signed out."))


def handle_password_change_ui(
    username: str,
    old_password: str,
    new_password: str,
    confirm_password: str,
):
    return change_password(
        username, old_password, new_password, confirm_password
    ).message


def refresh_pending_account_requests(
    session_token: str | None = None,
    request: gr.Request | None = None,
):
    admin_username = auth_storage.current_username(request, session_token)
    admin = auth_storage.get_user(admin_username)
    if (
        not auth_storage.is_admin(admin_username)
        or admin is None
        or str(admin.get("account_status") or "")
        != auth_storage.ACCOUNT_STATUS_APPROVED
    ):
        return (
            gr.update(value=[]),
            gr.update(choices=[], value=None),
            "Only an approved administrator can view account requests.",
        )
    rows, choices = _pending_account_controls()
    return (
        gr.update(value=rows),
        gr.update(choices=choices, value=choices[0] if choices else None),
        f"{len(choices)} pending account request(s).",
    )


def handle_admin_account_review(
    session_token: str,
    pending_username: str,
    decision: str,
    request: gr.Request | None = None,
):
    admin_username = ""
    try:
        admin_username = auth_storage.require_authenticated_username(
            request,
            session_token,
        )
        reviewed = auth_storage.review_account_request(
            admin_username,
            pending_username,
            decision,
        )
        message = (
            f"Account request for {reviewed['username']} was "
            f"{reviewed['account_status']} by {admin_username}."
        )
    except (PermissionError, ValueError) as exc:
        message = f"Could not review account request: {exc}"
    except Exception:
        message = "Could not review the account request. Try again later."
    rows, choices = (
        _pending_account_controls()
        if auth_storage.is_admin(admin_username)
        else ([], [])
    )
    return (
        gr.update(value=rows),
        gr.update(choices=choices, value=choices[0] if choices else None),
        message,
    )


def handle_admin_account_approval(
    session_token: str,
    pending_username: str,
    request: gr.Request | None = None,
):
    return handle_admin_account_review(
        session_token,
        pending_username,
        auth_storage.ACCOUNT_STATUS_APPROVED,
        request,
    )


def handle_admin_account_denial(
    session_token: str,
    pending_username: str,
    request: gr.Request | None = None,
):
    return handle_admin_account_review(
        session_token,
        pending_username,
        auth_storage.ACCOUNT_STATUS_DENIED,
        request,
    )


def _model_info(manifest: Mapping[str, Any], artifact_type: str):
    target = str(
        manifest.get("target_column") or manifest.get("label_column") or "Unknown"
    )
    if artifact_type == "image":
        task = normalize_image_task(manifest.get("task_type") or "classification")
        names = {
            "classification": ("Image classifier", "class label"),
            "object_detection": ("Object detection model", "bounding boxes"),
            "instance_segmentation": ("Instance segmentation model", "masks and bounding boxes"),
            "semantic_segmentation": ("Semantic segmentation model", "segmentation masks"),
        }
        model_type, output = names[task]
        info = f"### Loaded Model\n\n**Model type:** {model_type}\n\n**Output:** {output}\n\n**Prediction target:** `{target}`\n\n**Required input:** one image file"
        cpu_decision = route_prediction(
            {"image_device": "cpu"},
            manifest,
            routing_mode=routing_mode_from_env(),
        )
        default_device = (
            "cpu"
            if cpu_decision.profile == "cpu"
            else "cuda - requires GPU hardware"
        )
        return (
            info,
            gr.update(visible=False),
            gr.update(visible=True),
            gr.update(visible=True, value=default_device),
            gr.update(visible=True),
            gr.update(interactive=True),
            dict(manifest),
            replay_route_preview(manifest, default_device),
        )
    protected = "enabled" if manifest.get("input_schema_file") else "legacy package"
    info = f"### Loaded Model\n\n**Model type:** Tabular model\n\n**Prediction target:** `{target}`\n\n**Input schema protection:** {protected}\n\n**Required input:** one CSV containing one row"
    return (
        info,
        gr.update(visible=True),
        gr.update(visible=False),
        gr.update(visible=False, value="cpu"),
        gr.update(visible=False),
        gr.update(interactive=True),
        dict(manifest),
        replay_route_preview(manifest, "cpu"),
    )


def update_single_input_controls(
    model_file: Any,
    saved_model_id: str | None,
    session_token: str | None,
    request: gr.Request | None = None,
):
    username = auth_storage.current_username(request, session_token)
    if not username:
        return (
            "Sign in before loading a model package.",
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(interactive=False),
            {},
            "Sign in to inspect the model compute route.",
        )
    if not saved_model_id and not model_file:
        return (
            "Choose a saved cloud model or upload a trained model package.",
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(interactive=False),
            {},
            "Choose a model to inspect its compute route.",
        )
    try:
        if saved_model_id:
            record = auth_storage.get_saved_model(username, saved_model_id)
            if record is None:
                raise PermissionError("Saved model not found for this account.")
            manifest = auth_storage.manifest_from_record(record)
            artifact_type = str(
                record.get("artifact_type")
                or manifest.get("artifact_type")
                or "tabular"
            )
            return _model_info(manifest, artifact_type)
        package_path = _file_path(model_file)
        if not package_path:
            raise ValueError("Choose a saved cloud model or upload a package.")
        manifest, artifact_type = inspect_package_manifest(package_path)
        return _model_info(manifest, artifact_type)
    except Exception as exc:
        return (
            f"❌ Could not inspect model package: {exc}",
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(interactive=False),
            {},
            _route_preview_error(exc),
        )


def _training_outputs(
    status: str,
    metrics: str = "",
    predictions: pd.DataFrame | None = None,
    summary: str = "",
    tools: str = "",
    reasoning: str = "",
    package: str | None = None,
):
    return status, metrics, predictions if predictions is not None else pd.DataFrame(), summary, tools, reasoning, package


def _training_error(message: str):
    safe = str(message or "The request failed.")
    if not safe.startswith("❌"):
        safe = f"❌ Error: {safe}"
    return _training_outputs(safe)


def _gateway_egress_root() -> Path:
    return Path(
        os.getenv("AUTOML_GATEWAY_EGRESS_DIR", "/tmp/automl_gateway_egress")
    ).expanduser()


def _gateway_egress_max_age() -> int:
    try:
        return max(
            60,
            int(os.getenv("AUTOML_GATEWAY_EGRESS_MAX_AGE_SECONDS", "3600")),
        )
    except ValueError:
        return 3600


def prune_gateway_egress(
    root: str | Path | None = None,
    *,
    max_age_seconds: int | None = None,
    now: float | None = None,
) -> int:
    base = Path(root) if root is not None else _gateway_egress_root()
    if not base.exists():
        return 0
    current = float(now if now is not None else time.time())
    maximum = max(60, int(max_age_seconds or _gateway_egress_max_age()))
    removed = 0
    for child in list(base.iterdir()):
        try:
            expired = current - child.stat().st_mtime > maximum
        except OSError:
            continue
        if not expired:
            continue
        try:
            shutil.rmtree(child) if child.is_dir() else child.unlink()
            removed += 1
        except OSError:
            continue
    return removed


def _copy_to_gateway_egress(source_path: str) -> str:
    source = Path(source_path)
    if not source.is_file():
        raise FileNotFoundError("The backend model package was not downloaded.")
    root = _gateway_egress_root()
    root.mkdir(parents=True, exist_ok=True)
    request_dir = root / uuid.uuid4().hex
    request_dir.mkdir(mode=0o700)
    destination = request_dir / (source.name or "automl_model.zip")
    shutil.copy2(source, destination)
    os.utime(request_dir, None)
    return str(destination)


def _cleanup_files(paths: tuple[str, ...] | list[str]) -> None:
    for value in paths:
        try:
            path = Path(value)
            if path.is_file():
                path.unlink()
        except OSError:
            continue


def _display_name(payload: Mapping[str, Any]) -> str:
    source = str(payload.get("dataset") or payload.get("data_type") or "AutoML model")
    target = str(payload.get("target_column") or "").strip()
    stamp = datetime.now().strftime("%Y-%m-%d %H:%M")
    return f"{source} ({target}) {stamp}" if target else f"{source} {stamp}"


def render_training_result(
    username: str,
    envelope: Mapping[str, Any],
    package_path: str | None,
    display_name: str,
):
    if envelope.get("ok") is not True:
        error = envelope.get("error") or {}
        return _training_error(str(error.get("message") or "The backend job failed."))
    summary = str(envelope.get("summary") or "")
    download = None
    if package_path:
        try:
            manifest, artifact_type = inspect_package_manifest(package_path)
            download = _copy_to_gateway_egress(package_path)
            try:
                record = auth_storage.save_model_for_user(
                    username,
                    download,
                    display_name=display_name,
                    manifest=manifest,
                    artifact_type=artifact_type,
                )
                summary += f"\n\nSaved cloud model: `{record.get('display_name')}` (encrypted, id `{record.get('model_id')}`)."
            except Exception:
                summary += "\n\nSaved cloud model: cloud save failed; use the immediate download instead."
        except Exception as exc:
            summary += f"\n\nModel package rejected by the gateway: {exc}"
            download = None
    return _training_outputs(
        str(envelope.get("status") or "Complete"),
        str(envelope.get("metrics_markdown") or ""),
        predictions_frame(envelope),
        summary,
        str(envelope.get("tools_used") or ""),
        str(envelope.get("reasoning") or ""),
        download,
    )


def run_existing_remote(
    payload: Mapping[str, Any],
    session_token: str | None,
    request: gr.Request | None = None,
):
    """Authenticated generator for one existing-dataset backend job."""

    try:
        username = auth_storage.require_authenticated_username(request, session_token)
    except PermissionError:
        yield _training_error("Sign in before starting an analysis.")
        return
    result: BackendResult | None = None
    decision: RouteDecision | None = None
    try:
        routed_payload = dict(payload)
        metadata = _dataset_metadata(str(routed_payload.get("dataset") or ""))
        dataset_is_image = bool(metadata.get("image"))
        routed_payload["data_type"] = (
            "Image Dataset" if dataset_is_image else "Tabular CSV"
        )
        mode = routing_mode_from_env()
        decision = route_existing(
            routed_payload,
            dataset_is_image=dataset_is_image,
            routing_mode=mode,
        )
        request_payload = declare_route(routed_payload, decision, mode)
        profile_label = decision.profile.upper()
        yield _training_outputs(
            f"Queued for the private {profile_label} backend…"
        )
        with ROUTED_JOB_ADMISSION.lease(
            decision.profile, username, timeout=30
        ):
            yield _training_outputs(f"{profile_label} backend is waking up…")
            client = GatewayBackendClient.from_env(decision.profile)
            result = client.run_existing(request_payload)
            prune_gateway_egress()
            yield render_training_result(
                username,
                result.envelope,
                result.package_path,
                _display_name(request_payload),
            )
    except BackendUnavailable as exc:
        if decision is not None and decision.profile == "cpu":
            yield _training_error(
                "The private CPU backend is unavailable; GPU fallback was not attempted."
            )
        else:
            yield _training_error(str(exc))
    except (ContractError, RoutingError, JobRejected) as exc:
        yield _training_error(str(exc))
    except Exception:
        yield _training_error("The gateway could not complete the request.")
    finally:
        if result is not None:
            _cleanup_files(result.cleanup_files)


def run_uploaded_remote(
    payload: Mapping[str, Any],
    train_file: Any,
    test_file: Any,
    session_token: str | None,
    request: gr.Request | None = None,
):
    """Authenticated generator for one uploaded-dataset backend job."""

    try:
        username = auth_storage.require_authenticated_username(request, session_token)
    except PermissionError:
        yield _training_error("Sign in before starting an analysis.")
        return
    train_path = _file_path(train_file)
    test_path = _file_path(test_file)
    result: BackendResult | None = None
    decision: RouteDecision | None = None
    try:
        if not train_path:
            raise ContractError("Upload a training dataset first.")
        check_upload_size(train_path)
        if test_path:
            check_upload_size(test_path)
        mode = routing_mode_from_env()
        decision = route_uploaded(payload, routing_mode=mode)
        request_payload = declare_route(payload, decision, mode)
        profile_label = decision.profile.upper()
        yield _training_outputs(
            f"Queued for the private {profile_label} backend…"
        )
        with ROUTED_JOB_ADMISSION.lease(
            decision.profile, username, timeout=30
        ):
            yield _training_outputs(f"{profile_label} backend is waking up…")
            client = GatewayBackendClient.from_env(decision.profile)
            result = client.run_uploaded(
                request_payload, train_path, test_path or None
            )
            prune_gateway_egress()
            yield render_training_result(
                username,
                result.envelope,
                result.package_path,
                _display_name(request_payload),
            )
    except BackendUnavailable as exc:
        if decision is not None and decision.profile == "cpu":
            yield _training_error(
                "The private CPU backend is unavailable; GPU fallback was not attempted."
            )
        else:
            yield _training_error(str(exc))
    except (ContractError, RoutingError, JobRejected) as exc:
        yield _training_error(str(exc))
    except Exception:
        yield _training_error("The gateway could not complete the request.")
    finally:
        if result is not None:
            _cleanup_files(result.cleanup_files)


def predict_remote(
    model_file: Any,
    tabular_text: str,
    tabular_file: Any,
    image_file: Any,
    image_device: str,
    image_batch_size: int,
    saved_model_id: str | None,
    session_token: str | None,
    request: gr.Request | None = None,
):
    """Authenticate, materialize one package, and replay it privately."""

    cleanup_root = ""
    decision: RouteDecision | None = None
    try:
        username = auth_storage.require_authenticated_username(request, session_token)
        if saved_model_id:
            package_path, cleanup_root = auth_storage.materialize_saved_model(
                username, saved_model_id
            )
        else:
            package_path = _file_path(model_file)
        if not package_path:
            raise ContractError("Choose a saved model or upload a package first.")
        manifest, artifact_type = inspect_package_manifest(package_path)
        if not manifest.get("artifact_type"):
            manifest = {**manifest, "artifact_type": artifact_type}
        tabular_path = _file_path(tabular_file)
        image_path = _file_path(image_file)
        if artifact_type == "image" and not image_path:
            raise ContractError("Upload one image for this model.")
        if artifact_type != "image" and not tabular_path and not str(tabular_text or "").strip():
            raise ContractError("Upload one tabular CSV row for this model.")
        if tabular_path:
            check_upload_size(tabular_path)
        if image_path:
            check_upload_size(image_path)
        payload = {
            "protocol_version": PROTOCOL_VERSION,
            "tabular_text": str(tabular_text or ""),
            "image_device": image_device or AUTO_DEVICE,
            "image_batch_size": int(image_batch_size or 16),
        }
        mode = routing_mode_from_env()
        decision = route_prediction(payload, manifest, routing_mode=mode)
        request_payload = declare_route(payload, decision, mode)
        with ROUTED_JOB_ADMISSION.lease(
            decision.profile, username, timeout=30
        ):
            client = GatewayBackendClient.from_env(decision.profile)
            result = client.predict_package(
                request_payload,
                package_path,
                tabular_path=tabular_path or None,
                image_path=image_path or None,
            )
        if result.envelope.get("ok") is not True:
            error = result.envelope.get("error") or {}
            return f"❌ Error: {error.get('message') or 'Prediction failed.'}", "", pd.DataFrame()
        return (
            str(result.envelope.get("status") or "✅ Prediction complete."),
            str(result.envelope.get("summary") or ""),
            predictions_frame(result.envelope),
        )
    except BackendUnavailable as exc:
        if decision is not None and decision.profile == "cpu":
            return (
                "❌ Error: The private CPU backend is unavailable; GPU fallback was not attempted.",
                "",
                pd.DataFrame(),
            )
        return f"❌ Error: {exc}", "", pd.DataFrame()
    except (ContractError, RoutingError, JobRejected, PermissionError) as exc:
        return f"❌ Error: {exc}", "", pd.DataFrame()
    except Exception:
        return "❌ Error: The gateway could not complete the prediction.", "", pd.DataFrame()
    finally:
        if cleanup_root:
            shutil.rmtree(cleanup_root, ignore_errors=True)


def _existing_payload(
    dataset,
    target_column,
    split_training_data,
    test_size,
    preset,
    time_limit,
    trials,
    feature_selection_tool,
    transform_tool,
    ml_tool,
    image_backbone,
    image_classifier,
    image_size,
    image_device,
    image_batch_size,
    image_pretrained,
    image_task,
    image_backend,
    image_dataset_format,
    image_epochs,
    image_imgsz,
    image_benchmark_backends,
    cancer_preset,
):
    return {
        "protocol_version": PROTOCOL_VERSION,
        "dataset": dataset,
        "target_column": target_column,
        "split_training_data": bool(split_training_data),
        "test_size": float(test_size),
        "preset": preset,
        "time_limit": int(time_limit),
        "trials": int(trials),
        "feature_selection_tool": feature_selection_tool,
        "transform_tool": transform_tool,
        "ml_tool": ml_tool,
        "image_backbone": image_backbone,
        "image_classifier": image_classifier,
        "image_size": image_size,
        "image_device": image_device,
        "image_batch_size": int(image_batch_size),
        "image_pretrained": bool(image_pretrained),
        "image_task": image_task,
        "image_backend": image_backend,
        "image_dataset_format": image_dataset_format,
        "image_data_path": "",
        "image_epochs": int(image_epochs),
        "image_imgsz": int(image_imgsz),
        "image_benchmark_backends": image_benchmark_backends,
        "cancer_preset": cancer_preset,
    }


def existing_ui_handler(
    dataset,
    target_column,
    split_training_data,
    test_size,
    preset,
    time_limit,
    trials,
    feature_selection_tool,
    transform_tool,
    ml_tool,
    image_backbone,
    image_classifier,
    image_size,
    image_device,
    image_batch_size,
    image_pretrained,
    image_task,
    image_backend,
    image_dataset_format,
    image_epochs,
    image_imgsz,
    image_benchmark_backends,
    cancer_preset,
    session_token,
    request: gr.Request | None = None,
):
    payload = _existing_payload(
        dataset,
        target_column,
        split_training_data,
        test_size,
        preset,
        time_limit,
        trials,
        feature_selection_tool,
        transform_tool,
        ml_tool,
        image_backbone,
        image_classifier,
        image_size,
        image_device,
        image_batch_size,
        image_pretrained,
        image_task,
        image_backend,
        image_dataset_format,
        image_epochs,
        image_imgsz,
        image_benchmark_backends,
        cancer_preset,
    )
    yield from run_existing_remote(payload, session_token, request)


def _uploaded_payload(
    data_type,
    target_column,
    split_training_data,
    test_size,
    preset,
    time_limit,
    trials,
    feature_selection_tool,
    transform_tool,
    ml_tool,
    image_backbone,
    image_classifier,
    image_size,
    image_device,
    image_batch_size,
    image_pretrained,
    uploaded_image_backend,
    cancer_preset,
):
    return {
        "protocol_version": PROTOCOL_VERSION,
        "data_type": data_type,
        "target_column": target_column,
        "split_training_data": bool(split_training_data),
        "test_size": float(test_size),
        "preset": preset,
        "time_limit": int(time_limit),
        "trials": int(trials),
        "feature_selection_tool": feature_selection_tool,
        "transform_tool": transform_tool,
        "ml_tool": ml_tool,
        "image_backbone": image_backbone,
        "image_classifier": image_classifier,
        "image_size": image_size,
        "image_device": image_device,
        "image_batch_size": int(image_batch_size),
        "image_pretrained": bool(image_pretrained),
        "uploaded_image_backend": uploaded_image_backend,
        "cancer_preset": cancer_preset,
        "image_epochs": 1,
        "image_imgsz": 64,
    }


def uploaded_ui_handler(
    data_type,
    target_column,
    split_training_data,
    test_size,
    preset,
    time_limit,
    trials,
    feature_selection_tool,
    transform_tool,
    ml_tool,
    image_backbone,
    image_classifier,
    image_size,
    image_device,
    image_batch_size,
    image_pretrained,
    uploaded_image_backend,
    cancer_preset,
    train_file,
    test_file,
    session_token,
    request: gr.Request | None = None,
):
    payload = _uploaded_payload(
        data_type,
        target_column,
        split_training_data,
        test_size,
        preset,
        time_limit,
        trials,
        feature_selection_tool,
        transform_tool,
        ml_tool,
        image_backbone,
        image_classifier,
        image_size,
        image_device,
        image_batch_size,
        image_pretrained,
        uploaded_image_backend,
        cancer_preset,
    )
    yield from run_uploaded_remote(
        payload, train_file, test_file, session_token, request
    )


def _add_results_section():
    gr.Markdown("---\n## 📊 Results & Analysis")
    status = gr.Markdown("Status: _idle_")
    metrics = gr.Markdown("### 📊 Model Performance Metrics\n\n_Run a pipeline to see metrics._")
    predictions = gr.DataFrame(label="Sample Predictions", interactive=False)
    package = gr.File(label="Trained Model Package", interactive=False)
    summary = gr.Textbox(label="Dataset Summary", interactive=False, max_lines=18)
    tools = gr.Textbox(label="Tools Used", interactive=False)
    reasoning = gr.Textbox(label="🤖 Agent Reasoning", interactive=False, max_lines=20)
    return status, metrics, predictions, summary, tools, reasoning, package


def create_interface() -> gr.Blocks:
    """Build the public frontend without importing or contacting ML code."""

    routing_mode_from_env()
    catalog = _catalog()["datasets"]
    dataset_choices = list(catalog) or ["No datasets configured"]
    default_dataset = "titanic" if "titanic" in catalog else dataset_choices[0]
    metadata = dict(catalog.get(default_dataset) or {})
    targets = list(metadata.get("targets") or [])
    default_target = targets[0] if targets else None
    initial_image = bool(metadata.get("image"))
    initial_state = get_image_task_ui_state("classification", "Auto")

    with gr.Blocks(title="AutoML Pipeline") as demo:
        session_token = gr.BrowserState(
            "",
            storage_key="automl-auth-session-v1",
            secret=auth_storage.browser_state_secret(),
        )
        gr.Markdown("# 🤖 AutoML Pipeline")
        gr.Markdown("Sign in or request an account, configure a workflow, and run it on the appropriate private CPU or GPU backend.")

        with gr.Group(visible=True) as auth_panel:
            gr.Markdown("## Account")
            auth_status = gr.Markdown("Sign in or create an account to continue.")
            with gr.Tabs():
                with gr.TabItem("Log In"):
                    login_username = gr.Textbox(label="Username")
                    login_password = gr.Textbox(label="Password", type="password")
                    login_button = gr.Button("Log In")
                with gr.TabItem(
                    "Sign Up",
                    visible=auth_storage.signups_allowed(),
                ):
                    signup_username = gr.Textbox(label="Username")
                    signup_password = gr.Textbox(label="Password", type="password")
                    signup_confirm_password = gr.Textbox(
                        label="Confirm Password",
                        type="password",
                    )
                    signup_button = gr.Button("Submit Account Request")
                with gr.TabItem("Change Password"):
                    password_username = gr.Textbox(label="Username")
                    old_password = gr.Textbox(label="Current Password", type="password")
                    new_password = gr.Textbox(label="New Password", type="password")
                    confirm_password = gr.Textbox(label="Confirm New Password", type="password")
                    password_button = gr.Button("Update Password")
                    password_status = gr.Markdown("")
            denied_username_state = gr.State("")
            with gr.Group(visible=False) as denied_request_group:
                gr.Markdown(
                    "### Denied account request\n"
                    "Re-enter the password used for signup, then choose whether to "
                    "submit the request for another review or permanently delete it."
                )
                denied_password = gr.Textbox(
                    label="Signup Password",
                    type="password",
                )
                denied_request_choice = gr.Radio(
                    choices=[
                        "Re-send request",
                        "Do not re-send; delete my account request",
                    ],
                    label="What would you like to do?",
                )
                denied_request_button = gr.Button("Submit Choice")

        with gr.Row(visible=False) as account_row:
            account_status = gr.Markdown("Account: _signed out_")
            logout_button = gr.Button("Log Out", min_width=100)

        with gr.Tabs(visible=False) as main_tabs:
            with gr.TabItem("📁 Use Existing Datasets"):
                with gr.Row():
                    with gr.Column():
                        dataset = gr.Dropdown(dataset_choices, value=default_dataset, label="Dataset")
                        target = gr.Dropdown(targets, value=default_target, label="🎯 Target Column to Predict", allow_custom_value=True)
                        split = gr.Checkbox(False, label="🔄 Split Training Data", visible=not initial_image)
                        test_size = gr.Slider(0.1, 0.5, value=0.2, step=0.05, label="📊 Validation Size" if initial_image else "📊 Test Size")
                        preset = gr.Dropdown(QUALITY_PRESETS, value="medium_quality", label="Quality Preset")
                        time_limit = gr.Slider(60, 3600, value=300, step=60, label="Time Limit (seconds)")
                        trials = gr.Dropdown(["1", "2", "3", "4"], value="1", label="Trials")
                        with gr.Group(visible=not initial_image) as tabular_group:
                            gr.Markdown("### 🛠️ Tool Selection (Optional)")
                            feature = gr.Dropdown(FEATURE_SELECTION_CHOICES, value="🤖 Agent Decides", label="Feature Selection Tool")
                            transform = gr.Dropdown(TRANSFORM_CHOICES, value="🤖 Agent Decides", label="Data Transformation Tool")
                            ml_tool = gr.Dropdown(ML_CHOICES, value="🤖 Agent Decides", label="Machine Learning Tool")
                        with gr.Group(visible=initial_image) as image_group:
                            gr.Markdown("### 🖼️ Image Pipeline Controls")
                            image_task = gr.Dropdown(IMAGE_TASK_CHOICES, value="classification", label="Image Task")
                            image_backend = gr.Dropdown(IMAGE_BACKEND_CHOICES, value="Auto", label="Image Backend")
                            image_format = gr.Dropdown(IMAGE_FORMAT_CHOICES, value="auto", label="Dataset Format")
                            gr.Textbox(value="Managed existing-dataset source", label="Dataset Source", interactive=False)
                            image_note = gr.Markdown(initial_state["note"])
                            image_device = gr.Dropdown(IMAGE_DEVICE_CHOICES, value=AUTO_DEVICE, label="Image Device")
                            image_batch = gr.Slider(1, 128, value=64, step=1, label="Image Batch Size")
                            with gr.Group(visible=initial_state["classification"]) as classification_group:
                                image_backbone = gr.Dropdown(IMAGE_BACKBONE_CHOICES, value=AUTO_CHOICE, label="Image Backbone")
                                image_classifier = gr.Dropdown(IMAGE_CLASSIFIER_CHOICES, value=AUTO_CHOICE, label="Image Classifier")
                                image_size = gr.Dropdown(IMAGE_SIZE_CHOICES, value=AUTO_CHOICE, label="Image Size")
                                image_pretrained = gr.Checkbox(True, label="Use Pretrained CNN Weights")
                            with gr.Group(visible=initial_state["medical"]) as medical_group:
                                cancer_preset = gr.Dropdown(CANCER_PRESET_CHOICES, value="Auto - agent decides", label="Cancer Feature Preset")
                            with gr.Group(visible=initial_state["detection"]) as detection_group:
                                gr.Markdown("### Detection Controls\nCOCO annotations use backend defaults.")
                            with gr.Group(visible=initial_state["instance"]) as instance_group:
                                gr.Markdown("### Instance Segmentation Controls\nCOCO polygons/RLE are preserved.")
                            with gr.Group(visible=initial_state["semantic"]) as semantic_group:
                                gr.Markdown("### Semantic Segmentation Controls\nUses a managed image/mask manifest.")
                            with gr.Group(visible=initial_state["optional"]) as optional_group:
                                image_epochs = gr.Slider(1, 100, value=1, step=1, label="Backend Epochs")
                                image_imgsz = gr.Slider(32, 1024, value=64, step=32, label="Backend Image Size")
                            with gr.Group(visible=initial_state["benchmark"]) as benchmark_group:
                                benchmark_backends = gr.Textbox("", label="Benchmark Backends", placeholder="yolo,automm")
                        existing_route = gr.Markdown(
                            existing_route_preview(
                                default_dataset,
                                AUTO_CHOICE,
                                AUTO_DEVICE,
                                "classification",
                                "Auto",
                                "",
                                "Auto - agent decides",
                            )
                        )
                        run_button = gr.Button("🚀 Start Analysis", interactive=False)
                        dataset.change(
                            update_existing_dataset_controls,
                            dataset,
                            [target, split, test_size, tabular_group, image_group],
                            api_name=False,
                        )
                        for selector in (image_task, image_backend):
                            selector.change(
                                image_task_ui_updates,
                                [image_task, image_backend],
                                [classification_group, medical_group, detection_group, instance_group, semantic_group, benchmark_group, optional_group, image_note],
                                api_name=False,
                            )
                        existing_route_inputs = [
                            dataset,
                            image_backbone,
                            image_device,
                            image_task,
                            image_backend,
                            benchmark_backends,
                            cancer_preset,
                        ]
                        for selector in (
                            dataset,
                            image_backbone,
                            image_device,
                            image_task,
                            image_backend,
                            benchmark_backends,
                            cancer_preset,
                        ):
                            selector.change(
                                existing_route_preview,
                                existing_route_inputs,
                                existing_route,
                                api_name=False,
                            )
                existing_outputs = _add_results_section()

            with gr.TabItem("📤 Upload Your Data"):
                with gr.Row():
                    with gr.Column():
                        upload_type = gr.Dropdown(["Tabular CSV", "Image Dataset"], value="Tabular CSV", label="Data Type")
                        train_upload = gr.File(label="📁 Training Data (CSV)", file_types=[".csv"])
                        test_upload = gr.File(label="📁 Test Data (CSV) - Optional", file_types=[".csv"])
                        upload_target = gr.Dropdown([], label="🎯 Target Column to Predict", allow_custom_value=True)
                        upload_split = gr.Checkbox(True, label="🔄 Split Training Data")
                        upload_test_size = gr.Slider(0.1, 0.5, value=0.2, step=0.05, label="📊 Test Size")
                        upload_preset = gr.Dropdown(QUALITY_PRESETS, value="medium_quality", label="Quality Preset")
                        upload_time = gr.Slider(60, 3600, value=300, step=60, label="Time Limit (seconds)")
                        upload_trials = gr.Dropdown(["1", "2", "3", "4"], value="1", label="Trials")
                        with gr.Group(visible=True) as upload_tabular_group:
                            upload_feature = gr.Dropdown(FEATURE_SELECTION_CHOICES, value="🤖 Agent Decides", label="Feature Selection Tool")
                            upload_transform = gr.Dropdown(TRANSFORM_CHOICES, value="🤖 Agent Decides", label="Data Transformation Tool")
                            upload_ml = gr.Dropdown(ML_CHOICES, value="🤖 Agent Decides", label="Machine Learning Tool")
                        with gr.Group(visible=False) as upload_image_group:
                            upload_image_backend = gr.Dropdown(["Existing classification pipeline", "Medical image agent"], value="Existing classification pipeline", label="Image Backend")
                            with gr.Group(visible=True) as upload_existing_group:
                                upload_backbone = gr.Dropdown(IMAGE_BACKBONE_CHOICES, value=AUTO_CHOICE, label="Image Backbone")
                                upload_classifier = gr.Dropdown(IMAGE_CLASSIFIER_CHOICES, value=AUTO_CHOICE, label="Image Classifier")
                                upload_size = gr.Dropdown(IMAGE_SIZE_CHOICES, value=AUTO_CHOICE, label="Image Size")
                            with gr.Group(visible=False) as upload_medical_group:
                                upload_cancer_preset = gr.Dropdown(CANCER_PRESET_CHOICES, value="Auto - agent decides", label="Cancer Feature Preset")
                            upload_device = gr.Dropdown(IMAGE_DEVICE_CHOICES, value=AUTO_DEVICE, label="Image Device")
                            upload_batch = gr.Slider(8, 128, value=64, step=8, label="Image Batch Size")
                            upload_pretrained = gr.Checkbox(True, label="Use Pretrained CNN Weights")
                        uploaded_route = gr.Markdown(
                            uploaded_route_preview(
                                "Tabular CSV",
                                AUTO_CHOICE,
                                AUTO_DEVICE,
                                "Existing classification pipeline",
                                "Auto - agent decides",
                            )
                        )
                        upload_run_button = gr.Button("🚀 Start Analysis on Uploaded Data", interactive=False)
                        upload_type.change(
                            update_uploaded_type_controls,
                            upload_type,
                            [train_upload, test_upload, upload_target, upload_split, upload_test_size, upload_tabular_group, upload_image_group],
                            api_name=False,
                        ).then(update_uploaded_targets, [upload_type, train_upload], upload_target, api_name=False)
                        train_upload.change(update_uploaded_targets, [upload_type, train_upload], upload_target, api_name=False)
                        upload_image_backend.change(uploaded_image_backend_updates, upload_image_backend, [upload_existing_group, upload_medical_group], api_name=False)
                        uploaded_route_inputs = [
                            upload_type,
                            upload_backbone,
                            upload_device,
                            upload_image_backend,
                            upload_cancer_preset,
                        ]
                        for selector in (
                            upload_type,
                            upload_backbone,
                            upload_device,
                            upload_image_backend,
                            upload_cancer_preset,
                        ):
                            selector.change(
                                uploaded_route_preview,
                                uploaded_route_inputs,
                                uploaded_route,
                                api_name=False,
                            )
                uploaded_outputs = _add_results_section()

            with gr.TabItem("🔮 Use Trained Model") as replay_tab:
                with gr.Row():
                    with gr.Column():
                        saved_models = gr.Dropdown([], label="Saved Cloud Model", interactive=False)
                        refresh_models = gr.Button("Refresh Saved Models", interactive=False)
                        model_upload = gr.File(label="Trained Model Package (.zip)", file_types=[".zip"])
                        model_info = gr.Markdown("Choose a saved model or upload a package.")
                        tabular_text = gr.Textbox("", visible=False)
                        with gr.Group(visible=False) as tabular_input_group:
                            tabular_input = gr.File(label="Single Tabular Input (CSV, one row)", file_types=[".csv"])
                        with gr.Group(visible=False) as image_input_group:
                            image_input = gr.File(
                                label="Single Image Input",
                                file_types=[".png", ".jpg", ".jpeg", ".bmp", ".tif", ".tiff"],
                            )
                        replay_device = gr.Dropdown(IMAGE_DEVICE_CHOICES, value=AUTO_DEVICE, label="Image Device", visible=False)
                        replay_batch = gr.Slider(1, 128, value=16, step=1, label="Image Batch Size", visible=False)
                        replay_manifest_state = gr.State({})
                        replay_route = gr.Markdown("Choose a model to inspect its compute route.")
                        predict_button = gr.Button("Predict", interactive=False)
                    with gr.Column():
                        predict_status = gr.Markdown("Status: _idle_")
                        predict_summary = gr.Markdown("")
                        predict_frame = gr.DataFrame(label="Prediction", interactive=False)

            with gr.TabItem("👥 Account Approvals", visible=False) as admin_tab:
                gr.Markdown(
                    "## Pending account requests\n"
                    "Approve a request to let that user log in, or deny it to let "
                    "the user choose whether to re-submit or remove their information."
                )
                pending_requests_df = gr.Dataframe(
                    headers=["Username", "Requested At (UTC)"],
                    datatype=["str", "str"],
                    value=[],
                    interactive=False,
                    label="Pending Requests",
                )
                pending_username = gr.Dropdown(
                    choices=[],
                    value=None,
                    label="Account Request",
                )
                with gr.Row():
                    approve_account_button = gr.Button("Approve", variant="primary")
                    deny_account_button = gr.Button("Deny", variant="stop")
                    refresh_accounts_button = gr.Button("Refresh")
                admin_status = gr.Markdown("")

        auth_outputs = [
            auth_status,
            session_token,
            auth_panel,
            account_row,
            main_tabs,
            account_status,
            saved_models,
            password_status,
            run_button,
            upload_run_button,
            refresh_models,
            predict_button,
            model_info,
            login_username,
            login_password,
            tabular_input_group,
            image_input_group,
            replay_device,
            replay_batch,
            replay_manifest_state,
            replay_route,
            predict_status,
            predict_summary,
            predict_frame,
            denied_request_group,
            denied_username_state,
            denied_password,
            denied_request_choice,
            admin_tab,
            pending_requests_df,
            pending_username,
            admin_status,
        ]
        demo.load(
            restore_auth_ui,
            session_token,
            auth_outputs,
            api_name=False,
        )
        login_button.click(handle_login_ui, [login_username, login_password], auth_outputs, api_name=False)
        signup_button.click(
            handle_signup_ui,
            [signup_username, signup_password, signup_confirm_password],
            auth_outputs,
            api_name=False,
        )
        denied_request_button.click(
            handle_denied_request_ui,
            [
                denied_username_state,
                denied_password,
                denied_request_choice,
            ],
            [
                auth_status,
                denied_request_group,
                denied_username_state,
                denied_password,
                denied_request_choice,
            ],
            api_name=False,
        )
        logout_button.click(handle_logout_ui, None, auth_outputs, api_name=False)
        password_button.click(handle_password_change_ui, [password_username, old_password, new_password, confirm_password], password_status, api_name=False)
        refresh_accounts_button.click(
            refresh_pending_account_requests,
            session_token,
            [pending_requests_df, pending_username, admin_status],
            api_name=False,
        )
        approve_account_button.click(
            handle_admin_account_approval,
            [session_token, pending_username],
            [pending_requests_df, pending_username, admin_status],
            api_name=False,
        )
        deny_account_button.click(
            handle_admin_account_denial,
            [session_token, pending_username],
            [pending_requests_df, pending_username, admin_status],
            api_name=False,
        )

        inspect_outputs = [model_info, tabular_input_group, image_input_group, replay_device, replay_batch, predict_button, replay_manifest_state, replay_route]
        model_upload.change(update_single_input_controls, [model_upload, saved_models, session_token], inspect_outputs, api_name=False)
        saved_models.change(update_single_input_controls, [model_upload, saved_models, session_token], inspect_outputs, api_name=False)
        refresh_models.click(refresh_saved_model_choices, [saved_models, session_token], saved_models, api_name=False)
        replay_tab.select(refresh_saved_model_choices, [saved_models, session_token], saved_models, api_name=False)
        replay_device.change(
            replay_route_preview,
            [replay_manifest_state, replay_device],
            replay_route,
            api_name=False,
        )

        existing_inputs = [dataset, target, split, test_size, preset, time_limit, trials, feature, transform, ml_tool, image_backbone, image_classifier, image_size, image_device, image_batch, image_pretrained, image_task, image_backend, image_format, image_epochs, image_imgsz, benchmark_backends, cancer_preset, session_token]
        run_button.click(existing_ui_handler, existing_inputs, list(existing_outputs), api_name=False)

        uploaded_inputs = [upload_type, upload_target, upload_split, upload_test_size, upload_preset, upload_time, upload_trials, upload_feature, upload_transform, upload_ml, upload_backbone, upload_classifier, upload_size, upload_device, upload_batch, upload_pretrained, upload_image_backend, upload_cancer_preset, train_upload, test_upload, session_token]
        upload_run_button.click(uploaded_ui_handler, uploaded_inputs, list(uploaded_outputs), api_name=False)

        predict_button.click(predict_remote, [model_upload, tabular_text, tabular_input, image_input, replay_device, replay_batch, saved_models, session_token], [predict_status, predict_summary, predict_frame], api_name=False)
    return demo


def validate_gateway_security_config() -> None:
    """Fail closed when the public deployment weakens its security boundary."""

    routing_mode_from_env()
    if auth_storage.env_truthy("AUTOML_AUTH_DISABLED", False):
        raise RuntimeError("Public gateway does not permit AUTOML_AUTH_DISABLED=1.")
    if not auth_storage.app_auth_required():
        raise RuntimeError("Public gateway requires AUTOML_REQUIRE_AUTH=1.")
    if not auth_storage.encryption_enabled():
        raise RuntimeError(
            "Public gateway requires AUTOML_MODEL_STORAGE_KEY for encrypted model storage."
        )


if __name__ == "__main__":
    validate_gateway_security_config()
    create_interface().queue(default_concurrency_limit=4).launch(
        server_name="0.0.0.0", server_port=7860, show_error=False
    )


__all__ = [
    "GATEWAY_CATALOG_PATH",
    "create_interface",
    "run_existing_remote",
    "run_uploaded_remote",
    "predict_remote",
    "render_training_result",
    "prune_gateway_egress",
    "routing_mode_from_env",
    "format_route_preview",
    "existing_route_preview",
    "uploaded_route_preview",
    "replay_route_preview",
    "validate_gateway_security_config",
]