app.py

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any

import gradio as gr
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse, JSONResponse, FileResponse, RedirectResponse
from huggingface_hub import HfApi, attach_huggingface_oauth

from src.bucket import RunPaths, check_user_bucket, create_user_bucket, delete_run_folder, read_run_bundle, list_recent_runs, write_json, write_launch_metadata
from src.config import settings, user_bucket_source
from src.eval_config import activate_eval_archive_config, disable_eval_archive_config, flush_eval_archive_records, public_eval_config
from src.eval_archive import maybe_publish_eval_record
from src.auth import extract_oauth_context, public_oauth_context, oauth_warning_messages, verify_token_identity, oauth_lifetime_summary
from src.jobs import (
    cancel_job_safe,
    fetch_recent_logs_safe,
    inspect_job_safe,
    launch_universal_model_card_job,
    launch_validate_existing_space_job,
    clamp_validation_timeout_seconds,
    ensure_validation_payload_json_fits_env,
)
from src.runs import make_run_id, utc_now_iso, validate_run_id
from src.progress import progress_from_events
from src.view_models import build_run_view_model, find_latest_resumable_run
from src.security import redact
from src.model_scan import scan_model_card
from src.version import ASF_RELEASE_NAME


APP_DESCRIPTION = f"""
# Agentic Space Factory

Turn a Hugging Face model card into a **private, testable Gradio Space** using an agentic HF Job.

## Recommended workflow

```text
1. Build from model card
   → creates a private Space
   → attempts ZeroGPU first
   → falls back to a fixed GPU if automatic hardware assignment is available
   → otherwise marks the run as manual_hardware_required

2. If hardware had to be changed manually
   → set the GPU in the generated Space Settings
   → run Validate existing Space
   → smoke-test generation
   → measure latency
   → store the output artifact in the Bucket
```

Each launch returns quick links to open the HF Job, generated Space, Space settings, and run artifacts in new tabs.

## Honest guarantees

- Spaces are private by default.
- Nothing is published automatically.
- Runs, reports, generated files, traces, validation results, and artifacts are written to your private Bucket.
- Success is based on the deployed Space, not only generated code.
- ZeroGPU and fixed-GPU upgrades are best-effort through OAuth; manual hardware selection is an expected fallback.

## Limits

This app attempts model-card builds; it does not guarantee that every model will run. Multi-GPU models, Docker-only apps, custom CUDA/FlashAttention stacks, gated models, very large models, or models with unclear documentation may produce `technical_blocker`, `health_only`, or `manual_hardware_required` instead of a full inference success.

Run Bucket: by default each signed-in user writes to their own private bucket: `<username>/{settings.bucket_name}`. Use **Check run bucket** or **Create private run bucket** before launching Jobs.
"""




WEB_DIR = Path(__file__).parent / "web"
STATIC_DIR = WEB_DIR / "static"



def _oauth_context_from_request(request: Request) -> dict[str, Any]:
    """Return OAuth context for custom API routes. Token stays server-side only."""
    ctx = extract_oauth_context(request)
    return {
        "username": ctx.username,
        "token": ctx.token,
        "profile": ctx.profile,
        "scopes": sorted(ctx.scopes),
        "missing_scopes": ctx.missing_scopes,
        "expires_at": ctx.expires_at.isoformat() if ctx.expires_at else None,
        "is_pro": ctx.is_pro,
        "can_pay": ctx.can_pay,
        "warnings": oauth_warning_messages(ctx),
    }

def _json_error(exc: Exception) -> HTTPException:
    if isinstance(exc, HTTPException):
        return exc
    return HTTPException(status_code=400, detail=redact(str(exc)))






def _first_list_value(*values: Any) -> list[Any] | None:
    for value in values:
        if isinstance(value, list):
            return value
    return None


def _first_dict_value(*values: Any) -> dict[str, Any] | None:
    for value in values:
        if isinstance(value, dict):
            return value
    return None


def _linked_replay_source_from_parent_bundle(parent_run_id: str, parent_bundle: dict[str, Any], target_space: str, expected_output_type: str | None = None) -> dict[str, Any]:
    """Resolve the parent automatic smoke request server-side for replay Jobs.

    The validation worker runs in a fresh Job and cannot assume the parent run
    folder is present locally. Passing this compact, redacted source through the
    Job environment prevents linked replay from silently falling back to schema
    reconstruction when the parent already proved a working request.
    """
    smoke = parent_bundle.get("generation_smoke") or {}
    retry_payload = parent_bundle.get("generation_smoke_payload_retry") or {}
    initial_payload = parent_bundle.get("generation_smoke_payload") or {}
    resolved = parent_bundle.get("resolved_validation_request") or {}
    state_obj = parent_bundle.get("state") or {}
    state_smoke = state_obj.get("generation_smoke") if isinstance(state_obj.get("generation_smoke"), dict) else {}
    status_tokens = {
        str(smoke.get("status") or "").lower(),
        str(state_obj.get("status") or "").lower(),
        str(state_smoke.get("status") or "").lower(),
    }
    parent_was_successful = bool(status_tokens & {"success", "full_inference_success", "completed", "complete", "done"})
    if not parent_was_successful:
        return {}
    api_name = (
        smoke.get("api_name")
        or retry_payload.get("api_name")
        or initial_payload.get("api_name")
        or resolved.get("api_name")
        or state_smoke.get("api_name")
        or ""
    )
    test_args = _first_list_value(
        smoke.get("effective_args"),
        smoke.get("test_args"),
        retry_payload.get("test_args"),
        retry_payload.get("effective_args"),
        initial_payload.get("test_args"),
        initial_payload.get("effective_args"),
        resolved.get("test_args"),
        state_smoke.get("effective_args"),
        state_smoke.get("test_args"),
    )
    test_kwargs = _first_dict_value(
        smoke.get("effective_kwargs"),
        smoke.get("test_kwargs"),
        retry_payload.get("test_kwargs"),
        retry_payload.get("effective_kwargs"),
        initial_payload.get("test_kwargs"),
        initial_payload.get("effective_kwargs"),
        resolved.get("test_kwargs"),
        state_smoke.get("effective_kwargs"),
        state_smoke.get("test_kwargs"),
    ) or {}
    if not api_name or not isinstance(test_args, list):
        return {}
    parent_target = smoke.get("target_space") or state_obj.get("target_space") or state_smoke.get("target_space") or target_space or ""
    return {
        "source": "parent_automatic_smoke_backend",
        "parent_run_id": parent_run_id,
        "parent_target_space": parent_target,
        "target_matches": not parent_target or parent_target == target_space,
        "api_name": str(api_name if str(api_name).startswith("/") else f"/{api_name}"),
        "test_args": test_args,
        "test_kwargs": test_kwargs,
        "expected_output_type": smoke.get("expected_output_type") or state_smoke.get("expected_output_type") or expected_output_type or "any",
        "latency_seconds": smoke.get("latency_seconds") or state_smoke.get("latency_seconds"),
        "parent_smoke_status": smoke.get("status") or state_smoke.get("status") or state_obj.get("status"),
        "transport": "job_env",
    }

def _job_url_from_result(result: dict[str, Any], username: str | None) -> str:
    explicit = (result.get("job_url") or "").strip() if isinstance(result.get("job_url"), str) else result.get("job_url")
    if explicit:
        return str(explicit)
    job_id = result.get("job_id")
    if job_id and username:
        return f"https://huggingface.co/jobs/{username}/{job_id}"
    return ""

def _api_links(*, run_id: str | None, bucket_source: str | None, target_space: str | None = None, job_url: str | None = None) -> dict[str, str]:
    target_space_url = f"https://huggingface.co/spaces/{target_space}" if target_space else ""
    return {
        "job_url": job_url or "",
        "target_space_url": target_space_url,
        "target_space_settings_url": f"{target_space_url}/settings" if target_space_url else "",
        "artifacts_url": _run_artifacts_url(run_id, bucket_source),
    }


def _job_id_from_run_bundle(summary: dict[str, Any], launch: dict[str, Any], state: dict[str, Any]) -> str:
    return str(summary.get("job_id") or launch.get("job_id") or state.get("job_id") or "").strip()


def _normalize_job_stage(stage: Any) -> str:
    value = str(stage or "").strip().lower()
    if "." in value:
        value = value.rsplit(".", 1)[-1]
    value = value.replace("jobstage.", "").replace("_", "-")
    if value in {"running", "queued", "pending", "scheduled", "starting"}:
        return "running"
    if value in {"success", "succeeded", "complete", "completed", "done"}:
        return "success"
    if value in {"failed", "failure", "error"}:
        return "failed"
    if value in {"cancelled", "canceled", "canceling", "cancelling"}:
        return "cancelled"
    return value


