app.py

from __future__ import annotations

import json
from typing import Any

import gradio as gr

from src.bucket import check_user_bucket, create_user_bucket, read_run_bundle
from src.config import settings, user_bucket_source
from src.jobs import (
    fetch_recent_logs_safe,
    inspect_job_safe,
    launch_universal_model_card_job,
    launch_validate_existing_space_job,
)
from src.runs import make_run_id, validate_run_id
from src.security import redact


APP_DESCRIPTION = f"""
# Agentic Space Factory

Turn a Hugging Face model card into a **private, testable Gradio Space** using an agentic HF Job.

## Recommended workflow

```text
1. Build from model card
   → creates a private Space
   → attempts ZeroGPU first
   → falls back to a fixed GPU if automatic hardware assignment is available
   → otherwise marks the run as manual_hardware_required

2. If hardware had to be changed manually
   → set the GPU in the generated Space Settings
   → run Validate existing Space
   → smoke-test generation
   → measure latency
   → store the output artifact in the Bucket
```

Each launch returns quick links to open the HF Job, generated Space, Space settings, and run artifacts in new tabs.

## Honest guarantees

- Spaces are private by default.
- Nothing is published automatically.
- Runs, reports, generated files, traces, validation results, and artifacts are written to your private Bucket.
- Success is based on the deployed Space, not only generated code.
- ZeroGPU and fixed-GPU upgrades are best-effort through OAuth; manual hardware selection is an expected fallback.

## Limits

This app attempts model-card builds; it does not guarantee that every model will run. Multi-GPU models, Docker-only apps, custom CUDA/FlashAttention stacks, gated models, very large models, or models with unclear documentation may produce `technical_blocker`, `health_only`, or `manual_hardware_required` instead of a full inference success.

Run Bucket: by default each signed-in user writes to their own private bucket: `<username>/{settings.bucket_name}`. Use **Check run bucket** or **Create private run bucket** before launching Jobs.
"""


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}**. Generated Spaces are created under `{username}/...` and remain private."




def _safe_url(url: str | None) -> str:
    return (url or "").strip()


def _run_artifacts_url(run_id: str | None, bucket_source: str | None) -> str:
    if not run_id or not bucket_source:
        return ""
    return f"https://huggingface.co/buckets/{bucket_source}/tree/main/runs/{run_id}"


def _button_link(label: str, url: str | None):
    url = _safe_url(url)
    return gr.update(value=label, link=url or None, visible=bool(url))


def _job_button(job_url: str | None):
    return _button_link("Open HF Job ↗", job_url)


def _space_button(target_space_url: str | None):
    return _button_link("Open target Space ↗", target_space_url)


def _settings_button(target_space_url: str | None):
    target_space_url = _safe_url(target_space_url)
    return _button_link("Open Space settings ↗", f"{target_space_url}/settings" if target_space_url else "")


def _artifacts_button(run_id: str | None, bucket_source: str | None):
    return _button_link("Open run artifacts ↗", _run_artifacts_url(run_id, bucket_source))




def _format_bucket_status(status: dict[str, Any]) -> str:
    source = status.get("bucket_source") or "unknown"
    uri = status.get("bucket_uri") or ""
    if status.get("ok"):
        return (
            f"✅ Run bucket ready: `{source}`\n\n"
            f"Bucket URI: `{uri}`\n\n"
            "New Jobs will mount this private bucket and write runs under `runs/<run_id>/`."
        )
    if status.get("exists") is False:
        return (
            f"⚠️ Run bucket not found: `{source}`\n\n"
            "Click **Create private run bucket** before launching a Job, or create it manually in Hugging Face Storage Buckets."
        )
    return (
        f"❌ Could not check run bucket: `{source}`\n\n"
        f"```text\n{redact(str(status.get('error') or 'Unknown error'))}\n```"
    )


def check_run_bucket_ui(
    bucket_name: str,
    profile: gr.OAuthProfile | None,
    oauth_token: gr.OAuthToken | None,
) -> 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.")
    return _format_bucket_status(check_user_bucket(username=username, bucket_name=bucket_name, token=token))


def create_run_bucket_ui(
    bucket_name: str,
    profile: gr.OAuthProfile | None,
    oauth_token: gr.OAuthToken | None,
) -> 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.")
    return _format_bucket_status(create_user_bucket(username=username, bucket_name=bucket_name, token=token))


def propose_universal_run_id() -> str:
    return make_run_id("universal")


