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, ) from src.runs import make_run_id, validate_run_id from src.security import redact APP_DESCRIPTION = f""" # Agentic Space Factory — V2 Foundation 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 ``` 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 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 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 2 passes, the next step is Phase 3: 1. install and configure Pi inside the Job, 2. ask Pi to modify/repair the simple target Space, 3. collect Pi traces into the bucket, 4. keep the same live API validation gate. """ ) return demo if __name__ == "__main__": build_demo().launch()