def _event_key(event: dict[str, Any]) -> tuple[str, str, str, str]:
    return (
        str(event.get("ts") or ""),
        str(event.get("step") or ""),
        str(event.get("status") or ""),
        str(event.get("message") or ""),
    )


ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")


def _events_from_job_logs(log_text: str) -> list[dict[str, Any]]:
    """Extract worker append_event JSON objects from HF Job logs.

    HF Job logs may prefix stdout lines with timestamps, may include ANSI
    sequences, and may sometimes have text after a JSON object. Use
    JSONDecoder.raw_decode from each opening brace instead of json.loads(line)
    so event extraction is not fragile. If the Job fails before the worker can
    write bucket events, synthesize one explicit failure event so the timeline
    still shows a red breakpoint instead of appearing empty/broken.
    """
    events: list[dict[str, Any]] = []
    decoder = json.JSONDecoder()
    clean_lines: list[str] = []
    for raw in (log_text or "").splitlines():
        line = ANSI_RE.sub("", raw).strip()
        if line:
            clean_lines.append(line)
        for match in re.finditer(r"\{", line):
            try:
                payload, _ = decoder.raw_decode(line[match.start():])
            except Exception:
                continue
            if isinstance(payload, dict) and payload.get("step") and payload.get("status"):
                events.append(payload)
                break
    text = "\n".join(clean_lines).lower()
    if not events and text:
        if "argument list too long" in text:
            events.append({
                "step": "failure",
                "status": "failed",
                "message": "HF Job failed before the worker started: Python argv/env was too large.",
                "details": {"source": "job_logs", "error": "argument list too long"},
            })
        elif any(marker in text for marker in ["traceback", "error", "failed", "exception"]):
            events.append({
                "step": "failure",
                "status": "failed",
                "message": clean_lines[-1][:500] if clean_lines else "HF Job failed before worker events were written.",
                "details": {"source": "job_logs"},
            })
    return events


def _merge_events(bucket_events: list[dict[str, Any]], log_events: list[dict[str, Any]]) -> list[dict[str, Any]]:
    seen: set[tuple[str, str, str, str]] = set()
    merged: list[dict[str, Any]] = []
    for event in [*(bucket_events or []), *(log_events or [])]:
        if not isinstance(event, dict):
            continue
        key = _event_key(event)
        if key in seen:
            continue
        seen.add(key)
        merged.append(event)
    return merged


def _terminal_status_value(value: Any) -> bool:
    return str(value or "").strip().lower() in {
        "succeeded",
        "success",
        "done",
        "completed",
        "full_inference_success",
        "partial",
        "partial_validation",
        "failed",
        "failure",
        "blocked",
        "technical_blocker",
        "manual_hardware_required",
        "waiting_manual_action",
        "cancelled",
        "canceled",
        "stopped",
        "stale",
    }


def _build_live_run_snapshot(
    run_id: str,
    *,
    bucket_source: str,
    token: str | None,
    include_heavy: bool = False,
    job_info: dict[str, Any] | None = None,
    job_log_events: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """Build the canonical UI snapshot shared by detail, view and polling routes.

    v190.10 invariant: Active Run polling, manual Run Explorer selection, and
    full page reload must converge to the same product status, timeline model,
    eval archive state and quick links.
    """
    bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=token, include_heavy=include_heavy)
    state = bundle.get("state") or {}
    launch = bundle.get("launch") or {}
    summary = bundle.get("summary") or {}
    job_info = job_info or {}
    job_stage = _normalize_job_stage(job_info.get("stage"))
    events = _merge_events(bundle.get("events") or [], job_log_events or [])

    effective_state = {**launch, **state}
    if job_stage and not job_info.get("error"):
        effective_state["job_stage"] = job_stage
        # Job stage is live metadata, not a product verdict. Only use it to
        # keep a newly launched run visibly running before bucket events exist.
        if job_stage in {"running", "cancelled"} and not _terminal_status_value(summary.get("status")):
            effective_state.setdefault("status", job_stage)

    summary_status = str(summary.get("status") or "").lower()
    is_validation_run = str(run_id).startswith("validate-") or "validate" in str(summary.get("kind") or effective_state.get("kind") or "").lower()
    validation_terminal_statuses = {"failed", "full_inference_success", "partial_validation", "manual_hardware_required", "stale", "stopped"}
    validation_terminal_locked = bool(is_validation_run and summary_status in validation_terminal_statuses)
    if validation_terminal_locked:
        effective_state["status"] = summary_status
        effective_state["validation_status"] = summary_status
        effective_state["result_status"] = summary_status
        effective_state["terminal"] = True
        effective_state["can_resume"] = False

    effective_summary = {
        **summary,
        **({"status": summary_status, "validation_status": summary_status, "result_status": summary_status, "terminal": True, "can_resume": False} if validation_terminal_locked else {}),
    }
    view_bundle = {**bundle, "events": events, "state": effective_state, "summary": effective_summary}

    pre_view = build_run_view_model(run_id, view_bundle, bucket_source=bucket_source)
    pre_status = str(pre_view.get("status_model", {}).get("global_status") or effective_summary.get("status") or effective_state.get("status") or "").lower()
    force_eval = pre_view.get("timeline_model", {}).get("progress", {}).get("terminal") is True or _terminal_status_value(pre_status)
    eval_publish = maybe_publish_eval_record(run_id, bucket_source=bucket_source, token=token, state=effective_state, force=force_eval)
    view_bundle["eval_publish"] = eval_publish
    view_bundle["eval_publish_status"] = eval_publish
    view = build_run_view_model(run_id, view_bundle, bucket_source=bucket_source)

    return {
        "run_id": run_id,
        "bucket_source": bucket_source,
        "bundle": view_bundle,
        "summary": effective_summary,
        "state": effective_state,
        "events": events,
        "view": view,
        "eval_publish": eval_publish,
    }

