Upload 6 files

src/jobs.py

@@ -11,6 +11,8 @@ from .worker_payload import (
     encoded_create_space_worker_script,
     encoded_pi_gist_worker_script,
     encoded_pi_model_card_worker_script,
+    encoded_runtime_recommender_worker_script,
+    encoded_longcat_article_worker_script,
     encoded_pi_space_worker_script,
     encoded_worker_script,
     python_decode_and_run_command,
@@ -30,12 +32,12 @@ def _base_env(*, run_id: str, username: str, worker_script_b64: str) -> dict[str
     }
 
 
-def _launch_job(*, token: str, env: dict[str, str]) -> Any:
+def _launch_job(*, token: str, env: dict[str, str], flavor: str | None = None, timeout: str | None = None) -> Any:
     return run_job(
         image=settings.job_image,
         command=python_decode_and_run_command(),
-        flavor=settings.job_flavor,
-        timeout=settings.job_timeout,
+        flavor=flavor or settings.job_flavor,
+        timeout=timeout or settings.job_timeout,
         env=env,
         secrets={"HF_TOKEN": token},
         volumes=[Volume(type="bucket", source=settings.bucket_source, mount_path=settings.bucket_mount)],
@@ -222,6 +224,84 @@ def launch_pi_model_card_job(
     )
 
 
+
+def launch_runtime_recommender_job(
+    *,
+    token: str,
+    username: str,
+    model_id: str | None = None,
+    run_id: str | None = None,
+) -> dict[str, Any]:
+    """Launch the Phase 6 Job: analyze model metadata and recommend safe runtime/hardware before building."""
+    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("runtime")
+    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.")
+    env = _base_env(
+        run_id=safe_run_id,
+        username=username,
+        worker_script_b64=encoded_runtime_recommender_worker_script(),
+    )
+    env["MODEL_ID"] = clean_model_id
+    job = _launch_job(token=token, env=env)
+    return _job_result(
+        job,
+        run_id=safe_run_id,
+        kind="runtime_recommender",
+        extra={"model_id": clean_model_id},
+    )
+
+
+def launch_longcat_article_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,
+    run_id: str | None = None,
+) -> dict[str, Any]:
+    """Launch Phase 7: attempt an article-style LongCat Space with ZeroGPU + fixed GPU fallback."""
+    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("longcat")
+    target_space_id = normalize_target_space(username=username, target_slug=target_slug, run_id=safe_run_id)
+    clean_model_id = (model_id or "meituan-longcat/LongCat-Video-Avatar-1.5").strip().replace("https://huggingface.co/", "").strip("/")
+    if "/" not in clean_model_id:
+        raise ValueError("Model ID must look like owner/model-name.")
+
+    env = _base_env(
+        run_id=safe_run_id,
+        username=username,
+        worker_script_b64=encoded_longcat_article_worker_script(),
+    )
+    env["TARGET_SPACE_ID"] = target_space_id
+    env["MODEL_ID"] = clean_model_id
+    env["PI_MODEL"] = (pi_model or "moonshotai/Kimi-K2.5").strip()
+    env["PREFERRED_SPACE_HARDWARE"] = (preferred_space_hardware or "zero-a10g").strip()
+    env["FALLBACK_SPACE_HARDWARE"] = (fallback_space_hardware or "l40sx1").strip()
+    env["ALLOW_FIXED_GPU_FALLBACK"] = "true" if allow_fixed_gpu_fallback else "false"
+    job = _launch_job(token=token, env=env, timeout="60m")
+    return _job_result(
+        job,
+        run_id=safe_run_id,
+        kind="longcat_article_reproduction",
+        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,
+        },
+    )
+
 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

