Upload 8 files

src/auth.py

@@ -0,0 +1,199 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi import HTTPException, Request
+from huggingface_hub import HfApi
+
+try:  # huggingface_hub>=1.0 provides official FastAPI helpers.
+    from huggingface_hub import parse_huggingface_oauth
+except Exception:  # pragma: no cover - compatibility fallback for older local envs.
+    parse_huggingface_oauth = None  # type: ignore[assignment]
+
+from .security import redact
+
+REQUIRED_OAUTH_SCOPES: set[str] = {
+    "read-repos",
+    "write-repos",
+    "manage-repos",
+    "gated-repos",
+    "inference-api",
+    "jobs",
+    "read-billing",
+}
+
+
+@dataclass(frozen=True)
+class OAuthContext:
+    username: str
+    token: str
+    profile: dict[str, Any] = field(default_factory=dict)
+    scopes: set[str] = field(default_factory=set)
+    expires_at: datetime | None = None
+    is_pro: bool | None = None
+    can_pay: bool | None = None
+
+    @property
+    def missing_scopes(self) -> list[str]:
+        return sorted(REQUIRED_OAUTH_SCOPES - self.scopes)
+
+    @property
+    def is_expired(self) -> bool:
+        if self.expires_at is None:
+            return False
+        return self.expires_at <= datetime.now(timezone.utc)
+
+
+def _parse_scope(scope: Any) -> set[str]:
+    if not scope:
+        return set()
+    if isinstance(scope, str):
+        # HF OAuth scope strings are space-separated; be tolerant of comma lists.
+        return {part for chunk in scope.split(",") for part in chunk.split() if part}
+    if isinstance(scope, (list, tuple, set)):
+        return {str(part) for part in scope if part}
+    return {str(scope)}
+
+
+def _normalize_expires_at(value: Any) -> datetime | None:
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        if value.tzinfo is None:
+            return value.replace(tzinfo=timezone.utc)
+        return value.astimezone(timezone.utc)
+    try:
+        return datetime.fromtimestamp(float(value), tz=timezone.utc)
+    except Exception:
+        return None
+
+
+def _ctx_from_official_parser(request: Request) -> OAuthContext | None:
+    if parse_huggingface_oauth is None:
+        return None
+    info = parse_huggingface_oauth(request)  # type: ignore[misc]
+    if info is None:
+        return None
+
+    user_info = getattr(info, "user_info", None)
+    username = getattr(user_info, "preferred_username", None) or getattr(user_info, "name", None)
+    token = getattr(info, "access_token", None)
+    if not username or not token:
+        return None
+
+    profile = {
+        "name": getattr(user_info, "name", None),
+        "preferred_username": getattr(user_info, "preferred_username", None),
+        "picture": getattr(user_info, "picture", None),
+        "email": getattr(user_info, "email", None),
+        "is_pro": getattr(user_info, "is_pro", None),
+        "can_pay": getattr(user_info, "can_pay", None),
+    }
+    return OAuthContext(
+        username=str(username),
+        token=str(token),
+        profile={k: v for k, v in profile.items() if v is not None},
+        scopes=_parse_scope(getattr(info, "scope", None)),
+        expires_at=_normalize_expires_at(getattr(info, "access_token_expires_at", None)),
+        is_pro=getattr(user_info, "is_pro", None),
+        can_pay=getattr(user_info, "can_pay", None),
+    )
+
+
+def _ctx_from_raw_session(request: Request) -> OAuthContext | None:
+    try:
+        oauth_info = request.session.get("oauth_info")  # type: ignore[attr-defined]
+    except Exception:
+        oauth_info = None
+    if not oauth_info:
+        return None
+
+    userinfo = oauth_info.get("userinfo") or {}
+    username = userinfo.get("preferred_username") or userinfo.get("username") or userinfo.get("name")
+    token = oauth_info.get("access_token")
+    if not username or not token:
+        return None
+
+    profile = {
+        "name": userinfo.get("name"),
+        "preferred_username": userinfo.get("preferred_username") or userinfo.get("username"),
+        "picture": userinfo.get("picture"),
+        "email": userinfo.get("email"),
+        "is_pro": userinfo.get("isPro") or userinfo.get("is_pro"),
+        "can_pay": userinfo.get("canPay") or userinfo.get("can_pay"),
+    }
+    return OAuthContext(
+        username=str(username),
+        token=str(token),
+        profile={k: v for k, v in profile.items() if v is not None},
+        scopes=_parse_scope(oauth_info.get("scope")),
+        expires_at=_normalize_expires_at(oauth_info.get("expires_at")),
+        is_pro=profile.get("is_pro"),
+        can_pay=profile.get("can_pay"),
+    )
+
+
+def extract_oauth_context(request: Request) -> OAuthContext:
+    """Extract and validate the signed-in HF user from the Gradio/FastAPI OAuth session.
+
+    Uses the official `huggingface_hub.parse_huggingface_oauth` helper first, then falls
+    back to the raw Gradio session shape for compatibility. The token is kept server-side
+    only and must never be returned by API responses.
+    """
+    ctx = _ctx_from_official_parser(request) or _ctx_from_raw_session(request)
+    if ctx is None:
+        raise HTTPException(status_code=401, detail="Please sign in with Hugging Face first.")
+    if ctx.is_expired:
+        raise HTTPException(status_code=401, detail="Your Hugging Face OAuth session expired. Please sign in again.")
+    return ctx
+
+
+def public_oauth_context(ctx: OAuthContext) -> dict[str, Any]:
+    return {
+        "username": ctx.username,
+        "profile": {
+            "name": ctx.profile.get("name"),
+            "preferred_username": ctx.profile.get("preferred_username") or ctx.username,
+            "picture": ctx.profile.get("picture"),
+            "is_pro": ctx.is_pro,
+            "can_pay": ctx.can_pay,
+        },
+        "scopes": sorted(ctx.scopes),
+        "missing_scopes": ctx.missing_scopes,
+        "expires_at": ctx.expires_at.isoformat() if ctx.expires_at else None,
+        "authenticated": True,
+    }
+
+
+def oauth_warning_messages(ctx: OAuthContext) -> list[str]:
+    warnings: list[str] = []
+    if ctx.missing_scopes:
+        warnings.append("Missing OAuth scopes: " + ", ".join(ctx.missing_scopes))
+    if ctx.can_pay is False:
+        warnings.append("No billing/payment method is visible through OAuth; fixed GPU hardware may require manual action.")
+    return warnings
+
+
+def verify_token_identity(ctx: OAuthContext) -> dict[str, Any]:
+    """Best-effort diagnostics endpoint helper. Never returns the raw token."""
+    try:
+        info = HfApi(token=ctx.token).whoami()
+        name = info.get("name") or info.get("fullname") or info.get("preferred_username")
+        return {
+            "ok": True,
+            "oauth_username": ctx.username,
+            "whoami_name": name,
+            "matches_oauth_user": name == ctx.username if name else None,
+            "can_pay": ctx.can_pay,
+            "is_pro": ctx.is_pro,
+            "missing_scopes": ctx.missing_scopes,
+        }
+    except Exception as exc:  # noqa: BLE001
+        return {
+            "ok": False,
+            "oauth_username": ctx.username,
+            "error": redact(str(exc)),
+            "missing_scopes": ctx.missing_scopes,
+        }

src/bucket.py

@@ -0,0 +1,300 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from typing import Any
+
+from huggingface_hub import HfFileSystem, bucket_info, create_bucket
+
+from .config import bucket_uri_from_source, normalize_bucket_name, settings, user_bucket_source
+from .security import redact
+
+
+@dataclass(frozen=True)
+class RunPaths:
+    run_id: str
+    bucket_source: str
+
+    @property
+    def bucket_uri(self) -> str:
+        return bucket_uri_from_source(self.bucket_source)
+
+    @property
+    def root(self) -> str:
+        return f"{self.bucket_uri}/runs/{self.run_id}"
+
+    @property
+    def state(self) -> str:
+        return f"{self.root}/state.json"
+
+    @property
+    def events(self) -> str:
+        return f"{self.root}/events.jsonl"
+
+    @property
+    def report(self) -> str:
+        return f"{self.root}/report.md"
+
+
+def _fs(token: str | None = None) -> HfFileSystem:
+    return HfFileSystem(token=token)
+
+
+def check_user_bucket(*, username: str, bucket_name: str | None = None, token: str | None = None) -> dict[str, Any]:
+    """Return bucket readiness for the signed-in user without creating resources."""
+    bucket_source = user_bucket_source(username=username, bucket_name=bucket_name)
+    bucket_uri = bucket_uri_from_source(bucket_source)
+    try:
+        info = bucket_info(bucket_source, token=token)
+        return {
+            "ok": True,
+            "exists": True,
+            "bucket_source": bucket_source,
+            "bucket_uri": bucket_uri,
+            "name": getattr(info, "name", normalize_bucket_name(bucket_name)),
+            "private": getattr(info, "private", None),
+        }
+    except Exception as exc:  # noqa: BLE001 - report readable status in UI
+        error = str(exc)
+        not_found = any(marker in error.lower() for marker in ["404", "not found", "repository not found", "bucket not found"])
+        return {
+            "ok": False,
+            "exists": False if not_found else None,
+            "bucket_source": bucket_source,
+            "bucket_uri": bucket_uri,
+            "error": error,
+        }
+
+
+def create_user_bucket(*, username: str, bucket_name: str | None = None, token: str | None = None) -> dict[str, Any]:
+    """Create the signed-in user's private run bucket, then return readiness."""
+    bucket_source = user_bucket_source(username=username, bucket_name=bucket_name)
+    try:
+        url = create_bucket(bucket_source, private=True, exist_ok=True, token=token)
+    except Exception as exc:  # noqa: BLE001
+        return {
+            "ok": False,
+            "exists": None,
+            "bucket_source": bucket_source,
+            "bucket_uri": bucket_uri_from_source(bucket_source),
+            "error": str(exc),
+        }
+    status = check_user_bucket(username=username, bucket_name=bucket_name, token=token)
+    status["created_or_existing"] = True
+    status["create_url"] = str(url)
+    return status
+
+
+def assert_user_bucket_ready(*, username: str, bucket_name: str | None = None, token: str | None = None) -> dict[str, Any]:
+    """Raise a clear error if the user's run bucket cannot be used."""
+    status = check_user_bucket(username=username, bucket_name=bucket_name, token=token)
+    if not status.get("ok"):
+        source = status.get("bucket_source") or user_bucket_source(username=username, bucket_name=bucket_name)
+        error = status.get("error") or "Bucket does not exist or is not accessible."
+        raise ValueError(
+            f"Run bucket `{source}` is not ready. Click 'Create private run bucket' first, "
+            f"or create it manually in Hugging Face Storage Buckets. Details: {error}"
+        )
+    return status
+
+
+def read_text(path: str, token: str | None = None) -> str | None:
+    fs = _fs(token)
+    try:
+        with fs.open(path, "r") as f:
+            return f.read()
+    except FileNotFoundError:
+        return None
+    except Exception as exc:  # noqa: BLE001 - surface readable error in UI
+        return f"[Could not read {path}: {exc}]"
+
+
+def read_json(path: str, token: str | None = None) -> dict[str, Any] | None:
+    content = read_text(path, token=token)
+    if not content or content.startswith("[Could not read"):
+        return None
+    try:
+        return json.loads(content)
+    except json.JSONDecodeError:
+        return {"_error": "Invalid JSON", "raw": redact(content)}
+
+
+def read_events(run_id: str, *, bucket_source: str, token: str | None = None) -> list[dict[str, Any]]:
+    paths = RunPaths(run_id, bucket_source=bucket_source)
+    content = read_text(paths.events, token=token)
+    if not content:
+        return []
+    events: list[dict[str, Any]] = []
+    for line in content.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            events.append(json.loads(line))
+        except json.JSONDecodeError:
+            events.append({"step": "parse_events", "status": "warning", "message": redact(line)})
+    return events
+
+
+def _safe_read_json(path: str, token: str | None = None) -> dict[str, Any]:
+    return read_json(path, token=token) or {}
+
+
+def _safe_read_text(path: str, token: str | None = None) -> str:
+    return redact(read_text(path, token=token) or "")
+
+
+def _list_run_files(
+    run_id: str,
+    *,
+    bucket_source: str,
+    token: str | None = None,
+    prefixes: tuple[str, ...] = ("generated", "tests", "logs", "artifacts", "traces", "repair"),
+    max_files: int = 120,
+) -> list[dict[str, Any]]:
+    """Return a compact, best-effort file index for a run.
+
+    The Run Explorer should not fail when a Bucket contains partial or old runs.
+    This function therefore treats every listing error as non-fatal and limits
+    recursion so the UI stays responsive even when traces are large.
+    """
+    fs = _fs(token)
+    base = f"{bucket_uri_from_source(bucket_source)}/runs/{run_id}"
+    files: list[dict[str, Any]] = []
+
+    def walk(prefix: str, depth: int = 0) -> None:
+        if len(files) >= max_files or depth > 3:
+            return
+        path = f"{base}/{prefix}"
+        try:
+            entries = fs.ls(path, detail=True)
+        except Exception:
+            return
+        for entry in entries:
+            if len(files) >= max_files:
+                break
+            name = entry.get("name") if isinstance(entry, dict) else str(entry)
+            if not name:
+                continue
+            typ = str(entry.get("type") or "") if isinstance(entry, dict) else ""
+            size = entry.get("size") if isinstance(entry, dict) else None
+            rel = name.replace(base + "/", "", 1)
+            if typ == "directory":
+                walk(rel, depth + 1)
+                continue
+            files.append(
+                {
+                    "path": rel,
+                    "name": rel.split("/")[-1],
+                    "size": size,
+                    "url": f"https://huggingface.co/buckets/{bucket_source}/blob/main/runs/{run_id}/{rel}",
+                }
+            )
+
+    for prefix in prefixes:
+        walk(prefix)
+    return files
+
+
+def summarize_run_bundle(run_id: str, bundle: dict[str, Any], *, bucket_source: str) -> dict[str, Any]:
+    state = bundle.get("state") or {}
+    gate = bundle.get("inference_gate") or {}
+    smoke = bundle.get("generation_smoke") or {}
+    hardware = bundle.get("hardware_strategy") or {}
+    target_space = state.get("target_space") or ""
+    status = state.get("status") or gate.get("status") or smoke.get("status") or "unknown"
+    return {
+        "run_id": run_id,
+        "kind": state.get("kind") or "unknown",
+        "status": status,
+        "model_id": state.get("model_id") or state.get("model") or "",
+        "target_space": target_space,
+        "target_space_url": state.get("target_space_url") or (f"https://huggingface.co/spaces/{target_space}" if target_space else ""),
+        "job_url": state.get("job_url") or "",
+        "created_at": state.get("created_at") or "",
+        "updated_at": state.get("updated_at") or state.get("created_at") or "",
+        "selected_hardware": hardware.get("selected_hardware") or state.get("selected_hardware") or state.get("hardware") or "",
+        "manual_hardware_required": bool(gate.get("manual_hardware_required") or hardware.get("manual_action_required")),
+        "health_passed": bool((gate.get("implementation_signals") or {}).get("health_passed") or smoke.get("health_passed")),
+        "smoke_test_passed": bool(smoke.get("ok") or smoke.get("status") == "success"),
+        "latency_seconds": smoke.get("latency_seconds"),
+        "expected_output_type": smoke.get("expected_output_type") or state.get("expected_output_type") or "",
+        "artifacts_url": f"https://huggingface.co/buckets/{bucket_source}/tree/main/runs/{run_id}",
+    }
+
+
+def read_run_bundle(run_id: str, *, bucket_source: str, token: str | None = None) -> dict[str, Any]:
+    paths = RunPaths(run_id, bucket_source=bucket_source)
+    bundle = {
+        "paths": {
+            "root": paths.root,
+            "state": paths.state,
+            "events": paths.events,
+            "report": paths.report,
+        },
+        "state": read_json(paths.state, token=token) or {},
+        "events": read_events(run_id, bucket_source=bucket_source, token=token),
+        "report": _safe_read_text(paths.report, token=token),
+        "inference_gate": _safe_read_json(f"{paths.root}/inference_gate.json", token=token),
+        "generation_smoke": _safe_read_json(f"{paths.root}/tests/generation_smoke.json", token=token) or _safe_read_json(f"{paths.root}/generation_smoke.json", token=token),
+        "hardware_strategy": _safe_read_json(f"{paths.root}/hardware_strategy.json", token=token),
+        "hardware_attempts": _safe_read_json(f"{paths.root}/hardware_attempts.json", token=token),
+        "technical_blockers": _safe_read_json(f"{paths.root}/TECHNICAL_BLOCKERS.json", token=token),
+        "model_analysis": _safe_read_json(f"{paths.root}/model_analysis.json", token=token),
+        "space_runtime": _safe_read_json(f"{paths.root}/space_runtime.json", token=token),
+        "files": _list_run_files(run_id, bucket_source=bucket_source, token=token),
+    }
+    bundle["summary"] = summarize_run_bundle(run_id, bundle, bucket_source=bucket_source)
+    return bundle
+
+
+def _run_id_from_path(path: str) -> str:
+    return path.rstrip('/').split('/')[-1]
+
+
+def list_recent_runs(
+    *,
+    bucket_source: str,
+    token: str | None = None,
+    limit: int = 50,
+    query: str | None = None,
+    status: str | None = None,
+) -> list[dict[str, Any]]:
+    """List recent run summaries from a user's bucket.
+
+    Best effort: if a run folder is incomplete, return the information that is
+    available instead of failing the entire explorer.
+    """
+    fs = _fs(token)
+    root = f"{bucket_uri_from_source(bucket_source)}/runs"
+    try:
+        entries = fs.ls(root, detail=True)
+    except FileNotFoundError:
+        return []
+    except Exception:
+        return []
+
+    runs: list[dict[str, Any]] = []
+    for entry in entries:
+        name = entry.get("name") if isinstance(entry, dict) else str(entry)
+        if not name:
+            continue
+        run_id = _run_id_from_path(name)
+        if run_id in {"runs", ""}:
+            continue
+        state = read_json(f"{root}/{run_id}/state.json", token=token) or {}
+        gate = read_json(f"{root}/{run_id}/inference_gate.json", token=token) or {}
+        smoke = read_json(f"{root}/{run_id}/tests/generation_smoke.json", token=token) or read_json(f"{root}/{run_id}/generation_smoke.json", token=token) or {}
+        hardware = read_json(f"{root}/{run_id}/hardware_strategy.json", token=token) or {}
+        partial_bundle = {"state": state, "inference_gate": gate, "generation_smoke": smoke, "hardware_strategy": hardware}
+        item = summarize_run_bundle(run_id, partial_bundle, bucket_source=bucket_source)
+        item["bucket_source"] = bucket_source
+        haystack = " ".join(str(item.get(k, "")) for k in ["run_id", "model_id", "target_space", "status", "kind"]).lower()
+        if query and query.lower() not in haystack:
+            continue
+        if status and status not in {"all", ""} and item["status"] != status:
+            continue
+        runs.append(item)
+
+    runs.sort(key=lambda r: str(r.get("updated_at") or r.get("created_at") or r.get("run_id")), reverse=True)
+    return runs[: max(1, min(int(limit or 50), 200))]