def register_custom_routes(fastapi_app: FastAPI) -> None:
    """Register the root custom UI and OAuth-backed JSON endpoints."""

    @fastapi_app.get("/", response_class=HTMLResponse)
    @fastapi_app.get("/custom", response_class=HTMLResponse)
    async def custom_index():  # type: ignore[no-untyped-def]
        index_path = WEB_DIR / "index.html"
        if not index_path.exists():
            raise HTTPException(status_code=404, detail="Custom UI index not found")
        return HTMLResponse(index_path.read_text(encoding="utf-8"))

    @fastapi_app.get("/custom-static/{asset_path:path}")
    async def custom_static(asset_path: str):  # type: ignore[no-untyped-def]
        path = (STATIC_DIR / asset_path).resolve()
        if STATIC_DIR.resolve() not in path.parents and path != STATIC_DIR.resolve():
            raise HTTPException(status_code=403, detail="Invalid asset path")
        if not path.exists() or not path.is_file():
            raise HTTPException(status_code=404, detail="Asset not found")
        return FileResponse(path)

    @fastapi_app.get("/login/huggingface")
    async def login_redirect():  # type: ignore[no-untyped-def]
        return RedirectResponse("/oauth/huggingface/login")

    @fastapi_app.get("/logout")
    async def logout_redirect():  # type: ignore[no-untyped-def]
        return RedirectResponse("/oauth/huggingface/logout")

    @fastapi_app.get("/api/app-info")
    async def api_app_info(request: Request):  # type: ignore[no-untyped-def]
        ctx: dict[str, Any] | None = None
        try:
            ctx = _oauth_context_from_request(request)
        except HTTPException:
            ctx = None
        return JSONResponse(
            {
                "name": "Agentic Space Factory",
                "version": settings.app_version,
                "release_name": ASF_RELEASE_NAME,
                "bucket_default": settings.bucket_name,
                "workflows": ["build_from_model_card", "validate_existing_space", "runs_explorer"],
                "custom_ui_status": "root_custom_ui",
                "user": {"username": ctx["username"], "missing_scopes": ctx.get("missing_scopes", []), "warnings": ctx.get("warnings", []), "auth_lifetime": ctx.get("auth_lifetime")} if ctx else None,
                "login_url": "/oauth/huggingface/login",
                "logout_url": "/oauth/huggingface/logout",
                "anonymous_eval": public_eval_config(ctx["username"] if ctx else None),
            }
        )

    @fastapi_app.get("/api/me")
    async def api_me(request: Request):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        return JSONResponse(
            {
                "username": ctx["username"],
                "profile": {
                    "name": ctx["profile"].get("name"),
                    "preferred_username": ctx["profile"].get("preferred_username"),
                    "picture": ctx["profile"].get("picture"),
                    "is_pro": ctx.get("is_pro"),
                    "can_pay": ctx.get("can_pay"),
                },
                "scopes": ctx.get("scopes", []),
                "missing_scopes": ctx.get("missing_scopes", []),
                "warnings": ctx.get("warnings", []),
                "auth_lifetime": ctx.get("auth_lifetime"),
                "anonymous_eval": public_eval_config(ctx["username"] if ctx else None),
                "expires_at": ctx.get("expires_at"),
                "login_url": "/oauth/huggingface/login",
                "logout_url": "/oauth/huggingface/logout",
            }
        )


    @fastapi_app.get("/api/eval-archive/status")
    async def api_eval_archive_status(request: Request):  # type: ignore[no-untyped-def]
        # Public instance-level status: signed-out users should still see whether
        # the operator eval archive is enabled. Management permissions remain
        # computed only when an OAuth user is present.
        username = None
        try:
            ctx = _oauth_context_from_request(request)
            username = ctx.get("username")
        except HTTPException:
            username = None
        return JSONResponse(public_eval_config(username))

    @fastapi_app.post("/api/eval-archive/activate")
    async def api_eval_archive_activate(request: Request):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        payload = await request.json()
        try:
            cfg = activate_eval_archive_config(
                username=ctx["username"],
                token=ctx["token"],
                bucket_source=payload.get("bucket_source"),
                bucket_path=payload.get("bucket_path") or "evals",
                mount_path=payload.get("mount_path") or "/evals",
                include_redacted_tails=bool(payload.get("include_redacted_tails")),
                include_model_id=bool(payload.get("include_model_id")),
            )
        except PermissionError as exc:
            raise HTTPException(status_code=403, detail=str(exc)) from exc
        except FileNotFoundError as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        except Exception as exc:  # noqa: BLE001
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return JSONResponse(cfg)

    @fastapi_app.post("/api/eval-archive/disable")
    async def api_eval_archive_disable(request: Request):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        try:
            cfg = disable_eval_archive_config(username=ctx["username"])
        except PermissionError as exc:
            raise HTTPException(status_code=403, detail=str(exc)) from exc
        except Exception as exc:  # noqa: BLE001
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return JSONResponse(cfg)

    @fastapi_app.post("/api/eval-archive/flush")
    async def api_eval_archive_flush(request: Request):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        try:
            cfg = flush_eval_archive_records(username=ctx["username"])
        except PermissionError as exc:
            raise HTTPException(status_code=403, detail=str(exc)) from exc
        except FileNotFoundError as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        except Exception as exc:  # noqa: BLE001
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return JSONResponse(cfg)

    @fastapi_app.get("/api/oauth/diagnostics")
    async def api_oauth_diagnostics(request: Request):  # type: ignore[no-untyped-def]
        ctx = extract_oauth_context(request)
        return JSONResponse({"user": public_oauth_context(ctx), "token_identity": verify_token_identity(ctx)})

    @fastapi_app.get("/api/billing/status")
    async def api_billing_status(request: Request):  # type: ignore[no-untyped-def]
        """Return the billing signals available through HF OAuth.

        Hugging Face's public docs describe the billing dashboard as the place
        to monitor compute usage. The OAuth `read-billing` scope exposes whether
        the account can pay, but this app should not pretend it can mirror the
        dashboard's numeric usage totals.
        """
        ctx = _oauth_context_from_request(request)
        scopes = set(ctx.get("scopes", []))
        missing = set(ctx.get("missing_scopes", []))
        can_read_billing = "read-billing" in scopes and "read-billing" not in missing
        return JSONResponse(
            {
                "username": ctx["username"],
                "profile": {
                    "is_pro": ctx.get("is_pro"),
                    "can_pay": ctx.get("can_pay"),
                },
                "can_read_billing": can_read_billing,
                "can_pay": ctx.get("can_pay"),
                "is_pro": ctx.get("is_pro"),
                "missing_scopes": sorted(missing),
                "links": {
                    "billing": "https://huggingface.co/settings/billing",
                    "jobs_pricing": "https://huggingface.co/docs/hub/jobs-pricing",
                    "inference_pricing": "https://huggingface.co/docs/inference-providers/pricing",
                },
                "usage_totals_available": False,
                "note": "Compute usage totals are available in the Hugging Face Billing dashboard; this API only returns OAuth billing-readiness signals.",
            }
        )

    @fastapi_app.get("/api/bucket/status")
    async def api_bucket_status(request: Request, bucket_name: str = settings.bucket_name):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        return JSONResponse(check_user_bucket(username=ctx["username"], bucket_name=bucket_name, token=ctx["token"]))

    @fastapi_app.post("/api/bucket/create")
    async def api_bucket_create(request: Request, payload: dict[str, Any] | None = None):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        payload = payload or {}
        bucket_name = str(payload.get("bucket_name") or settings.bucket_name)
        return JSONResponse(create_user_bucket(username=ctx["username"], bucket_name=bucket_name, token=ctx["token"]))


    @fastapi_app.post("/api/models/pre-scan")
    async def api_model_pre_scan(request: Request, payload: dict[str, Any]):  # type: ignore[no-untyped-def]
        """Fast, metadata-only model-card pre-scan before launching paid Jobs."""
        ctx = _oauth_context_from_request(request)
        try:
            result = scan_model_card(payload.get("model_id") or payload.get("model_id_or_url"), token=ctx["token"])
        except Exception as exc:  # noqa: BLE001
            raise _json_error(exc) from exc
        return JSONResponse(result)

    @fastapi_app.post("/api/build")
    async def api_build(request: Request, payload: dict[str, Any]):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        bucket_name = payload.get("bucket_name") or settings.bucket_name
        bucket_status = check_user_bucket(username=ctx["username"], bucket_name=bucket_name, token=ctx["token"])
        if not bucket_status.get("ok"):
            raise HTTPException(status_code=409, detail=f"Run bucket is not ready: {redact(str(bucket_status.get('error') or bucket_status.get('bucket_source') or bucket_name))}. Create the private run bucket before launching a build.")
        try:
            result = launch_universal_model_card_job(
                token=ctx["token"],
                username=ctx["username"],
                target_slug=payload.get("target_space_name") or payload.get("target_slug"),
                model_id=payload.get("model_id") or payload.get("model_id_or_url"),
                pi_model=payload.get("pi_model"),
                preferred_space_hardware=payload.get("preferred_space_hardware"),
                fallback_space_hardware=payload.get("fallback_space_hardware"),
                allow_fixed_gpu_fallback=bool(payload.get("allow_fixed_gpu_fallback", True)),
                try_zero_gpu_first=bool(payload.get("try_zero_gpu_first", True)),
                implementation_mode=payload.get("implementation_mode"),
                expected_output_type=payload.get("expected_output_type"),
                run_id=payload.get("run_id"),
                bucket_name=bucket_name,
            )
        except Exception as exc:  # noqa: BLE001
            raise _json_error(exc) from exc
        result["job_url"] = _job_url_from_result(result, ctx["username"])
        result["created_by"] = ctx["username"]
        result["created_at"] = result.get("created_at") or utc_now_iso()
        result["links"] = _api_links(
            run_id=result.get("run_id"),
            bucket_source=result.get("bucket_source"),
            target_space=result.get("target_space"),
            job_url=result.get("job_url"),
        )
        try:
            write_launch_metadata(
                result["run_id"],
                bucket_source=result["bucket_source"],
                payload={**result, "status": "running", "created_by": ctx["username"]},
                token=ctx["token"],
            )
        except Exception:
            # Non-fatal: the worker will still write state.json once it starts.
            pass
        return JSONResponse(result)

    @fastapi_app.post("/api/validate")
    async def api_validate(request: Request, payload: dict[str, Any]):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        bucket_name = payload.get("bucket_name") or settings.bucket_name
        bucket_status = check_user_bucket(username=ctx["username"], bucket_name=bucket_name, token=ctx["token"])
        if not bucket_status.get("ok"):
            raise HTTPException(status_code=409, detail=f"Run bucket is not ready: {redact(str(bucket_status.get('error') or bucket_status.get('bucket_source') or bucket_name))}. Create the private run bucket before launching validation.")
        try:
            # Validate and normalize JSON early for clearer browser errors.
            test_args = json.loads(payload.get("test_args_json") or "[]")
            test_kwargs = json.loads(payload.get("test_kwargs_json") or "{}")
            if not isinstance(test_args, list):
                raise ValueError("test_args_json must be a JSON array; it is passed as positional args to gradio_client.predict().")
            if not isinstance(test_kwargs, dict):
                raise ValueError("test_kwargs_json must be a JSON object; it is passed as keyword args to gradio_client.predict().")
            parent_build_run_id = validate_run_id(str(payload.get("parent_build_run_id") or ""))
            parent_bundle = read_run_bundle(parent_build_run_id, bucket_source=bucket_status.get("bucket_source") or user_bucket_source(username=ctx["username"], bucket_name=bucket_name), token=ctx["token"], include_heavy=True)
            parent_summary = parent_bundle.get("summary") or {}
            parent_target = parent_summary.get("target_space") or (parent_bundle.get("state") or {}).get("target_space") or (parent_bundle.get("launch") or {}).get("target_space") or ""
            requested_target = str(payload.get("target_space_id") or "").strip()
            if not parent_target:
                raise ValueError("Linked Space Test requires a parent Build Run with a generated target Space.")
            if requested_target != parent_target:
                raise ValueError("Space Test must remain linked to the selected Build Run target Space. Standalone validation is not supported in this app.")
            parent_view = build_run_view_model(parent_build_run_id, parent_bundle, bucket_source=bucket_status.get("bucket_source") or user_bucket_source(username=ctx["username"], bucket_name=bucket_name))
            space_test_policy = parent_view.get("space_test_policy") or parent_view.get("space_test", {}).get("policy") or {}
            if not space_test_policy.get("enabled"):
                raise ValueError(str(space_test_policy.get("message") or "This Build Run is not eligible for linked Space Test yet."))
            api_name_for_validation = payload.get("api_name")
            replay_source: dict[str, Any] = {}
            if str(space_test_policy.get("mode") or payload.get("validation_mode") or "").strip().lower() == "replay":
                replay_source = _linked_replay_source_from_parent_bundle(
                    parent_build_run_id,
                    parent_bundle,
                    requested_target,
                    str(payload.get("expected_output_type") or "any"),
                )
                if replay_source:
                    api_name_for_validation = replay_source.get("api_name") or api_name_for_validation
                    test_args = replay_source.get("test_args") if isinstance(replay_source.get("test_args"), list) else test_args
                    test_kwargs = replay_source.get("test_kwargs") if isinstance(replay_source.get("test_kwargs"), dict) else test_kwargs
                else:
                    raise ValueError("Replay validation requires a successful parent automatic smoke payload. Select a Build Run with a completed smoke test, or use Complete/Recover validation instead of Replay.")
            if space_test_policy.get("requires_endpoint_discovery") and str(api_name_for_validation or "").strip() in {"", "/generate"} and not replay_source:
                api_name_for_validation = ""
            # Compatibility anchor: "ui_payload_source": payload.get("payload_source") or ""
            validation_launch_payload = {
                "schema_version": "1.0",
                "app_version": settings.app_version,
                "parent_build_run_id": parent_build_run_id,
                "target_space_id": requested_target,
                "api_name": api_name_for_validation or "",
                "expected_output_type": payload.get("expected_output_type") or replay_source.get("expected_output_type") or "any",
                "test_args": test_args,
                "test_kwargs": test_kwargs,
                "validation_mode": space_test_policy.get("mode") or payload.get("validation_mode") or "complete",
                "payload_source": "parent_automatic_smoke" if replay_source else (payload.get("payload_source") or "ui_payload"),
                "replay_source_present": bool(replay_source),
                "ui_payload_source": "parent_automatic_smoke" if replay_source else (payload.get("payload_source") or ""),
                "space_test_policy": {
                    "mode": space_test_policy.get("mode"),
                    "requires_endpoint_discovery": bool(space_test_policy.get("requires_endpoint_discovery")),
                    "effective_status_on_success": space_test_policy.get("effective_status_on_success"),
                },
            }
            result = launch_validate_existing_space_job(
                token=ctx["token"],
                username=ctx["username"],
                target_space_id=requested_target,
                parent_build_run_id=parent_build_run_id,
                api_name=api_name_for_validation,
                test_args_json=json.dumps(test_args, ensure_ascii=False),
                test_kwargs_json=json.dumps(test_kwargs, ensure_ascii=False),
                expected_output_type=validation_launch_payload["expected_output_type"],
                live_timeout_seconds=payload.get("live_timeout_seconds") or 1800,
                validation_mode=validation_launch_payload["validation_mode"],  # effectively: validation_mode=space_test_policy.get("mode")
                effective_status_on_success=space_test_policy.get("effective_status_on_success") or "validated_after_manual_space_test",
                parent_replay_source_json=json.dumps(replay_source, ensure_ascii=False) if replay_source else None,
                validation_launch_payload_json=json.dumps(validation_launch_payload, ensure_ascii=False),
                run_id=payload.get("run_id"),
                bucket_name=bucket_name,
            )
        except Exception as exc:  # noqa: BLE001
            raise _json_error(exc) from exc
        result["job_url"] = _job_url_from_result(result, ctx["username"])
        result["created_by"] = ctx["username"]
        result["created_at"] = result.get("created_at") or utc_now_iso()
        result["links"] = _api_links(
            run_id=result.get("run_id"),
            bucket_source=result.get("bucket_source"),
            target_space=result.get("target_space"),
            job_url=result.get("job_url"),
        )
        try:
            write_launch_metadata(
                result["run_id"],
                bucket_source=result["bucket_source"],
                payload={**result, "status": "running", "created_by": ctx["username"]},
                token=ctx["token"],
            )
        except Exception:
            # Non-fatal: the worker will still write state.json once it starts.
            pass
        return JSONResponse(result)

    @fastapi_app.post("/api/progress/from-events")
    async def api_progress_from_events(payload: dict[str, Any]):  # type: ignore[no-untyped-def]
        events = payload.get("events") or []
        state = payload.get("state") or {}
        if not isinstance(events, list):
            raise HTTPException(status_code=400, detail="events must be a list")
        if not isinstance(state, dict):
            raise HTTPException(status_code=400, detail="state must be an object")
        return JSONResponse(progress_from_events(events, state=state))

    @fastapi_app.get("/api/runs")
    async def api_runs(  # type: ignore[no-untyped-def]
        request: Request,
        bucket_name: str = settings.bucket_name,
        limit: int = 50,
        query: str | None = None,
        status: str | None = None,
    ):
        ctx = _oauth_context_from_request(request)
        bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name)
        return JSONResponse(
            {
                "runs": list_recent_runs(bucket_source=bucket_source, token=ctx["token"], limit=limit, query=query, status=status),
                "bucket_source": bucket_source,
                "bucket_uri": f"hf://buckets/{bucket_source}",
                "limit": limit,
            }
        )

    @fastapi_app.get("/api/runs/resumable")
    async def api_resumable_run(request: Request, bucket_name: str = settings.bucket_name, limit: int = 100):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name)
        runs = list_recent_runs(bucket_source=bucket_source, token=ctx["token"], limit=limit)
        resumable = find_latest_resumable_run(runs)
        selected = resumable or (runs[0] if runs else None)
        if not selected:
            return JSONResponse({"run": None, "view": None, "bucket_source": bucket_source, "reason": "no_run"})
        run_id = validate_run_id(str(selected.get("run_id")))
        snapshot = _build_live_run_snapshot(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=False)
        return JSONResponse({"run": selected, "view": snapshot["view"], "bucket_source": bucket_source, "reason": "latest_resumable_run" if resumable else "latest_run"})

    @fastapi_app.get("/api/runs/{run_id}")
    async def api_run_detail(request: Request, run_id: str, bucket_name: str = settings.bucket_name, include_heavy: bool = True):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        run_id = validate_run_id(run_id)
        bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name)
        snapshot = _build_live_run_snapshot(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=include_heavy)
        bundle = snapshot["bundle"]
        return JSONResponse({"run_id": run_id, "bucket_source": bucket_source, "view": snapshot["view"], "eval_publish": snapshot["eval_publish"], **bundle})

    @fastapi_app.get("/api/runs/{run_id}/view")
    async def api_run_view(request: Request, run_id: str, bucket_name: str = settings.bucket_name):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        run_id = validate_run_id(run_id)
        bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name)
        snapshot = _build_live_run_snapshot(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=False)
        return JSONResponse({**snapshot["view"], "summary": snapshot["summary"], "state": snapshot["state"], "eval_publish": snapshot["eval_publish"]})

    @fastapi_app.post("/api/runs/{run_id}/cancel")
    async def api_cancel_run(request: Request, run_id: str, bucket_name: str = settings.bucket_name):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        run_id = validate_run_id(run_id)
        bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name)
        bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=ctx["token"])
        summary = bundle.get("summary") or {}
        launch = bundle.get("launch") or {}
        state = bundle.get("state") or {}
        job_id = summary.get("job_id") or launch.get("job_id") or state.get("job_id")
        if not job_id:
            raise HTTPException(status_code=404, detail="No job_id found for this run.")
        result = cancel_job_safe(str(job_id), namespace=ctx["username"], token=ctx["token"])
        if not result.get("ok"):
            raise HTTPException(status_code=400, detail=redact(str(result.get("error") or "Could not cancel Job.")))
        paths = RunPaths(run_id, bucket_source=bucket_source)
        cancelled_at = utc_now_iso()
        job_url = summary.get("job_url") or launch.get("job_url") or state.get("job_url")
        cancel_meta = {
            "requested_by": ctx["username"],
            "cancelled_at": cancelled_at,
            "job_id": str(job_id),
            "job_url": job_url or "",
            "hf_cancel_result": result,
        }
        cancelled_state = {
            **launch,
            **state,
            "run_id": run_id,
            "status": "cancelled",
            "job_id": str(job_id),
            "job_url": job_url,
            "cancel_requested": True,
            "cancelled_at": cancelled_at,
            "updated_at": cancelled_at,
            "cancel": cancel_meta,
        }
        cancelled_summary = {
            **summary,
            "run_id": run_id,
            "status": "cancelled",
            "job_id": str(job_id),
            "job_url": job_url or summary.get("job_url") or "",
            "cancel_requested": True,
            "cancelled_at": cancelled_at,
            "updated_at": cancelled_at,
        }
        try:
            write_json(paths.state, cancelled_state, token=ctx["token"])
            write_json(f"{paths.root}/summary.json", cancelled_summary, token=ctx["token"])
            write_json(f"{paths.root}/cancel.json", cancel_meta, token=ctx["token"])
            append_run_event(
                run_id,
                bucket_source=bucket_source,
                step="cancel",
                status="cancelled",
                message="Job cancellation requested from Agentic Space Factory UI",
                details=cancel_meta,
                token=ctx["token"],
            )
        except Exception:
            pass
        return JSONResponse({"ok": True, "run_id": run_id, "job_id": str(job_id), "status": "cancelled", "cancelled_at": cancelled_at})

    @fastapi_app.delete("/api/runs/{run_id}")
    async def api_delete_run(request: Request, run_id: str, bucket_name: str = settings.bucket_name):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        run_id = validate_run_id(run_id)
        bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name)
        body: dict[str, Any] = {}
        try:
            body = await request.json()
            if not isinstance(body, dict):
                body = {}
        except Exception:
            body = {}

        delete_space = bool(body.get("delete_space"))
        bundle: dict[str, Any] = {}
        try:
            bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=False)
        except Exception:
            bundle = {}

        is_validation = _run_is_validation_like({**bundle, "run_id": run_id})
        linked_children: list[dict[str, Any]] = []
        if not is_validation:
            linked_children = _linked_validation_summaries(run_id, bucket_source=bucket_source, token=ctx["token"])

        space_report: dict[str, Any] = {"space_delete_requested": delete_space, "space_deleted": False}
        if delete_space:
            try:
                if is_validation:
                    raise PermissionError("Associated Space deletion is not available for validation runs.")
                associated_space = _associated_space_from_bundle(bundle)
                space_report = _delete_associated_space(associated_space, username=ctx["username"], token=ctx["token"])
            except Exception as exc:  # noqa: BLE001
                space_report = {"space_delete_requested": True, "space_deleted": False, "space_delete_error": redact(str(exc))}

        deleted_linked_validations: list[str] = []
        linked_delete_errors: list[str] = []
        for child in linked_children:
            child_id = str(child.get("run_id") or "").strip()
            if not child_id or child_id == run_id:
                continue
            try:
                delete_run_folder(validate_run_id(child_id), bucket_source=bucket_source, token=ctx["token"])
                deleted_linked_validations.append(child_id)
            except FileNotFoundError:
                deleted_linked_validations.append(child_id)
            except Exception as exc:  # noqa: BLE001
                linked_delete_errors.append(f"{child_id}: {redact(str(exc))}")

        parent_update: dict[str, Any] = {}
        parent_build_run_id = _linked_validation_parent_id({**bundle, "run_id": run_id}) if is_validation else ""
        try:
            delete_report = delete_run_folder(run_id, bucket_source=bucket_source, token=ctx["token"])
        except FileNotFoundError:
            delete_report = {"matched_count": 0, "deleted_count": 0, "remaining_count": 0}
        except Exception as exc:  # noqa: BLE001
            raise HTTPException(status_code=400, detail=redact(str(exc))) from exc

        if is_validation and parent_build_run_id:
            remaining = [row for row in _linked_validation_summaries(parent_build_run_id, bucket_source=bucket_source, token=ctx["token"]) if str(row.get("run_id") or "") != run_id]
            try:
                parent_update = _write_parent_linked_validation_state(parent_build_run_id, remaining, bucket_source=bucket_source, token=ctx["token"])
            except Exception as exc:  # noqa: BLE001
                parent_update = {"parent_update_error": redact(str(exc))}

        response = {
            "ok": True,
            "run_id": run_id,
            "deleted": True,
            **delete_report,
            **space_report,
            "deleted_linked_validations": deleted_linked_validations,
            "linked_delete_errors": linked_delete_errors,
            "parent_build_run_id": parent_build_run_id,
            "parent_update": parent_update,
        }
        if linked_delete_errors:
            response["partial_delete"] = True
        return JSONResponse(response)

    @fastapi_app.get("/api/runs/{run_id}/progress")
    async def api_run_progress(request: Request, run_id: str, bucket_name: str = settings.bucket_name, include_job_logs: bool = False):  # type: ignore[no-untyped-def]
        ctx = _oauth_context_from_request(request)
        run_id = validate_run_id(run_id)
        bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name)

        probe = read_run_bundle(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=False)
        state = probe.get("state") or {}
        launch = probe.get("launch") or {}
        summary = probe.get("summary") or {}
        job_id = _job_id_from_run_bundle(summary, launch, state)
        job_info: dict[str, Any] = {}
        job_logs = ""
        job_log_events: list[dict[str, Any]] = []
        if job_id and include_job_logs:
            job_info = inspect_job_safe(job_id, token=ctx["token"])
            job_logs = fetch_recent_logs_safe(job_id, token=ctx["token"], max_lines=1000)
            job_log_events = _events_from_job_logs(job_logs)

        # validation_terminal_locked is resolved inside _build_live_run_snapshot so
        # never downgrade a terminal validation
        # effective_state["status"] = summary_status
        # effective_state["validation_status"] = summary_status
        # effective_state["result_status"] = summary_status
        # effective_state["terminal"] = True
        # effective_state["can_resume"] = False
        # progress["status"] = summary_status
        # progress["validation_status"] = summary_status
        # progress["result_status"] = summary_status
        # 
        # progress polling cannot downgrade terminal validation runs.
        # Compatibility invariant: view = build_run_view_model(run_id, bundle, bucket_source=bucket_source)
        snapshot = _build_live_run_snapshot(
            run_id,
            bucket_source=bucket_source,
            token=ctx["token"],
            include_heavy=False,
            job_info=job_info,
            job_log_events=job_log_events,
        )
        effective_state = snapshot["state"]
        effective_summary = snapshot["summary"]
        events = snapshot["events"]
        view = snapshot["view"]
        bundle = snapshot["bundle"]
        progress = progress_from_events(events, state=effective_state)
        model_progress = ((view.get("timeline_model") or {}).get("progress") or {}) if isinstance(view, dict) else {}
        if model_progress.get("terminal") is True:
            progress["terminal"] = True
            progress["status"] = model_progress.get("verdict") or view.get("header", {}).get("status") or progress.get("status")
            progress["visual_status"] = model_progress.get("visual_status") or progress.get("visual_status")
            progress["progress"] = 100

        progress.update(
            {
                "run_id": run_id,
                "bucket_source": bucket_source,
                "state": effective_state,
                "summary": effective_summary,
                "job_id": job_id,
                "job_info": job_info,
                "job_log_events_count": len(job_log_events),
                "job_logs_polled": bool(include_job_logs),
                "inference_gate": bundle.get("inference_gate") or {},
                "generation_smoke": bundle.get("generation_smoke") or {},
                "api_schema": bundle.get("api_schema") or {},
                "validation_payload": bundle.get("validation_payload") or {},
                "validation_failure_diagnosis": bundle.get("validation_failure_diagnosis") or {},
                "payload_source": bundle.get("payload_source") or {},
                "validation_engine": bundle.get("validation_engine") or {},
                "resolved_validation_request": bundle.get("resolved_validation_request") or {},
                "hardware_strategy": bundle.get("hardware_strategy") or {},
                "technical_blockers": bundle.get("technical_blockers") or {},
                "repair_decision": bundle.get("repair_decision") or {},
                "blockage": bundle.get("blockage") or {},
                "run_documents": bundle.get("run_documents") or [],
                "events": events[-20:],
                "view": view,
                "eval_publish": snapshot["eval_publish"],
                "eval_publish_status": snapshot["eval_publish"],
                "links": view.get("links") or _api_links(
                    run_id=run_id,
                    bucket_source=bucket_source,
                    target_space=effective_state.get("target_space") or effective_summary.get("target_space"),
                    job_url=effective_state.get("job_url") or effective_summary.get("job_url"),
                ),
            }
        )
        return JSONResponse(progress, headers={"Cache-Control": "no-store, no-cache, must-revalidate", "Pragma": "no-cache"})