@@ -568,6 +568,509 @@ def _encode(script: str) -> str:
 
 PI_MODEL_CARD_WORKER_SCRIPT = 'import json\nimport os\nimport re\nimport shutil\nimport subprocess\nimport sys\nimport time\nfrom datetime import datetime, timezone\nfrom pathlib import Path\nfrom textwrap import dedent\n\n\nTARGET_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{1,95}/[A-Za-z0-9][A-Za-z0-9._-]{1,95}$")\nSUPPORTED_TASKS = {"text-generation", "text2text-generation", "fill-mask", "text-classification", "sentiment-analysis"}\n\n\ndef now():\n    return datetime.now(timezone.utc).isoformat()\n\n\ndef write_json(path: Path, payload: dict):\n    path.parent.mkdir(parents=True, exist_ok=True)\n    path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\\n", encoding="utf-8")\n\n\ndef append_event(path: Path, step: str, status: str, message: str, data: dict | None = None):\n    path.parent.mkdir(parents=True, exist_ok=True)\n    event = {"ts": now(), "step": step, "status": status, "message": message, "data": data or {}}\n    line = json.dumps(event, ensure_ascii=False)\n    with path.open("a", encoding="utf-8") as f:\n        f.write(line + "\\n")\n    print(line, flush=True)\n\n\ndef redact_text(text: str | None) -> str:\n    if not text:\n        return ""\n    value = text\n    for secret_name in ["HF_TOKEN", "HUGGING_FACE_HUB_TOKEN"]:\n        secret = os.environ.get(secret_name)\n        if secret:\n            value = value.replace(secret, "[REDACTED]")\n    value = re.sub(r"Bearer\\s+[A-Za-z0-9_\\-.=]+", "Bearer [REDACTED]", value)\n    value = re.sub(r"hf_[A-Za-z0-9_\\-]{10,}", "hf_[REDACTED]", value)\n    return value\n\n\ndef safe_details(details: dict | None) -> dict:\n    if not details:\n        return {}\n    try:\n        return json.loads(redact_text(json.dumps(details, ensure_ascii=False)))\n    except Exception:\n        return {"redacted_details": redact_text(str(details))[-4000:]}\n\n\ndef fail(run_dir: Path, events_path: Path, message: str, details: dict | None = None, status: str = "failed"):\n    safe = safe_details(details)\n    append_event(events_path, "failure", "failed", message, safe)\n    write_json(run_dir / "state.json", {\n        "run_id": os.environ.get("RUN_ID"),\n        "kind": "pi_model_card",\n        "status": status,\n        "message": message,\n        "updated_at": now(),\n        "details": safe,\n    })\n    report = f"""# Agentic Space Factory — Model Card Space Report\n\nStatus: **{status}**\n\n{message}\n\n```json\n{json.dumps(safe, indent=2, ensure_ascii=False)}\n```\n"""\n    (run_dir / "report.md").write_text(report, encoding="utf-8")\n    raise SystemExit(1)\n\n\ndef run_cmd(cmd: list[str], *, cwd: Path | None = None, env: dict | None = None, timeout: int = 600):\n    result = subprocess.run(cmd, cwd=str(cwd) if cwd else None, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout)\n    return result.returncode, redact_text(result.stdout)\n\n\ndef install_python_deps(events_path: Path):\n    append_event(events_path, "dependencies", "started", "Installing Python worker dependencies")\n    code, out = run_cmd([sys.executable, "-m", "pip", "install", "-q", "--upgrade", "huggingface_hub>=1.0.0", "gradio_client>=2.0.0"], timeout=600)\n    if code != 0:\n        append_event(events_path, "dependencies", "failed", "Python dependency installation failed", {"output_tail": out[-4000:]})\n        raise RuntimeError(out)\n    append_event(events_path, "dependencies", "success", "Python worker dependencies installed")\n\n\ndef ensure_node(events_path: Path):\n    node = shutil.which("node")\n    npm = shutil.which("npm")\n    if node and npm:\n        _, node_v = run_cmd([node, "--version"], timeout=30)\n        _, npm_v = run_cmd([npm, "--version"], timeout=30)\n        append_event(events_path, "node", "success", "Node/npm already available", {"node": node_v.strip(), "npm": npm_v.strip()})\n        return\n    append_event(events_path, "node", "started", "Installing nodejs/npm through apt-get")\n    code, out = run_cmd(["bash", "-lc", "apt-get update -qq && apt-get install -y -qq nodejs npm"], timeout=600)\n    if code != 0:\n        append_event(events_path, "node", "failed", "Could not install nodejs/npm", {"output_tail": out[-4000:]})\n        raise RuntimeError(out)\n    append_event(events_path, "node", "success", "Installed nodejs/npm")\n\n\ndef install_pi(events_path: Path):\n    ensure_node(events_path)\n    append_event(events_path, "pi_install", "started", "Installing Pi coding agent from npm")\n    code, out = run_cmd(["npm", "install", "-g", "@mariozechner/pi-coding-agent"], timeout=900)\n    if code != 0:\n        append_event(events_path, "pi_install", "failed", "Pi npm installation failed", {"output_tail": out[-4000:]})\n        raise RuntimeError(out)\n    code, version = run_cmd(["pi", "--version"], timeout=60)\n    append_event(events_path, "pi_install", "success", "Pi installed", {"version_output": version.strip()[-300:]})\n\n\ndef configure_pi(events_path: Path, model: str):\n    pi_dir = Path.home() / ".pi" / "agent"\n    pi_dir.mkdir(parents=True, exist_ok=True)\n    (pi_dir / "auth.json").write_text(json.dumps({"huggingface": {"type": "api_key", "key": os.environ.get("HF_TOKEN", "")}}, indent=2), encoding="utf-8")\n    (pi_dir / "settings.json").write_text(json.dumps({"model": model, "provider": "huggingface", "autoRun": True, "autoApply": True}, indent=2), encoding="utf-8")\n    append_event(events_path, "pi_config", "success", "Configured Pi", {"model": model})\n\n\ndef sanitize_model_id(model_id: str) -> str:\n    model_id = (model_id or "").strip().replace("https://huggingface.co/", "")\n    model_id = model_id.split("?", 1)[0].strip("/")\n    if not re.match(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", model_id):\n        raise ValueError("MODEL_ID must look like owner/model-name")\n    return model_id\n\n\ndef analyze_model(model_id: str, token: str, run_dir: Path, events_path: Path) -> dict:\n    from huggingface_hub import HfApi, hf_hub_download\n    append_event(events_path, "model_analysis", "started", "Fetching model metadata", {"model_id": model_id})\n    api = HfApi(token=token)\n    info = api.model_info(model_id, token=token, files_metadata=False)\n    siblings = [getattr(s, "rfilename", "") for s in (info.siblings or [])]\n    pipeline_tag = getattr(info, "pipeline_tag", None)\n    library_name = getattr(info, "library_name", None)\n    tags = list(getattr(info, "tags", []) or [])\n    readme_excerpt = ""\n    try:\n        readme_path = hf_hub_download(repo_id=model_id, filename="README.md", token=token)\n        readme_text = Path(readme_path).read_text(encoding="utf-8", errors="ignore")\n        readme_excerpt = readme_text[:6000]\n    except Exception as exc:\n        readme_excerpt = f"Could not download README.md: {exc}"\n    task = pipeline_tag or "text-generation"\n    if task == "sentiment-analysis":\n        task = "text-classification"\n    supported = task in SUPPORTED_TASKS\n    analysis = {\n        "model_id": model_id,\n        "pipeline_tag": pipeline_tag,\n        "library_name": library_name,\n        "tags": tags[:80],\n        "siblings": siblings[:120],\n        "selected_task": task,\n        "template": "transformers_text_pipeline" if supported else "unsupported",\n        "supported": supported,\n        "confidence": 0.8 if supported else 0.25,\n        "risks": [],\n        "readme_excerpt": readme_excerpt,\n        "evidence": [f"pipeline_tag={pipeline_tag}", f"library_name={library_name}", f"files={\', \'.join(siblings[:12])}"],\n    }\n    if not supported:\n        analysis["risks"].append("Phase 5 only supports simple Transformers text pipeline tasks.")\n    if "gated" in tags:\n        analysis["risks"].append("Model appears gated; generated Space will not receive OAuth token as a secret in Phase 5.")\n    write_json(run_dir / "model_analysis.json", analysis)\n    append_event(events_path, "model_analysis", "success" if supported else "unsupported", "Model metadata analyzed", {"selected_task": task, "supported": supported, "confidence": analysis["confidence"]})\n    return analysis\n\n\ndef render_app(model_id: str, task: str) -> str:\n    return dedent(f\'\'\'\n    import gradio as gr\n    from transformers import pipeline\n\n    MODEL_ID = {model_id!r}\n    TASK = {task!r}\n\n    pipe = pipeline(TASK, model=MODEL_ID)\n\n    def run_model(text: str) -> str:\n        text = (text or "Hello from Agentic Space Factory").strip() or "Hello from Agentic Space Factory"\n        if TASK == "text-generation":\n            result = pipe(text, max_new_tokens=32, do_sample=False)\n            return result[0].get("generated_text", str(result))\n        if TASK == "text2text-generation":\n            result = pipe(text, max_new_tokens=64)\n            return result[0].get("generated_text", str(result))\n        if TASK == "fill-mask":\n            mask = getattr(getattr(pipe, "tokenizer", None), "mask_token", None) or "<mask>"\n            if mask not in text:\n                text = f"Hugging Face is {{mask}}."\n            result = pipe(text)\n            return str(result[:3] if isinstance(result, list) else result)\n        if TASK in {"text-classification", "sentiment-analysis"}:\n            return str(pipe(text))\n        return str(pipe(text))\n\n    demo = gr.Interface(\n        fn=run_model,\n        inputs=gr.Textbox(label="Input", value="Hello from Agentic Space Factory"),\n        outputs=gr.Textbox(label="Model output"),\n        title=f"Model demo: {{MODEL_ID}}",\n        description="Generated by Agentic Space Factory from model metadata. Pi adapted this model app.",\n        examples=[["Hello from Agentic Space Factory"], ["Hugging Face is awesome"]],\n    )\n\n    if __name__ == "__main__":\n        demo.launch()\n    \'\'\').strip() + "\\n"\n\n\ndef render_readme(model_id: str, task: str, target_space_id: str) -> str:\n    return dedent(f\'\'\'\n    ---\n    title: Model Card Generated Space\n    emoji: 🤖\n    colorFrom: green\n    colorTo: blue\n    sdk: gradio\n    app_file: app.py\n    python_version: "3.11"\n    pinned: false\n    ---\n\n    # Model Card Generated Space\n\n    This private Space was generated by Agentic Space Factory from `{model_id}`.\n\n    - Target Space: `{target_space_id}`\n    - Selected task: `{task}`\n    - Template: `transformers_text_pipeline`\n\n    Phase 5 is intentionally limited to simple Transformers text pipelines.\n    \'\'\').strip() + "\\n"\n\n\ndef prepare_workspace(workspace: Path, run_dir: Path, model_id: str, task: str, target_space_id: str, analysis: dict, events_path: Path):\n    workspace.mkdir(parents=True, exist_ok=True)\n    (workspace / "app.py").write_text(render_app(model_id, task), encoding="utf-8")\n    (workspace / "README.md").write_text(render_readme(model_id, task, target_space_id), encoding="utf-8")\n    (workspace / "requirements.txt").write_text("gradio>=5.0.0\\nhuggingface_hub>=0.34.0,<1.0.0\\ntransformers>=4.45.0\\ntorch\\nsafetensors\\n", encoding="utf-8")\n    goal = f"""You are running inside a Hugging Face Job as a coding agent.\n\nGoal: adapt the provided minimal Gradio app for the model `{model_id}` and task `{task}`.\n\nFirst, read the HF Spaces Agent Quickstart gist:\nhttps://gist.github.com/gary149/2aba2962375fa9ca56bb9ef53f00b73d\n\nRules for this Phase 5 smoke test:\n- Work only in the current workspace.\n- Do not create, delete, publish, or modify Hugging Face repos. The wrapper will create/upload the private Space.\n- Preserve `app.py`, `README.md`, and `requirements.txt`.\n- Do not remove the `huggingface_hub>=0.34.0,<1.0.0` compatibility pin from requirements.txt.\n- Preserve the `run_model` function and a Gradio Interface or Blocks app.\n- Preserve the exact marker phrase: Pi adapted this model app.\n- Keep the app simple and CPU-friendly.\n- Do not print secrets.\n- Write a short summary to `PI_SUMMARY.md`.\n\nModel analysis:\n```json\n{json.dumps({k: v for k, v in analysis.items() if k != \'readme_excerpt\'}, indent=2, ensure_ascii=False)}\n```\n\nREADME excerpt:\n{analysis.get(\'readme_excerpt\', \'\')[:3000]}\n"""\n    (workspace / "GOAL.md").write_text(goal, encoding="utf-8")\n    save_generated_files(run_dir, workspace)\n    append_event(events_path, "workspace", "success", "Prepared model app workspace", {"files": ["app.py", "README.md", "requirements.txt", "GOAL.md"]})\n\n\ndef save_generated_files(run_dir: Path, workspace: Path):\n    generated_dir = run_dir / "generated"\n    generated_dir.mkdir(parents=True, exist_ok=True)\n    for filename in ["app.py", "README.md", "requirements.txt", "GOAL.md", "PI_SUMMARY.md"]:\n        path = workspace / filename\n        if path.exists():\n            (generated_dir / filename).write_text(path.read_text(encoding="utf-8", errors="ignore"), encoding="utf-8")\n\n\ndef run_pi(workspace: Path, run_dir: Path, events_path: Path, model: str):\n    append_event(events_path, "pi_run", "started", "Running Pi on model-card workspace", {"model": model})\n    env = os.environ.copy()\n    env["HF_TOKEN"] = os.environ.get("HF_TOKEN", "")\n    code, out = run_cmd(["pi", "-p", (workspace / "GOAL.md").read_text(encoding="utf-8")], cwd=workspace, env=env, timeout=1800)\n    logs_dir = run_dir / "logs"\n    logs_dir.mkdir(parents=True, exist_ok=True)\n    (logs_dir / "pi_output.txt").write_text(out, encoding="utf-8")\n    save_generated_files(run_dir, workspace)\n    if code != 0:\n        append_event(events_path, "pi_run", "failed", "Pi returned a non-zero exit code", {"returncode": code, "output_tail": out[-4000:]})\n        raise RuntimeError("Pi failed. See logs/pi_output.txt")\n    app_text = (workspace / "app.py").read_text(encoding="utf-8", errors="ignore")\n    if "Pi adapted this model app" not in app_text:\n        raise RuntimeError("Pi/app verification failed: expected marker phrase missing from app.py")\n    append_event(events_path, "pi_run", "success", "Pi completed and preserved required marker")\n\n\ndef collect_pi_traces(run_dir: Path, events_path: Path):\n    src = Path.home() / ".pi" / "agent" / "sessions"\n    raw_dir = run_dir / "traces" / "raw"\n    redacted_dir = run_dir / "traces" / "redacted"\n    raw_dir.mkdir(parents=True, exist_ok=True)\n    redacted_dir.mkdir(parents=True, exist_ok=True)\n    count = 0\n    if src.exists():\n        for path in src.rglob("*.jsonl"):\n            count += 1\n            text = path.read_text(encoding="utf-8", errors="ignore")\n            (raw_dir / path.name).write_text(redact_text(text), encoding="utf-8")\n            (redacted_dir / path.name).write_text(redact_text(text), encoding="utf-8")\n    append_event(events_path, "traces", "success", "Collected Pi traces", {"count": count})\n\n\ndef make_gradio_client(target_space_id: str, token: str):\n    import inspect\n    from gradio_client import Client\n    params = inspect.signature(Client).parameters\n    if "token" in params:\n        return Client(target_space_id, token=token)\n    if "hf_token" in params:\n        return Client(target_space_id, hf_token=token)\n    if "api_key" in params:\n        return Client(target_space_id, api_key=token)\n    if "headers" in params:\n        return Client(target_space_id, headers={"Authorization": f"Bearer {token}"})\n    return Client(target_space_id)\n\n\ndef get_api_schema(client):\n    try:\n        return client.view_api(return_format="dict")\n    except TypeError:\n        return client.view_api()\n\n\ndef extract_api_names(api_schema) -> list[str]:\n    names = []\n    def add(value):\n        if not value or not isinstance(value, str):\n            return\n        name = value if value.startswith("/") else f"/{value}"\n        if name not in names:\n            names.append(name)\n    def walk(obj):\n        if isinstance(obj, dict):\n            for key, value in obj.items():\n                if key in {"api_name", "apiName"}:\n                    add(value)\n                if isinstance(key, str) and key.startswith("/"):\n                    add(key)\n                walk(value)\n        elif isinstance(obj, list):\n            for item in obj:\n                walk(item)\n    walk(api_schema)\n    return names\n\n\ndef predict_with_available_endpoint(client, api_schema, value: str):\n    candidates = extract_api_names(api_schema)\n    for fallback in ["/run_model", "/predict", "/greet"]:\n        if fallback not in candidates:\n            candidates.append(fallback)\n    errors = []\n    for api_name in candidates:\n        try:\n            return api_name, client.predict(value, api_name=api_name)\n        except Exception as exc:\n            errors.append({"api_name": api_name, "error": str(exc)[-500:]})\n    try:\n        return None, client.predict(value)\n    except Exception as exc:\n        errors.append({"api_name": None, "error": str(exc)[-500:]})\n        raise RuntimeError(f"No candidate Gradio endpoint worked: {json.dumps(errors, ensure_ascii=False)}")\n\n\ndef validate_live_api(target_space_id: str, token: str, events_path: Path, tests_dir: Path, timeout_seconds: int = 900):\n    tests_dir.mkdir(parents=True, exist_ok=True)\n    deadline = time.time() + timeout_seconds\n    last_error = None\n    attempt = 0\n    append_event(events_path, "api_validation", "started", "Waiting for live model Gradio API to become available")\n    while time.time() < deadline:\n        attempt += 1\n        try:\n            client = make_gradio_client(target_space_id, token)\n            api_schema = get_api_schema(client)\n            api_names = extract_api_names(api_schema)\n            write_json(tests_dir / "api_schema.json", {"schema": api_schema, "api_names": api_names})\n            used_api_name, result = predict_with_available_endpoint(client, api_schema, "Hello from Agentic Space Factory")\n            result_text = str(result)\n            ok = bool(result_text and len(result_text.strip()) >= 2)\n            payload = {"attempt": attempt, "target_space": target_space_id, "api_test_passed": ok, "api_name": used_api_name, "discovered_api_names": api_names, "result": result_text[:4000], "validated_at": now()}\n            write_json(tests_dir / "test_result.json", payload)\n            if ok:\n                append_event(events_path, "api_validation", "success", "Live model API test passed", {"attempt": attempt, "api_name": used_api_name, "discovered_api_names": api_names})\n                return payload\n            last_error = f"Unexpected empty API result from {used_api_name}: {result_text}"\n        except Exception as exc:\n            last_error = str(exc)\n            append_event(events_path, "api_validation", "waiting", "Live API not ready yet", {"attempt": attempt, "error": last_error[-1000:]})\n        time.sleep(20)\n    payload = {"target_space": target_space_id, "api_test_passed": False, "error": last_error, "validated_at": now()}\n    write_json(tests_dir / "test_result.json", payload)\n    raise RuntimeError(f"Live API validation did not pass before timeout: {last_error}")\n\n\ndef create_and_upload_space(api, token: str, target_space_id: str, workspace: Path, events_path: Path):\n    append_event(events_path, "create_space", "started", f"Creating private target Space {target_space_id}")\n    api.create_repo(repo_id=target_space_id, repo_type="space", space_sdk="gradio", private=True, exist_ok=False, token=token)\n    append_event(events_path, "create_space", "success", "Private target Space created", {"target_space": target_space_id})\n    append_event(events_path, "upload_files", "started", "Uploading model app files to target Space")\n    for path_in_repo in ["app.py", "README.md", "requirements.txt"]:\n        api.upload_file(path_or_fileobj=(workspace / path_in_repo).read_bytes(), path_in_repo=path_in_repo, repo_id=target_space_id, repo_type="space", token=token)\n        append_event(events_path, "upload_files", "success", f"Uploaded {path_in_repo}")\n\n\ndef main():\n    run_id = os.environ["RUN_ID"]\n    hf_username = os.environ.get("HF_USERNAME", "unknown")\n    bucket_source = os.environ.get("BUCKET_SOURCE", "unknown")\n    output_root = Path(os.environ.get("OUTPUT_ROOT", "/output"))\n    target_space_id = os.environ["TARGET_SPACE_ID"]\n    model_id = sanitize_model_id(os.environ.get("MODEL_ID", ""))\n    pi_model = os.environ.get("PI_MODEL") or "moonshotai/Kimi-K2.5"\n    token = os.environ.get("HF_TOKEN")\n    run_dir = output_root / "runs" / run_id\n    events_path = run_dir / "events.jsonl"\n    state_path = run_dir / "state.json"\n    workspace = Path("/tmp") / f"space-factory-model-{run_id}"\n    append_event(events_path, "bootstrap", "started", "Pi model-card worker started", {"model_id": model_id})\n    write_json(state_path, {"run_id": run_id, "kind": "pi_model_card", "status": "running", "message": "Analyzing model card and generating a private model demo Space", "model_id": model_id, "target_space": target_space_id, "created_by": hf_username, "bucket_source": bucket_source, "created_at": now(), "updated_at": now()})\n    if not token:\n        fail(run_dir, events_path, "HF_TOKEN is missing from Job secrets")\n    if not TARGET_RE.match(target_space_id):\n        fail(run_dir, events_path, "Invalid TARGET_SPACE_ID", {"target_space": target_space_id})\n    if not target_space_id.startswith(f"{hf_username}/"):\n        fail(run_dir, events_path, "Target Space must be in the signed-in user\'s namespace", {"target_space": target_space_id, "username": hf_username})\n    try:\n        install_python_deps(events_path)\n        from huggingface_hub import HfApi\n        api = HfApi(token=token)\n        whoami = api.whoami(token=token)\n        append_event(events_path, "auth", "success", "Authenticated inside Job", {"whoami_name": whoami.get("name")})\n        analysis = analyze_model(model_id, token, run_dir, events_path)\n        if not analysis.get("supported"):\n            fail(run_dir, events_path, "Model task is unsupported by Phase 5", {"model_analysis": {k: v for k, v in analysis.items() if k != "readme_excerpt"}}, status="unsupported")\n        prepare_workspace(workspace, run_dir, model_id, analysis["selected_task"], target_space_id, analysis, events_path)\n        install_pi(events_path)\n        configure_pi(events_path, pi_model)\n        run_pi(workspace, run_dir, events_path, pi_model)\n        collect_pi_traces(run_dir, events_path)\n        create_and_upload_space(api, token, target_space_id, workspace, events_path)\n        write_json(run_dir / "target_space.json", {"target_space": target_space_id, "url": f"https://huggingface.co/spaces/{target_space_id}", "private": True, "sdk": "gradio", "created_by": hf_username, "model_id": model_id})\n        validation = validate_live_api(target_space_id, token, events_path, run_dir / "tests")\n        final_state = {"run_id": run_id, "kind": "pi_model_card", "status": "success", "message": "Model-card generated private Space created and validated through the live API.", "model_id": model_id, "target_space": target_space_id, "target_space_url": f"https://huggingface.co/spaces/{target_space_id}", "created_by": hf_username, "bucket_source": bucket_source, "model_analysis": {k: v for k, v in analysis.items() if k != "readme_excerpt"}, "validation": validation, "updated_at": now(), "security_notes": ["The target Space was created as private.", "The HF token was not printed or intentionally persisted.", "Phase 5 supports only simple public text pipeline models.", "Success was declared only after the wrapper live API test passed."]}\n        write_json(state_path, final_state)\n        report = f"""# Agentic Space Factory — Model Card Space Report\n\nRun ID: `{run_id}`\n\nStatus: **success**\n\nCreated private model demo Space: [`{target_space_id}`](https://huggingface.co/spaces/{target_space_id})\n\n## Model\n\n- Model ID: `{model_id}`\n- Selected task: `{analysis[\'selected_task\']}`\n- Template: `{analysis[\'template\']}`\n- Pi model: `{pi_model}`\n\n## What happened\n\n```text\nOAuth user → HF Job → model metadata analysis → Pi adapts app.py → private Space creation → live API validation → Bucket report\n```\n\n## Live API validation\n\n```json\n{json.dumps(validation, indent=2, ensure_ascii=False)}\n```\n\n## Security posture\n\n- The target Space was created as private.\n- No token was printed or intentionally persisted.\n- Pi was instructed not to create/delete/publish repos; the wrapper performed Hub operations.\n- Success was declared only after the live API returned a non-empty result.\n\n## Next step\n\nPhase 6 should add a ZeroGPU Diffusers template and stricter model compatibility gating.\n"""\n        (run_dir / "report.md").write_text(report, encoding="utf-8")\n        append_event(events_path, "report_write", "success", "Wrote report.md")\n        append_event(events_path, "done", "success", "Pi model-card worker completed")\n    except SystemExit:\n        raise\n    except Exception as exc:\n        collect_pi_traces(run_dir, events_path)\n        fail(run_dir, events_path, "Pi model-card worker failed", {"error": str(exc)})\n\n\nif __name__ == "__main__":\n    main()\n'
 