src/config.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Settings:
+    """Runtime configuration for the orchestrator Space."""
+
+    bucket_name: str = os.getenv("SPACE_FACTORY_BUCKET_NAME", "space-factory-runs")
+    bucket_mount: str = os.getenv("SPACE_FACTORY_BUCKET_MOUNT", "/output")
+    job_flavor: str = os.getenv("SPACE_FACTORY_JOB_FLAVOR", "cpu-basic")
+    job_timeout: str = os.getenv("SPACE_FACTORY_JOB_TIMEOUT", "30m")
+    job_image: str = os.getenv("SPACE_FACTORY_JOB_IMAGE", "python:3.12")
+
+
+def normalize_bucket_name(value: str | None) -> str:
+    """Return a safe bucket name, defaulting to the product bucket name."""
+    bucket = (value or settings.bucket_name).strip().strip("/")
+    if not bucket:
+        bucket = settings.bucket_name
+    if "/" in bucket:
+        raise ValueError("Bucket name must be a name like 'space-factory-runs', not owner/name.")
+    return bucket
+
+
+def user_bucket_source(*, username: str, bucket_name: str | None = None) -> str:
+    """Return the per-user bucket source expected by HF Jobs volumes."""
+    clean_user = (username or "").strip()
+    if not clean_user:
+        raise ValueError("Missing username for per-user bucket source.")
+    return f"{clean_user}/{normalize_bucket_name(bucket_name)}"
+
+
+def bucket_uri_from_source(bucket_source: str) -> str:
+    return f"hf://buckets/{bucket_source}"
+
+
+settings = Settings()

src/jobs.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from huggingface_hub import Volume, fetch_job_logs, inspect_job, run_job
+
+from .config import bucket_uri_from_source, user_bucket_source, settings
+from .bucket import assert_user_bucket_ready
+from .runs import make_run_id, utc_now_iso, validate_run_id
+from .worker_payload import (
+    encoded_universal_model_card_worker_script,
+    encoded_validate_existing_space_worker_script,
+    python_decode_and_run_command,
+)
+
+SPACE_SLUG_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{1,95}$")
+
+
+def _base_env(*, run_id: str, username: str, bucket_source: str, worker_script_b64: str) -> dict[str, str]:
+    return {
+        "RUN_ID": run_id,
+        "HF_USERNAME": username or "unknown",
+        "BUCKET_SOURCE": bucket_source,
+        "OUTPUT_ROOT": settings.bucket_mount,
+        "WORKER_SCRIPT_B64": worker_script_b64,
+        "LAUNCHED_AT": utc_now_iso(),
+    }
+
+
+def _launch_job(*, token: str, env: dict[str, str], bucket_source: str, flavor: str | None = None, timeout: str | None = None) -> Any:
+    return run_job(
+        image=settings.job_image,
+        command=python_decode_and_run_command(),
+        flavor=flavor or settings.job_flavor,
+        timeout=timeout or settings.job_timeout,
+        env=env,
+        secrets={"HF_TOKEN": token},
+        volumes=[Volume(type="bucket", source=bucket_source, mount_path=settings.bucket_mount)],
+        token=token,
+    )
+
+
+def _job_result(job: Any, *, run_id: str, kind: str, bucket_source: str, extra: dict[str, Any] | None = None) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        "run_id": run_id,
+        "kind": kind,
+        "job_id": job.id,
+        "job_url": getattr(job, "url", None),
+        "status": getattr(getattr(job, "status", None), "stage", None),
+        "bucket_source": bucket_source,
+        "bucket_uri": bucket_uri_from_source(bucket_source),
+    }
+    if extra:
+        payload.update(extra)
+    return payload
+
+
+def normalize_target_space(*, username: str, target_slug: str | None, run_id: str) -> str:
+    """Return `username/slug`, constrained to the signed-in user's namespace."""
+    slug = (target_slug or "").strip()
+    if not slug:
+        slug = f"space-factory-{run_id}".lower()[:80]
+    if "/" in slug:
+        namespace, repo = slug.split("/", 1)
+        if namespace != username:
+            raise ValueError("The target Space must be created in your own namespace.")
+        slug = repo
+    if not SPACE_SLUG_RE.match(slug):
+        raise ValueError("Invalid target Space name. Use letters, numbers, dots, underscores, or dashes.")
+    return f"{username}/{slug}"
+
+
+def _clean_repo_id(value: str | None, *, repo_kind: str) -> str:
+    cleaned = (value or "").strip()
+    cleaned = cleaned.replace("https://huggingface.co/spaces/", "")
+    cleaned = cleaned.replace("https://huggingface.co/", "")
+    cleaned = cleaned.strip("/")
+    if "/" not in cleaned:
+        raise ValueError(f"{repo_kind} must look like owner/name or a Hugging Face URL.")
+    return cleaned
+
+
+def launch_universal_model_card_job(
+    *,
+    token: str,
+    username: str,
+    target_slug: str | None = None,
+    model_id: str | None = None,
+    pi_model: str | None = None,
+    preferred_space_hardware: str | None = None,
+    fallback_space_hardware: str | None = None,
+    allow_fixed_gpu_fallback: bool = True,
+    implementation_mode: str | None = None,
+    run_id: str | None = None,
+    bucket_name: str | None = None,
+) -> dict[str, Any]:
+    """Launch the public product builder: model card → private Space attempt."""
+    if not token:
+        raise ValueError("Missing OAuth token. Please sign in with Hugging Face first.")
+    safe_run_id = validate_run_id(run_id) if run_id else make_run_id("universal")
+    target_space_id = normalize_target_space(username=username, target_slug=target_slug, run_id=safe_run_id)
+    clean_model_id = _clean_repo_id(model_id, repo_kind="Model ID")
+    bucket_source = user_bucket_source(username=username, bucket_name=bucket_name)
+    assert_user_bucket_ready(username=username, bucket_name=bucket_name, token=token)
+
+    env = _base_env(
+        run_id=safe_run_id,
+        username=username,
+        bucket_source=bucket_source,
+        worker_script_b64=encoded_universal_model_card_worker_script(),
+    )
+    env["TARGET_SPACE_ID"] = target_space_id
+    env["MODEL_ID"] = clean_model_id
+    env["PI_MODEL"] = (pi_model or "Qwen/Qwen3-Coder-Next").strip()
+    env["PREFERRED_SPACE_HARDWARE"] = (preferred_space_hardware or "zero-a10g").strip()
+    env["FALLBACK_SPACE_HARDWARE"] = (fallback_space_hardware or "l40sx1").strip()
+    env["ALLOW_FIXED_GPU_FALLBACK"] = "true" if allow_fixed_gpu_fallback else "false"
+    env["IMPLEMENTATION_MODE"] = (implementation_mode or "full-inference-gated").strip()
+    job = _launch_job(token=token, env=env, bucket_source=bucket_source, timeout="60m")
+    return _job_result(
+        job,
+        run_id=safe_run_id,
+        kind="universal_model_card_builder",
+        bucket_source=bucket_source,
+        extra={
+            "target_space": target_space_id,
+            "target_space_url": f"https://huggingface.co/spaces/{target_space_id}",
+            "model_id": clean_model_id,
+            "pi_model": env["PI_MODEL"],
+            "preferred_space_hardware": env["PREFERRED_SPACE_HARDWARE"],
+            "fallback_space_hardware": env["FALLBACK_SPACE_HARDWARE"],
+            "allow_fixed_gpu_fallback": allow_fixed_gpu_fallback,
+            "implementation_mode": env["IMPLEMENTATION_MODE"],
+        },
+    )
+
+
+def launch_validate_existing_space_job(
+    *,
+    token: str,
+    username: str,
+    target_space_id: str,
+    api_name: str | None = None,
+    test_args_json: str | None = None,
+    test_kwargs_json: str | None = None,
+    expected_output_type: str | None = None,
+    live_timeout_seconds: int = 1800,
+    run_id: str | None = None,
+    bucket_name: str | None = None,
+) -> dict[str, Any]:
+    """Launch the public product validator for an existing generated Space."""
+    if not token:
+        raise ValueError("Missing OAuth token. Please sign in with Hugging Face first.")
+    safe_run_id = validate_run_id(run_id) if run_id else make_run_id("validate")
+    target = _clean_repo_id(target_space_id, repo_kind="Target Space")
+    namespace, _ = target.split("/", 1)
+    if namespace != username:
+        raise ValueError("For this version, target Space validation is limited to your own namespace.")
+    bucket_source = user_bucket_source(username=username, bucket_name=bucket_name)
+    assert_user_bucket_ready(username=username, bucket_name=bucket_name, token=token)
+    env = _base_env(
+        run_id=safe_run_id,
+        username=username,
+        bucket_source=bucket_source,
+        worker_script_b64=encoded_validate_existing_space_worker_script(),
+    )
+    env["TARGET_SPACE_ID"] = target
+    env["API_NAME"] = (api_name or "/generate").strip()
+    env["TEST_ARGS_JSON"] = (test_args_json or '["a cinematic robot cat astronaut, detailed, studio lighting"]').strip()
+    env["TEST_KWARGS_JSON"] = (test_kwargs_json or "{}").strip()
+    env["EXPECTED_OUTPUT_TYPE"] = (expected_output_type or "image").strip()
+    env["LIVE_TIMEOUT_SECONDS"] = str(int(live_timeout_seconds or 1800))
+    job = _launch_job(token=token, env=env, bucket_source=bucket_source, timeout="60m")
+    return _job_result(
+        job,
+        run_id=safe_run_id,
+        kind="validate_existing_space",
+        bucket_source=bucket_source,
+        extra={
+            "target_space": target,
+            "target_space_url": f"https://huggingface.co/spaces/{target}",
+            "api_name": env["API_NAME"],
+            "expected_output_type": env["EXPECTED_OUTPUT_TYPE"],
+            "test_args_json": env["TEST_ARGS_JSON"],
+            "test_kwargs_json": env["TEST_KWARGS_JSON"],
+        },
+    )
+
+
+def inspect_job_safe(job_id: str, token: str | None = None) -> dict[str, Any]:
+    if not job_id:
+        return {"error": "Missing job_id"}
+    try:
+        info = inspect_job(job_id=job_id, token=token)
+        status = getattr(info, "status", None)
+        return {
+            "id": info.id,
+            "url": getattr(info, "url", None),
+            "stage": getattr(status, "stage", None),
+            "message": getattr(status, "message", None),
+            "flavor": getattr(info, "flavor", None),
+            "created_at": str(getattr(info, "created_at", "")),
+            "started_at": str(getattr(info, "started_at", "")),
+            "finished_at": str(getattr(info, "finished_at", "")),
+        }
+    except Exception as exc:  # noqa: BLE001
+        return {"error": str(exc)}
+
+
+def fetch_recent_logs_safe(job_id: str, token: str | None = None, max_lines: int = 120) -> str:
+    if not job_id:
+        return ""
+    try:
+        logs = list(fetch_job_logs(job_id=job_id, token=token))
+        return "\n".join(str(line).rstrip("\n") for line in logs[-max_lines:])
+    except Exception as exc:  # noqa: BLE001
+        return f"Could not fetch job logs: {exc}"