def _profile_username(profile: Any) -> str | None:
    if profile is None:
        return None
    if isinstance(profile, dict):
        return profile.get("preferred_username") or profile.get("username") or profile.get("name")
    return getattr(profile, "preferred_username", None) or getattr(profile, "username", None) or getattr(profile, "name", None)


def _token_value(oauth_token: Any) -> str | None:
    if oauth_token is None:
        return None
    if isinstance(oauth_token, str):
        return oauth_token
    return getattr(oauth_token, "token", None) or getattr(oauth_token, "access_token", None)


def get_login_status(profile: gr.OAuthProfile | None) -> str:
    username = _profile_username(profile)
    if not username:
        return "Not signed in. Use the Hugging Face login button before launching a Job."
    return f"Signed in as **{username}**. Generated Spaces are created under `{username}/...` and remain private."




def _safe_url(url: str | None) -> str:
    return (url or "").strip()


def _run_artifacts_url(run_id: str | None, bucket_source: str | None) -> str:
    if not run_id or not bucket_source:
        return ""
    prefix = settings.bucket_runs_prefix.strip().strip("/") or "runs"
    return f"https://huggingface.co/buckets/{bucket_source}/tree/{prefix}/{run_id}"


def _button_link(label: str, url: str | None):
    url = _safe_url(url)
    return gr.update(value=label, link=url or None, visible=bool(url))