def propose_validate_run_id() -> str:
    return make_run_id("validate")


def launch_universal_model_card_job_ui(
    requested_run_id: str,
    model_id: str,
    target_space_name: str,
    pi_model: str,
    preferred_hardware: str,
    allow_fixed_gpu_fallback: bool,
    fallback_hardware: str,
    implementation_mode: str,
    bucket_name: str,
    profile: gr.OAuthProfile | None,
    oauth_token: gr.OAuthToken | None,
) -> tuple[str, str, str, str, str, Any, Any, Any, Any, 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_universal_run_id())
    result = launch_universal_model_card_job(
        token=token,
        username=username,
        target_slug=target_space_name,
        model_id=model_id,
        pi_model=pi_model,
        preferred_space_hardware=preferred_hardware,
        fallback_space_hardware=fallback_hardware,
        allow_fixed_gpu_fallback=allow_fixed_gpu_fallback,
        implementation_mode=implementation_mode,
        run_id=run_id,
        bucket_name=bucket_name,
    )
    job_url = result.get("job_url") or ""
    target_space_url = result.get("target_space_url") or ""
    bucket_source = result.get("bucket_source") or user_bucket_source(username=username, bucket_name=bucket_name)
    return (
        run_id,
        result["job_id"],
        job_url,
        result.get("target_space") or "",
        target_space_url,
        _job_button(job_url),
        _space_button(target_space_url),
        _settings_button(target_space_url),
        _artifacts_button(run_id, bucket_source),
        json.dumps(result, indent=2),
    )


def launch_validate_existing_space_job_ui(
    requested_run_id: str,
    target_space_id: str,
    api_name: str,
    test_args_json: str,
    test_kwargs_json: str,
    expected_output_type: str,
    live_timeout_seconds: float,
    bucket_name: str,
    profile: gr.OAuthProfile | None,
    oauth_token: gr.OAuthToken | None,
) -> tuple[str, str, str, str, Any, Any, Any, Any, 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_validate_run_id())
    try:
        json.loads(test_args_json or "[]")
        json.loads(test_kwargs_json or "{}")
    except Exception as exc:
        raise gr.Error(f"Invalid JSON test args/kwargs: {exc}") from exc

    result = launch_validate_existing_space_job(
        token=token,
        username=username,
        target_space_id=target_space_id,
        api_name=api_name,
        test_args_json=test_args_json,
        test_kwargs_json=test_kwargs_json,
        expected_output_type=expected_output_type,
        live_timeout_seconds=int(live_timeout_seconds or 1800),
        run_id=run_id,
        bucket_name=bucket_name,
    )
    job_url = result.get("job_url") or ""
    target_space_url = result.get("target_space_url") or f"https://huggingface.co/spaces/{result.get('target_space', target_space_id)}"
    bucket_source = result.get("bucket_source") or user_bucket_source(username=username, bucket_name=bucket_name)
    return (
        run_id,
        result["job_id"],
        job_url,
        target_space_url,
        _job_button(job_url),
        _space_button(target_space_url),
        _settings_button(target_space_url),
        _artifacts_button(run_id, bucket_source),
        json.dumps(result, indent=2),
    )


