Upload 6 files

src/jobs.py

@@ -13,6 +13,7 @@ from .worker_payload import (
     encoded_pi_model_card_worker_script,
     encoded_runtime_recommender_worker_script,
     encoded_longcat_article_worker_script,
+    encoded_universal_model_card_worker_script,
     encoded_pi_space_worker_script,
     encoded_worker_script,
     python_decode_and_run_command,
@@ -305,6 +306,58 @@ def launch_longcat_article_job(
         },
     )
 
+
+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,
+) -> dict[str, Any]:
+    """Launch Phase 10: universal model-card → private Space builder with Pi and gated validation."""
+    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 = (model_id or "").strip().replace("https://huggingface.co/", "").strip("/")
+    if "/" not in clean_model_id:
+        raise ValueError("Model ID must look like owner/model-name or a Hugging Face model URL.")
+
+    env = _base_env(
+        run_id=safe_run_id,
+        username=username,
+        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 "cpu-basic").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, timeout="60m")
+    return _job_result(
+        job,
+        run_id=safe_run_id,
+        kind="universal_model_card_builder",
+        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 inspect_job_safe(job_id: str, token: str | None = None) -> dict[str, Any]:
     if not job_id:
         return {"error": "Missing job_id"}

src/worker_payload.py

@@ -1346,6 +1346,801 @@ if __name__ == "__main__":
 '''
 
 
+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 request_hardware(api, target_space_id: str, hardware: str, token: str, events_path: Path, step: str, retries: int = 4):
+    """Best-effort hardware request.
+
+    OAuth tokens can create/write Spaces but may still be unable to trigger
+    hardware changes, especially paid GPU upgrades. Treat 401/auth/billing
+    failures as manual-action-required instead of burning retries.
+    """
+    if not hardware:
+        return {"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 = {
+                "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 = any(marker in last_error for marker in ["401", "Invalid username or password", "Unauthorized", "Repository Not Found"])
+            payload = {"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 {"requested": True, "hardware": hardware, "ok": False, "attempts": attempt, "error": last_error, "manual_action_required": True}
+            if attempt < retries:
+                time.sleep(8 * attempt)
+    return {"requested": True, "hardware": hardware, "ok": False, "attempts": retries, "error": last_error, "manual_action_required": False}
+
+
+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>=5.0.0
+huggingface_hub>=0.34.0,<1.0.0
+spaces
+transformers>=4.45.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 Phase 10. 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 Phase 10.
+
+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 Phase 10 worker.
+- Preserve a cheap health endpoint named `health` with `api_name="health"`. It must not load weights, run GPU work, or download large files.
+- Keep the huggingface_hub pin in requirements.txt: huggingface_hub>=0.34.0,<1.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 = "model video avatar 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": "model video avatar demo"})
+
+
+def upload_workspace(api, workspace: Path, target_space_id: str, token: str, run_dir: Path, events_path: Path):
+    sanitize_readme_metadata(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, "create_space", "started", "Creating private target Space", {"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)
+        append_event(events_path, "create_space", "success", "Private target Space created", {"target_space": target_space_id})
+
+        # Upload before requesting hardware. Newly created private Spaces may not be
+        # immediately available on the hardware endpoint; uploading first also ensures
+        # the repo has valid Space metadata before any restart is triggered.
+        upload_workspace(api, workspace, target_space_id, token, run_dir, events_path)
+
+        hardware_attempts = []
+        preferred = request_hardware(api, target_space_id, preferred_hardware, token, events_path, "hardware_preferred")
+        hardware_attempts.append(preferred)
+        selected_hardware = preferred_hardware if preferred.get("ok") else None
+        if not selected_hardware and allow_fixed_gpu_fallback and fallback_hardware:
+            fallback = request_hardware(api, target_space_id, fallback_hardware, token, events_path, "hardware_fallback")
+            hardware_attempts.append(fallback)
+            selected_hardware = fallback_hardware if fallback.get("ok") else None
+        if not selected_hardware:
+            append_event(events_path, "hardware", "warning", "Could not request preferred/fallback hardware; Space may remain on default CPU", {"attempts": hardware_attempts})
+            selected_hardware = "default-cpu-or-existing"
+        write_json(run_dir / "hardware_attempts.json", {"selected_hardware": selected_hardware, "attempts": hardware_attempts})
+
+        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)
+        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()
+
+'''
+
+
 def encoded_worker_script() -> str:
     """Return the base64-encoded Phase 1 hello worker script."""
     return _encode(HELLO_WORKER_SCRIPT)
@@ -1381,6 +2176,11 @@ def encoded_longcat_article_worker_script() -> str:
     """Return the base64-encoded Phase 9 LongCat article reproduction worker script."""
     return _encode(LONGCAT_ARTICLE_WORKER_SCRIPT)
 
+
+def encoded_universal_model_card_worker_script() -> str:
+    """Return the base64-encoded Phase 10 universal model-card builder worker script."""
+    return _encode(UNIVERSAL_MODEL_CARD_WORKER_SCRIPT)
+
 def python_decode_and_run_command() -> list[str]:
     """Command list for `run_job`.