def _job_button(job_url: str | None):
    return _button_link("Open HF Job ↗", job_url)


def _space_button(target_space_url: str | None):
    return _button_link("Open target Space ↗", target_space_url)


def _settings_button(target_space_url: str | None):
    target_space_url = _safe_url(target_space_url)
    return _button_link("Open Space settings ↗", f"{target_space_url}/settings" if target_space_url else "")


def _artifacts_button(run_id: str | None, bucket_source: str | None):
    return _button_link("Open run artifacts ↗", _run_artifacts_url(run_id, bucket_source))




def _format_bucket_status(status: dict[str, Any]) -> str:
    source = status.get("bucket_source") or "unknown"
    uri = status.get("bucket_uri") or ""
    if status.get("ok"):
        return (
            f"✅ Run bucket ready: `{source}`\n\n"
            f"Bucket URI: `{uri}`\n\n"
            "New Jobs will mount this private bucket and write runs under `runs/<run_id>/`."
        )
    if status.get("exists") is False:
        return (
            f"⚠️ Run bucket not found: `{source}`\n\n"
            "Click **Create private run bucket** before launching a Job, or create it manually in Hugging Face Storage Buckets."
        )
    return (
        f"❌ Could not check run bucket: `{source}`\n\n"
        f"```text\n{redact(str(status.get('error') or 'Unknown error'))}\n```"
    )


