from __future__ import annotations import json import os 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, append_run_event, check_user_bucket, create_user_bucket, delete_run_folder, read_run_bundle, list_recent_runs, read_json, upsert_run_index_entry, write_json, write_launch_metadata from src.target_space_identity import ( TARGET_SPACE_OVERRIDE_FILENAME, normalize_target_space_id as normalize_effective_target_space_id, target_space_identity_overlay, target_space_url as effective_target_space_url, ) 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 from src.effective_status import compute_effective_run_status 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: `/{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, "auth_lifetime": oauth_lifetime_summary(ctx), "is_pro": ctx.is_pro, "can_pay": ctx.can_pay, "warnings": oauth_warning_messages(ctx), } def _clear_local_oauth_session(request: Request) -> dict[str, Any]: """Clear ASF/HF OAuth session state without requiring a valid token. This is intentionally tolerant: expired OAuth sessions must be recoverable even when parse_huggingface_oauth/extract_oauth_context can no longer deserialize or validate the token. """ cleared_keys: list[str] = [] had_session = False try: session = request.session # type: ignore[attr-defined] had_session = bool(session) for key in list(session.keys()): key_l = str(key).lower() if "oauth" in key_l or "huggingface" in key_l or key_l in {"state", "nonce", "next", "redirect_uri"}: cleared_keys.append(str(key)) session.clear() except Exception: had_session = False return {"cleared": True, "had_session": had_session, "cleared_keys": sorted(set(cleared_keys))} def _auth_recovery_payload(reason: str = "not_authenticated", detail: str | None = None) -> dict[str, Any]: reason_l = (reason or "not_authenticated").lower() if "expired" in reason_l or (detail and "expired" in detail.lower()): reason_l = "oauth_expired" message = "Your Hugging Face OAuth session expired. Refresh sign-in to continue." else: reason_l = "not_signed_in" if reason_l in {"not_authenticated", "please sign in with hugging face first."} else reason_l message = "Sign in with Hugging Face to continue." return { "authenticated": False, "reason": reason_l, "message": message, "login_url": "/oauth/huggingface/login", "refresh_login_url": "/auth/refresh-login", "reset_url": "/auth/logout-local", "logout_url": "/auth/logout-local", "recovery": { "primary_action": "refresh_sign_in" if reason_l == "oauth_expired" else "sign_in", "refresh_login_url": "/auth/refresh-login", "reset_local_auth_url": "/auth/logout-local", "open_in_new_tab_recommended": True, }, } def _api_me_payload(ctx: dict[str, Any]) -> dict[str, Any]: return { "authenticated": True, "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", "refresh_login_url": "/auth/refresh-login", "reset_url": "/auth/logout-local", "logout_url": "/auth/logout-local", } def _resumable_payload_from_runs( runs: list[dict[str, Any]], *, bucket_source: str, token: str | None, ) -> dict[str, Any]: resumable = find_latest_resumable_run(runs) selected = resumable or (runs[0] if runs else None) if not selected: return {"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=token, include_heavy=False) return { "run": selected, "view": snapshot["view"], "summary": snapshot["summary"], "state": snapshot["state"], "bucket_source": bucket_source, "reason": "latest_resumable_run" if resumable else "latest_run", } 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 _normalize_target_space_id(value: Any) -> str: text = str(value or "").strip() if not text: return "" text = text.replace("https://huggingface.co/spaces/", "").strip("/") text = text.split("/settings", 1)[0].strip("/") if ".hf.space" in text and "/" not in text: # Subdomains are not reversible for all names; keep them out of repo-id # fields and use only owner/name repo ids as canonical target_space. return "" if re.match(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", text): return text return "" def _target_space_from_events(events: list[dict[str, Any]] | None, *, require_success: bool = False) -> str: for event in reversed(events or []): if not isinstance(event, dict): continue step = str(event.get("step") or "") status = str(event.get("status") or "").lower() if step not in {"create_space", "create_space_hardware", "upload_files", "runtime_upload_epoch"}: continue if require_success and status != "success": continue data = event.get("data") if isinstance(event.get("data"), dict) else {} target = _normalize_target_space_id(data.get("target_space") or data.get("target_space_id")) if target: return target return "" def _resolve_target_space_identity(run_id: str | None, bundle: dict[str, Any], *, bucket_source: str | None, job_url: str | None = None) -> dict[str, Any]: """Return canonical Space links from all stable run sources. v198.26.123: apply an auditable Target Space identity override before building UI links or validation prefill. The raw/generated target remains available as target_space_original for audit; target_space is the effective repo id to use for future validations and buttons. """ summary = bundle.get("summary") or bundle.get("summary_file") or {} state = bundle.get("state") or {} launch = bundle.get("launch") or {} runtime_upload_epoch = bundle.get("runtime_upload_epoch") or {} explicit_url = _first_non_empty( summary.get("target_space_url"), state.get("target_space_url"), launch.get("target_space_url") ) raw_target = _normalize_target_space_id( _first_non_empty( summary.get("target_space"), summary.get("target_space_id"), state.get("target_space"), state.get("target_space_id"), launch.get("target_space"), launch.get("target_space_id"), runtime_upload_epoch.get("target_space_id"), runtime_upload_epoch.get("target_space"), _target_space_from_events(bundle.get("events") or []), ) ) if explicit_url and not raw_target: raw_target = _normalize_target_space_id(explicit_url) overlay = target_space_identity_overlay({**bundle, "summary": summary, "state": state, "launch": launch}) target = overlay.get("target_space") or raw_target target_space_url = overlay.get("target_space_url") or (f"https://huggingface.co/spaces/{target}" if target else "") target_space_settings_url = f"{target_space_url}/settings" if target_space_url else "" events = bundle.get("events") or [] space_runtime = bundle.get("space_runtime") or {} runtime_history = runtime_upload_epoch.get("history") if isinstance(runtime_upload_epoch.get("history"), list) else [] create_event_ok = bool(raw_target and any(str((e or {}).get("step") or "") in {"create_space", "create_space_hardware"} and str((e or {}).get("status") or "").lower() == "success" for e in events if isinstance(e, dict))) upload_event_ok = bool(any(str((e or {}).get("step") or "") in {"upload_files", "runtime_upload_epoch"} and str((e or {}).get("status") or "").lower() == "success" for e in events if isinstance(e, dict))) runtime_uploaded = bool(runtime_upload_epoch.get("last_upload_completed_at") or runtime_upload_epoch.get("upload_sequence") or runtime_history or upload_event_ok) space_runtime_known = bool(isinstance(space_runtime, dict) and (space_runtime.get("stage") or space_runtime.get("status") or space_runtime.get("runtime_status") or space_runtime.get("updated_at") or space_runtime.get("url"))) space_created = bool(target and (create_event_ok or runtime_uploaded or space_runtime_known or overlay.get("target_space_override_active"))) sources = [] if create_event_ok: sources.append("create_space_event") if upload_event_ok: sources.append("upload_event") if runtime_uploaded: sources.append("runtime_upload_epoch") if space_runtime_known: sources.append("space_runtime") if explicit_url: sources.append("explicit_target_space_url") if overlay.get("target_space_override_active"): sources.insert(0, "target_space_override") links_ready = bool(target_space_url and (space_created or runtime_uploaded or space_runtime_known or overlay.get("target_space_override_active"))) return { "schema_version": "space_identity.v198_26_122", "target_space": target, "target_space_id": target, "target_space_known": bool(target), "target_space_url": target_space_url, "target_space_settings_url": target_space_settings_url, "target_space_original": overlay.get("target_space_original") or raw_target, "target_space_original_url": overlay.get("target_space_original_url") or (f"https://huggingface.co/spaces/{raw_target}" if raw_target else ""), "target_space_override_active": bool(overlay.get("target_space_override_active")), "target_space_override": overlay.get("target_space_override") or {}, "target_space_source": overlay.get("target_space_source") or "generated_run", "space_created": bool(space_created), "runtime_uploaded": bool(runtime_uploaded), "space_uploaded": bool(runtime_uploaded), "space_runtime_known": bool(space_runtime_known), "links_ready": bool(links_ready), "source": "+".join(sources) or "resolved_from_bundle", "sources": sources, "job_url": job_url or "", "artifacts_url": _run_artifacts_url(run_id, bucket_source), } 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 = _normalize_target_space_id(target_space) target_space_url = f"https://huggingface.co/spaces/{target}" if target 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 _run_target_override_path(run_id: str, bucket_source: str) -> str: return f"{RunPaths(run_id, bucket_source=bucket_source).root}/{TARGET_SPACE_OVERRIDE_FILENAME}" def _target_space_override_response(run_id: str, bucket_source: str, bundle: dict[str, Any]) -> dict[str, Any]: identity = _resolve_target_space_identity(run_id, bundle, bucket_source=bucket_source, job_url=(bundle.get("state") or {}).get("job_url") or (bundle.get("summary") or {}).get("job_url") or "") return { "run_id": run_id, "bucket_source": bucket_source, "target_space": identity.get("target_space") or "", "target_space_id": identity.get("target_space") or "", "target_space_url": identity.get("target_space_url") or "", "target_space_settings_url": identity.get("target_space_settings_url") or "", "target_space_original": identity.get("target_space_original") or "", "target_space_original_url": identity.get("target_space_original_url") or "", "target_space_override_active": bool(identity.get("target_space_override_active")), "target_space_override": identity.get("target_space_override") or {}, "target_space_source": identity.get("target_space_source") or "generated_run", "space_identity": identity, "links": {**identity, "run_id": run_id, "bucket_source": bucket_source}, } def _assert_override_repo_allowed(*, target_space: str, username: str, token: str | None) -> None: owner = target_space.split("/", 1)[0] if "/" in target_space else "" if username and owner and owner != username: raise ValueError(f"Target Space override must stay in your namespace for now ({username}/...).") try: HfApi(token=token).repo_info(repo_id=target_space, repo_type="space", token=token) except Exception as exc: # noqa: BLE001 raise ValueError(f"Could not verify target Space `{target_space}`. Make sure it exists and your OAuth token can access it. {redact(str(exc))}") from exc def _write_target_space_override( *, run_id: str, bucket_source: str, username: str, token: str | None, target_space: str, reason: str, ) -> dict[str, Any]: bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=token, include_heavy=False) if _run_is_validation_like({"run_id": run_id, **bundle}): raise ValueError("Target Space override can only be set on a Build Run, not on a validation run.") original_identity = target_space_identity_overlay(bundle) original = original_identity.get("target_space_original") or original_identity.get("target_space") or "" payload = { "schema_version": "target_space_override.v1", "status": "active", "parent_build_run_id": run_id, "original_target_space_id": original, "target_space_id": target_space, "target_space": target_space, "target_space_url": effective_target_space_url(target_space), "target_space_settings_url": f"{effective_target_space_url(target_space)}/settings", "reason": reason, "set_by": username, "verified_repo_exists": True, "updated_at": utc_now_iso(), } write_json(_run_target_override_path(run_id, bucket_source), payload, token=token) try: append_run_event( run_id, bucket_source=bucket_source, step="target_space_override", status="success", message="Target Space identity override saved", details={"target_space": target_space, "original_target_space": original, "reason": reason}, token=token, ) except Exception: pass refreshed = read_run_bundle(run_id, bucket_source=bucket_source, token=token, include_heavy=False) response = _target_space_override_response(run_id, bucket_source, refreshed) try: summary = refreshed.get("summary") or {} upsert_run_index_entry(run_id, bucket_source=bucket_source, summary={**summary, **response, "updated_at": utc_now_iso()}, token=token) except Exception: pass return response def _clear_target_space_override(*, run_id: str, bucket_source: str, username: str, token: str | None, reason: str = "") -> dict[str, Any]: bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=token, include_heavy=False) if _run_is_validation_like({"run_id": run_id, **bundle}): raise ValueError("Target Space override can only be cleared on a Build Run, not on a validation run.") current = bundle.get("target_space_override") or {} original_identity = target_space_identity_overlay(bundle) payload = { "schema_version": "target_space_override.v1", "status": "cleared", "parent_build_run_id": run_id, "previous_target_space_id": current.get("target_space_id") or current.get("target_space") or "", "original_target_space_id": original_identity.get("target_space_original") or "", "reason": reason or "Target Space override cleared by user.", "cleared_by": username, "updated_at": utc_now_iso(), } write_json(_run_target_override_path(run_id, bucket_source), payload, token=token) try: append_run_event( run_id, bucket_source=bucket_source, step="target_space_override", status="cleared", message="Target Space identity override cleared", details={"previous_target_space": payload["previous_target_space_id"], "reason": payload["reason"]}, token=token, ) except Exception: pass refreshed = read_run_bundle(run_id, bucket_source=bucket_source, token=token, include_heavy=False) response = _target_space_override_response(run_id, bucket_source, refreshed) try: summary = refreshed.get("summary") or {} upsert_run_index_entry(run_id, bucket_source=bucket_source, summary={**summary, **response, "updated_at": utc_now_iso()}, token=token) except Exception: pass return response 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 {}), } # v198.26.34: repair_outcome.json is a terminal evidence source. A run can # be left with stale summary/state="running" if the repair path writes the # outcome after the last lightweight state flush. Active Run polling should # derive the product verdict from this artifact instead of showing endless # running. Protected partial summaries remain partial; otherwise terminal # repair failure becomes failed. repair_outcome = bundle.get("repair_outcome") or {} if isinstance(repair_outcome, dict) and repair_outcome: repair_status = str(repair_outcome.get("post_repair_validation") or repair_outcome.get("failure_type") or repair_outcome.get("status") or "").strip().lower() terminal_repair_failures = {"failed", "repair_validation_failed", "upload_failed", "smoke_still_failed", "repair_failed_same_error", "repair_exhausted", "no_patch_produced_by_pi", "no_relevant_patch_produced_by_pi", "pi_patch_rejected_by_guard"} if repair_status in terminal_repair_failures and str(effective_summary.get("status") or "").lower() in {"", "running", "queued", "pending", "started", "scheduled"}: effective_summary["status"] = "failed" effective_summary["display_status"] = "failed" effective_summary["effective_status"] = "failed" effective_summary["repair_outcome_status"] = repair_outcome.get("post_repair_validation") or "" effective_summary["failure_type"] = repair_outcome.get("failure_type") or "repair_validation_failed" effective_summary["terminal_evidence"] = "repair_outcome.json" effective_summary["message"] = repair_outcome.get("final_user_message") or effective_summary.get("message") or "Repair failed terminally." effective_state["status"] = "failed" effective_state["repair_outcome"] = repair_outcome effective_state["terminal_evidence"] = "repair_outcome.json" else: effective_state.setdefault("repair_outcome", repair_outcome) view_bundle = {**bundle, "events": events, "state": effective_state, "summary": effective_summary} # v198.26.1: build/read snapshots must be read-only. Loading a run used to # call the eval publisher and therefore rewrite eval_publish_status.json # on a simple GET. Keep eval status as a view annotation only; an explicit # POST endpoint below owns publication side effects. eval_publish = _read_eval_publish_status_for_view(run_id, bundle, bucket_source=bucket_source, token=token) view_bundle["eval_publish"] = eval_publish view_bundle["eval_publish_status"] = eval_publish space_identity = _resolve_target_space_identity(run_id, view_bundle, bucket_source=bucket_source, job_url=effective_state.get("job_url") or effective_summary.get("job_url")) if space_identity.get("target_space"): effective_state["target_space"] = space_identity["target_space"] effective_state["target_space_id"] = space_identity["target_space"] effective_state["target_space_original"] = space_identity.get("target_space_original") or effective_state.get("target_space_original") or "" effective_state["target_space_override_active"] = bool(space_identity.get("target_space_override_active")) effective_summary["target_space"] = space_identity["target_space"] effective_summary["target_space_id"] = space_identity["target_space"] effective_summary["target_space_original"] = space_identity.get("target_space_original") or effective_summary.get("target_space_original") or "" effective_summary["target_space_override_active"] = bool(space_identity.get("target_space_override_active")) if space_identity.get("target_space_url"): effective_state["target_space_url"] = space_identity["target_space_url"] effective_summary["target_space_url"] = space_identity["target_space_url"] view_bundle["state"] = effective_state view_bundle["summary"] = effective_summary view_bundle["space_identity"] = space_identity view_bundle["links"] = {**space_identity, "job_url": space_identity.get("job_url") or effective_state.get("job_url") or effective_summary.get("job_url") or ""} effective_run_status = compute_effective_run_status(view_bundle, build_status=effective_summary.get("status") or effective_state.get("status")) view_bundle["effective_run_status"] = effective_run_status effective_summary = {**effective_summary, "effective_run_status": effective_run_status, "display_status": effective_run_status.get("display_status"), "effective_status": effective_run_status.get("effective_status"), "effective_verdict": effective_run_status.get("effective_verdict")} view_bundle["summary"] = effective_summary 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 _read_eval_publish_status_for_view(run_id: str, bundle: dict[str, Any], *, bucket_source: str, token: str | None = None) -> dict[str, Any]: """Return eval publish metadata for UI reads without writing to the Bucket.""" existing = bundle.get("eval_publish_status") or bundle.get("eval_publish") or {} if isinstance(existing, dict) and existing: return {**existing, "read_only_view": True} try: paths = RunPaths(run_id, bucket_source=bucket_source) persisted = read_json(f"{paths.root}/eval_publish_status.json", token=token) or {} except Exception: persisted = {} if isinstance(persisted, dict) and persisted: return {**persisted, "read_only_view": True} return { "schema_version": "eval_publish_status.virtual.v198_26_1", "publish_mode": "backend_read_only", "attempted": False, "published": False, "virtual": True, "read_only_view": True, "reason": "not_published_or_not_checked", } def register_custom_routes(fastapi_app: FastAPI) -> None: """Register the root custom UI and OAuth-backed JSON endpoints.""" @fastapi_app.get("/login/huggingface") async def oauth_login_compat(): # type: ignore[no-untyped-def] # Backward-compatible route for browsers that cached older ASF JS/HTML. # The canonical OAuth endpoint remains /oauth/huggingface/login. return RedirectResponse("/oauth/huggingface/login", status_code=307) @fastapi_app.get("/logout") async def oauth_logout_compat(): # type: ignore[no-untyped-def] # Backward-compatible route for cached clients. Keep it as a plain # redirect without custom target parameters to avoid OAuth loops. return RedirectResponse("/oauth/huggingface/logout", status_code=307) @fastapi_app.get("/auth/logout-local") async def auth_logout_local(request: Request): # type: ignore[no-untyped-def] # Local reset must work even when the HF OAuth token is expired and the # official helper can no longer parse the session. _clear_local_oauth_session(request) response = RedirectResponse("/", status_code=303) response.delete_cookie("session") response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate" response.headers["Pragma"] = "no-cache" return response @fastapi_app.get("/auth/refresh-login") async def auth_refresh_login(request: Request): # type: ignore[no-untyped-def] # Force a clean re-authorization path. This avoids stale oauth_info # state after token expiry and should make Space duplication unnecessary. _clear_local_oauth_session(request) response = RedirectResponse("/oauth/huggingface/login", status_code=303) response.delete_cookie("session") response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate" response.headers["Pragma"] = "no-cache" return response @fastapi_app.get("/auth/logout-and-refresh") async def auth_logout_and_refresh(request: Request): # type: ignore[no-untyped-def] _clear_local_oauth_session(request) response = RedirectResponse("/auth/refresh-login", status_code=303) response.delete_cookie("session") response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate" response.headers["Pragma"] = "no-cache" return response @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"), headers={"Cache-Control": "no-store, no-cache, must-revalidate", "Pragma": "no-cache"}, ) @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, headers={"Cache-Control": "no-store, no-cache, must-revalidate", "Pragma": "no-cache"}, ) @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", "refresh_login_url": "/auth/refresh-login", "reset_auth_url": "/auth/logout-local", "logout_url": "/auth/logout-local", "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] try: ctx = _oauth_context_from_request(request) except HTTPException as exc: payload = _auth_recovery_payload(str(exc.detail), str(exc.detail)) payload["anonymous_eval"] = public_eval_config(None) return JSONResponse(payload, status_code=401) return JSONResponse(_api_me_payload(ctx)) @fastapi_app.get("/api/bootstrap") async def api_bootstrap(request: Request, bucket_name: str = settings.bucket_name, limit: int = 100): # type: ignore[no-untyped-def] try: ctx = _oauth_context_from_request(request) except HTTPException as exc: payload = _auth_recovery_payload(str(exc.detail), str(exc.detail)) payload["anonymous_eval"] = public_eval_config(None) return JSONResponse(payload, status_code=401) bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name) bucket_status = check_user_bucket(username=ctx["username"], bucket_name=bucket_name, token=ctx["token"]) runs: list[dict[str, Any]] = [] resumable = {"run": None, "view": None, "bucket_source": bucket_source, "reason": "bucket_not_ready"} if bucket_status.get("ok"): runs = list_recent_runs(bucket_source=bucket_source, token=ctx["token"], limit=limit) resumable = _resumable_payload_from_runs(runs, bucket_source=bucket_source, token=ctx["token"]) return JSONResponse( { "schema_version": "asf_bootstrap.v1", "me": _api_me_payload(ctx), "bucket": bucket_status, "runs": { "runs": runs, "bucket_source": bucket_source, "bucket_uri": f"hf://buckets/{bucket_source}", "limit": limit, }, "resumable": resumable, "bucket_source": bucket_source, }, headers={"Cache-Control": "no-store, no-cache, must-revalidate", "Pragma": "no-cache"}, ) @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.post("/api/runs/{run_id}/eval-publish") async def api_run_eval_publish(request: Request, run_id: str, bucket_name: str = settings.bucket_name): # type: ignore[no-untyped-def] """Explicit eval publish action. GET/detail routes are intentionally read-only.""" 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: payload = await request.json() body = payload if isinstance(payload, dict) else {} except Exception: body = {} bundle = read_run_bundle(run_id, bucket_source=bucket_source, token=ctx["token"], include_heavy=False) state = bundle.get("state") or bundle.get("summary") or {} try: result = maybe_publish_eval_record(run_id, bucket_source=bucket_source, token=ctx["token"], state=state, force=bool(body.get("force"))) except Exception as exc: # noqa: BLE001 raise HTTPException(status_code=400, detail=redact(str(exc))) from exc return JSONResponse({"ok": True, "run_id": run_id, "bucket_source": bucket_source, "eval_publish": result, "eval_publish_status": result}) @fastapi_app.get("/api/oauth/diagnostics") async def api_oauth_diagnostics(request: Request): # type: ignore[no-untyped-def] """Public-safe OAuth diagnostics for iframe/sign-in troubleshooting. This endpoint never exposes secrets. It is intentionally readable while signed out so the UI/operator can tell whether OAuth routes and Space metadata-derived environment variables are present before login works. """ env_status = { "space_id_present": bool(os.getenv("SPACE_ID")), "space_host_present": bool(os.getenv("SPACE_HOST")), "oauth_client_id_present": bool(os.getenv("OAUTH_CLIENT_ID")), "oauth_client_secret_present": bool(os.getenv("OAUTH_CLIENT_SECRET")), "oauth_scopes_present": bool(os.getenv("OAUTH_SCOPES")), "openid_provider_url_present": bool(os.getenv("OPENID_PROVIDER_URL")), } try: ctx = extract_oauth_context(request) except HTTPException as exc: return JSONResponse( { "authenticated": False, "reason": exc.detail, "oauth_env": env_status, "login_url": "/oauth/huggingface/login", "refresh_login_url": "/auth/refresh-login", "reset_url": "/auth/logout-local", "logout_url": "/auth/logout-local", "recommended_action": "refresh_sign_in" if "expired" in str(exc.detail).lower() else "sign_in", } ) return JSONResponse( { "authenticated": True, "user": public_oauth_context(ctx), "token_identity": verify_token_identity(ctx), "oauth_env": env_status, "login_url": "/oauth/huggingface/login", "refresh_login_url": "/auth/refresh-login", "reset_url": "/auth/logout-local", "logout_url": "/auth/logout-local", } ) @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 "")) if parent_build_run_id.startswith("validate-"): raise ValueError("Space Test must be linked to a Build Run, not to another validation run. Select the parent Build Run and retry validation.") 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) if _run_is_validation_like({"run_id": parent_build_run_id, **parent_bundle}): raise ValueError("Space Test must be linked to a Build Run, not to another validation run. Select the parent Build Run and retry validation.") parent_identity = _resolve_target_space_identity( parent_build_run_id, parent_bundle, bucket_source=bucket_status.get("bucket_source") or user_bucket_source(username=ctx["username"], bucket_name=bucket_name), job_url=(parent_bundle.get("state") or {}).get("job_url") or (parent_bundle.get("summary") or {}).get("job_url"), ) parent_target = parent_identity.get("target_space") or "" requested_target = normalize_effective_target_space_id(payload.get("target_space_id") or "") 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 effective target Space. Save a Target Space override first if the Space was renamed.") 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.")) expert_intervention_required = bool(payload.get("expert_intervention_required")) expert_intervention_note = str(payload.get("expert_intervention_note") or "").strip() if expert_intervention_required and not expert_intervention_note: raise ValueError("Expert intervention acceptance requires a short note describing the manual repair before validation.") payload_source_raw = str(payload.get("payload_source") or "").strip() manual_endpoint_override = bool(payload.get("endpoint_override")) or payload_source_raw == "endpoint_registry_manual_override" api_name_for_validation = str(payload.get("api_name") or "").strip() replay_source: dict[str, Any] = {} replay_mode_requested = str(space_test_policy.get("mode") or payload.get("validation_mode") or "").strip().lower() == "replay" if replay_mode_requested and not manual_endpoint_override: 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: if not manual_endpoint_override: 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 and not manual_endpoint_override: api_name_for_validation = "" requested_validation_mode = space_test_policy.get("mode") or payload.get("validation_mode") or "complete" validation_mode_for_launch = "complete" if manual_endpoint_override and str(requested_validation_mode or "").strip().lower() == "replay" else requested_validation_mode base_validation_mode_for_launch = validation_mode_for_launch if expert_intervention_required: validation_mode_for_launch = "expert_acceptance" # 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": validation_mode_for_launch, "base_validation_mode": base_validation_mode_for_launch, "expert_intervention_required": expert_intervention_required, "expert_intervention_note": expert_intervention_note, "expert_acceptance_requires_validation": bool(expert_intervention_required), "payload_source": (payload.get("payload_source") if manual_endpoint_override else "parent_automatic_smoke") if replay_source else (payload.get("payload_source") or "ui_payload"), "endpoint_override": bool(manual_endpoint_override), "endpoint_registry_source_run_id": payload.get("endpoint_registry_source_run_id") or "", "replay_source_present": bool(replay_source), "ui_payload_source": (payload.get("payload_source") if manual_endpoint_override else "parent_automatic_smoke") if replay_source else (payload.get("payload_source") or ""), "space_test_policy": { "mode": space_test_policy.get("mode"), "launch_mode": validation_mode_for_launch, "base_launch_mode": base_validation_mode_for_launch, "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="success_with_expert_intervention" if expert_intervention_required else (space_test_policy.get("effective_status_on_success") or "validated_after_manual_space_test"), expert_intervention_required=expert_intervention_required, expert_intervention_note=expert_intervention_note, parent_replay_source_json=json.dumps(replay_source, ensure_ascii=False) if replay_source and not manual_endpoint_override 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/runs/{run_id}/target-space-override") async def api_set_target_space_override(request: Request, run_id: str, 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_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name) run_id = validate_run_id(run_id) target = normalize_effective_target_space_id(payload.get("target_space_id") or payload.get("target_space") or "") if not target: raise HTTPException(status_code=400, detail="Enter a valid Hugging Face Space repo ID like owner/space-name.") reason = str(payload.get("reason") or "").strip() if not reason: raise HTTPException(status_code=400, detail="Target Space override requires a short reason, for example: renamed manually in HF Space settings.") try: _assert_override_repo_allowed(target_space=target, username=ctx["username"], token=ctx["token"]) result = _write_target_space_override( run_id=run_id, bucket_source=bucket_source, username=ctx["username"], token=ctx["token"], target_space=target, reason=reason, ) return JSONResponse(result) except Exception as exc: # noqa: BLE001 raise _json_error(exc) from exc @fastapi_app.post("/api/runs/{run_id}/target-space-override/clear") async def api_clear_target_space_override(request: Request, run_id: str, payload: dict[str, Any] | None = None): # type: ignore[no-untyped-def] ctx = _oauth_context_from_request(request) payload = payload or {} bucket_name = payload.get("bucket_name") or settings.bucket_name bucket_source = user_bucket_source(username=ctx["username"], bucket_name=bucket_name) run_id = validate_run_id(run_id) try: result = _clear_target_space_override( run_id=run_id, bucket_source=bucket_source, username=ctx["username"], token=ctx["token"], reason=str(payload.get("reason") or "").strip(), ) return JSONResponse(result) except Exception as exc: # noqa: BLE001 raise _json_error(exc) from exc @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"], "events": snapshot["events"], "events_recent": snapshot["events"][-100:], "events_count": len(snapshot["events"]), "repair_attempts": snapshot["bundle"].get("repair_attempts") or {}, "eval_publish": snapshot["eval_publish"], "final_status_reconciliation": snapshot["bundle"].get("final_status_reconciliation") or {}, "eval_record": snapshot["bundle"].get("eval_record") or {}, "validation_metrics": snapshot["view"].get("validation_metrics") or {}, "run_documents": snapshot["view"].get("run_documents") or snapshot["bundle"].get("run_documents") or [], "space_link_state": snapshot["view"].get("space_link_state") or {}, "runtime_stage_history": snapshot["view"].get("runtime_stage_history") or {}, "space_identity": snapshot["bundle"].get("space_identity") or {}, "links": snapshot["view"].get("links") or {}}) @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"]) upsert_run_index_entry(run_id, bucket_source=bucket_source, summary=cancelled_summary, 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 {}, "final_status_reconciliation": bundle.get("final_status_reconciliation") or {}, "eval_record": bundle.get("eval_record") or {}, "api_schema": bundle.get("api_schema") or {}, "space_test_endpoint_registry": bundle.get("space_test_endpoint_registry") or {}, "gradio_endpoint_discovery": bundle.get("gradio_endpoint_discovery") or {}, "gradio_schema": bundle.get("gradio_schema") or {}, "endpoint_candidates": bundle.get("endpoint_candidates") or {}, "selected_endpoint": bundle.get("selected_endpoint") 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 {}, "repair_attempts": bundle.get("repair_attempts") or {}, "run_documents": view.get("run_documents") or bundle.get("run_documents") or [], "validation_metrics": view.get("validation_metrics") or {}, "post_build_validation_status": bundle.get("post_build_validation_status") or {}, "manual_validation_status": bundle.get("manual_validation_status") or {}, "linked_validations": bundle.get("linked_validations") or {}, "events": events[-100:], "events_recent": events[-100:], "events_count": len(events), "view": view, "eval_publish": snapshot["eval_publish"], "eval_publish_status": snapshot["eval_publish"], "links": view.get("links") or bundle.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"), ), "space_link_state": view.get("space_link_state") or {}, "runtime_stage_history": view.get("runtime_stage_history") or {}, "space_identity": bundle.get("space_identity") or {}, } ) 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//`." ) 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 /. 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=[ "zai-org/GLM-5.2", "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="zai-org/GLM-5.2", 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]: success_tokens = {"full_inference_success", "success", "succeeded", "passed", "validated_after_space_test", "validated_after_manual_space_test", "recovered_by_space_test", "recovered_by_manual_validation", "manual_validation_passed", "manual_validated"} failure_tokens = {"partial_validation", "manual_hardware_required", "generated_needs_manual_hardware", "technical_blocker", "technical_blocker_boot_only", "blocked", "completed_with_warnings", "failed", "failure", "error", "auth_refresh_required", "stale", "stopped", "cancelled", "canceled"} successes = [row for row in linked if str(row.get("status") or row.get("effective_status") or "").lower() in success_tokens] failures = [row for row in linked if str(row.get("status") or row.get("validation_status") or row.get("result_status") or row.get("effective_status") or "").lower() in failure_tokens] 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 _cleared_space_test_endpoint_registry(parent_run_id: str) -> dict[str, Any]: return { "schema_version": "1.0", "status": "cleared", "parent_build_run_id": parent_run_id, "target_space_id": "", "validation_run_id": "", "requested_api_name": "", "preferred_api_name": "", "selected_api_name": "", "discovered_api_names": [], "endpoint_candidates": [], "endpoint_discovery_required": False, "payload_source": "", "schema": {}, "cleared_reason": "no_linked_validation_runs", "updated_at": utc_now_iso(), } 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 row.get("validation_status") or row.get("result_status") or "unknown", "effective_status": row.get("effective_status") or "", "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) write_json(f"{paths.root}/post_build_validation_status.json", manual_status, token=token) if not compact: write_json(f"{paths.root}/space_test_endpoint_registry.json", _cleared_space_test_endpoint_registry(parent_run_id), token=token) try: index_summary = { "run_id": parent_run_id, "updated_at": linked_payload.get("updated_at"), "linked_validation_count": len(compact), "linked_validations_count": len(compact), "manual_validation_passed": False, } if manual_status.get("status") == "success": index_summary.update({ "manual_validation_passed": True, "post_build_status": linked_payload.get("effective_status") or "validated_after_manual_space_test", "effective_status": linked_payload.get("effective_status") or "validated_after_manual_space_test", "latency_seconds": manual_status.get("latency_seconds"), "observed_latency_seconds": manual_status.get("observed_latency_seconds"), "recommended_zero_gpu_duration_seconds": manual_status.get("recommended_zero_gpu_duration_seconds"), "recommendation_source": manual_status.get("recommendation_source") or "linked_space_test", "recommendation_hardware": manual_status.get("recommendation_hardware"), "hardware_used_for_validation": manual_status.get("hardware_used_for_validation"), }) else: parent_state = read_json(f"{paths.root}/state.json", token=token) or {} parent_summary = read_json(f"{paths.root}/summary.json", token=token) or {} parent_status = str(parent_state.get("status") or parent_summary.get("status") or "unknown").strip() or "unknown" index_summary.update({ "post_build_status": "none", "effective_status": parent_status, }) upsert_run_index_entry(parent_run_id, bucket_source=bucket_source, summary=index_summary, token=token) except Exception: pass 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()