src/worker_payload.py

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]