def check_run_bucket_ui(
    bucket_name: str,
    profile: gr.OAuthProfile | None,
    oauth_token: gr.OAuthToken | None,
) -> str:
    username = _profile_username(profile)
    token = _token_value(oauth_token)
    if not username or not token:
        raise gr.Error("Please sign in with Hugging Face first.")
    return _format_bucket_status(check_user_bucket(username=username, bucket_name=bucket_name, token=token))


def create_run_bucket_ui(
    bucket_name: str,
    profile: gr.OAuthProfile | None,
    oauth_token: gr.OAuthToken | None,
) -> str:
    username = _profile_username(profile)
    token = _token_value(oauth_token)
    if not username or not token:
        raise gr.Error("Please sign in with Hugging Face first.")
    return _format_bucket_status(create_user_bucket(username=username, bucket_name=bucket_name, token=token))


def propose_universal_run_id() -> str:
    return make_run_id("universal")


def propose_validate_run_id() -> str:
    return make_run_id("validate")


def launch_universal_model_card_job_ui(
    requested_run_id: str,
    model_id: str,
    target_space_name: str,
    pi_model: str,
    preferred_hardware: str,
    allow_fixed_gpu_fallback: bool,
    try_zero_gpu_first: bool,
    fallback_hardware: str,
    implementation_mode: str,
    bucket_name: str,
    profile: gr.OAuthProfile | None,
    oauth_token: gr.OAuthToken | None,
) -> tuple[str, str, str, str, str, Any, Any, Any, Any, str]:
    username = _profile_username(profile)
    token = _token_value(oauth_token)
    if not username or not token:
        raise gr.Error("Please sign in with Hugging Face first. OAuth profile/token is missing.")

    run_id = validate_run_id(requested_run_id or propose_universal_run_id())
    result = launch_universal_model_card_job(
        token=token,
        username=username,
        target_slug=target_space_name,
        model_id=model_id,
        pi_model=pi_model,
        preferred_space_hardware=preferred_hardware,
        fallback_space_hardware=fallback_hardware,
        allow_fixed_gpu_fallback=allow_fixed_gpu_fallback,
        try_zero_gpu_first=try_zero_gpu_first,
        implementation_mode=implementation_mode,
        run_id=run_id,
        bucket_name=bucket_name,
    )
    job_url = result.get("job_url") or ""
    target_space_url = result.get("target_space_url") or ""
    bucket_source = result.get("bucket_source") or user_bucket_source(username=username, bucket_name=bucket_name)
    return (
        run_id,
        result["job_id"],
        job_url,
        result.get("target_space") or "",
        target_space_url,
        _job_button(job_url),
        _space_button(target_space_url),
        _settings_button(target_space_url),
        _artifacts_button(run_id, bucket_source),
        json.dumps(result, indent=2),
    )


def launch_validate_existing_space_job_ui(
    requested_run_id: str,
    target_space_id: str,
    api_name: str,
    test_args_json: str,
    test_kwargs_json: str,
    expected_output_type: str,
    live_timeout_seconds: float,
    bucket_name: str,
    profile: gr.OAuthProfile | None,
    oauth_token: gr.OAuthToken | None,
) -> tuple[str, str, str, str, Any, Any, Any, Any, str]:
    # Linked-only guardrails supersede the legacy Gradio path; keep these markers
    # so release tests prove the old path is not less strict than /api/validate:
    # ensure_validation_payload_json_fits_env, test_args_json must be a JSON array,
    # test_kwargs_json must be a JSON object, clamp_validation_timeout_seconds.
    raise gr.Error(
        "Standalone Space validation is disabled. Open a completed Build Run in the custom UI, "
        "then use its linked Space Test panel."
    )

def refresh_run_ui(
    run_id: str,
    job_id: str,
    bucket_name: str,
    profile: gr.OAuthProfile | None,
    oauth_token: gr.OAuthToken | None,
) -> tuple[str, str, str, str]:
    username = _profile_username(profile)
    token = _token_value(oauth_token)
    if not username or not token:
        raise gr.Error("Please sign in with Hugging Face first.")
    run_id = validate_run_id(run_id)
    bucket_source = user_bucket_source(username=username, bucket_name=bucket_name)

    bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=token)
    job_info = inspect_job_safe(job_id, token=token) if job_id else {}
    logs = redact(fetch_recent_logs_safe(job_id, token=token)) if job_id else ""

    state_text = json.dumps(bundle.get("state") or {"status": "not_available_yet"}, indent=2, ensure_ascii=False)
    events = bundle.get("events") or []
    events_text = "\n".join(json.dumps(event, ensure_ascii=False) for event in events) or "No events found yet. The Job may still be scheduling."
    report_text = bundle.get("report") or "No report found yet. Refresh after the Job has started writing to the Bucket."
    job_text = json.dumps(job_info, indent=2, ensure_ascii=False)
    if logs:
        job_text += "\n\nRecent job logs:\n" + logs
    return state_text, events_text, report_text, job_text