src/progress.py

@@ -0,0 +1,197 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any
+
+STEP_ORDER = [
+    "bucket_ready",
+    "job_launched",
+    "dependencies",
+    "auth",
+    "model_analysis",
+    "workspace",
+    "node",
+    "pi_install",
+    "pi_config",
+    "pi_run",
+    "create_space",
+    "upload_files",
+    "hardware",
+    "api_validation",
+    "inference_gate",
+    "report_write",
+    "done",
+]
+
+STEP_LABELS = {
+    "bucket_ready": "Bucket ready",
+    "job_launched": "Job launched",
+    "dependencies": "Dependencies",
+    "auth": "Authenticated",
+    "model_analysis": "Model analysis",
+    "workspace": "Workspace",
+    "node": "Node/npm",
+    "pi_install": "Pi installed",
+    "pi_config": "Pi configured",
+    "pi_run": "Pi running",
+    "create_space": "Create Space",
+    "upload_files": "Upload files",
+    "hardware": "Hardware request",
+    "api_validation": "API validation",
+    "inference_gate": "Gate",
+    "report_write": "Report",
+    "done": "Done",
+}
+
+STEP_PROGRESS = {
+    "bucket_ready": 5,
+    "job_launched": 8,
+    "dependencies": 12,
+    "auth": 16,
+    "model_analysis": 22,
+    "workspace": 28,
+    "node": 34,
+    "pi_install": 40,
+    "pi_config": 44,
+    "pi_run": 62,
+    "create_space": 72,
+    "upload_files": 80,
+    "hardware": 86,
+    "api_validation": 92,
+    "inference_gate": 96,
+    "report_write": 98,
+    "done": 100,
+}
+
+STEP_ALIASES = {
+    "bootstrap": "job_launched",
+    "dependencies": "dependencies",
+    "auth": "auth",
+    "model_analysis": "model_analysis",
+    "workspace": "workspace",
+    "node": "node",
+    "pi_install": "pi_install",
+    "pi_config": "pi_config",
+    "pi_run": "pi_run",
+    "create_space": "create_space",
+    "upload_files": "upload_files",
+    "hardware_preferred": "hardware",
+    "hardware_fallback": "hardware",
+    "hardware": "hardware",
+    "api_validation": "api_validation",
+    "inference_gate": "inference_gate",
+    "report_write": "report_write",
+    "done": "done",
+    "failure": "done",
+}
+
+DONE_STATUSES = {"success", "done", "completed", "passed", "full_inference_success", "full_inference_candidate_health_passed", "manual_hardware_required", "technical_blocker"}
+RUNNING_STATUSES = {"started", "running", "waiting"}
+FAILED_STATUSES = {"failed", "error"}
+
+
+def _parse_ts(ts: str | None) -> datetime | None:
+    if not ts:
+        return None
+    try:
+        return datetime.fromisoformat(ts.replace("Z", "+00:00"))
+    except Exception:
+        return None
+
+
+def _canonical_step(step: str | None) -> str | None:
+    if not step:
+        return None
+    return STEP_ALIASES.get(step, step if step in STEP_PROGRESS else None)
+
+
+def progress_from_events(events: list[dict[str, Any]] | None, *, state: dict[str, Any] | None = None) -> dict[str, Any]:
+    """Build a stable UI progress model from worker events.jsonl + optional state.json."""
+    events = events or []
+    state = state or {}
+    step_status = {step: "pending" for step in STEP_ORDER}
+    last_event = None
+    current_step = "job_launched"
+    terminal_status = None
+    first_ts = None
+    last_ts = None
+
+    for event in events:
+        if not isinstance(event, dict):
+            continue
+        last_event = event
+        step = _canonical_step(event.get("step"))
+        status = str(event.get("status") or "").lower()
+        ts = _parse_ts(event.get("ts"))
+        if ts and not first_ts:
+            first_ts = ts
+        if ts:
+            last_ts = ts
+        if not step:
+            continue
+        current_step = step
+        if status in FAILED_STATUSES:
+            step_status[step] = "failed"
+            terminal_status = "failed"
+        elif status in RUNNING_STATUSES:
+            if step_status.get(step) != "done":
+                step_status[step] = "running"
+        elif status in DONE_STATUSES or status:
+            step_status[step] = "done"
+
+        if step == "done" and status:
+            terminal_status = status
+
+    # Mark all previous steps as done up to the current running/done step.
+    current_index = STEP_ORDER.index(current_step) if current_step in STEP_ORDER else 0
+    for step in STEP_ORDER[:current_index]:
+        if step_status[step] == "pending":
+            step_status[step] = "done"
+
+    if terminal_status and terminal_status != "failed":
+        current_step = "done"
+        for step in STEP_ORDER:
+            if step_status[step] != "failed":
+                step_status[step] = "done"
+
+    status_from_state = state.get("status") or state.get("gate_status")
+    overall_status = terminal_status or status_from_state or ("running" if events else "not_started")
+
+    if overall_status in {"failed", "error"}:
+        progress = max(STEP_PROGRESS.get(current_step, 8), 8)
+    elif current_step == "done" or overall_status in DONE_STATUSES:
+        progress = 100
+    else:
+        progress = STEP_PROGRESS.get(current_step, 8)
+        if step_status.get(current_step) == "running":
+            previous = STEP_ORDER[max(0, current_index - 1)] if current_index > 0 else current_step
+            progress = max(STEP_PROGRESS.get(previous, 0) + 2, progress - 8)
+
+    now = datetime.now(timezone.utc)
+    elapsed = int(((last_ts or now) - first_ts).total_seconds()) if first_ts else 0
+    if not first_ts and state.get("created_at"):
+        created = _parse_ts(str(state.get("created_at")))
+        if created:
+            elapsed = int((now - created).total_seconds())
+
+    timeline = [
+        {"step": step, "label": STEP_LABELS[step], "status": step_status[step]}
+        for step in STEP_ORDER
+    ]
+
+    last_message = None
+    if last_event:
+        last_message = last_event.get("message") or last_event.get("step")
+
+    return {
+        "status": overall_status,
+        "progress": int(max(0, min(100, progress))),
+        "current_step": current_step,
+        "current_step_label": STEP_LABELS.get(current_step, current_step),
+        "last_event": last_message or "No events yet",
+        "elapsed_seconds": max(0, elapsed),
+        "eta_seconds": None,
+        "timeline": timeline,
+        "last_event_raw": last_event or {},
+    }

src/runs.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+import re
+import uuid
+from datetime import datetime, timezone
+
+RUN_ID_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_.-]{2,80}$")
+
+
+def utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def make_run_id(prefix: str = "run") -> str:
+    stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
+    return f"{prefix}-{stamp}-{uuid.uuid4().hex[:8]}"
+
+
+def validate_run_id(run_id: str) -> str:
+    cleaned = (run_id or "").strip()
+    if not RUN_ID_RE.match(cleaned):
+        raise ValueError("Invalid run_id. Use 3-80 characters: letters, numbers, dots, underscores or dashes.")
+    return cleaned

src/security.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import re
+
+SECRET_PATTERNS = [
+    re.compile(r"hf_[A-Za-z0-9_\-]{20,}"),
+    re.compile(r"Bearer\s+[A-Za-z0-9_\.\-]+", re.IGNORECASE),
+    re.compile(r"(HF_TOKEN|OAUTH_TOKEN|ACCESS_TOKEN|AUTHORIZATION|PASSWORD|SECRET)\s*[:=]\s*[^\s]+", re.IGNORECASE),
+]
+
+
+def redact(text: str | None) -> str:
+    """Best-effort redaction for logs/reports shown in the UI.
+
+    This is intentionally conservative. It is not a complete DLP system,
+    but it protects against obvious token leaks in first-version outputs.
+    """
+    if not text:
+        return ""
+
+    redacted = text
+    for pattern in SECRET_PATTERNS:
+        redacted = pattern.sub("[REDACTED]", redacted)
+    return redacted

src/worker_payload.py

