Blame for app.py

"""
AI API failure states: one Gradio app, three tabs.

  1. Rate limits          429 -> Retry-After / exponential backoff -> success
  2. Stalled streams      silence detection, keep the partial answer
  3. Refusal vs. error    a declined request is not a broken request

Default mode is Simulated: free, no API key, every failure on demand.
"Real API" mode streams from OpenAI or Anthropic through the same code.
"""

import html
import os
import time

import gradio as gr
import spaces

from core import Event, stream_with_retries
from providers import RealClient, SimulatedClient


@spaces.GPU
def _warmup():
    """No-op. Exists only so Hugging Face's ZeroGPU startup check finds an
    @spaces.GPU function. This app never actually needs a GPU."""
    return None


COLORS = {"ok": "#5fb87a", "wait": "#e0a336", "error": "#d9695f", "refusal": "#8b909c"}
DEFAULT_PROMPT = "Explain how a TCP handshake works in three sentences."


# ---------------------------------------------------------------- helpers
def panel(kind: str, title: str, body: str = "", code: str = "") -> str:
    c = COLORS[kind]
    chip = (f'<span style="margin-left:auto;font-family:monospace;font-size:0.75rem;'
            f'opacity:.7">{html.escape(code)}</span>') if code else ""
    body_html = (f'<div style="margin-top:6px;font-size:0.9rem;opacity:.9">'
                 f'{html.escape(body)}</div>') if body else ""
    return (f'<div style="border:1px solid {c}55;background:{c}1a;border-radius:8px;'
            f'padding:12px 14px;"><div style="display:flex;align-items:center;gap:8px;'
            f'font-weight:600"><span style="width:8px;height:8px;border-radius:50%;'
            f'background:{c}"></span>{html.escape(title)}{chip}</div>{body_html}</div>')


def make_client(mode, provider, key, scenario):
    """Return a client with .stream(prompt). Raises ValueError with a
    human-readable message if real mode is missing a key."""
    if mode.startswith("Simulated"):
        return SimulatedClient(scenario)
    key = (key or "").strip() or os.environ.get(
        "OPENAI_API_KEY" if provider == "openai" else "ANTHROPIC_API_KEY", "")
    if not key:
        raise ValueError(f"Real API mode needs a {provider} API key.")
    return RealClient(provider, key)


class Log:
    def __init__(self):
        self.t0, self.lines = time.time(), []

    def add(self, line):
        self.lines.append(f"+{time.time() - self.t0:4.1f}s  {line}")
        return "\n".join(self.lines)

    def text(self):
        return "\n".join(self.lines)


# ------------------------------------------------------------ tab 1: 429s
def run_rate_limit(mode, provider, key, prompt):
    try:
        client = make_client(mode, provider, key, "rate_limit")
    except ValueError as e:
        yield "", panel("error", "Missing API key", str(e)), ""
        return
    log, text = Log(), ""
    for ev in stream_with_retries(lambda: client.stream(prompt), max_retries=3):
        text = ev.text or text
        if ev.kind == "start":
            yield text, panel("wait", f"Sending request (attempt {ev.attempt})"), log.add(f"attempt {ev.attempt}: request sent")
        elif ev.kind == "chunk":
            yield text, panel("wait", "Streaming\u2026"), log.text()
        elif ev.kind == "retry_wait" and ev.message:
            yield text, panel("wait", f"Rate limited. Retrying in {ev.seconds}s", ev.message, "429"), log.add(f"[429] {ev.message}")
        elif ev.kind == "retry_wait":
            yield text, panel("wait", f"Rate limited. Retrying in {ev.seconds}s", "Your request is queued; nothing was lost.", "429"), log.text()
        elif ev.kind == "done":
            yield text, panel("ok", "Done", f"Succeeded on attempt {ev.attempt}.", "200"), log.add("[200] stream completed")
        else:  # error
            yield text, panel("error", "Request failed", ev.message, str(ev.status_code or "")), log.add(f"[error] {ev.message}")


# ------------------------------------------------------- tab 2: stalled stream
def run_stall(mode, provider, key, prompt, stall_seconds, force_ok=False):
    hide = gr.update(visible=False)
    try:
        client = make_client(mode, provider, key, "ok" if force_ok else "stall")
    except ValueError as e:
        yield "", panel("error", "Missing API key", str(e)), "", hide, hide
        return
    log, text = Log(), ""
    for ev in stream_with_retries(lambda: client.stream(prompt), stall_seconds=stall_seconds):
        text = ev.text or text
        if ev.kind == "start":
            yield "", panel("wait", "Streaming\u2026"), log.add("request sent"), hide, hide
        elif ev.kind == "chunk":
            yield text, panel("wait", "Streaming\u2026"), log.text(), hide, hide
        elif ev.kind == "done":
            yield text, panel("ok", "Stream completed", code="200"), log.add("[200] stream completed"), hide, hide
        elif ev.kind == "stalled":
            n = len(text.split())
            yield (text,
                   panel("error", f"Response stalled after {n} words",
                         f"{ev.message}. The partial answer is kept. Retry, or keep what you have."),
                   log.add(f"[stall] {ev.message} (kept {n} words)"),
                   gr.update(visible=True), gr.update(visible=True))
        else:
            yield text, panel("error", "Request failed", ev.message), log.add(f"[error] {ev.message}"), hide, hide