def refresh_run_ui(
    run_id: str,
    job_id: str,
    bucket_name: str,
    profile: gr.OAuthProfile | None,
    oauth_token: gr.OAuthToken | None,
) -> tuple[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.")
    run_id = validate_run_id(run_id)
    bucket_source = user_bucket_source(username=username, bucket_name=bucket_name)

    bundle = read_run_bundle(run_id, bucket_source=bucket_source, 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)

        gr.Markdown("## Run storage")
        gr.Markdown(
            "Runs are stored in a private Storage Bucket under the signed-in user's namespace. "
            "Create it once here, then use the same bucket name for Build and Validate."
        )
        global_bucket_name = gr.Textbox(
            label="Run Bucket name",
            value=settings.bucket_name,
            info="The app uses <your-username>/<bucket-name>. Default: space-factory-runs.",
        )
        with gr.Row():
            check_bucket_btn = gr.Button("Check run bucket")
            create_bucket_btn = gr.Button("Create private run bucket", variant="primary")
        bucket_status = gr.Markdown("Sign in, then check or create your private run bucket before launching Jobs.")
        check_bucket_btn.click(fn=check_run_bucket_ui, inputs=[global_bucket_name], outputs=bucket_status)
        create_bucket_btn.click(fn=create_run_bucket_ui, inputs=[global_bucket_name], outputs=bucket_status)

        with gr.Tab("Build from model card"):
            gr.Markdown(
                """
Paste a Hugging Face model ID or model-card URL. The worker creates a **private** Space, asks Pi + Qwen Coder to build the best Gradio app it can, attempts ZeroGPU first, then a fixed-GPU fallback if enabled. If automatic hardware assignment fails, set the hardware manually in the generated Space settings and run the validation tab.
"""
            )
            with gr.Row():
                build_run_id = gr.Textbox(label="Run ID", value=propose_universal_run_id, interactive=True)
                gr.Button("Generate new run id").click(fn=propose_universal_run_id, inputs=None, outputs=build_run_id)
            model_id = gr.Textbox(
                label="Model card URL or model ID",
                value="Tongyi-MAI/Z-Image-Turbo",
                info="Examples: owner/model, https://huggingface.co/owner/model",
            )
            target_space_name = gr.Textbox(
                label="Target Space name",
                placeholder="e.g. space-factory-z-image-v1",
                info="Use a fresh name. The Space is created under your username and remains private.",
            )
            pi_model = gr.Textbox(
                label="Pi model",
                value="Qwen/Qwen3-Coder-Next",
                info="Model used by Pi through Hugging Face Inference Providers.",
            )
            implementation_mode = gr.Dropdown(
                label="Implementation mode",
                choices=["full-inference-gated", "full-inference-attempt", "safe-scaffold"],
                value="full-inference-gated",
                info="Gated mode forbids placeholder success; impossible models must produce technical blockers.",
            )
            with gr.Row():
                preferred_hw = gr.Dropdown(
                    label="Preferred Space hardware",
                    choices=["zero-a10g", "cpu-basic", "t4-small", "t4-medium", "a10g-large", "l40sx1", "a100-large", "h200"],
                    value="zero-a10g",
                    info="ZeroGPU is attempted first by the worker. If your quota is exceeded, use manual hardware selection after generation.",
                )
                allow_fallback = gr.Checkbox(label="Allow fixed GPU fallback", value=True)
                fallback_hw = gr.Dropdown(
                    label="Fallback Space hardware",
                    choices=["l40sx1", "a10g-large", "a100-large", "h200", "t4-medium"],
                    value="l40sx1",
                )

            build_btn = gr.Button("Build private Space", variant="primary")
            build_job_id = gr.Textbox(label="Job ID", interactive=True)
            build_job_url = gr.Textbox(label="Job URL", interactive=False)
            generated_space = gr.Textbox(label="Generated Space", interactive=False)
            generated_space_url = gr.Textbox(label="Generated Space URL", interactive=False)
            gr.Markdown("Quick links")
            with gr.Row():
                build_job_button = gr.Button("Open HF Job ↗", link=None, link_target="_blank", visible=False)
                build_space_button = gr.Button("Open target Space ↗", link=None, link_target="_blank", visible=False)
                build_settings_button = gr.Button("Open Space settings ↗", link=None, link_target="_blank", visible=False)
                build_artifacts_button = gr.Button("Open run artifacts ↗", link=None, link_target="_blank", visible=False)
            build_result = gr.Code(label="Launch result", language="json")

            build_btn.click(
                fn=launch_universal_model_card_job_ui,
                inputs=[build_run_id, model_id, target_space_name, pi_model, preferred_hw, allow_fallback, fallback_hw, implementation_mode, global_bucket_name],
                outputs=[
                    build_run_id,
                    build_job_id,
                    build_job_url,
                    generated_space,
                    generated_space_url,
                    build_job_button,
                    build_space_button,
                    build_settings_button,
                    build_artifacts_button,
                    build_result,
                ],
            )

            build_refresh = gr.Button("Refresh build run status")
            with gr.Tab("Build state"):
                build_state = gr.Code(label="state.json", language="json")
            with gr.Tab("Build events"):
                build_events = gr.Code(label="events.jsonl", language="json")
            with gr.Tab("Build report"):
                build_report = gr.Markdown()
            with gr.Tab("Build job"):
                build_job_info = gr.Code(label="Job info/logs", language="json")

            build_refresh.click(fn=refresh_run_ui, inputs=[build_run_id, build_job_id, global_bucket_name], outputs=[build_state, build_events, build_report, build_job_info])

        with gr.Tab("Validate existing Space"):
            gr.Markdown(
                """
Use this after the builder generated a Space, especially if you had to set the GPU manually. This job does not rerun Pi. It waits for the existing Space, calls a live generation endpoint, checks the output type, stores returned artifacts in the Bucket, measures latency, and recommends a conservative ZeroGPU duration.
"""
            )
            with gr.Row():
                validate_run_id = gr.Textbox(label="Run ID", value=propose_validate_run_id, interactive=True)
                gr.Button("Generate new validation run id").click(fn=propose_validate_run_id, inputs=None, outputs=validate_run_id)
            target_space = gr.Textbox(
                label="Existing target Space",
                placeholder="fffiloni/space-factory-... or https://huggingface.co/spaces/...",
            )
            with gr.Row():
                api_name = gr.Textbox(label="Generation API name", value="/generate")
                expected_type = gr.Dropdown(label="Expected output type", choices=["image", "video", "audio", "text", "any"], value="image")
            test_args = gr.Code(label="Test args JSON list", language="json", value='["a cinematic robot cat astronaut, detailed, studio lighting"]')
            test_kwargs = gr.Code(label="Test kwargs JSON object", language="json", value="{}")
            timeout_s = gr.Number(label="Live wait timeout seconds", value=1800, precision=0)

            validate_btn = gr.Button("Validate Space + smoke-test generation", variant="primary")
            validate_job_id = gr.Textbox(label="Job ID", interactive=True)
            validate_job_url = gr.Textbox(label="Job URL", interactive=False)
            validate_space_url = gr.Textbox(label="Target Space URL", interactive=False)
            gr.Markdown("Quick links")
            with gr.Row():
                validate_job_button = gr.Button("Open HF Job ↗", link=None, link_target="_blank", visible=False)
                validate_space_button = gr.Button("Open target Space ↗", link=None, link_target="_blank", visible=False)
                validate_settings_button = gr.Button("Open Space settings ↗", link=None, link_target="_blank", visible=False)
                validate_artifacts_button = gr.Button("Open run artifacts ↗", link=None, link_target="_blank", visible=False)
            validate_result = gr.Code(label="Launch result", language="json")

            validate_btn.click(
                fn=launch_validate_existing_space_job_ui,
                inputs=[validate_run_id, target_space, api_name, test_args, test_kwargs, expected_type, timeout_s, global_bucket_name],
                outputs=[
                    validate_run_id,
                    validate_job_id,
                    validate_job_url,
                    validate_space_url,
                    validate_job_button,
                    validate_space_button,
                    validate_settings_button,
                    validate_artifacts_button,
                    validate_result,
                ],
            )

            validate_refresh = gr.Button("Refresh validation run status")
            with gr.Tab("Validation state"):
                validate_state = gr.Code(label="state.json", language="json")
            with gr.Tab("Validation events"):
                validate_events = gr.Code(label="events.jsonl", language="json")
            with gr.Tab("Validation report"):
                validate_report = gr.Markdown()
            with gr.Tab("Validation job"):
                validate_job_info = gr.Code(label="Job info/logs", language="json")

            validate_refresh.click(fn=refresh_run_ui, inputs=[validate_run_id, validate_job_id, global_bucket_name], outputs=[validate_state, validate_events, validate_report, validate_job_info])

        with gr.Tab("About & limits"):
            gr.Markdown(
                """
## Result statuses

- `full_inference_success`: a live generation smoke test returned the expected output type.
- `manual_hardware_required`: the Space was generated but automatic ZeroGPU/fixed-GPU assignment failed; set hardware manually, then validate.
- `full_inference_candidate_health_passed`: the Space boots and contains inference signals, but generation was not smoke-tested yet.
- `health_only`: the Space boots, but no real inference path was validated.
- `technical_blocker`: the agent found concrete blockers such as multi-GPU requirements, missing licenses, custom CUDA, or unclear usage.
- `failed`: the build, runtime, or validation job failed.

## Hardware policy

The builder tries to create an app optimized for ZeroGPU when GPU is needed. It attempts ZeroGPU first, then a fixed-GPU fallback if enabled. Hardware assignment through OAuth may fail because of quota, billing, or permission limits; manual hardware selection is a supported path.

## What this app cannot guarantee

It cannot guarantee that every model card becomes a working Space. It cannot bypass model licenses, ZeroGPU quota, billing requirements, custom CUDA build failures, multi-GPU needs, or missing model documentation.
"""
            )

    return demo


if __name__ == "__main__":
    build_demo().launch()