@@ -0,0 +1,1428 @@
+from __future__ import annotations
+
+import base64
+import textwrap
+
+
+def _encode(script: str) -> str:
+    return base64.b64encode(script.encode("utf-8")).decode("ascii")
+
+
+UNIVERSAL_MODEL_CARD_WORKER_SCRIPT = r'''
+
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from textwrap import dedent
+
+TARGET_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{1,95}/[A-Za-z0-9][A-Za-z0-9._-]{1,95}$")
+GIST_URL = "https://gist.github.com/gary149/2aba2962375fa9ca56bb9ef53f00b73d"
+DEFAULT_MODEL_ID = "sshleifer/tiny-gpt2"
+
+
+def now():
+    return datetime.now(timezone.utc).isoformat()
+
+
+def write_json(path: Path, payload: dict):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+
+
+def append_event(path: Path, step: str, status: str, message: str, data: dict | None = None):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    event = {"ts": now(), "step": step, "status": status, "message": message, "data": data or {}}
+    line = json.dumps(event, ensure_ascii=False)
+    with path.open("a", encoding="utf-8") as f:
+        f.write(line + "\n")
+    print(line, flush=True)
+
+
+def redact_text(text: str | None) -> str:
+    if not text:
+        return ""
+    value = text
+    for secret_name in ["HF_TOKEN", "HUGGING_FACE_HUB_TOKEN"]:
+        secret = os.environ.get(secret_name)
+        if secret:
+            value = value.replace(secret, "[REDACTED]")
+    value = re.sub(r"Bearer\s+[A-Za-z0-9_\-.=]+", "Bearer [REDACTED]", value)
+    value = re.sub(r"hf_[A-Za-z0-9_\-]{10,}", "hf_[REDACTED]", value)
+    return value
+
+
+def safe_details(details: dict | None) -> dict:
+    if not details:
+        return {}
+    try:
+        return json.loads(redact_text(json.dumps(details, ensure_ascii=False)))
+    except Exception:
+        return {"redacted_details": redact_text(str(details))[-4000:]}
+
+
+def fail(run_dir: Path, events_path: Path, message: str, details: dict | None = None, status: str = "failed"):
+    safe = safe_details(details)
+    append_event(events_path, "failure", "failed", message, safe)
+    write_json(run_dir / "state.json", {
+        "run_id": os.environ.get("RUN_ID"),
+        "kind": "universal_model_card_builder",
+        "status": status,
+        "message": message,
+        "updated_at": now(),
+        "details": safe,
+    })
+    report = f"""# Agentic Space Factory — model Article Reproduction Report
+
+Status: **{status}**
+
+{message}
+
+```json
+{json.dumps(safe, indent=2, ensure_ascii=False)}
+```
+"""
+    (run_dir / "report.md").write_text(report, encoding="utf-8")
+    raise SystemExit(1)
+
+
+def run_cmd(cmd: list[str], *, cwd: Path | None = None, env: dict | None = None, timeout: int = 600):
+    result = subprocess.run(cmd, cwd=str(cwd) if cwd else None, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout)
+    return result.returncode, redact_text(result.stdout)
+
+
+def install_python_deps(events_path: Path):
+    append_event(events_path, "dependencies", "started", "Installing Python worker dependencies")
+    code, out = run_cmd([sys.executable, "-m", "pip", "install", "-q", "--upgrade", "huggingface_hub>=1.0.0", "gradio_client>=2.0.0", "requests>=2.31.0"], timeout=600)
+    if code != 0:
+        append_event(events_path, "dependencies", "failed", "Python dependency installation failed", {"output_tail": out[-4000:]})
+        raise RuntimeError(out)
+    append_event(events_path, "dependencies", "success", "Python worker dependencies installed")
+
+
+def ensure_node(events_path: Path):
+    node = shutil.which("node")
+    npm = shutil.which("npm")
+    if node and npm:
+        _, node_v = run_cmd([node, "--version"], timeout=30)
+        _, npm_v = run_cmd([npm, "--version"], timeout=30)
+        append_event(events_path, "node", "success", "Node/npm already available", {"node": node_v.strip(), "npm": npm_v.strip()})
+        return
+    append_event(events_path, "node", "started", "Installing nodejs/npm through apt-get")
+    code, out = run_cmd(["bash", "-lc", "apt-get update -qq && apt-get install -y -qq nodejs npm"], timeout=600)
+    if code != 0:
+        append_event(events_path, "node", "failed", "Could not install nodejs/npm", {"output_tail": out[-4000:]})
+        raise RuntimeError(out)
+    append_event(events_path, "node", "success", "Installed nodejs/npm")
+
+
+def install_pi(events_path: Path):
+    ensure_node(events_path)
+    append_event(events_path, "pi_install", "started", "Installing Pi coding agent from npm")
+    code, out = run_cmd(["npm", "install", "-g", "@mariozechner/pi-coding-agent"], timeout=900)
+    if code != 0:
+        append_event(events_path, "pi_install", "failed", "Pi npm installation failed", {"output_tail": out[-4000:]})
+        raise RuntimeError(out)
+    code, version = run_cmd(["pi", "--version"], timeout=60)
+    append_event(events_path, "pi_install", "success", "Pi installed", {"version_output": version.strip()[-300:]})
+
+
+def configure_pi(events_path: Path, model: str):
+    pi_dir = Path.home() / ".pi" / "agent"
+    pi_dir.mkdir(parents=True, exist_ok=True)
+    (pi_dir / "auth.json").write_text(json.dumps({"huggingface": {"type": "api_key", "key": os.environ.get("HF_TOKEN", "")}}, indent=2), encoding="utf-8")
+    (pi_dir / "settings.json").write_text(json.dumps({"model": model, "provider": "huggingface", "autoRun": True, "autoApply": True}, indent=2), encoding="utf-8")
+    append_event(events_path, "pi_config", "success", "Configured Pi", {"model": model})
+
+
+def collect_pi_traces(run_dir: Path, events_path: Path):
+    traces_dir = Path.home() / ".pi" / "agent" / "sessions"
+    raw_dir = run_dir / "traces" / "raw"
+    redacted_dir = run_dir / "traces" / "redacted"
+    raw_dir.mkdir(parents=True, exist_ok=True)
+    redacted_dir.mkdir(parents=True, exist_ok=True)
+    count = 0
+    if traces_dir.exists():
+        for path in traces_dir.rglob("*.jsonl"):
+            rel = path.relative_to(traces_dir)
+            target_raw = raw_dir / rel
+            target_raw.parent.mkdir(parents=True, exist_ok=True)
+            text = path.read_text(encoding="utf-8", errors="ignore")
+            target_raw.write_text(text, encoding="utf-8")
+            target_redacted = redacted_dir / rel
+            target_redacted.parent.mkdir(parents=True, exist_ok=True)
+            target_redacted.write_text(redact_text(text), encoding="utf-8")
+            count += 1
+    append_event(events_path, "traces", "success", "Collected Pi traces", {"count": count})
+    return count
+
+
+def sanitize_model_id(model_id: str) -> str:
+    model_id = (model_id or DEFAULT_MODEL_ID).strip().replace("https://huggingface.co/", "")
+    model_id = model_id.split("?", 1)[0].strip("/")
+    if not re.match(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", model_id):
+        raise ValueError("MODEL_ID must look like owner/model-name")
+    return model_id
+
+
+def make_gradio_client(target_space_id: str, token: str):
+    import inspect
+    from gradio_client import Client
+    params = inspect.signature(Client).parameters
+    if "token" in params:
+        return Client(target_space_id, token=token)
+    if "hf_token" in params:
+        return Client(target_space_id, hf_token=token)
+    if "api_key" in params:
+        return Client(target_space_id, api_key=token)
+    if "headers" in params:
+        return Client(target_space_id, headers={"Authorization": f"Bearer {token}"})
+    return Client(target_space_id)
+
+
+def api_names_from_schema(schema) -> list[str]:
+    names: list[str] = []
+    if isinstance(schema, dict):
+        endpoints = schema.get("named_endpoints") or schema.get("endpoints") or {}
+        if isinstance(endpoints, dict):
+            for key, value in endpoints.items():
+                if isinstance(key, str) and key.startswith("/"):
+                    names.append(key)
+                if isinstance(value, dict):
+                    api_name = value.get("api_name")
+                    if isinstance(api_name, str) and api_name.startswith("/"):
+                        names.append(api_name)
+        if isinstance(schema.get("dependencies"), list):
+            for dep in schema["dependencies"]:
+                if isinstance(dep, dict):
+                    api_name = dep.get("api_name")
+                    if isinstance(api_name, str):
+                        names.append(api_name if api_name.startswith("/") else f"/{api_name}")
+    return list(dict.fromkeys(names))
+
+
+def space_subdomain_url(target_space_id: str) -> str:
+    owner, name = target_space_id.split("/", 1)
+    # This matches the common Spaces app URL pattern. Keep conservative: our
+    # generated slugs are ASCII and hyphen-friendly.
+    return f"https://{owner}-{name}.hf.space".replace("_", "-").lower()
+
+
+def runtime_to_dict(runtime) -> dict:
+    payload = {}
+    for attr in ["stage", "hardware", "requested_hardware", "sleep_time", "storage", "gc_timeout"]:
+        value = getattr(runtime, attr, None)
+        payload[attr] = getattr(value, "value", value)
+    return {k: str(v) if v is not None else None for k, v in payload.items()}
+
+
+def write_space_runtime(api, target_space_id: str, token: str, run_dir: Path, events_path: Path, attempt: int | None = None) -> dict:
+    try:
+        runtime = api.get_space_runtime(repo_id=target_space_id, token=token)
+        payload = runtime_to_dict(runtime)
+        payload["attempt"] = attempt
+        write_json(run_dir / "space_runtime.json", payload)
+        return payload
+    except Exception as exc:
+        payload = {"error": str(exc)[:2000], "attempt": attempt}
+        write_json(run_dir / "space_runtime.json", payload)
+        append_event(events_path, "space_runtime", "warning", "Could not fetch Space runtime", payload)
+        return payload
+
+
+def collect_space_logs(target_space_id: str, token: str, run_dir: Path, events_path: Path):
+    logs_dir = run_dir / "logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    env = os.environ.copy()
+    env["HF_TOKEN"] = token
+    commands = {
+        "space_logs_runtime.txt": ["hf", "spaces", "logs", target_space_id],
+        "space_logs_build.txt": ["hf", "spaces", "logs", target_space_id, "--build"],
+    }
+    written = []
+    for filename, cmd in commands.items():
+        try:
+            code, out = run_cmd(cmd, env=env, timeout=75)
+            (logs_dir / filename).write_text(out, encoding="utf-8")
+            written.append({"file": filename, "returncode": code, "tail": out[-1000:]})
+        except Exception as exc:
+            written.append({"file": filename, "error": str(exc)[:1000]})
+    append_event(events_path, "space_logs", "success", "Collected best-effort Space logs", {"files": written})
+    return written
+
+
+def validate_http_health(target_space_id: str, token: str, run_dir: Path, events_path: Path, attempt: int):
+    import requests
+    base_url = space_subdomain_url(target_space_id)
+    url = base_url.rstrip("/") + "/health"
+    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json,text/plain,*/*"}
+    response = requests.get(url, headers=headers, timeout=20)
+    payload = {
+        "status": "success" if response.ok else "failed",
+        "attempt": attempt,
+        "url": url,
+        "status_code": response.status_code,
+        "content_type": response.headers.get("content-type"),
+        "text": response.text[:2000],
+    }
+    if response.ok:
+        try:
+            payload["json"] = response.json()
+        except Exception:
+            pass
+        write_json(run_dir / "tests" / "http_health.json", payload)
+        write_json(run_dir / "tests" / "test_result.json", payload | {"validator": "http_get_health"})
+        append_event(events_path, "api_validation", "success", "HTTP /health validation passed", {"attempt": attempt, "url": url, "status_code": response.status_code})
+        return payload | {"validator": "http_get_health"}
+    raise RuntimeError(f"HTTP /health returned {response.status_code}: {response.text[:500]}")
+
+
+def validate_gradio_api(target_space_id: str, token: str, run_dir: Path, events_path: Path, attempt: int):
+    client = make_gradio_client(target_space_id, token)
+    schema = client.view_api(return_format="dict")
+    write_json(run_dir / "tests" / "api_schema.json", schema if isinstance(schema, dict) else {"schema": str(schema)})
+    discovered = api_names_from_schema(schema)
+    candidates = []
+    for name in ["/health", "/predict", "/greet"] + discovered:
+        if name not in candidates:
+            candidates.append(name)
+    errors = []
+    for api_name in candidates:
+        try:
+            if api_name == "/greet":
+                result = client.predict("Agentic Space Factory", api_name=api_name)
+            else:
+                result = client.predict(api_name=api_name)
+            payload = {"status": "success", "attempt": attempt, "api_name": api_name, "discovered_api_names": discovered, "result_repr": repr(result)[:2000], "validator": "gradio_client"}
+            write_json(run_dir / "tests" / "test_result.json", payload)
+            append_event(events_path, "api_validation", "success", "Gradio API validation passed", {"attempt": attempt, "api_name": api_name, "discovered_api_names": discovered})
+            return payload
+        except Exception as exc:
+            errors.append({"api_name": api_name, "error": str(exc)[:1000]})
+    raise RuntimeError("; ".join(f"{e['api_name']}: {e['error']}" for e in errors[:5]) or "No callable API endpoints found")
+
+
+def validate_live_api(api, target_space_id: str, token: str, run_dir: Path, events_path: Path, timeout_s: int = 900):
+    append_event(events_path, "api_validation", "started", "Waiting for live HTTP /health or Gradio API to become available")
+    deadline = time.time() + timeout_s
+    attempt = 0
+    last_error = None
+    runtime_error_count = 0
+    while time.time() < deadline:
+        attempt += 1
+        runtime_payload = write_space_runtime(api, target_space_id, token, run_dir, events_path, attempt)
+        stage = str(runtime_payload.get("stage") or "").upper()
+        if "RUNTIME_ERROR" in stage:
+            runtime_error_count += 1
+            collect_space_logs(target_space_id, token, run_dir, events_path)
+            last_error = f"Space runtime stage is {stage}"
+            if runtime_error_count >= 2:
+                raise RuntimeError(f"Space is in RUNTIME_ERROR. See logs/space_logs_runtime.txt and logs/space_logs_build.txt. Last runtime: {runtime_payload}")
+        try:
+            return validate_http_health(target_space_id, token, run_dir, events_path, attempt)
+        except Exception as exc:
+            last_error = f"HTTP /health failed: {exc}"
+        try:
+            return validate_gradio_api(target_space_id, token, run_dir, events_path, attempt)
+        except Exception as exc:
+            last_error = (last_error or "") + f"; Gradio API failed: {exc}"
+        append_event(events_path, "api_validation", "waiting", "Live health/API not ready yet", {"attempt": attempt, "runtime": runtime_payload, "error": last_error[-1500:] if last_error else None})
+        time.sleep(30)
+    collect_space_logs(target_space_id, token, run_dir, events_path)
+    raise RuntimeError(f"Live health/API validation did not pass before timeout: {last_error}")
+
+
+def is_auth_or_billing_like_error(error: str | None) -> bool:
+    value = error or ""
+    markers = [
+        "401",
+        "402",
+        "403",
+        "Invalid username or password",
+        "Unauthorized",
+        "Repository Not Found",
+        "payment",
+        "billing",
+        "quota",
+        "grant",
+    ]
+    return any(marker.lower() in value.lower() for marker in markers)
+
+
+def request_hardware(api, target_space_id: str, hardware: str, token: str, events_path: Path, step: str, retries: int = 2):
+    """Best-effort hardware request after Space creation.
+
+    V23 tries hardware at create_repo time first. This function remains as a
+    fallback for cases where a Space was created on CPU and the Hub later
+    accepts a hardware switch. Auth/billing/quota errors are not retried.
+    """
+    if not hardware:
+        return {"phase": "post_create_request", "requested": False, "hardware": hardware, "ok": False, "error": "empty hardware"}
+    last_error = None
+    for attempt in range(1, retries + 1):
+        try:
+            runtime = api.request_space_hardware(repo_id=target_space_id, hardware=hardware, token=token)
+            payload = {
+                "phase": "post_create_request",
+                "requested": True,
+                "hardware": hardware,
+                "ok": True,
+                "attempt": attempt,
+                "runtime_stage": getattr(getattr(runtime, "stage", None), "value", str(getattr(runtime, "stage", None))),
+                "requested_hardware": getattr(runtime, "requested_hardware", None),
+                "hardware_current": getattr(runtime, "hardware", None),
+            }
+            append_event(events_path, step, "success", f"Requested Space hardware {hardware}", payload)
+            return payload
+        except Exception as exc:
+            last_error = str(exc)[:2000]
+            auth_like = is_auth_or_billing_like_error(last_error)
+            payload = {"phase": "post_create_request", "attempt": attempt, "hardware": hardware, "error": last_error, "manual_action_required": auth_like}
+            append_event(events_path, step, "failed" if auth_like or attempt == retries else "waiting", f"Could not request Space hardware {hardware}", payload)
+            if auth_like:
+                return {"phase": "post_create_request", "requested": True, "hardware": hardware, "ok": False, "attempts": attempt, "error": last_error, "manual_action_required": True}
+            if attempt < retries:
+                time.sleep(8 * attempt)
+    return {"phase": "post_create_request", "requested": True, "hardware": hardware, "ok": False, "attempts": retries, "error": last_error, "manual_action_required": False}
+
+
+def build_hardware_sequence(preferred_hardware: str, fallback_hardware: str, allow_fixed_gpu_fallback: bool) -> list[str]:
+    sequence = []
+    for hw in ["zero-a10g", preferred_hardware, fallback_hardware if allow_fixed_gpu_fallback else None]:
+        value = (hw or "").strip()
+        if value and value not in sequence:
+            sequence.append(value)
+    return sequence
+
+
+def create_space_with_hardware_strategy(api, target_space_id: str, token: str, preferred_hardware: str, fallback_hardware: str, allow_fixed_gpu_fallback: bool, events_path: Path):
+    """Create a private Space and request hardware as early as possible.
+
+    HF supports `space_hardware` directly on create_repo. This is the cleanest
+    moment to request hardware because the Space does not need a second restart.
+    If OAuth/billing/quota prevents automatic hardware selection, fall back to
+    a normal private CPU Space and mark manual hardware as required.
+    """
+    sequence = build_hardware_sequence(preferred_hardware, fallback_hardware, allow_fixed_gpu_fallback)
+    attempts = []
+
+    for hardware in sequence:
+        try:
+            append_event(events_path, "create_space_hardware", "started", f"Creating private Space with requested hardware {hardware}", {"target_space": target_space_id, "hardware": hardware})
+            api.create_repo(
+                repo_id=target_space_id,
+                repo_type="space",
+                space_sdk="gradio",
+                private=True,
+                exist_ok=False,
+                space_hardware=hardware,
+                token=token,
+            )
+            payload = {"phase": "create_repo_space_hardware", "hardware": hardware, "ok": True, "target_space": target_space_id}
+            append_event(events_path, "create_space", "success", f"Private target Space created with requested hardware {hardware}", payload)
+            return {"created": True, "selected_hardware": hardware, "requested_sequence": sequence, "attempts": attempts + [payload], "manual_action_required": False}
+        except Exception as exc:
+            error = str(exc)[:2500]
+            manual = is_auth_or_billing_like_error(error)
+            payload = {"phase": "create_repo_space_hardware", "hardware": hardware, "ok": False, "error": error, "manual_action_required": manual}
+            attempts.append(payload)
+            append_event(events_path, "create_space_hardware", "failed", f"Could not create Space with requested hardware {hardware}", payload)
+            # Continue through the sequence: ZeroGPU quota/auth can fail while a fixed GPU
+            # may still be worth trying. If fixed GPU also fails, we'll create CPU below.
+
+    append_event(events_path, "create_space", "started", "Creating private target Space on default CPU after hardware-at-creation attempts failed", {"target_space": target_space_id})
+    api.create_repo(repo_id=target_space_id, repo_type="space", space_sdk="gradio", private=True, exist_ok=False, token=token)
+    cpu_payload = {"phase": "create_repo_default_cpu", "hardware": "cpu-basic", "ok": True, "target_space": target_space_id, "manual_action_required": True}
+    append_event(events_path, "create_space", "success", "Private target Space created on default CPU; manual hardware selection may be required", cpu_payload)
+    return {"created": True, "selected_hardware": "default-cpu-or-existing", "requested_sequence": sequence, "attempts": attempts + [cpu_payload], "manual_action_required": True}
+
+
+def create_initial_workspace(workspace: Path, model_id: str, target_space_id: str, preferred_hardware: str, fallback_hardware: str, allow_fallback: bool, implementation_mode: str, model_analysis: dict | None = None):
+    workspace.mkdir(parents=True, exist_ok=True)
+    model_analysis = model_analysis or {}
+    pipeline_tag = model_analysis.get("pipeline_tag")
+    library_name = model_analysis.get("library_name")
+    tags = model_analysis.get("tags", [])[:40]
+    siblings = model_analysis.get("siblings", [])[:60]
+    app_py = f"""import gradio as gr
+from huggingface_hub import model_info, list_repo_files
+
+MODEL_ID = {model_id!r}
+TARGET_SPACE_ID = {target_space_id!r}
+
+
+def health():
+    return {{
+        "status": "booted",
+        "model_id": MODEL_ID,
+        "target_space_id": TARGET_SPACE_ID,
+        "stage": "initial-scaffold",
+        "note": "Pi should replace this scaffold with a model-specific demo while preserving a cheap health endpoint.",
+    }}
+
+
+def placeholder(*args):
+    return "Initial scaffold. Pi should replace this with a model-specific inference path, or write TECHNICAL_BLOCKERS.json."
+
+with gr.Blocks(title="Generated Model Space — Agentic Space Factory") as demo:
+    gr.Markdown("# Generated Model Space — Agentic Space Factory")
+    gr.Markdown(f"Private generated Space for `{{MODEL_ID}}`.")
+    gr.JSON(label="Health", value=health(), every=None)
+    gr.Button("Health check").click(fn=health, inputs=None, outputs=gr.JSON(), api_name="health")
+    gr.Textbox(label="Input", value="Hello from Agentic Space Factory").submit(fn=placeholder, inputs=None, outputs=gr.Textbox(), api_name="predict")
+    gr.Button("Run placeholder").click(fn=placeholder, inputs=None, outputs=gr.Textbox(), api_name="predict")
+
+if __name__ == "__main__":
+    demo.launch()
+"""
+    (workspace / "app.py").write_text(app_py, encoding="utf-8")
+    req = """gradio>=6.0.0
+huggingface_hub>=0.34.0,<2.0.0
+spaces
+transformers>=4.45.0,<6.0.0
+diffusers
+accelerate
+safetensors
+torch
+kernels
+pillow
+numpy
+requests
+"""
+    (workspace / "requirements.txt").write_text(req, encoding="utf-8")
+    readme = f"""---
+title: Generated Model Space
+sdk: gradio
+app_file: app.py
+python_version: "3.10"
+suggested_hardware: {preferred_hardware or fallback_hardware or "cpu-basic"}
+short_description: "Agent-built model demo"
+---
+
+# Generated Model Space — Agentic Space Factory
+
+Private generated Space for `{model_id}`.
+
+This Space is created by Agentic Space Factory. It should remain private until manually reviewed.
+"""
+    (workspace / "README.md").write_text(readme, encoding="utf-8")
+    analysis_json = json.dumps({"pipeline_tag": pipeline_tag, "library_name": library_name, "tags": tags, "siblings": siblings}, indent=2, ensure_ascii=False)
+    goal = f"""You are Pi running inside a Hugging Face Job for Agentic Space Factory.
+
+Goal: build the best possible private Hugging Face Space demo for an arbitrary model card.
+
+MODEL_ID: {model_id}
+TARGET_SPACE_ID: {target_space_id}
+IMPLEMENTATION_MODE: {implementation_mode}
+MODEL_METADATA:
+```json
+{analysis_json}
+```
+
+First read and follow the operational rules from this gist:
+{GIST_URL}
+
+Non-negotiable safety and product constraints:
+- The target Space must remain private.
+- Do not delete any user resources.
+- Do not print secrets or tokens.
+- Work only inside the current workspace.
+- The wrapper will create the private Space, request hardware best-effort, upload files, and validate the live app. Do not create/delete repos yourself in this builder worker.
+- Preserve a cheap health endpoint named `health` with `api_name="health"`. It must not load weights, run GPU work, or download large files.
+- Do not pin huggingface_hub below 1.0. Use huggingface_hub>=0.34.0,<2.0.0 unless the model card requires a narrower compatible range. If transformers>=5 is used, keep huggingface_hub compatible with it, for example huggingface_hub>=1.5.0,<2.0.0.
+- README.md frontmatter must remain valid; if it uses short_description, it must be 60 characters or fewer.
+
+Implementation contract:
+- If IMPLEMENTATION_MODE is `full-inference-gated`, you are not allowed to silently replace generation with a placeholder and call it success.
+- Try to implement the closest real inference path for the model card using evidence from README, model metadata, config files, and repo files.
+- You may choose an appropriate Gradio UI for the task: text, image, audio, video, multimodal, embeddings, classification, etc.
+- If the model is standard and feasible, implement a real generate/predict function and expose it as a Gradio endpoint.
+- If the model requires GPU, add ZeroGPU-compatible `@spaces.GPU(...)` only around the inference function. Do not decorate health.
+- If the model requires special dependencies, include them only when needed and document risks.
+- Investigate compatibility fallbacks before declaring a blocker: PyTorch SDPA, xformers, HF Kernels where relevant, CPU/offload/lazy loading, smaller resolution/steps, safe smoke-test inputs.
+- If real inference is impossible or unsafe in a Space, write TECHNICAL_BLOCKERS.json with concrete evidence for every blocker.
+
+Deliverables:
+- app.py must boot on Hugging Face Spaces.
+- app.py must expose health/api_name="health".
+- If real generation is implemented, generate/predict must attempt a real model call, not only return a textual diagnostic.
+- If real generation is not implemented, write TECHNICAL_BLOCKERS.json with: full_inference_implemented=false, blockers[], evidence[], minimum_runtime, and suggested_next_step.
+- Write INFERENCE_CONTRACT.json with: full_inference_implemented, health_endpoint, primary_api_name, expected_output_type, validation_level, requires_gpu, estimated_vram, and blockers_count.
+- README.md must explain the runtime strategy, task, limitations, and how to test.
+- Write a concise PI_SUMMARY.md with what you changed and whether full inference is implemented.
+"""
+    (workspace / "GOAL.md").write_text(goal, encoding="utf-8")
+    return ["app.py", "requirements.txt", "README.md", "GOAL.md"]
+
+
+def sanitize_readme_metadata(workspace: Path, events_path: Path):
+    readme_path = workspace / "README.md"
+    if not readme_path.exists():
+        return
+    text = readme_path.read_text(encoding="utf-8", errors="ignore")
+    if not text.startswith("---"):
+        return
+    parts = text.split("---", 2)
+    if len(parts) < 3:
+        return
+    _, frontmatter, body = parts
+    changed = False
+    sanitized_lines = []
+    for line in frontmatter.splitlines():
+        if line.strip().startswith("short_description:"):
+            value = "Generated model demo"
+            sanitized_lines.append(f"short_description: {value}")
+            changed = True
+        else:
+            sanitized_lines.append(line)
+    # If Pi added other unexpectedly long one-line metadata values, leave them alone:
+    # the known Hub validation blocker for this run was short_description > 60 chars.
+    if changed:
+        new_text = "---\n" + "\n".join(sanitized_lines).strip() + "\n---" + body
+        readme_path.write_text(new_text, encoding="utf-8")
+        append_event(events_path, "metadata_sanitize", "success", "Sanitized README metadata", {"short_description": "Generated model demo"})
+
+
+
+def normalize_requirements_for_modern_hub(workspace: Path, events_path: Path):
+    """Prevent a known resolver conflict in generated Spaces.
+
+    Older builder versions forced `huggingface_hub<1.0.0` to avoid old Gradio
+    import issues. Modern Spaces can use Gradio 6 and recent Transformers;
+    Transformers 5.x requires huggingface-hub >=1.5.0, so the old pin breaks
+    builds. Keep the constraint broad and modern unless Pi intentionally uses a
+    different compatible stack.
+    """
+    req_path = workspace / "requirements.txt"
+    if not req_path.exists():
+        return
+    raw = req_path.read_text(encoding="utf-8", errors="ignore")
+    lines = [line.rstrip() for line in raw.splitlines()]
+    changed = False
+    filtered = []
+    transformers_needs_hub_15 = False
+    for line in lines:
+        stripped = line.strip()
+        low = stripped.lower().replace("_", "-")
+        if low.startswith("huggingface-hub"):
+            if "<1" in low or "< 1" in low or ",<1" in low:
+                changed = True
+            # Always replace with the policy line to avoid duplicate/conflicting pins.
+            changed = True
+            continue
+        if low.startswith("transformers") and (">=5" in low or "==5" in low or "~=5" in low):
+            transformers_needs_hub_15 = True
+        filtered.append(line)
+    hub_line = "huggingface_hub>=1.5.0,<2.0.0" if transformers_needs_hub_15 else "huggingface_hub>=0.34.0,<2.0.0"
+    # Put hub near the top, after any --extra-index-url lines.
+    insert_at = 0
+    while insert_at < len(filtered) and filtered[insert_at].strip().startswith("--"):
+        insert_at += 1
+    filtered.insert(insert_at, hub_line)
+    new = "\n".join(line for line in filtered if line.strip()) + "\n"
+    if new != raw:
+        req_path.write_text(new, encoding="utf-8")
+        append_event(events_path, "requirements_sanitize", "success", "Normalized huggingface_hub requirement for modern dependency resolution", {"huggingface_hub": hub_line})
+
+
+def repair_workspace_with_pi(workspace: Path, run_dir: Path, events_path: Path, pi_model: str, target_space_id: str, model_id: str, failure_reason: str):
+    """Ask Pi for one minimal build/runtime repair pass based on collected logs."""
+    logs_dir = run_dir / "logs"
+    build_log = (logs_dir / "space_logs_build.txt").read_text(encoding="utf-8", errors="ignore") if (logs_dir / "space_logs_build.txt").exists() else ""
+    runtime_log = (logs_dir / "space_logs_runtime.txt").read_text(encoding="utf-8", errors="ignore") if (logs_dir / "space_logs_runtime.txt").exists() else ""
+    repair_dir = run_dir / "repair"
+    before_dir = repair_dir / "before"
+    after_dir = repair_dir / "after"
+    if before_dir.exists():
+        shutil.rmtree(before_dir)
+    shutil.copytree(workspace, before_dir, ignore=shutil.ignore_patterns(".git", "node_modules", "__pycache__", "*.pyc"))
+    goal = f"""You are Pi repairing a Hugging Face Space generated by Agentic Space Factory.
+
+MODEL_ID: {model_id}
+TARGET_SPACE_ID: {target_space_id}
+
+The first build/runtime validation failed.
+
+Failure summary:
+{failure_reason[:4000]}
+
+Build log tail:
+```text
+{build_log[-12000:]}
+```
+
+Runtime log tail:
+```text
+{runtime_log[-12000:]}
+```
+
+Repair contract:
+- Make the smallest patch possible.
+- Prefer fixing dependency resolver conflicts, missing imports, invalid metadata, Gradio endpoint bugs, and import-order issues.
+- Do not replace real inference with a placeholder unless TECHNICAL_BLOCKERS.json clearly explains why full inference is impossible.
+- Preserve a cheap health endpoint with api_name="health".
+- Keep README frontmatter valid, short_description <= 60 chars.
+- Do not pin huggingface_hub below 1.0. For modern generated Spaces use huggingface_hub>=0.34.0,<2.0.0. If transformers>=5 is present, use huggingface_hub>=1.5.0,<2.0.0.
+- Do not delete the app. Do not publish anything. Work only in the current workspace.
+
+Deliverables:
+- patched app.py / requirements.txt / README.md as needed
+- REPAIR_SUMMARY.md explaining the patch
+- keep or update INFERENCE_CONTRACT.json if the inference contract changed
+"""
+    (workspace / "REPAIR_GOAL.md").write_text(goal, encoding="utf-8")
+    append_event(events_path, "repair", "started", "Running Pi repair pass using build/runtime logs", {"model": pi_model})
+    code, out = run_cmd(["pi", "-p", goal], cwd=workspace, timeout=1500)
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    (logs_dir / "pi_repair_output.txt").write_text(out, encoding="utf-8")
+    if code != 0:
+        append_event(events_path, "repair", "failed", "Pi repair returned a non-zero exit code", {"returncode": code, "output_tail": out[-3000:]})
+        return False
+    normalize_requirements_for_modern_hub(workspace, events_path)
+    if after_dir.exists():
+        shutil.rmtree(after_dir)
+    shutil.copytree(workspace, after_dir, ignore=shutil.ignore_patterns(".git", "node_modules", "__pycache__", "*.pyc"))
+    append_event(events_path, "repair", "success", "Pi repair pass completed", {"output_tail": out[-3000:]})
+    return True
+
+def upload_workspace(api, workspace: Path, target_space_id: str, token: str, run_dir: Path, events_path: Path):
+    sanitize_readme_metadata(workspace, events_path)
+    normalize_requirements_for_modern_hub(workspace, events_path)
+    append_event(events_path, "upload_files", "started", "Uploading generated universal model-card workspace recursively")
+    gen_dir = run_dir / "generated"
+    if gen_dir.exists():
+        shutil.rmtree(gen_dir)
+    shutil.copytree(workspace, gen_dir, ignore=shutil.ignore_patterns(".git", "node_modules", "__pycache__", "*.pyc"))
+    for filename in ["app.py", "README.md", "requirements.txt"]:
+        if not (workspace / filename).exists():
+            raise RuntimeError(f"Missing required generated file: {filename}")
+    api.upload_folder(
+        folder_path=str(workspace),
+        repo_id=target_space_id,
+        repo_type="space",
+        token=token,
+        ignore_patterns=[".git/*", "node_modules/*", "__pycache__/*", "*.pyc", "GOAL.md"],
+    )
+    uploaded_files = sorted(str(p.relative_to(workspace)) for p in workspace.rglob("*") if p.is_file() and "node_modules" not in p.parts and "__pycache__" not in p.parts)
+    append_event(events_path, "upload_files", "success", "Uploaded generated workspace folder", {"file_count": len(uploaded_files), "files_sample": uploaded_files[:50]})
+
+
+def load_json_if_exists(path: Path) -> dict:
+    if not path.exists():
+        return {}
+    try:
+        return json.loads(path.read_text(encoding="utf-8", errors="replace"))
+    except Exception as exc:
+        return {"parse_error": str(exc), "raw_tail": path.read_text(encoding="utf-8", errors="replace")[-2000:]}
+
+
+def infer_generation_gate(workspace: Path, implementation_mode: str, validation: dict, run_dir: Path, events_path: Path) -> dict:
+    """Classify the run separately from process success.
+
+    /health passing means the Space boots. It does not mean the generated Space
+    performs model inference. In full-inference-gated mode we require either
+    an actual implementation signal or a machine-readable blocker report.
+    """
+    app_text = (workspace / "app.py").read_text(encoding="utf-8", errors="ignore") if (workspace / "app.py").exists() else ""
+    summary_text = (workspace / "PI_SUMMARY.md").read_text(encoding="utf-8", errors="ignore") if (workspace / "PI_SUMMARY.md").exists() else ""
+    req_text = (workspace / "requirements.txt").read_text(encoding="utf-8", errors="ignore") if (workspace / "requirements.txt").exists() else ""
+    blockers_path = workspace / "TECHNICAL_BLOCKERS.json"
+    blockers = load_json_if_exists(blockers_path)
+
+    combined = (app_text + "\n" + summary_text).lower()
+    blocked_markers = [
+        "full generation is not implemented",
+        "full generation is intentionally not wired",
+        "full inference is blocked",
+        "returns a detailed diagnostic",
+        "diagnostic report instead",
+        "placeholder generator",
+        "placeholder generation",
+        "info-only",
+        "not implemented",
+        "cannot run in this environment",
+        "out of scope",
+    ]
+    blocker_detected = bool(blockers) or any(m in combined for m in blocked_markers)
+    implementation_signals = {
+        "has_spaces_gpu": "@spaces.GPU" in app_text,
+        "has_torch": "torch" in req_text or "import torch" in app_text,
+        "has_diffusers": "diffusers" in req_text or "diffusers" in app_text,
+        "has_video_output_hint": any(x in app_text.lower() for x in ["gr.video", "video", ".mp4", "ffmpeg"]),
+        "health_passed": validation.get("method") in {"http_health", "gradio"},
+    }
+
+    if blocker_detected:
+        status = "technical_blocker"
+        message = "Space boots, but full model inference was not implemented. See TECHNICAL_BLOCKERS.json / PI_SUMMARY.md."
+    elif implementation_mode in {"full-inference-gated", "full-inference-attempt"}:
+        # Without a video smoke test, do not claim real inference success.
+        status = "full_inference_candidate_health_passed"
+        message = "Space boots and contains inference signals, but no generation smoke test has validated a real video output."
+    else:
+        status = "health_only"
+        message = "Safe scaffold health validation passed. Full inference was not requested."
+
+    if blocker_detected and not blockers:
+        blockers = {
+            "full_inference_implemented": False,
+            "source": "worker_heuristic_from_PI_SUMMARY_or_app.py",
+            "blockers": [
+                {
+                    "type": "agent_declared_or_detected_blocker",
+                    "claim": "Pi-generated artifacts state that full inference is blocked/not implemented or generation returns diagnostics/placeholders.",
+                    "evidence": "See PI_SUMMARY.md and app.py in generated artifacts.",
+                    "severity": "blocking",
+                }
+            ],
+            "required_investigations_for_next_run": [
+                "Check whether PyTorch SDPA can replace flash-attn calls.",
+                "Check whether HF Kernels flash-attn2/3/4 can replace required flash-attn APIs.",
+                "Verify whether 2-GPU context parallelism is strictly required or can be reduced to a single-GPU smoke test.",
+            ],
+        }
+        (workspace / "TECHNICAL_BLOCKERS.json").write_text(json.dumps(blockers, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+        (run_dir / "generated" / "TECHNICAL_BLOCKERS.json").write_text(json.dumps(blockers, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+
+    gate = {
+        "status": status,
+        "message": message,
+        "implementation_mode": implementation_mode,
+        "blocker_detected": blocker_detected,
+        "implementation_signals": implementation_signals,
+        "validation_method": validation.get("method"),
+        "blockers": blockers,
+    }
+    write_json(run_dir / "inference_gate.json", gate)
+    append_event(events_path, "inference_gate", status, message, gate)
+    return gate
+
+
+def main():
+    run_id = os.environ["RUN_ID"]
+    hf_username = os.environ.get("HF_USERNAME", "unknown")
+    bucket_source = os.environ.get("BUCKET_SOURCE", "unknown")
+    output_root = Path(os.environ.get("OUTPUT_ROOT", "/output"))
+    target_space_id = os.environ.get("TARGET_SPACE_ID", "")
+    model_id = sanitize_model_id(os.environ.get("MODEL_ID", DEFAULT_MODEL_ID))
+    pi_model = os.environ.get("PI_MODEL", "Qwen/Qwen3-Coder-Next")
+    preferred_hardware = os.environ.get("PREFERRED_SPACE_HARDWARE", "zero-a10g")
+    fallback_hardware = os.environ.get("FALLBACK_SPACE_HARDWARE", "l40sx1")
+    allow_fixed_gpu_fallback = os.environ.get("ALLOW_FIXED_GPU_FALLBACK", "true").lower() in {"1", "true", "yes", "on"}
+    implementation_mode = os.environ.get("IMPLEMENTATION_MODE", "full-inference-attempt")
+    token = os.environ.get("HF_TOKEN")
+
+    run_dir = output_root / "runs" / run_id
+    events_path = run_dir / "events.jsonl"
+    state_path = run_dir / "state.json"
+    workspace = Path("/tmp/universal_workspace")
+
+    append_event(events_path, "bootstrap", "started", "Universal model-card builder worker started", {"model_id": model_id, "target_space_id": target_space_id})
+    write_json(state_path, {"run_id": run_id, "kind": "universal_model_card_builder", "status": "running", "message": "Attempting Universal model-card builderd Space creation", "model_id": model_id, "target_space": target_space_id, "created_by": hf_username, "bucket_source": bucket_source, "created_at": now(), "updated_at": now()})
+    if not token:
+        fail(run_dir, events_path, "HF_TOKEN is missing from Job secrets")
+    if not TARGET_RE.match(target_space_id):
+        fail(run_dir, events_path, "Invalid TARGET_SPACE_ID", {"target_space_id": target_space_id})
+
+    try:
+        install_python_deps(events_path)
+        from huggingface_hub import HfApi
+        api = HfApi(token=token)
+        whoami = api.whoami(token=token)
+        append_event(events_path, "auth", "success", "Authenticated inside Job", {"whoami_name": whoami.get("name")})
+
+        append_event(events_path, "model_analysis", "started", "Fetching model metadata", {"model_id": model_id})
+        info = api.model_info(model_id, token=token, files_metadata=True)
+        siblings = [getattr(s, "rfilename", "") for s in (info.siblings or [])]
+        analysis = {"model_id": model_id, "pipeline_tag": getattr(info, "pipeline_tag", None), "library_name": getattr(info, "library_name", None), "tags": list(getattr(info, "tags", []) or [])[:100], "siblings": siblings[:160], "default_model_target": model_id == DEFAULT_MODEL_ID, "preferred_hardware": preferred_hardware, "fallback_hardware": fallback_hardware, "allow_fixed_gpu_fallback": allow_fixed_gpu_fallback, "implementation_mode": implementation_mode}
+        write_json(run_dir / "model_analysis.json", analysis)
+        append_event(events_path, "model_analysis", "success", "Model metadata fetched", {"pipeline_tag": analysis["pipeline_tag"], "library_name": analysis["library_name"]})
+
+        create_initial_workspace(workspace, model_id, target_space_id, preferred_hardware, fallback_hardware, allow_fixed_gpu_fallback, implementation_mode, analysis)
+        append_event(events_path, "workspace", "success", "Prepared universal model-card workspace", {"files": sorted(p.name for p in workspace.iterdir())})
+
+        install_pi(events_path)
+        configure_pi(events_path, pi_model)
+        append_event(events_path, "pi_run", "started", "Running Pi on universal model-card workspace", {"model": pi_model})
+        code, pi_out = run_cmd(["pi", "-p", (workspace / "GOAL.md").read_text(encoding="utf-8")], cwd=workspace, timeout=2400)
+        (run_dir / "logs").mkdir(parents=True, exist_ok=True)
+        (run_dir / "logs" / "pi_output.txt").write_text(pi_out, encoding="utf-8")
+        if code != 0:
+            append_event(events_path, "pi_run", "failed", "Pi returned a non-zero exit code", {"returncode": code, "output_tail": pi_out[-4000:]})
+            collect_pi_traces(run_dir, events_path)
+            fail(run_dir, events_path, "Pi failed before Space upload", {"returncode": code, "output_tail": pi_out[-4000:]})
+        append_event(events_path, "pi_run", "success", "Pi completed universal model-card workspace pass", {"output_tail": pi_out[-2000:]})
+        if not (workspace / "PI_SUMMARY.md").exists():
+            (workspace / "PI_SUMMARY.md").write_text("# Pi Summary\n\nPi did not create a PI_SUMMARY.md. See logs/pi_output.txt.\n", encoding="utf-8")
+
+        app_text = (workspace / "app.py").read_text(encoding="utf-8", errors="ignore")
+        if "/health" not in app_text and "api_name=\"health\"" not in app_text and "api_name='health'" not in app_text:
+            append_event(events_path, "pi_verification", "failed", "app.py does not appear to expose /health; injecting safe health endpoint is not implemented")
+            fail(run_dir, events_path, "Pi output did not preserve a /health endpoint")
+        append_event(events_path, "pi_verification", "success", "Pi output preserved health validation endpoint")
+
+        append_event(events_path, "hardware_strategy", "started", "Creating Space with hardware-at-creation strategy", {"preferred_hardware": preferred_hardware, "fallback_hardware": fallback_hardware, "allow_fixed_gpu_fallback": allow_fixed_gpu_fallback})
+        hardware_strategy = create_space_with_hardware_strategy(
+            api,
+            target_space_id,
+            token,
+            preferred_hardware,
+            fallback_hardware,
+            allow_fixed_gpu_fallback,
+            events_path,
+        )
+        selected_hardware = hardware_strategy.get("selected_hardware") or "default-cpu-or-existing"
+        hardware_attempts = list(hardware_strategy.get("attempts") or [])
+        requested_hardware_sequence = list(hardware_strategy.get("requested_sequence") or [])
+
+        # Upload after create. If create_repo(space_hardware=...) succeeded, the build
+        # starts directly on the requested hardware. If it fell back to CPU, the run
+        # remains valid but will be marked manual_hardware_required when inference
+        # signals indicate GPU is needed.
+        upload_workspace(api, workspace, target_space_id, token, run_dir, events_path)
+
+        if selected_hardware == "default-cpu-or-existing":
+            append_event(events_path, "hardware", "warning", "Automatic hardware-at-creation failed; Space is on default CPU unless user changes it manually", {"attempts": hardware_attempts})
+
+        write_json(run_dir / "hardware_attempts.json", {"selected_hardware": selected_hardware, "requested_sequence": requested_hardware_sequence, "attempts": hardware_attempts, "strategy": "create_repo_space_hardware_first"})
+        write_json(run_dir / "hardware_strategy.json", {"selected_hardware": selected_hardware, "requested_sequence": requested_hardware_sequence, "attempts": hardware_attempts, "manual_action_required": selected_hardware == "default-cpu-or-existing", "strategy": "create_repo_space_hardware_first"})
+
+        try:
+            validation = validate_live_api(api, target_space_id, token, run_dir, events_path, timeout_s=1200)
+        except Exception as validation_error:
+            append_event(events_path, "repair", "started", "Initial live validation failed; attempting one repair pass", {"error": str(validation_error)[:2000]})
+            collect_space_logs(target_space_id, token, run_dir, events_path)
+            repaired = repair_workspace_with_pi(workspace, run_dir, events_path, pi_model, target_space_id, model_id, str(validation_error))
+            if not repaired:
+                raise
+            upload_workspace(api, workspace, target_space_id, token, run_dir, events_path)
+            validation = validate_live_api(api, target_space_id, token, run_dir, events_path, timeout_s=1200)
+        inference_gate = infer_generation_gate(workspace, implementation_mode, validation, run_dir, events_path)
+
+        # If the generated app looks like real GPU inference but automatic
+        # hardware requests failed, classify the run honestly as needing manual
+        # hardware instead of pretending CPU/default hardware is enough. the existing-Space validation workflow
+        # can then smoke-test generation after the user sets a GPU manually.
+        manual_hw_required = selected_hardware == "default-cpu-or-existing" and inference_gate.get("status") not in {"technical_blocker", "health_only"} and (
+            inference_gate.get("implementation_signals", {}).get("has_spaces_gpu")
+            or inference_gate.get("implementation_signals", {}).get("has_torch")
+            or any((a.get("manual_action_required") for a in hardware_attempts if isinstance(a, dict)))
+        )
+        if manual_hw_required:
+            inference_gate = dict(inference_gate)
+            inference_gate["status"] = "manual_hardware_required"
+            inference_gate["message"] = "Space was generated and boots, but automatic ZeroGPU/fixed-GPU assignment failed. Set hardware manually, then run the existing-Space validation workflow."
+            inference_gate["manual_hardware_required"] = True
+            inference_gate["hardware_attempts"] = hardware_attempts
+            write_json(run_dir / "inference_gate.json", inference_gate)
+            append_event(events_path, "inference_gate", "manual_hardware_required", inference_gate["message"], inference_gate)
+
+        collect_pi_traces(run_dir, events_path)
+
+        final_state = {
+            "run_id": run_id,
+            "kind": "universal_model_card_builder",
+            "status": inference_gate["status"],
+            "message": inference_gate["message"],
+            "model_id": model_id,
+            "target_space": target_space_id,
+            "target_space_url": f"https://huggingface.co/spaces/{target_space_id}",
+            "selected_hardware": selected_hardware,
+            "hardware_attempts": hardware_attempts,
+            "validation": validation,
+            "inference_gate": inference_gate,
+            "updated_at": now(),
+            "created_by": hf_username,
+            "bucket_source": bucket_source,
+        }
+        write_json(state_path, final_state)
+        report = f"""# Agentic Space Factory — Universal Model-Card Builder Report
+
+Run ID: `{run_id}`
+
+Status: **{inference_gate['status']}**
+
+{inference_gate['message']}
+
+Target Space: https://huggingface.co/spaces/{target_space_id}
+
+Model: `{model_id}`
+
+## Hardware
+
+Selected/requested hardware: `{selected_hardware}`
+
+Hardware changes are best-effort with OAuth. If requests fail with 401/auth/billing errors, set the Space hardware manually and rerun validation.
+
+```json
+{json.dumps(hardware_attempts, indent=2, ensure_ascii=False)}
+```
+
+## Health validation
+
+The wrapper validated the live Space using HTTP `/health` first, with Gradio Client as fallback. This only proves bootability.
+
+```json
+{json.dumps(validation, indent=2, ensure_ascii=False)}
+```
+
+## Full-inference gate
+
+```json
+{json.dumps(inference_gate, indent=2, ensure_ascii=False)}
+```
+
+## Pi summary
+
+{(workspace / 'PI_SUMMARY.md').read_text(encoding='utf-8', errors='ignore') if (workspace / 'PI_SUMMARY.md').exists() else 'No PI_SUMMARY.md was produced.'}
+
+## Safety
+
+- The target Space was created private.
+- No public publication was attempted.
+- Raw traces should remain private; redacted traces are stored separately.
+- If fallback fixed GPU was used or selected manually, review billing/hardware settings manually after the run.
+"""
+        (run_dir / "report.md").write_text(report, encoding="utf-8")
+        append_event(events_path, "report_write", "success", "Wrote report.md")
+        append_event(events_path, "done", inference_gate["status"], "Universal model-card builder completed", {"target_space": target_space_id, "selected_hardware": selected_hardware, "gate_status": inference_gate["status"]})
+    except SystemExit:
+        raise
+    except Exception as exc:
+        try:
+            collect_pi_traces(run_dir, events_path)
+        except Exception:
+            pass
+        fail(run_dir, events_path, "Universal model-card builder worker failed", {"error": str(exc)})
+
+
+if __name__ == "__main__":
+    main()
+
+'''
+
+
+VALIDATE_EXISTING_SPACE_WORKER_SCRIPT = r'''
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+TARGET_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{1,95}/[A-Za-z0-9][A-Za-z0-9._-]{1,95}$")
+
+
+def now():
+    return datetime.now(timezone.utc).isoformat()
+
+
+def write_json(path: Path, payload: dict):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+
+
+def append_event(path: Path, step: str, status: str, message: str, data: dict | None = None):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    event = {"ts": now(), "step": step, "status": status, "message": message, "data": data or {}}
+    line = json.dumps(event, ensure_ascii=False)
+    with path.open("a", encoding="utf-8") as f:
+        f.write(line + "\n")
+    print(line, flush=True)
+
+
+def redact_text(text: str | None) -> str:
+    if not text:
+        return ""
+    value = text
+    for secret_name in ["HF_TOKEN", "HUGGING_FACE_HUB_TOKEN"]:
+        secret = os.environ.get(secret_name)
+        if secret:
+            value = value.replace(secret, "[REDACTED]")
+    value = re.sub(r"Bearer\s+[A-Za-z0-9_\-.=]+", "Bearer [REDACTED]", value)
+    value = re.sub(r"hf_[A-Za-z0-9_\-]{10,}", "hf_[REDACTED]", value)
+    return value
+
+
+def run_cmd(cmd: list[str], *, env: dict | None = None, timeout: int = 120):
+    result = subprocess.run(cmd, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout)
+    return result.returncode, redact_text(result.stdout)
+
+
+def install_deps(events_path: Path):
+    append_event(events_path, "dependencies", "started", "Installing validation dependencies")
+    code, out = run_cmd([sys.executable, "-m", "pip", "install", "-q", "--upgrade", "huggingface_hub>=1.0.0", "gradio_client>=2.0.0", "requests>=2.31.0"], timeout=600)
+    if code != 0:
+        append_event(events_path, "dependencies", "failed", "Dependency installation failed", {"output_tail": out[-4000:]})
+        raise RuntimeError(out)
+    append_event(events_path, "dependencies", "success", "Validation dependencies installed")
+
+
+def make_gradio_client(target_space_id: str, token: str):
+    import inspect
+    from gradio_client import Client
+    params = inspect.signature(Client).parameters
+    if "token" in params:
+        return Client(target_space_id, token=token)
+    if "hf_token" in params:
+        return Client(target_space_id, hf_token=token)
+    if "api_key" in params:
+        return Client(target_space_id, api_key=token)
+    if "headers" in params:
+        return Client(target_space_id, headers={"Authorization": f"Bearer {token}"})
+    return Client(target_space_id)
+
+
+def api_names_from_schema(schema) -> list[str]:
+    names: list[str] = []
+    def add(name):
+        if not isinstance(name, str) or not name:
+            return
+        if not name.startswith("/"):
+            name = "/" + name
+        if name not in names:
+            names.append(name)
+    def walk(obj):
+        if isinstance(obj, dict):
+            for k, v in obj.items():
+                if k in {"api_name", "apiName"}:
+                    add(v)
+                if isinstance(k, str) and k.startswith("/"):
+                    add(k)
+                walk(v)
+        elif isinstance(obj, list):
+            for item in obj:
+                walk(item)
+    walk(schema)
+    return names
+
+
+def runtime_to_dict(runtime) -> dict:
+    payload = {}
+    for attr in ["stage", "hardware", "requested_hardware", "sleep_time", "storage", "gc_timeout"]:
+        value = getattr(runtime, attr, None)
+        payload[attr] = getattr(value, "value", value)
+    return {k: str(v) if v is not None else None for k, v in payload.items()}
+
+
+def write_space_runtime(api, target_space_id: str, token: str, run_dir: Path, events_path: Path, attempt: int | None = None) -> dict:
+    try:
+        runtime = api.get_space_runtime(repo_id=target_space_id, token=token)
+        payload = runtime_to_dict(runtime)
+        payload["attempt"] = attempt
+        write_json(run_dir / "space_runtime.json", payload)
+        return payload
+    except Exception as exc:
+        payload = {"error": str(exc)[:2000], "attempt": attempt}
+        write_json(run_dir / "space_runtime.json", payload)
+        append_event(events_path, "space_runtime", "warning", "Could not fetch Space runtime", payload)
+        return payload
+
+
+def collect_space_logs(target_space_id: str, token: str, run_dir: Path, events_path: Path):
+    logs_dir = run_dir / "logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    env = os.environ.copy()
+    env["HF_TOKEN"] = token
+    commands = {
+        "space_logs_runtime.txt": ["hf", "spaces", "logs", target_space_id],
+        "space_logs_build.txt": ["hf", "spaces", "logs", target_space_id, "--build"],
+    }
+    written = []
+    for filename, cmd in commands.items():
+        try:
+            code, out = run_cmd(cmd, env=env, timeout=75)
+            (logs_dir / filename).write_text(out, encoding="utf-8")
+            written.append({"file": filename, "returncode": code, "tail": out[-1000:]})
+        except Exception as exc:
+            written.append({"file": filename, "error": str(exc)[:1000]})
+    append_event(events_path, "space_logs", "success", "Collected best-effort Space logs", {"files": written})
+    return written
+
+
+def space_subdomain_url(target_space_id: str) -> str:
+    owner, name = target_space_id.split("/", 1)
+    return f"https://{owner}-{name}.hf.space".replace("_", "-").lower()
+
+
+def validate_http_health(target_space_id: str, token: str, run_dir: Path, attempt: int):
+    import requests
+    url = space_subdomain_url(target_space_id).rstrip("/") + "/health"
+    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json,text/plain,*/*"}
+    response = requests.get(url, headers=headers, timeout=20)
+    payload = {
+        "status": "success" if response.ok else "failed",
+        "attempt": attempt,
+        "url": url,
+        "status_code": response.status_code,
+        "content_type": response.headers.get("content-type"),
+        "text": response.text[:2000],
+    }
+    if response.ok:
+        try:
+            payload["json"] = response.json()
+        except Exception:
+            pass
+        write_json(run_dir / "tests" / "http_health.json", payload)
+        return payload
+    raise RuntimeError(f"HTTP /health returned {response.status_code}: {response.text[:500]}")
+
+
+def wait_until_live(api, target_space_id: str, token: str, run_dir: Path, events_path: Path, timeout_s: int = 1800):
+    append_event(events_path, "live_wait", "started", "Waiting for existing Space to become live")
+    deadline = time.time() + timeout_s
+    attempt = 0
+    last_error = None
+    while time.time() < deadline:
+        attempt += 1
+        runtime_payload = write_space_runtime(api, target_space_id, token, run_dir, events_path, attempt)
+        stage = str(runtime_payload.get("stage") or "").upper()
+        if "RUNTIME_ERROR" in stage:
+            collect_space_logs(target_space_id, token, run_dir, events_path)
+            last_error = f"Space is in RUNTIME_ERROR: {runtime_payload}"
+            append_event(events_path, "live_wait", "waiting", "Space is in runtime error; still waiting in case hardware was changed manually", {"attempt": attempt, "runtime": runtime_payload})
+            time.sleep(30)
+            continue
+        try:
+            health = validate_http_health(target_space_id, token, run_dir, attempt)
+            append_event(events_path, "live_wait", "success", "HTTP /health is live", {"attempt": attempt})
+            return {"validator": "http_health", "health": health, "runtime": runtime_payload}
+        except Exception as http_exc:
+            last_error = f"HTTP health failed: {http_exc}"
+        try:
+            client = make_gradio_client(target_space_id, token)
+            schema = client.view_api(return_format="dict")
+            names = api_names_from_schema(schema)
+            write_json(run_dir / "tests" / "api_schema.json", {"schema": schema, "api_names": names})
+            if names:
+                append_event(events_path, "live_wait", "success", "Gradio API schema is live", {"attempt": attempt, "api_names": names})
+                return {"validator": "gradio_schema", "api_names": names, "runtime": runtime_payload}
+        except Exception as gr_exc:
+            last_error = (last_error or "") + f"; Gradio schema failed: {gr_exc}"
+        append_event(events_path, "live_wait", "waiting", "Space not live yet", {"attempt": attempt, "runtime": runtime_payload, "error": last_error[-1500:] if last_error else None})
+        time.sleep(30)
+    collect_space_logs(target_space_id, token, run_dir, events_path)
+    raise RuntimeError(f"Space did not become live before timeout: {last_error}")
+
+
+def parse_json_env(name: str, default):
+    value = os.environ.get(name)
+    if not value:
+        return default
+    try:
+        return json.loads(value)
+    except Exception as exc:
+        raise ValueError(f"Invalid JSON for {name}: {exc}")
+
+
+def result_contains_expected_output(result, expected_output_type: str) -> tuple[bool, dict]:
+    expected = (expected_output_type or "any").lower().strip()
+    info = {"expected_output_type": expected, "result_type": type(result).__name__, "result_repr": repr(result)[:2000]}
+    paths = []
+    def visit(obj):
+        if isinstance(obj, (str, Path)):
+            text = str(obj)
+            if any(text.lower().endswith(ext) for ext in [".png", ".jpg", ".jpeg", ".webp", ".gif", ".mp4", ".wav", ".mp3", ".txt"]):
+                paths.append(text)
+        elif isinstance(obj, dict):
+            for key in ["path", "url", "name"]:
+                if key in obj:
+                    visit(obj[key])
+            for value in obj.values():
+                if isinstance(value, (dict, list, tuple)):
+                    visit(value)
+        elif isinstance(obj, (list, tuple)):
+            for item in obj:
+                visit(item)
+    visit(result)
+    info["detected_paths"] = paths[:20]
+    if expected == "any":
+        return result is not None, info
+    image_ext = [".png", ".jpg", ".jpeg", ".webp", ".gif"]
+    video_ext = [".mp4", ".mov", ".webm"]
+    audio_ext = [".wav", ".mp3", ".flac", ".ogg"]
+    if expected == "text":
+        return isinstance(result, str) and len(result.strip()) > 0, info
+    if expected == "image":
+        return any(str(p).lower().endswith(tuple(image_ext)) for p in paths), info
+    if expected == "video":
+        return any(str(p).lower().endswith(tuple(video_ext)) for p in paths), info
+    if expected == "audio":
+        return any(str(p).lower().endswith(tuple(audio_ext)) for p in paths), info
+    return result is not None, info
+
+
+def copy_result_artifacts(result, run_dir: Path):
+    artifacts = run_dir / "artifacts"
+    artifacts.mkdir(parents=True, exist_ok=True)
+    copied = []
+    def maybe_copy(obj):
+        if isinstance(obj, (str, Path)):
+            path = Path(str(obj))
+            if path.exists() and path.is_file():
+                target = artifacts / path.name
+                try:
+                    shutil.copy2(path, target)
+                    copied.append(str(target))
+                except Exception:
+                    pass
+        elif isinstance(obj, dict):
+            for key in ["path", "name"]:
+                if key in obj:
+                    maybe_copy(obj[key])
+            for value in obj.values():
+                if isinstance(value, (dict, list, tuple)):
+                    maybe_copy(value)
+        elif isinstance(obj, (list, tuple)):
+            for item in obj:
+                maybe_copy(item)
+    maybe_copy(result)
+    return copied
+
+
+def smoke_generate(target_space_id: str, token: str, run_dir: Path, events_path: Path):
+    api_name = (os.environ.get("API_NAME") or "/generate").strip()
+    expected_output_type = (os.environ.get("EXPECTED_OUTPUT_TYPE") or "any").strip()
+    test_args = parse_json_env("TEST_ARGS_JSON", ["a cinematic robot cat astronaut, detailed, studio lighting"])
+    test_kwargs = parse_json_env("TEST_KWARGS_JSON", {})
+    if not isinstance(test_args, list):
+        raise ValueError("TEST_ARGS_JSON must be a JSON list")
+    if not isinstance(test_kwargs, dict):
+        raise ValueError("TEST_KWARGS_JSON must be a JSON object")
+    append_event(events_path, "generation_smoke", "started", "Calling live generation endpoint", {"api_name": api_name, "expected_output_type": expected_output_type})
+    client = make_gradio_client(target_space_id, token)
+    schema = client.view_api(return_format="dict")
+    discovered = api_names_from_schema(schema)
+    write_json(run_dir / "tests" / "api_schema.json", {"schema": schema, "api_names": discovered})
+    started = time.time()
+    result = client.predict(*test_args, api_name=api_name, **test_kwargs)
+    latency = time.time() - started
+    ok, info = result_contains_expected_output(result, expected_output_type)
+    copied = copy_result_artifacts(result, run_dir)
+    payload = {
+        "status": "success" if ok else "failed",
+        "target_space": target_space_id,
+        "api_name": api_name,
+        "discovered_api_names": discovered,
+        "test_args": test_args,
+        "test_kwargs": test_kwargs,
+        "expected_output_type": expected_output_type,
+        "latency_seconds": round(latency, 3),
+        "result_info": info,
+        "copied_artifacts": copied,
+        "recommended_zero_gpu_duration_seconds": int(max(30, min(300, latency * 2 + 15))),
+        "validated_at": now(),
+    }
+    write_json(run_dir / "tests" / "generation_smoke.json", payload)
+    write_json(run_dir / "tests" / "test_result.json", payload)
+    if ok:
+        append_event(events_path, "generation_smoke", "success", "Live generation smoke test passed", {"latency_seconds": payload["latency_seconds"], "copied_artifacts": copied[:5]})
+        return payload
+    append_event(events_path, "generation_smoke", "failed", "Live generation returned an unexpected output type", payload)
+    raise RuntimeError("Generation smoke test failed: unexpected output type")
+
+
+def main():
+    run_id = os.environ["RUN_ID"]
+    username = os.environ.get("HF_USERNAME", "unknown")
+    output_root = Path(os.environ.get("OUTPUT_ROOT", "/output"))
+    target_space_id = os.environ["TARGET_SPACE_ID"].strip()
+    token = os.environ.get("HF_TOKEN")
+    run_dir = output_root / "runs" / run_id
+    events_path = run_dir / "events.jsonl"
+    state_path = run_dir / "state.json"
+    append_event(events_path, "bootstrap", "started", "Existing Space validation worker started", {"target_space_id": target_space_id})
+    write_json(state_path, {"run_id": run_id, "kind": "validate_existing_space", "status": "running", "target_space": target_space_id, "created_by": username, "updated_at": now()})
+    if not token:
+        raise RuntimeError("HF_TOKEN is missing")
+    if not TARGET_RE.match(target_space_id):
+        raise ValueError("TARGET_SPACE_ID must look like owner/space-name")
+    try:
+        install_deps(events_path)
+        from huggingface_hub import HfApi
+        api = HfApi(token=token)
+        whoami = api.whoami(token=token)
+        append_event(events_path, "auth", "success", "Authenticated inside validation Job", {"whoami_name": whoami.get("name")})
+        live = wait_until_live(api, target_space_id, token, run_dir, events_path, timeout_s=int(os.environ.get("LIVE_TIMEOUT_SECONDS", "1800")))
+        smoke = smoke_generate(target_space_id, token, run_dir, events_path)
+        final_state = {
+            "run_id": run_id,
+            "kind": "validate_existing_space",
+            "status": "full_inference_success",
+            "message": "Existing Space passed live health/schema validation and generation smoke test.",
+            "target_space": target_space_id,
+            "target_space_url": f"https://huggingface.co/spaces/{target_space_id}",
+            "live_validation": live,
+            "generation_smoke": smoke,
+            "updated_at": now(),
+        }
+        write_json(state_path, final_state)
+        report = f"""# Agentic Space Factory — Existing Space Validation Report
+
+Status: **full_inference_success**
+
+Target Space: [`{target_space_id}`](https://huggingface.co/spaces/{target_space_id})
+
+## Generation smoke test
+
+```json
+{json.dumps(smoke, indent=2, ensure_ascii=False)}
+```
+
+## Notes
+
+- This validation is intended for Spaces whose hardware was set manually after generation.
+- Latency is measured from the live Gradio endpoint call.
+- The recommended ZeroGPU duration is a rough estimate from this live run, not a guarantee.
+"""
+        (run_dir / "report.md").write_text(report, encoding="utf-8")
+        append_event(events_path, "report_write", "success", "Wrote report.md")
+        append_event(events_path, "done", "full_inference_success", "Existing Space validation completed", {"latency_seconds": smoke.get("latency_seconds")})
+    except Exception as exc:
+        collect_space_logs(target_space_id, token or "", run_dir, events_path)
+        details = {"error": str(exc)[:4000]}
+        write_json(state_path, {"run_id": run_id, "kind": "validate_existing_space", "status": "failed", "target_space": target_space_id, "details": details, "updated_at": now()})
+        (run_dir / "report.md").write_text(f"# Existing Space Validation Failed\n\n```json\n{json.dumps(details, indent=2, ensure_ascii=False)}\n```\n", encoding="utf-8")
+        append_event(events_path, "failure", "failed", "Existing Space validation failed", details)
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()
+'''
+
+
+def encoded_universal_model_card_worker_script() -> str:
+    """Return the base64-encoded universal model-card builder worker script."""
+    return _encode(UNIVERSAL_MODEL_CARD_WORKER_SCRIPT)
+
+
+def encoded_validate_existing_space_worker_script() -> str:
+    """Return the base64-encoded existing-Space validation worker script."""
+    return _encode(VALIDATE_EXISTING_SPACE_WORKER_SCRIPT)
+
+
+def python_decode_and_run_command() -> list[str]:
+    """Command list for `run_job`.
+
+    The Job image only needs Python. The script is passed via env as base64 and
+    executed from /tmp, which avoids persisting code or exposing secrets.
+    """
+    runner = textwrap.dedent(
+        """
+        import base64, os, pathlib, subprocess, sys
+        script = base64.b64decode(os.environ['WORKER_SCRIPT_B64']).decode('utf-8')
+        path = pathlib.Path('/tmp/space_factory_worker.py')
+        path.write_text(script, encoding='utf-8')
+        raise SystemExit(subprocess.call([sys.executable, str(path)]))
+        """
+    ).strip()
+    return ["python", "-c", runner]