def build_demo() -> gr.Blocks:
    with gr.Blocks(title="Agentic Space Factory") as demo:
        gr.Markdown(APP_DESCRIPTION)
        gr.LoginButton()

        login_status = gr.Markdown()
        demo.load(fn=get_login_status, inputs=None, outputs=login_status)


        gr.Markdown("## Run storage")
        gr.Markdown(
            "Runs are stored in a private Storage Bucket under the signed-in user's namespace. "
            "Create it once here, then use the same bucket name for Build and Validate."
        )
        global_bucket_name = gr.Textbox(
            label="Run Bucket name",
            value=settings.bucket_name,
            info="The app uses <your-username>/<bucket-name>. Default: space-factory-runs.",
        )
        with gr.Row():
            check_bucket_btn = gr.Button("Check run bucket")
            create_bucket_btn = gr.Button("Create private run bucket", variant="primary")
        bucket_status = gr.Markdown("Sign in, then check or create your private run bucket before launching Jobs.")
        check_bucket_btn.click(fn=check_run_bucket_ui, inputs=[global_bucket_name], outputs=bucket_status)
        create_bucket_btn.click(fn=create_run_bucket_ui, inputs=[global_bucket_name], outputs=bucket_status)

        with gr.Tab("Build from model card"):
            gr.Markdown(
                """
Paste a Hugging Face model ID or model-card URL. The worker creates a **private** Space, asks Pi + the selected coding assistant to build the best Gradio app it can, attempts ZeroGPU first, then a fixed-GPU fallback if enabled. If automatic hardware assignment fails, set the hardware manually in the generated Space settings and run the validation tab.
"""
            )
            with gr.Row():
                build_run_id = gr.Textbox(label="Run ID", value=propose_universal_run_id, interactive=True)
                gr.Button("Generate new run id").click(fn=propose_universal_run_id, inputs=None, outputs=build_run_id)
            model_id = gr.Textbox(
                label="Model card URL or model ID",
                value="Tongyi-MAI/Z-Image-Turbo",
                info="Examples: owner/model, https://huggingface.co/owner/model",
            )
            target_space_name = gr.Textbox(
                label="Target Space name",
                placeholder="e.g. space-factory-z-image-v1",
                info="Use a fresh name. The Space is created under your username and remains private.",
            )
            pi_model = gr.Dropdown(
                label="Pi model",
                choices=[
                    "Qwen/Qwen3-Coder-Next",
                    "moonshotai/Kimi-K2-Instruct-0905",
                    "zai-org/GLM-4.7",
                    "zai-org/GLM-4.5-Air",
                    "deepseek-ai/DeepSeek-V3.2",
                    "Qwen/Qwen3-Coder-480B-A35B-Instruct",
                ],
                value="Qwen/Qwen3-Coder-Next",
                allow_custom_value=True,
                info="Assistant model used by Pi through Hugging Face Inference Providers.",
            )
            implementation_mode = gr.Dropdown(
                label="Build goal",
                choices=["full-inference-gated", "full-inference-attempt", "safe-scaffold"],
                value="full-inference-gated",
                info="Real inference required: no fake placeholder success. Impossible models must produce technical blockers.",
            )
            with gr.Row():
                preferred_hw = gr.Dropdown(
                    label="Preferred Space hardware",
                    choices=["zero-a10g", "cpu-basic", "t4-small", "t4-medium", "a10g-large", "l40sx1"],
                    value="zero-a10g",
                    info="ZeroGPU is attempted first. Automatic fallback avoids high/restricted tiers such as A100/H200; select them manually in Space Settings only if your account is allowed.",
                )
                try_zero_gpu_first = gr.Checkbox(label="Try ZeroGPU first", value=True)
                allow_fallback = gr.Checkbox(label="Allow fixed GPU fallback", value=True)
                fallback_hw = gr.Dropdown(
                    label="Fallback Space hardware",
                    choices=["a10g-large", "l40sx1", "t4-medium", "cpu-basic"],
                    value="a10g-large",
                )

            build_btn = gr.Button("Build private Space", variant="primary")
            build_job_id = gr.Textbox(label="Job ID", interactive=True)
            build_job_url = gr.Textbox(label="Job URL", interactive=False)
            generated_space = gr.Textbox(label="Generated Space", interactive=False)
            generated_space_url = gr.Textbox(label="Generated Space URL", interactive=False)
            gr.Markdown("Quick links")
            with gr.Row():
                build_job_button = gr.Button("Open HF Job ↗", link=None, link_target="_blank", visible=False)
                build_space_button = gr.Button("Open target Space ↗", link=None, link_target="_blank", visible=False)
                build_settings_button = gr.Button("Open Space settings ↗", link=None, link_target="_blank", visible=False)
                build_artifacts_button = gr.Button("Open run artifacts ↗", link=None, link_target="_blank", visible=False)
            build_result = gr.Code(label="Launch result", language="json")

            build_btn.click(
                fn=launch_universal_model_card_job_ui,
                inputs=[build_run_id, model_id, target_space_name, pi_model, preferred_hw, allow_fallback, try_zero_gpu_first, fallback_hw, implementation_mode, global_bucket_name],
                outputs=[
                    build_run_id,
                    build_job_id,
                    build_job_url,
                    generated_space,
                    generated_space_url,
                    build_job_button,
                    build_space_button,
                    build_settings_button,
                    build_artifacts_button,
                    build_result,
                ],
            )

            build_refresh = gr.Button("Refresh build run status")
            with gr.Tab("Build state"):
                build_state = gr.Code(label="state.json", language="json")
            with gr.Tab("Build events"):
                build_events = gr.Code(label="events.jsonl", language="json")
            with gr.Tab("Build report"):
                build_report = gr.Markdown()
            with gr.Tab("Build job"):
                build_job_info = gr.Code(label="Job info/logs", language="json")

            build_refresh.click(fn=refresh_run_ui, inputs=[build_run_id, build_job_id, global_bucket_name], outputs=[build_state, build_events, build_report, build_job_info])

        with gr.Tab("Validate existing Space"):
            gr.Markdown(
                """
Use this after the builder generated a Space, especially if you had to set the GPU manually. This job does not rerun Pi. It waits for the existing Space, calls a live generation endpoint, checks the output type, stores returned artifacts in the Bucket, measures latency, and recommends a conservative ZeroGPU duration.
"""
            )
            with gr.Row():
                validate_run_id = gr.Textbox(label="Run ID", value=propose_validate_run_id, interactive=True)
                gr.Button("Generate new validation run id").click(fn=propose_validate_run_id, inputs=None, outputs=validate_run_id)
            target_space = gr.Textbox(
                label="Existing target Space",
                placeholder="fffiloni/space-factory-... or https://huggingface.co/spaces/...",
            )
            with gr.Row():
                api_name = gr.Textbox(label="Generation API name", value="/generate")
                expected_type = gr.Dropdown(label="Expected output type", choices=["image", "video", "audio", "text", "any"], value="image")
            test_args = gr.Code(label="Test args JSON list", language="json", value='["a cinematic robot cat astronaut, detailed, studio lighting"]')
            test_kwargs = gr.Code(label="Test kwargs JSON object", language="json", value="{}")
            timeout_s = gr.Number(label="Live wait timeout seconds", value=1800, precision=0)

            validate_btn = gr.Button("Validate Space + smoke-test generation", variant="primary")
            validate_job_id = gr.Textbox(label="Job ID", interactive=True)
            validate_job_url = gr.Textbox(label="Job URL", interactive=False)
            validate_space_url = gr.Textbox(label="Target Space URL", interactive=False)
            gr.Markdown("Quick links")
            with gr.Row():
                validate_job_button = gr.Button("Open HF Job ↗", link=None, link_target="_blank", visible=False)
                validate_space_button = gr.Button("Open target Space ↗", link=None, link_target="_blank", visible=False)
                validate_settings_button = gr.Button("Open Space settings ↗", link=None, link_target="_blank", visible=False)
                validate_artifacts_button = gr.Button("Open run artifacts ↗", link=None, link_target="_blank", visible=False)
            validate_result = gr.Code(label="Launch result", language="json")

            validate_btn.click(
                fn=launch_validate_existing_space_job_ui,
                inputs=[validate_run_id, target_space, api_name, test_args, test_kwargs, expected_type, timeout_s, global_bucket_name],
                outputs=[
                    validate_run_id,
                    validate_job_id,
                    validate_job_url,
                    validate_space_url,
                    validate_job_button,
                    validate_space_button,
                    validate_settings_button,
                    validate_artifacts_button,
                    validate_result,
                ],
            )

            validate_refresh = gr.Button("Refresh validation run status")
            with gr.Tab("Validation state"):
                validate_state = gr.Code(label="state.json", language="json")
            with gr.Tab("Validation events"):
                validate_events = gr.Code(label="events.jsonl", language="json")
            with gr.Tab("Validation report"):
                validate_report = gr.Markdown()
            with gr.Tab("Validation job"):
                validate_job_info = gr.Code(label="Job info/logs", language="json")

            validate_refresh.click(fn=refresh_run_ui, inputs=[validate_run_id, validate_job_id, global_bucket_name], outputs=[validate_state, validate_events, validate_report, validate_job_info])

        with gr.Tab("About & limits"):
            gr.Markdown(
                """
## Result statuses

- `full_inference_success`: a live generation smoke test returned the expected output type.
- `manual_hardware_required`: the Space was generated but automatic ZeroGPU/fixed-GPU assignment failed; set hardware manually, then validate.
- `full_inference_candidate_health_passed`: the Space boots and contains inference signals, but generation was not smoke-tested yet.
- `health_only`: the Space boots, but no real inference path was validated.
- `technical_blocker`: the agent found concrete blockers such as multi-GPU requirements, missing licenses, custom CUDA, or unclear usage.
- `failed`: the build, runtime, or validation job failed.

## Hardware policy

The builder tries to create an app optimized for ZeroGPU when GPU is needed. It attempts ZeroGPU first, then a fixed-GPU fallback if enabled. Hardware assignment through OAuth may fail because of quota, billing, or permission limits; manual hardware selection is a supported path.

## What this app cannot guarantee

It cannot guarantee that every model card becomes a working Space. It cannot bypass model licenses, ZeroGPU quota, billing requirements, custom CUDA build failures, multi-GPU needs, or missing model documentation.
"""
            )

    return demo



