| from __future__ import annotations |
|
|
| import json |
| from typing import Any |
|
|
| import gradio as gr |
|
|
| from src.bucket import read_run_bundle |
| from src.config import settings |
| from src.jobs import ( |
| fetch_recent_logs_safe, |
| inspect_job_safe, |
| launch_create_private_space_job, |
| launch_hello_job, |
| launch_pi_gist_recipe_job, |
| launch_pi_model_card_job, |
| launch_pi_space_smoke_job, |
| ) |
| from src.runs import make_run_id, validate_run_id |
| from src.security import redact |
|
|
|
|
| APP_DESCRIPTION = f""" |
| # Agentic Space Factory — V5 Model Card |
| |
| This version validates the two critical foundations: |
| |
| ```text |
| Phase 1: HF OAuth → HF Job → mounted private Bucket → run state/events/report → UI readback |
| Phase 2: HF OAuth → HF Job → private target Space → file upload → live Gradio API validation → Bucket report |
| Phase 3: HF OAuth → HF Job → Pi modifies app.py → private target Space → live API validation → Pi traces |
| Phase 4: HF OAuth → HF Job → Pi reads gist → uses hf CLI → private Space → wrapper live API validation |
| Phase 5: HF OAuth → HF Job → model-card analysis → Pi adapts template → private model Space → live API validation |
| ``` |
| |
| Configured bucket: `{settings.bucket_uri}` |
| """ |
|
|
|
|
| def _profile_username(profile: Any) -> str | None: |
| if profile is None: |
| return None |
| if isinstance(profile, dict): |
| return profile.get("preferred_username") or profile.get("username") or profile.get("name") |
| return getattr(profile, "preferred_username", None) or getattr(profile, "username", None) or getattr(profile, "name", None) |
|
|
|
|
| def _token_value(oauth_token: Any) -> str | None: |
| if oauth_token is None: |
| return None |
| if isinstance(oauth_token, str): |
| return oauth_token |
| return getattr(oauth_token, "token", None) or getattr(oauth_token, "access_token", None) |
|
|
|
|
| def get_login_status(profile: gr.OAuthProfile | None) -> str: |
| username = _profile_username(profile) |
| if not username: |
| return "Not signed in. Use the Hugging Face login button before launching a Job." |
| return f"Signed in as **{username}**. Phase 2 will only create target Spaces under `{username}/...`." |
|
|
|
|
| def propose_hello_run_id() -> str: |
| return make_run_id("hello") |
|
|
|
|
| def propose_space_run_id() -> str: |
| return make_run_id("space") |
|
|
|
|
|
|
| def propose_pi_run_id() -> str: |
| return make_run_id("pi") |
|
|
|
|
| def propose_gist_run_id() -> str: |
| return make_run_id("gist") |
|
|
|
|
| def propose_model_run_id() -> str: |
| return make_run_id("model") |
|
|
|
|
|
|
| def launch_pi_model_card_job_ui( |
| requested_run_id: str, |
| model_id: str, |
| target_space_name: str, |
| pi_model: str, |
| profile: gr.OAuthProfile | None, |
| oauth_token: gr.OAuthToken | None, |
| ) -> tuple[str, str, str, str, str, str]: |
| username = _profile_username(profile) |
| token = _token_value(oauth_token) |
| if not username or not token: |
| raise gr.Error("Please sign in with Hugging Face first. OAuth profile/token is missing.") |
|
|
| run_id = validate_run_id(requested_run_id or propose_model_run_id()) |
| result = launch_pi_model_card_job( |
| token=token, |
| username=username, |
| target_slug=target_space_name, |
| model_id=model_id, |
| pi_model=pi_model, |
| run_id=run_id, |
| ) |
|
|
| job_url = result.get("job_url") or "" |
| target_space = result.get("target_space") or "" |
| target_url = result.get("target_space_url") or "" |
| summary = json.dumps(result, indent=2) |
| return run_id, result["job_id"], job_url, target_space, target_url, summary |
|
|
|
|
| def launch_pi_gist_job_ui( |
| requested_run_id: str, |
| target_space_name: str, |
| pi_model: str, |
| profile: gr.OAuthProfile | None, |
| oauth_token: gr.OAuthToken | None, |
| ) -> tuple[str, str, str, str, str, str]: |
| username = _profile_username(profile) |
| token = _token_value(oauth_token) |
| if not username or not token: |
| raise gr.Error("Please sign in with Hugging Face first. OAuth profile/token is missing.") |
|
|
| run_id = validate_run_id(requested_run_id or propose_gist_run_id()) |
| result = launch_pi_gist_recipe_job( |
| token=token, |
| username=username, |
| target_slug=target_space_name, |
| pi_model=pi_model, |
| run_id=run_id, |
| ) |
|
|
| job_url = result.get("job_url") or "" |
| target_space = result.get("target_space") or "" |
| target_url = result.get("target_space_url") or "" |
| summary = json.dumps(result, indent=2) |
| return run_id, result["job_id"], job_url, target_space, target_url, summary |
|
|
|
|
| def launch_pi_space_job_ui( |
| requested_run_id: str, |
| target_space_name: str, |
| pi_model: str, |
| profile: gr.OAuthProfile | None, |
| oauth_token: gr.OAuthToken | None, |
| ) -> tuple[str, str, str, str, str, str]: |
| username = _profile_username(profile) |
| token = _token_value(oauth_token) |
| if not username or not token: |
| raise gr.Error("Please sign in with Hugging Face first. OAuth profile/token is missing.") |
|
|
| run_id = validate_run_id(requested_run_id or propose_pi_run_id()) |
| result = launch_pi_space_smoke_job( |
| token=token, |
| username=username, |
| target_slug=target_space_name, |
| pi_model=pi_model, |
| run_id=run_id, |
| ) |
|
|
| job_url = result.get("job_url") or "" |
| target_space = result.get("target_space") or "" |
| target_url = result.get("target_space_url") or "" |
| summary = json.dumps(result, indent=2) |
| return run_id, result["job_id"], job_url, target_space, target_url, summary |
|
|
| def launch_hello_job_ui( |
| requested_run_id: str, |
| profile: gr.OAuthProfile | None, |
| oauth_token: gr.OAuthToken | None, |
| ) -> tuple[str, str, str, str, str]: |
| username = _profile_username(profile) |
| token = _token_value(oauth_token) |
| if not username or not token: |
| raise gr.Error("Please sign in with Hugging Face first. OAuth profile/token is missing.") |
|
|
| run_id = validate_run_id(requested_run_id or propose_hello_run_id()) |
| result = launch_hello_job(token=token, username=username, run_id=run_id) |
|
|
| job_url = result.get("job_url") or "" |
| summary = json.dumps(result, indent=2) |
| return run_id, result["job_id"], job_url, summary, f"Launched hello run `{run_id}`." |
|
|
|
|
| def launch_private_space_job_ui( |
| requested_run_id: str, |
| target_space_name: str, |
| profile: gr.OAuthProfile | None, |
| oauth_token: gr.OAuthToken | None, |
| ) -> tuple[str, str, str, str, str, str]: |
| username = _profile_username(profile) |
| token = _token_value(oauth_token) |
| if not username or not token: |
| raise gr.Error("Please sign in with Hugging Face first. OAuth profile/token is missing.") |
|
|
| run_id = validate_run_id(requested_run_id or propose_space_run_id()) |
| result = launch_create_private_space_job( |
| token=token, |
| username=username, |
| target_slug=target_space_name, |
| run_id=run_id, |
| ) |
|
|
| job_url = result.get("job_url") or "" |
| target_space = result.get("target_space") or "" |
| target_url = result.get("target_space_url") or "" |
| summary = json.dumps(result, indent=2) |
| return run_id, result["job_id"], job_url, target_space, target_url, summary |
|
|
|
|
| def refresh_run_ui( |
| run_id: str, |
| job_id: str, |
| oauth_token: gr.OAuthToken | None, |
| ) -> tuple[str, str, str, str]: |
| token = _token_value(oauth_token) |
| if not token: |
| raise gr.Error("Please sign in with Hugging Face first.") |
| run_id = validate_run_id(run_id) |
|
|
| bundle = read_run_bundle(run_id, token=token) |
| job_info = inspect_job_safe(job_id, token=token) if job_id else {} |
| logs = redact(fetch_recent_logs_safe(job_id, token=token)) if job_id else "" |
|
|
| state_text = json.dumps(bundle.get("state") or {"status": "not_available_yet"}, indent=2, ensure_ascii=False) |
| events = bundle.get("events") or [] |
| events_text = "\n".join(json.dumps(event, ensure_ascii=False) for event in events) or "No events found yet. The Job may still be scheduling." |
| report_text = bundle.get("report") or "No report found yet. Refresh after the Job has started writing to the bucket." |
| job_text = json.dumps(job_info, indent=2, ensure_ascii=False) |
| if logs: |
| job_text += "\n\nRecent job logs:\n" + logs |
|
|
| return state_text, events_text, report_text, job_text |
|
|
|
|
| def build_demo() -> gr.Blocks: |
| with gr.Blocks(title="Agentic Space Factory") as demo: |
| gr.Markdown(APP_DESCRIPTION) |
| gr.LoginButton() |
|
|
| login_status = gr.Markdown() |
| demo.load(fn=get_login_status, inputs=None, outputs=login_status) |
|
|
|
|
| with gr.Tab("Phase 5 — Model card → private Space"): |
| gr.Markdown( |
| """ |
| This phase starts from a real Hugging Face `model_id`. The Job fetches model metadata/model card, gates to a small set of simple Transformers text-pipeline tasks, prepares a Gradio template, asks Pi to adapt the app, then creates a **private** target Space and validates the live API. |
| |
| Recommended first test model: `sshleifer/tiny-gpt2`. |
| |
| This phase is intentionally limited: simple text models only, private Space only, no ZeroGPU yet, no gated-model secret injection yet. |
| """ |
| ) |
| with gr.Row(): |
| model_run_id_box = gr.Textbox(label="Run ID", value=propose_model_run_id, interactive=True) |
| new_model_run_btn = gr.Button("Generate new run id") |
| new_model_run_btn.click(fn=propose_model_run_id, inputs=None, outputs=model_run_id_box) |
|
|
| model_id_box = gr.Textbox( |
| label="Model ID", |
| value="sshleifer/tiny-gpt2", |
| info="Use a small public model for the first test. Phase 5 supports simple Transformers text pipeline tasks only.", |
| ) |
| model_target_space_name = gr.Textbox( |
| label="Target Space name", |
| placeholder="e.g. space-factory-model-tiny-gpt2-v1", |
| info="Use a new name. The Space is created under your own username and is private.", |
| ) |
| model_pi_model_box = gr.Textbox( |
| label="Pi model", |
| value="moonshotai/Kimi-K2.5", |
| info="Model used by Pi through Hugging Face Inference Providers.", |
| ) |
| launch_model_btn = gr.Button("Generate model Space", variant="primary") |
|
|
| phase5_job_id_box = gr.Textbox(label="Job ID", interactive=True) |
| phase5_job_url_box = gr.Textbox(label="Job URL", interactive=False) |
| phase5_target_space_box = gr.Textbox(label="Target Space", interactive=False) |
| phase5_target_url_box = gr.Textbox(label="Target Space URL", interactive=False) |
| phase5_launch_result = gr.Code(label="Launch result", language="json") |
|
|
| launch_model_btn.click( |
| fn=launch_pi_model_card_job_ui, |
| inputs=[model_run_id_box, model_id_box, model_target_space_name, model_pi_model_box], |
| outputs=[ |
| model_run_id_box, |
| phase5_job_id_box, |
| phase5_job_url_box, |
| phase5_target_space_box, |
| phase5_target_url_box, |
| phase5_launch_result, |
| ], |
| ) |
|
|
| phase5_refresh_btn = gr.Button("Refresh Phase 5 run status") |
| with gr.Tab("Phase 5 state"): |
| phase5_state_code = gr.Code(label="state.json", language="json") |
| with gr.Tab("Phase 5 events"): |
| phase5_events_code = gr.Code(label="events.jsonl", language="json") |
| with gr.Tab("Phase 5 report"): |
| phase5_report_md = gr.Markdown() |
| with gr.Tab("Phase 5 job"): |
| phase5_job_code = gr.Code(label="Job status/logs", language="json") |
|
|
| phase5_refresh_btn.click( |
| fn=refresh_run_ui, |
| inputs=[model_run_id_box, phase5_job_id_box], |
| outputs=[phase5_state_code, phase5_events_code, phase5_report_md, phase5_job_code], |
| ) |
|
|
| with gr.Tab("Phase 4 — Pi gist recipe"): |
| gr.Markdown( |
| """ |
| This phase is the first closer reproduction of the article workflow. The HF Job installs Pi, gives it the HF Spaces Agent Quickstart gist, and asks Pi to use the `hf` CLI to create/upload a **private** target Space. |
| |
| The wrapper still stays in control of final success: it independently checks that the target Space exists and validates the live Gradio API. |
| |
| Use a fresh Space name. This phase intentionally fails if the target Space already exists. |
| """ |
| ) |
| with gr.Row(): |
| gist_run_id_box = gr.Textbox(label="Run ID", value=propose_gist_run_id, interactive=True) |
| new_gist_run_btn = gr.Button("Generate new run id") |
| new_gist_run_btn.click(fn=propose_gist_run_id, inputs=None, outputs=gist_run_id_box) |
|
|
| gist_target_space_name = gr.Textbox( |
| label="Target Space name", |
| placeholder="e.g. space-factory-pi-gist-v1", |
| info="Use a new name. Pi is asked to create this private Space under your username using hf CLI.", |
| ) |
| gist_model_box = gr.Textbox( |
| label="Pi model", |
| value="moonshotai/Kimi-K2.5", |
| info="Model used by Pi through Hugging Face Inference Providers.", |
| ) |
| launch_gist_btn = gr.Button("Run Pi gist recipe", variant="primary") |
|
|
| phase4_job_id_box = gr.Textbox(label="Job ID", interactive=True) |
| phase4_job_url_box = gr.Textbox(label="Job URL", interactive=False) |
| phase4_target_space_box = gr.Textbox(label="Target Space", interactive=False) |
| phase4_target_url_box = gr.Textbox(label="Target Space URL", interactive=False) |
| phase4_launch_result = gr.Code(label="Launch result", language="json") |
|
|
| launch_gist_btn.click( |
| fn=launch_pi_gist_job_ui, |
| inputs=[gist_run_id_box, gist_target_space_name, gist_model_box], |
| outputs=[ |
| gist_run_id_box, |
| phase4_job_id_box, |
| phase4_job_url_box, |
| phase4_target_space_box, |
| phase4_target_url_box, |
| phase4_launch_result, |
| ], |
| ) |
|
|
| phase4_refresh_btn = gr.Button("Refresh Phase 4 run status") |
| with gr.Tab("Phase 4 state"): |
| phase4_state_code = gr.Code(label="state.json", language="json") |
| with gr.Tab("Phase 4 events"): |
| phase4_events_code = gr.Code(label="events.jsonl", language="json") |
| with gr.Tab("Phase 4 report"): |
| phase4_report_md = gr.Markdown() |
| with gr.Tab("Phase 4 job"): |
| phase4_job_code = gr.Code(label="Job status/logs", language="json") |
|
|
| phase4_refresh_btn.click( |
| fn=refresh_run_ui, |
| inputs=[gist_run_id_box, phase4_job_id_box], |
| outputs=[phase4_state_code, phase4_events_code, phase4_report_md, phase4_job_code], |
| ) |
|
|
|
|
| with gr.Tab("Phase 3 — Pi smoke test"): |
| gr.Markdown( |
| """ |
| This is the first Pi integration. It launches an HF Job that installs Pi, configures it with Hugging Face Inference Providers, asks Pi to make a small safe edit to `app.py`, then creates a **private** target Space and validates the live API. |
| |
| This phase is intentionally narrow: Pi must only add the marker phrase `Pi modified this app` to the hello app. If Pi fails or does not make the expected edit, the Job fails safely before reporting success. |
| """ |
| ) |
| with gr.Row(): |
| pi_run_id_box = gr.Textbox(label="Run ID", value=propose_pi_run_id, interactive=True) |
| new_pi_run_btn = gr.Button("Generate new run id") |
| new_pi_run_btn.click(fn=propose_pi_run_id, inputs=None, outputs=pi_run_id_box) |
|
|
| pi_target_space_name = gr.Textbox( |
| label="Target Space name", |
| placeholder="e.g. space-factory-pi-smoke-v1", |
| info="Use a new name. The Space is created under your own username and is private.", |
| ) |
| pi_model_box = gr.Textbox( |
| label="Pi model", |
| value="moonshotai/Kimi-K2.5", |
| info="Model used by Pi through Hugging Face Inference Providers. You can replace it with another provider-backed coding model.", |
| ) |
| launch_pi_btn = gr.Button("Run Pi smoke test", variant="primary") |
|
|
| phase3_job_id_box = gr.Textbox(label="Job ID", interactive=True) |
| phase3_job_url_box = gr.Textbox(label="Job URL", interactive=False) |
| phase3_target_space_box = gr.Textbox(label="Target Space", interactive=False) |
| phase3_target_url_box = gr.Textbox(label="Target Space URL", interactive=False) |
| phase3_launch_result = gr.Code(label="Launch result", language="json") |
|
|
| launch_pi_btn.click( |
| fn=launch_pi_space_job_ui, |
| inputs=[pi_run_id_box, pi_target_space_name, pi_model_box], |
| outputs=[ |
| pi_run_id_box, |
| phase3_job_id_box, |
| phase3_job_url_box, |
| phase3_target_space_box, |
| phase3_target_url_box, |
| phase3_launch_result, |
| ], |
| ) |
|
|
| phase3_refresh_btn = gr.Button("Refresh Phase 3 run status") |
| with gr.Tab("Phase 3 state"): |
| phase3_state_code = gr.Code(label="state.json", language="json") |
| with gr.Tab("Phase 3 events"): |
| phase3_events_code = gr.Code(label="events.jsonl", language="json") |
| with gr.Tab("Phase 3 report"): |
| phase3_report_md = gr.Markdown() |
| with gr.Tab("Phase 3 job"): |
| phase3_job_code = gr.Code(label="Job status/logs", language="json") |
|
|
| phase3_refresh_btn.click( |
| fn=refresh_run_ui, |
| inputs=[pi_run_id_box, phase3_job_id_box], |
| outputs=[phase3_state_code, phase3_events_code, phase3_report_md, phase3_job_code], |
| ) |
|
|
| with gr.Tab("Phase 2 — Create private Space"): |
| gr.Markdown( |
| """ |
| This is the next safe increment. It creates a **private** hello-world Gradio Space under your own namespace, uploads files, waits for the live API, then writes the result to the bucket. |
| |
| It will fail safely if the target Space already exists, to avoid overwriting user resources unexpectedly. |
| """ |
| ) |
| with gr.Row(): |
| space_run_id_box = gr.Textbox(label="Run ID", value=propose_space_run_id, interactive=True) |
| new_space_run_btn = gr.Button("Generate new run id") |
| new_space_run_btn.click(fn=propose_space_run_id, inputs=None, outputs=space_run_id_box) |
|
|
| target_space_name = gr.Textbox( |
| label="Target Space name", |
| placeholder="e.g. space-factory-hello-test", |
| info="Use a new name. For Phase 2, the Space is always created under your own username and is private.", |
| ) |
| launch_space_btn = gr.Button("Create private hello Space", variant="primary") |
|
|
| phase2_job_id_box = gr.Textbox(label="Job ID", interactive=True) |
| phase2_job_url_box = gr.Textbox(label="Job URL", interactive=False) |
| phase2_target_space_box = gr.Textbox(label="Target Space", interactive=False) |
| phase2_target_url_box = gr.Textbox(label="Target Space URL", interactive=False) |
| phase2_launch_result = gr.Code(label="Launch result", language="json") |
|
|
| launch_space_btn.click( |
| fn=launch_private_space_job_ui, |
| inputs=[space_run_id_box, target_space_name], |
| outputs=[ |
| space_run_id_box, |
| phase2_job_id_box, |
| phase2_job_url_box, |
| phase2_target_space_box, |
| phase2_target_url_box, |
| phase2_launch_result, |
| ], |
| ) |
|
|
| phase2_refresh_btn = gr.Button("Refresh Phase 2 run status") |
| with gr.Tab("Phase 2 state"): |
| phase2_state_code = gr.Code(label="state.json", language="json") |
| with gr.Tab("Phase 2 events"): |
| phase2_events_code = gr.Code(label="events.jsonl", language="json") |
| with gr.Tab("Phase 2 report"): |
| phase2_report_md = gr.Markdown() |
| with gr.Tab("Phase 2 job"): |
| phase2_job_code = gr.Code(label="Job status/logs", language="json") |
|
|
| phase2_refresh_btn.click( |
| fn=refresh_run_ui, |
| inputs=[space_run_id_box, phase2_job_id_box], |
| outputs=[phase2_state_code, phase2_events_code, phase2_report_md, phase2_job_code], |
| ) |
|
|
| with gr.Tab("Phase 1 — Hello Job"): |
| gr.Markdown("Keep this tab as a regression test for OAuth → Job → Bucket only.") |
| with gr.Row(): |
| run_id_box = gr.Textbox(label="Run ID", value=propose_hello_run_id, interactive=True) |
| new_run_btn = gr.Button("Generate new run id") |
| new_run_btn.click(fn=propose_hello_run_id, inputs=None, outputs=run_id_box) |
|
|
| launch_btn = gr.Button("Launch hello HF Job", variant="secondary") |
| job_id_box = gr.Textbox(label="Job ID", interactive=True) |
| job_url_box = gr.Textbox(label="Job URL", interactive=False) |
| launch_result = gr.Code(label="Launch result", language="json") |
| status_md = gr.Markdown() |
|
|
| launch_btn.click( |
| fn=launch_hello_job_ui, |
| inputs=[run_id_box], |
| outputs=[run_id_box, job_id_box, job_url_box, launch_result, status_md], |
| ) |
|
|
| refresh_btn = gr.Button("Refresh hello run status") |
| with gr.Tab("Hello state"): |
| state_code = gr.Code(label="state.json", language="json") |
| with gr.Tab("Hello events"): |
| events_code = gr.Code(label="events.jsonl", language="json") |
| with gr.Tab("Hello report"): |
| report_md = gr.Markdown() |
| with gr.Tab("Hello job"): |
| job_code = gr.Code(label="Job status/logs", language="json") |
|
|
| refresh_btn.click( |
| fn=refresh_run_ui, |
| inputs=[run_id_box, job_id_box], |
| outputs=[state_code, events_code, report_md, job_code], |
| ) |
|
|
| gr.Markdown( |
| """ |
| ## Next increments |
| |
| After Phase 5 passes, the next step is Phase 6: |
| |
| 1. add a ZeroGPU Diffusers template, |
| 2. improve model compatibility gating, |
| 3. support richer validation by output type, |
| 4. keep the wrapper-owned live API validation gate. |
| """ |
| ) |
|
|
| return demo |
|
|
|
|
| if __name__ == "__main__": |
| build_demo().launch() |
|
|