def keep_partial(text):
    hide = gr.update(visible=False)
    return panel("refusal", "Kept partial answer", "Marked as incomplete so nobody mistakes it for a full answer."), hide, hide


# --------------------------------------------------- tab 3: refusal vs error
def run_refusal(mode, provider, key, scenario, prompt):
    sim = {"Normal answer": "ok", "Refusal (200, model declined)": "refusal",
           "Server error (500)": "server_error"}[scenario]
    try:
        client = make_client(mode, provider, key, sim)
    except ValueError as e:
        yield "", panel("error", "Missing API key", str(e))
        return
    text = ""
    for ev in stream_with_retries(lambda: client.stream(prompt)):
        text = ev.text or text
        if ev.kind == "chunk":
            yield text, panel("wait", "Streaming\u2026")
        elif ev.kind == "done":
            yield text, panel("ok", "Response", code="200")
        elif ev.kind == "refused":
            yield text, panel("refusal", "The model declined this request", ev.message + " Rewording the same ask won't help; try a different task.", "200 \u00b7 refusal")
        elif ev.kind == "error":
            yield text, panel("error", "Something went wrong on our end", ev.message + " This is a service problem, not a judgment about your prompt. Trying again is reasonable.", str(ev.status_code or "error"))


# ------------------------------------------------------------------- UI
with gr.Blocks(title="AI API failure states") as demo:
    gr.Markdown("## AI API failure states\nThree failure modes, one shared retry/stream layer. "
                "**Simulated** mode is free and needs no key; **Real API** streams from a live provider.")
    with gr.Row():
        mode = gr.Radio(["Simulated (free, no key)", "Real API"], value="Simulated (free, no key)", label="Mode")
        provider = gr.Dropdown(["openai", "anthropic"], value="openai", label="Provider (Real API only)")
        key = gr.Textbox(label="API key (Real API only)", type="password",
                         placeholder="Falls back to the Space secret if left empty")
    prompt = gr.Textbox(label="Prompt", value=DEFAULT_PROMPT, lines=2)

    with gr.Tabs():
        with gr.Tab("1 \u00b7 Rate limits (429)"):
            gr.Markdown("Two simulated 429s: the first has a `Retry-After`, the second doesn't (exponential backoff + jitter). Then it succeeds.")
            b1 = gr.Button("Send", variant="primary")
            o1, s1, l1 = gr.Textbox(label="Response", lines=5), gr.HTML(), gr.Textbox(label="Request log", lines=5, interactive=False)
            b1.click(run_rate_limit, [mode, provider, key, prompt], [o1, s1, l1])

        with gr.Tab("2 \u00b7 Stalled stream"):
            gr.Markdown("The stream goes silent mid-sentence. A watchdog notices, the partial answer is kept, and the user chooses.")
            stall_s = gr.Slider(1, 10, value=3, step=1, label="Silence threshold (seconds)")
            b2 = gr.Button("Send", variant="primary")
            o2, s2, l2 = gr.Textbox(label="Response", lines=5), gr.HTML(), gr.Textbox(label="Request log", lines=4, interactive=False)
            with gr.Row():
                retry2 = gr.Button("Retry request", visible=False)
                keep2 = gr.Button("Keep partial answer", visible=False)
            outs2 = [o2, s2, l2, retry2, keep2]
            b2.click(run_stall, [mode, provider, key, prompt, stall_s], outs2)
            retry2.click(lambda *a: run_stall(*a, force_ok=True), [mode, provider, key, prompt, stall_s], outs2)
            keep2.click(keep_partial, [o2], [s2, retry2, keep2])

        with gr.Tab("3 \u00b7 Refusal vs. error"):
            gr.Markdown("A refusal is a *successful* response that says no. A 500 is a failure. They need different copy and different next steps. "
                        "In Real API mode the scenario picker is ignored, and refusals are detected from the provider's structured signals.")
            scen = gr.Radio(["Normal answer", "Refusal (200, model declined)", "Server error (500)"],
                            value="Refusal (200, model declined)", label="Scenario (Simulated only)")
            b3 = gr.Button("Send", variant="primary")
            o3, s3 = gr.Textbox(label="Response", lines=3), gr.HTML()
            b3.click(run_refusal, [mode, provider, key, scen, prompt], [o3, s3])

if __name__ == "__main__":
    demo.queue().launch()