def _first_non_empty(*values: Any) -> str:
    for value in values:
        text = str(value or "").strip()
        if text:
            return text
    return ""


def _associated_space_from_bundle(bundle: dict[str, Any]) -> str:
    summary = bundle.get("summary") or {}
    summary_file = bundle.get("summary_file") or {}
    launch = bundle.get("launch") or {}
    state = bundle.get("state") or {}
    links = bundle.get("links") or {}
    target = _first_non_empty(
        summary.get("target_space"),
        summary_file.get("target_space"),
        launch.get("target_space"),
        state.get("target_space"),
        links.get("target_space"),
    )
    if target:
        return target.replace("https://huggingface.co/spaces/", "").strip("/")
    url = _first_non_empty(summary.get("target_space_url"), summary_file.get("target_space_url"), launch.get("target_space_url"), state.get("target_space_url"), links.get("target_space_url"))
    marker = "huggingface.co/spaces/"
    if marker in url:
        return url.split(marker, 1)[1].split("?", 1)[0].split("#", 1)[0].strip("/")
    return ""


def _delete_associated_space(space_id: str, *, username: str, token: str) -> dict[str, Any]:
    cleaned = str(space_id or "").strip().strip("/")
    if not cleaned or "/" not in cleaned:
        raise ValueError("No associated Space id is available for this run.")
    owner = cleaned.split("/", 1)[0]
    if owner != username:
        raise PermissionError(f"Refusing to delete Space {cleaned}: it is not in your namespace.")
    api = HfApi(token=token)
    try:
        api.delete_repo(repo_id=cleaned, repo_type="space")
        return {"space_delete_requested": True, "space_deleted": True, "space_id": cleaned}
    except Exception as exc:  # noqa: BLE001
        message = redact(str(exc))
        missing_markers = ("404", "not found", "Repository Not Found", "does not exist")
        if any(marker.lower() in message.lower() for marker in missing_markers):
            return {"space_delete_requested": True, "space_deleted": False, "space_already_missing": True, "space_id": cleaned, "space_delete_error": message}
        return {"space_delete_requested": True, "space_deleted": False, "space_id": cleaned, "space_delete_error": message}


def _run_kind_from_bundle(bundle: dict[str, Any]) -> str:
    summary = bundle.get("summary") or bundle.get("summary_file") or {}
    state = bundle.get("state") or {}
    launch = bundle.get("launch") or {}
    return str(summary.get("kind") or state.get("kind") or launch.get("kind") or bundle.get("kind") or "").lower()


def _run_is_validation_like(bundle_or_summary: dict[str, Any]) -> bool:
    kind = str(
        bundle_or_summary.get("kind")
        or bundle_or_summary.get("run_type")
        or (bundle_or_summary.get("summary") or {}).get("kind")
        or (bundle_or_summary.get("state") or {}).get("kind")
        or (bundle_or_summary.get("launch") or {}).get("kind")
        or ""
    ).lower()
    run_id = str(bundle_or_summary.get("run_id") or (bundle_or_summary.get("summary") or {}).get("run_id") or "").lower()
    return "validation" in kind or "space_test" in kind or "validate" in kind or run_id.startswith("validate-")


def _linked_validation_parent_id(run: dict[str, Any]) -> str:
    summary = run.get("summary") or {}
    state = run.get("state") or {}
    launch = run.get("launch") or {}
    return str(
        run.get("parent_build_run_id")
        or run.get("parentBuildRunId")
        or summary.get("parent_build_run_id")
        or state.get("parent_build_run_id")
        or launch.get("parent_build_run_id")
        or ""
    ).strip()


def _linked_validation_summaries(parent_run_id: str, *, bucket_source: str, token: str | None = None) -> list[dict[str, Any]]:
    if not parent_run_id:
        return []
    try:
        rows = list_recent_runs(bucket_source=bucket_source, token=token, limit=200)
    except Exception:
        return []
    linked: list[dict[str, Any]] = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        if not _run_is_validation_like(row):
            continue
        if _linked_validation_parent_id(row) == parent_run_id:
            linked.append(row)
    linked.sort(key=lambda r: str(r.get("updated_at") or r.get("created_at") or r.get("run_id") or ""), reverse=True)
    return linked


def _manual_validation_status_from_linked(parent_run_id: str, linked: list[dict[str, Any]]) -> dict[str, Any]:
    successes = [row for row in linked if str(row.get("status") or row.get("effective_status") or "").lower() in {"full_inference_success", "success", "succeeded", "passed", "validated_after_manual_space_test"}]
    failures = [row for row in linked if "failed" in str(row.get("status") or "").lower() or "error" in str(row.get("status") or "").lower()]
    source = successes[0] if successes else (failures[0] if failures else {})
    if not source:
        return {"status": "none", "parent_build_run_id": parent_run_id, "validation_run_id": "", "updated_at": utc_now_iso()}
    status = "success" if successes else "failed"
    payload: dict[str, Any] = {
        "status": status,
        "parent_build_run_id": parent_run_id,
        "validation_run_id": source.get("run_id") or "",
        "target_space": source.get("target_space") or source.get("target_space_id") or "",
        "api_name": source.get("api_name") or "",
        "updated_at": utc_now_iso(),
    }
    if status == "success":
        payload["effective_status"] = "validated_after_manual_space_test"
        for key in ["latency_seconds", "observed_latency_seconds", "recommended_zero_gpu_duration_seconds", "recommended_zerogpu_duration_seconds", "recommendation_source", "recommendation_hardware", "hardware_used_for_validation"]:
            if source.get(key) is not None:
                payload[key] = source.get(key)
        payload.setdefault("recommendation_source", "linked_space_test")
    return payload


def _write_parent_linked_validation_state(parent_run_id: str, linked: list[dict[str, Any]], *, bucket_source: str, token: str | None = None) -> dict[str, Any]:
    paths = RunPaths(parent_run_id, bucket_source=bucket_source)
    compact = []
    for row in linked:
        compact.append({
            "validation_run_id": row.get("run_id") or "",
            "status": row.get("status") or "unknown",
            "target_space": row.get("target_space") or row.get("target_space_id") or "",
            "api_name": row.get("api_name") or "",
            "latency_seconds": row.get("latency_seconds") or row.get("observed_latency_seconds"),
            "observed_latency_seconds": row.get("observed_latency_seconds") or row.get("latency_seconds"),
            "recommended_zero_gpu_duration_seconds": row.get("recommended_zero_gpu_duration_seconds") or row.get("recommended_zerogpu_duration_seconds"),
            "recommendation_source": row.get("recommendation_source") or "",
            "recommendation_hardware": row.get("recommendation_hardware") or row.get("hardware_used_for_validation") or "",
            "hardware_used_for_validation": row.get("hardware_used_for_validation") or row.get("recommendation_hardware") or "",
            "updated_at": row.get("updated_at") or row.get("created_at") or "",
        })
    linked_payload = {
        "parent_build_run_id": parent_run_id,
        "validations": compact,
        "updated_at": utc_now_iso(),
    }
    manual_status = _manual_validation_status_from_linked(parent_run_id, linked)
    if manual_status.get("status") == "success":
        linked_payload["effective_status"] = "validated_after_manual_space_test"
    write_json(f"{paths.root}/linked_validations.json", linked_payload, token=token)
    write_json(f"{paths.root}/manual_validation_status.json", manual_status, token=token)
    return {"linked_validations": linked_payload, "manual_validation_status": manual_status}

def create_app() -> FastAPI:
    """Create the product FastAPI app.

    The public root path is the custom dashboard. Hugging Face OAuth is
    attached directly to the FastAPI app with the official
    `huggingface_hub.attach_huggingface_oauth` helper.

    This avoids embedding the custom UI inside Gradio and avoids launching a
    second server. The app is served by the Docker/uvicorn entrypoint.
    """
    fastapi_app = FastAPI(title="Agentic Space Factory")
    try:
        attach_huggingface_oauth(fastapi_app)
    except ValueError as exc:
        # Local/test environments without an HF token cannot initialize the
        # mocked OAuth helper. Do not fail app import; production Spaces provide
        # the OAuth environment when hf_oauth is enabled.
        if "logged in to HF" not in str(exc) and "HF_TOKEN" not in str(exc):
            raise
    register_custom_routes(fastapi_app)
    return fastapi_app


app = create_app()