+
+RUNTIME_RECOMMENDER_WORKER_SCRIPT = 'import json\nimport os\nimport re\nimport sys\nfrom datetime import datetime, timezone\nfrom pathlib import Path\nimport subprocess\n\nSUPPORTED_TEXT_TASKS = {"text-generation", "text2text-generation", "fill-mask", "text-classification", "sentiment-analysis"}\nDIFFUSION_TASKS = {"text-to-image", "image-to-image", "image-to-video", "text-to-video"}\n\n\ndef now():\n    return datetime.now(timezone.utc).isoformat()\n\n\ndef write_json(path: Path, payload: dict):\n    path.parent.mkdir(parents=True, exist_ok=True)\n    path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\\n", encoding="utf-8")\n\n\ndef append_event(path: Path, step: str, status: str, message: str, data: dict | None = None):\n    path.parent.mkdir(parents=True, exist_ok=True)\n    event = {"ts": now(), "step": step, "status": status, "message": message, "data": data or {}}\n    line = json.dumps(event, ensure_ascii=False)\n    with path.open("a", encoding="utf-8") as f:\n        f.write(line + "\\n")\n    print(line, flush=True)\n\n\ndef redact_text(text: str | None) -> str:\n    if not text:\n        return ""\n    value = text\n    for secret_name in ["HF_TOKEN", "HUGGING_FACE_HUB_TOKEN"]:\n        secret = os.environ.get(secret_name)\n        if secret:\n            value = value.replace(secret, "[REDACTED]")\n    value = re.sub(r"Bearer\\s+[A-Za-z0-9_\\-.=]+", "Bearer [REDACTED]", value)\n    value = re.sub(r"hf_[A-Za-z0-9_\\-]{10,}", "hf_[REDACTED]", value)\n    return value\n\n\ndef run_cmd(cmd: list[str], timeout: int = 600):\n    result = subprocess.run(cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout)\n    return result.returncode, redact_text(result.stdout)\n\n\ndef fail(run_dir: Path, events_path: Path, message: str, details: dict | None = None, status: str = "failed"):\n    append_event(events_path, "failure", "failed", message, details or {})\n    write_json(run_dir / "state.json", {\n        "run_id": os.environ.get("RUN_ID"),\n        "kind": "runtime_recommender",\n        "status": status,\n        "message": message,\n        "updated_at": now(),\n        "details": details or {},\n    })\n    (run_dir / "report.md").write_text(f"# Runtime Recommendation Report\\n\\nStatus: **{status}**\\n\\n{message}\\n\\n```json\\n{json.dumps(details or {}, indent=2, ensure_ascii=False)}\\n```\\n", encoding="utf-8")\n    raise SystemExit(1)\n\n\ndef sanitize_model_id(model_id: str) -> str:\n    model_id = (model_id or "").strip().replace("https://huggingface.co/", "")\n    model_id = model_id.split("?", 1)[0].strip("/")\n    if not re.match(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", model_id):\n        raise ValueError("MODEL_ID must look like owner/model-name")\n    return model_id\n\n\ndef human_bytes(value: int | None) -> str:\n    if value is None:\n        return "unknown"\n    size = float(value)\n    for unit in ["B", "KB", "MB", "GB", "TB"]:\n        if size < 1024 or unit == "TB":\n            return f"{size:.1f} {unit}"\n        size /= 1024\n    return str(value)\n\n\ndef infer_task(pipeline_tag, library_name, tags, siblings):\n    if pipeline_tag:\n        return "text-classification" if pipeline_tag == "sentiment-analysis" else pipeline_tag\n    if library_name == "diffusers" or "model_index.json" in siblings:\n        return "text-to-image"\n    if "config.json" in siblings:\n        return "text-generation"\n    return "unknown"\n\n\ndef recommend_runtime(task: str, library_name: str | None, tags: list[str], siblings: list[str], total_size: int | None):\n    risks = []\n    evidence = []\n    supported_for_phase5 = task in SUPPORTED_TEXT_TASKS\n    is_diffusion = task in DIFFUSION_TASKS or library_name == "diffusers" or "model_index.json" in siblings\n    gated = any("gated" == t or "gated" in t for t in tags)\n    trust_remote = any("trust_remote_code" in t or "custom_code" in t for t in tags)\n\n    if gated:\n        risks.append("Model appears gated; generated Spaces will need explicit secret/token handling before runtime validation.")\n    if trust_remote:\n        risks.append("Model may require custom code/trust_remote_code; keep out of V1 auto-build unless manually approved.")\n    if total_size is not None:\n        evidence.append(f"Estimated repo file size: {human_bytes(total_size)}")\n    evidence.append(f"task={task}")\n    evidence.append(f"library_name={library_name}")\n\n    if is_diffusion:\n        if total_size and total_size > 15 * 1024**3:\n            return {\n                "target_runtime": "manual-review",\n                "target_space_hardware": "manual-review",\n                "job_flavor": "cpu-basic",\n                "confidence": 0.35,\n                "supported_by_current_builder": False,\n                "reason": "Diffusion/video-like model appears too large for the next ZeroGPU MVP without manual review.",\n                "risks": risks + ["Large diffusion/video models may exceed ZeroGPU duration/memory assumptions."],\n                "evidence": evidence,\n            }\n        return {\n            "target_runtime": "zerogpu-candidate",\n            "target_space_hardware": "zero-a10g / ZeroGPU if available for the user",\n            "job_flavor": "cpu-basic",\n            "confidence": 0.72,\n            "supported_by_current_builder": False,\n            "reason": "Diffusers-style model; good candidate for Phase 7 ZeroGPU template, but not for Phase 5 CPU text template.",\n            "risks": risks + ["ZeroGPU template is not implemented yet in this version."],\n            "evidence": evidence,\n        }\n\n    if task in SUPPORTED_TEXT_TASKS:\n        if total_size is None:\n            runtime = "cpu-basic"\n            confidence = 0.62\n            reason = "Simple text pipeline task; file sizes unavailable, starting with CPU is acceptable for a guarded MVP."\n        elif total_size <= 500 * 1024**2:\n            runtime = "cpu-basic"\n            confidence = 0.82\n            reason = "Small text model; CPU Basic should be acceptable for validation demos."\n        elif total_size <= 2 * 1024**3:\n            runtime = "cpu-upgrade recommended"\n            confidence = 0.68\n            reason = "Medium text model; CPU Basic may be slow, CPU Upgrade is safer for runtime."\n            risks.append("CPU Basic may cold-start or infer slowly.")\n        else:\n            runtime = "manual-review or GPU/Endpoint"\n            confidence = 0.38\n            reason = "Large text model; not suitable for automatic CPU Space generation in this MVP."\n            risks.append("Model appears too large for the current CPU-only template.")\n            supported_for_phase5 = False\n        return {\n            "target_runtime": runtime,\n            "target_space_hardware": runtime,\n            "job_flavor": "cpu-basic",\n            "confidence": confidence,\n            "supported_by_current_builder": supported_for_phase5,\n            "reason": reason,\n            "risks": risks,\n            "evidence": evidence,\n        }\n\n    return {\n        "target_runtime": "unsupported",\n        "target_space_hardware": "manual-review",\n        "job_flavor": "cpu-basic",\n        "confidence": 0.25,\n        "supported_by_current_builder": False,\n        "reason": "Task/library combination is outside the current safe templates.",\n        "risks": risks + ["Unsupported by current builder templates."],\n        "evidence": evidence,\n    }\n\n\ndef main():\n    run_id = os.environ["RUN_ID"]\n    hf_username = os.environ.get("HF_USERNAME", "unknown")\n    bucket_source = os.environ.get("BUCKET_SOURCE", "unknown")\n    output_root = Path(os.environ.get("OUTPUT_ROOT", "/output"))\n    model_id = sanitize_model_id(os.environ.get("MODEL_ID", ""))\n    token = os.environ.get("HF_TOKEN")\n    run_dir = output_root / "runs" / run_id\n    events_path = run_dir / "events.jsonl"\n    state_path = run_dir / "state.json"\n\n    append_event(events_path, "bootstrap", "started", "Runtime recommender worker started", {"model_id": model_id})\n    write_json(state_path, {"run_id": run_id, "kind": "runtime_recommender", "status": "running", "message": "Analyzing model compatibility and runtime recommendation", "model_id": model_id, "created_by": hf_username, "bucket_source": bucket_source, "created_at": now(), "updated_at": now()})\n    if not token:\n        fail(run_dir, events_path, "HF_TOKEN is missing from Job secrets")\n\n    try:\n        append_event(events_path, "dependencies", "started", "Installing Python worker dependencies")\n        code, out = run_cmd([sys.executable, "-m", "pip", "install", "-q", "--upgrade", "huggingface_hub>=1.0.0"], timeout=600)\n        if code != 0:\n            fail(run_dir, events_path, "Python dependency installation failed", {"output_tail": out[-4000:]})\n        append_event(events_path, "dependencies", "success", "Python worker dependencies installed")\n\n        from huggingface_hub import HfApi, hf_hub_download\n        api = HfApi(token=token)\n        whoami = api.whoami(token=token)\n        append_event(events_path, "auth", "success", "Authenticated inside Job", {"whoami_name": whoami.get("name")})\n\n        append_event(events_path, "model_analysis", "started", "Fetching model metadata", {"model_id": model_id})\n        info = api.model_info(model_id, token=token, files_metadata=True)\n        siblings_objs = list(info.siblings or [])\n        siblings = [getattr(s, "rfilename", "") for s in siblings_objs]\n        sizes = [getattr(s, "size", None) for s in siblings_objs]\n        total_size = sum(x for x in sizes if isinstance(x, int)) if any(isinstance(x, int) for x in sizes) else None\n        pipeline_tag = getattr(info, "pipeline_tag", None)\n        library_name = getattr(info, "library_name", None)\n        tags = list(getattr(info, "tags", []) or [])\n        task = infer_task(pipeline_tag, library_name, tags, siblings)\n        readme_excerpt = ""\n        try:\n            readme_path = hf_hub_download(repo_id=model_id, filename="README.md", token=token)\n            readme_excerpt = Path(readme_path).read_text(encoding="utf-8", errors="ignore")[:5000]\n        except Exception as exc:\n            readme_excerpt = f"README unavailable: {exc}"\n\n        recommendation = recommend_runtime(task, library_name, tags, siblings, total_size)\n        analysis = {\n            "model_id": model_id,\n            "pipeline_tag": pipeline_tag,\n            "library_name": library_name,\n            "tags": tags[:100],\n            "siblings": siblings[:160],\n            "estimated_total_file_size_bytes": total_size,\n            "estimated_total_file_size_human": human_bytes(total_size),\n            "selected_task": task,\n            "readme_excerpt": readme_excerpt,\n            "runtime_recommendation": recommendation,\n        }\n        write_json(run_dir / "model_analysis.json", analysis)\n        write_json(run_dir / "runtime_recommendation.json", recommendation)\n        append_event(events_path, "model_analysis", "success", "Model metadata analyzed", {"selected_task": task, "target_runtime": recommendation["target_runtime"], "confidence": recommendation["confidence"], "supported_by_current_builder": recommendation["supported_by_current_builder"]})\n\n        status = "success" if recommendation["confidence"] >= 0.35 else "needs_review"\n        final_state = {"run_id": run_id, "kind": "runtime_recommender", "status": status, "message": "Runtime recommendation completed", "model_id": model_id, "created_by": hf_username, "bucket_source": bucket_source, "model_analysis": {k: v for k, v in analysis.items() if k != "readme_excerpt"}, "updated_at": now()}\n        write_json(state_path, final_state)\n        report = f"""# Agentic Space Factory — Runtime Recommendation Report\n\nRun ID: `{run_id}`\n\nStatus: **{status}**\n\nModel: `{model_id}`\n\n## Recommendation\n\n```json\n{json.dumps(recommendation, indent=2, ensure_ascii=False)}\n```\n\n## Model metadata\n\n- Pipeline tag: `{pipeline_tag}`\n- Library: `{library_name}`\n- Selected task: `{task}`\n- Estimated total file size: `{human_bytes(total_size)}`\n\n## Interpretation\n\n- `supported_by_current_builder=true` means Phase 5 can attempt the current simple text-pipeline builder.\n- `zerogpu-candidate` means the model looks like a candidate for the upcoming ZeroGPU/Diffusers template, but should not be sent through the CPU text builder.\n- `manual-review` means the model should not be auto-built without an explicit user decision.\n\n## Next step\n\nUse this recommendation as a gate before launching Phase 5/Phase 7 builders.\n"""\n        (run_dir / "report.md").write_text(report, encoding="utf-8")\n        append_event(events_path, "report_write", "success", "Wrote report.md")\n        append_event(events_path, "done", "success", "Runtime recommendation completed")\n    except SystemExit:\n        raise\n    except Exception as exc:\n        fail(run_dir, events_path, "Runtime recommender worker failed", {"error": str(exc)})\n\n\nif __name__ == "__main__":\n    main()\n'
+
+
+LONGCAT_ARTICLE_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 = "meituan-longcat/LongCat-Video-Avatar-1.5"
+
+
+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": "longcat_article_reproduction",
+        "status": status,
+        "message": message,
+        "updated_at": now(),
+        "details": safe,
+    })
+    report = f"""# Agentic Space Factory — LongCat 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"], 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 validate_live_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 Gradio API to become available")
+    deadline = time.time() + timeout_s
+    attempt = 0
+    last_error = None
+    while time.time() < deadline:
+        attempt += 1
+        try:
+            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 == "/health":
+                        result = client.predict(api_name=api_name)
+                    elif 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]}
+                    write_json(run_dir / "tests" / "test_result.json", payload)
+                    append_event(events_path, "api_validation", "success", "Live 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]})
+            last_error = "; ".join(f"{e['api_name']}: {e['error']}" for e in errors[:5]) or "No callable API endpoints found"
+        except Exception as exc:
+            last_error = str(exc)
+        append_event(events_path, "api_validation", "waiting", "Live API not ready yet", {"attempt": attempt, "error": last_error[-1500:] if last_error else None})
+        time.sleep(30)
+    raise RuntimeError(f"Live 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):
+    if not hardware:
+        return {"requested": False, "hardware": hardware, "ok": False, "error": "empty hardware"}
+    try:
+        from huggingface_hub import SpaceHardware
+        hw = SpaceHardware(hardware)
+        runtime = api.request_space_hardware(repo_id=target_space_id, hardware=hw, token=token)
+        payload = {"requested": True, "hardware": hardware, "ok": True, "runtime_stage": getattr(getattr(runtime, "stage", None), "value", str(getattr(runtime, "stage", None)))}
+        append_event(events_path, step, "success", f"Requested Space hardware {hardware}", payload)
+        return payload
+    except Exception as exc:
+        payload = {"requested": True, "hardware": hardware, "ok": False, "error": str(exc)[:2000]}
+        append_event(events_path, step, "failed", f"Could not request Space hardware {hardware}", payload)
+        return payload
+
+
+def create_initial_workspace(workspace: Path, model_id: str, target_space_id: str, preferred_hardware: str, fallback_hardware: str, allow_fallback: bool):
+    workspace.mkdir(parents=True, exist_ok=True)
+    app_py = f"""import gradio as gr
+
+MODEL_ID = "{model_id}"
+TARGET_SPACE_ID = "{target_space_id}"
+
+
+def health():
+    return {{
+        "status": "booted",
+        "model_id": MODEL_ID,
+        "note": "Pi should replace this scaffold with a LongCat demo while preserving /health."
+    }}
+
+
+def placeholder():
+    return "LongCat demo scaffold booted. The full model pipeline was not wired yet."
+
+with gr.Blocks(title="LongCat Video Avatar — Agentic Space Factory") as demo:
+    gr.Markdown("# LongCat Video Avatar — Agentic Space Factory")
+    gr.Markdown("This private Space was generated by the Phase 7 article reproduction worker.")
+    gr.JSON(label="Health", value=health(), every=None)
+    gr.Button("Health check").click(fn=health, inputs=None, outputs=gr.JSON(), api_name="health")
+    gr.Button("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
+opencv-python-headless
+pillow
+numpy
+"""
+    (workspace / "requirements.txt").write_text(req, encoding="utf-8")
+    readme = f"""---
+title: LongCat Video Avatar Agentic Factory
+sdk: gradio
+app_file: app.py
+python_version: "3.10"
+suggested_hardware: {preferred_hardware or fallback_hardware or "zero-a10g"}
+---
+
+# LongCat Video Avatar — Agentic Space Factory
+
+Private generated Space for `{model_id}`.
+
+This Space is created by Phase 7. It should remain private until manually reviewed.
+"""
+    (workspace / "README.md").write_text(readme, encoding="utf-8")
+    goal = f"""You are Pi running inside a Hugging Face Job for Agentic Space Factory Phase 7.
+
+Goal: attempt to reproduce the workflow from the Hugging Face article by building a private Space demo for:
+
+MODEL_ID: {model_id}
+TARGET_SPACE_ID: {target_space_id}
+
+First read and follow the operational rules from this gist:
+{GIST_URL}
+
+Important 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, upload files, and validate the live API. Do not create/delete repos yourself in this Phase 7 worker.
+- You should edit app.py, requirements.txt, and README.md to make the best possible LongCat demo for the model card.
+- Preserve a cheap /health endpoint that does not run full video generation. The wrapper uses it for boot validation.
+- If full generation is too risky or requires unavailable inputs, keep the app bootable and clearly document what remains incomplete.
+- Prefer ZeroGPU-compatible code when possible, but allow fixed GPU runtime. Preferred hardware: {preferred_hardware}. Fallback hardware enabled: {allow_fallback}. Fallback hardware: {fallback_hardware}.
+- If you use @spaces.GPU, decorate only the inference/generation function. Do not decorate /health.
+- Keep the huggingface_hub pin in requirements.txt: huggingface_hub>=0.34.0,<1.0.0.
+
+Deliverables:
+- app.py must boot on Hugging Face Spaces.
+- app.py must expose /health.
+- README.md must explain the runtime strategy and limitations.
+- Write a concise PI_SUMMARY.md with what you changed and whether full generation is implemented.
+"""
+    (workspace / "GOAL.md").write_text(goal, encoding="utf-8")
+    return ["app.py", "requirements.txt", "README.md", "GOAL.md"]
+
+
+def upload_workspace(api, workspace: Path, target_space_id: str, token: str, run_dir: Path, events_path: Path):
+    append_event(events_path, "upload_files", "started", "Uploading generated LongCat files")
+    gen_dir = run_dir / "generated"
+    gen_dir.mkdir(parents=True, exist_ok=True)
+    for filename in ["app.py", "README.md", "requirements.txt", "GOAL.md", "PI_SUMMARY.md"]:
+        src = workspace / filename
+        if src.exists():
+            shutil.copy2(src, gen_dir / filename)
+    for filename in ["app.py", "README.md", "requirements.txt"]:
+        src = workspace / filename
+        if not src.exists():
+            raise RuntimeError(f"Missing required generated file: {filename}")
+        api.upload_file(path_or_fileobj=str(src), path_in_repo=filename, repo_id=target_space_id, repo_type="space", token=token)
+        append_event(events_path, "upload_files", "success", f"Uploaded {filename}")
+
+
+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", "moonshotai/Kimi-K2.5")
+    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"}
+    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/longcat_workspace")
+
+    append_event(events_path, "bootstrap", "started", "LongCat article reproduction worker started", {"model_id": model_id, "target_space_id": target_space_id})
+    write_json(state_path, {"run_id": run_id, "kind": "longcat_article_reproduction", "status": "running", "message": "Attempting LongCat article-style 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 LongCat 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], "article_target": model_id == DEFAULT_MODEL_ID, "preferred_hardware": preferred_hardware, "fallback_hardware": fallback_hardware, "allow_fixed_gpu_fallback": allow_fixed_gpu_fallback}
+        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)
+        append_event(events_path, "workspace", "success", "Prepared LongCat 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 LongCat 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 LongCat 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 LongCat 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})
+
+        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})
+
+        upload_workspace(api, workspace, target_space_id, token, run_dir, events_path)
+        validation = validate_live_api(target_space_id, token, run_dir, events_path, timeout_s=1200)
+        collect_pi_traces(run_dir, events_path)
+
+        final_state = {"run_id": run_id, "kind": "longcat_article_reproduction", "status": "success", "message": "LongCat article-style Space created and boot/API health validated", "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, "updated_at": now(), "created_by": hf_username, "bucket_source": bucket_source}
+        write_json(state_path, final_state)
+        report = f"""# Agentic Space Factory — LongCat Article Reproduction Report
+
+Run ID: `{run_id}`
+
+Status: **success**
+
+Target Space: https://huggingface.co/spaces/{target_space_id}
+
+Model: `{model_id}`
+
+## Hardware
+
+Selected/requested hardware: `{selected_hardware}`
+
+```json
+{json.dumps(hardware_attempts, indent=2, ensure_ascii=False)}
+```
+
+## Validation
+
+The wrapper validated the live Gradio API. This Phase 7 validation is intentionally a **boot/health API validation** first. Full video generation may still require manual review, real inputs, quota, and model-specific optimization.
+
+```json
+{json.dumps(validation, 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, 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", "success", "LongCat article reproduction worker completed", {"target_space": target_space_id, "selected_hardware": selected_hardware})
+    except SystemExit:
+        raise
+    except Exception as exc:
+        try:
+            collect_pi_traces(run_dir, events_path)
+        except Exception:
+            pass
+        fail(run_dir, events_path, "LongCat article reproduction 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)
@@ -593,6 +1096,16 @@ def encoded_pi_model_card_worker_script() -> str:
     return _encode(PI_MODEL_CARD_WORKER_SCRIPT)
 
 
+
+def encoded_runtime_recommender_worker_script() -> str:
+    """Return the base64-encoded Phase 6 runtime recommender worker script."""
+    return _encode(RUNTIME_RECOMMENDER_WORKER_SCRIPT)
+
+
+def encoded_longcat_article_worker_script() -> str:
+    """Return the base64-encoded Phase 7 LongCat article reproduction worker script."""
+    return _encode(LONGCAT_ARTICLE_WORKER_SCRIPT)
+
 def python_decode_and_run_command() -> list[str]:
     """Command list for `run_job`.