Generative UI

This looks like the known 3.1.0 reconnect storm. Possible duplicates: #143 (high: same
ORB-1001 flood after failover), #150 (medium: same error code) and #145 (medium). I'd
suggest marking #161 as a duplicate of #143 and pointing the reporter at 3.1.2.

{"issue_id": 7,
 "candidates": [
   {"issue_id": 3, "title": "Worker hangs on shutdown", "confidence": "high", "reason": "same stack trace"},
   {"issue_id": 5, "title": "SIGTERM ignored", "confidence": "medium", "reason": "same symptom"},
   {"issue_id": 6, "title": "Slow startup", "confidence": "low", "reason": "same component"}]}

# script_id: day_080_generative_ui/first_component
# Install: uv add "openai>=3.46" pydantic   Env: OPENAI_API_KEY, ATLAS_MODEL, ORBIT_DATA
import json
import os
from pathlib import Path
from typing import Literal

from openai import OpenAI, pydantic_responses_function_tool
from pydantic import BaseModel

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))

class IssueCard(BaseModel):
    """One Orbit issue: number, title, state, labels and link."""
    issue_id: int
    title: str
    state: Literal["open", "closed"]
    labels: list[str]
    url: str

issue = next(i for i in map(json.loads, (DATA_DIR / "issues/issues.jsonl").open()) if i["id"] == 143)
tool = pydantic_responses_function_tool(IssueCard, name="show_issue_card", description=IssueCard.__doc__)
resp = OpenAI().responses.create(
    model=MODEL, tools=[tool], tool_choice={"type": "function", "name": "show_issue_card"},
    input=f"Show this issue as a card. Its URL is https://issues.orbit.example/143\n{json.dumps(issue)[:3000]}")
call = next(i for i in resp.output if i.type == "function_call")   # forced, so it's there
print(repr(IssueCard.model_validate_json(call.arguments)))           # arguments is a JSON string

IssueCard(issue_id=143, title='3.1.0: reconnect storm after Redis failover, thousands of connections and an ORB-1001 flood take Redis down again', state='closed', labels=['type:bug', 'comp:broker-redis', 'severity:S1', 'version:3.1'], url='https://issues.orbit.example/143')

# script_id: day_080_generative_ui/break_it
# Install: uv add pydantic   No key needed: the "model output" below is hand-written.
import html
from typing import Literal
from urllib.parse import urlsplit

from pydantic import BaseModel, ValidationError

class IssueCard(BaseModel):              # same props model as first_component
    issue_id: int
    title: str
    state: Literal["open", "closed"]
    labels: list[str]
    url: str

bad = ('{"issue_id": 143, "title": "Reconnect storm", "state": "closed", '
       '"labels": "bug", "url": "https://issues.orbit.example/143"}')
try:
    IssueCard.model_validate_json(bad)
except ValidationError as exc:
    print(exc)                           # this is what you log; the user gets the text fallback

hostile = IssueCard(issue_id=999, title="<img src=x onerror=alert(1)>", state="open",
                    labels=["type:bug"], url="javascript:alert(document.cookie)")
print("naive:  ", f'<a href="{hostile.url}">{hostile.title}</a>')
print("escaped:", html.escape(hostile.title, quote=True))
print("url ok? ", urlsplit(hostile.url).scheme == "https")

1 validation error for IssueCard
labels
  Input should be a valid array [type=list_type, input_value='bug', input_type=str]
    For further information visit https://errors.pydantic.dev/2.14/v/list_type
naive:   <a href="javascript:alert(document.cookie)"><img src=x onerror=alert(1)></a>
escaped: &lt;img src=x onerror=alert(1)&gt;
url ok?  False

# script_id: day_080_generative_ui/components
# Install: uv add "openai>=3.46" pydantic   No key needed: this file validates and renders only.
# Becomes atlas/ui/components.py in R5. stream_app, gradio_view and test_renderer import it.
import html
import json
import logging
import re
from collections.abc import Callable
from typing import Literal
from urllib.parse import urlsplit

from openai import pydantic_responses_function_tool
from pydantic import BaseModel, Field, ValidationError

log = logging.getLogger("atlas.ui")

class AgentEvent(BaseModel):            # compact copy of Day 77's envelope (atlas/api/events.py)
    seq: int = 0
    type: Literal["status", "token", "tool_call", "tool_result", "citation", "component",
                  "final", "error", "cancelled", "done"]
    data: dict = Field(default_factory=dict)

class IssueCard(BaseModel):
    """One Orbit issue: number, title, state, labels and link."""
    issue_id: int
    title: str
    state: Literal["open", "closed"]
    labels: list[str]
    url: str

class Citation(BaseModel):              # compact copy of Day 15 / Day 28
    doc_id: str
    quote: str

class CitationList(BaseModel):
    """The Orbit docs an answer relied on, each with the passage quoted."""
    citations: list[Citation]

# script_id: day_080_generative_ui/components
class DuplicateCandidate(BaseModel):
    issue_id: int
    title: str
    confidence: Literal["high", "medium", "low"]
    reason: str

class DuplicateCandidates(BaseModel):
    """Earlier issues that may describe the same problem as issue_id, best match first."""
    issue_id: int
    candidates: list[DuplicateCandidate]

class ConfigDiff(BaseModel):
    """A before/after config change, such as orbit.ini (2.x) to orbit.toml (3.x)."""
    before: str
    after: str
    notes: list[str]

class ApprovalArgs(BaseModel):          # typed, not `dict`: strict tools need closed objects
    issue_id: int
    value: str                          # the label, the duplicate's number, or the reply text

class ApprovalRequest(BaseModel):
    """Ask a maintainer to approve one write action. Showing it never executes anything."""
    action: Literal["add_label", "mark_duplicate", "post_reply"]
    args: ApprovalArgs
    risk: Literal["low", "high"]

VERSION = 1                             # bump when a props model changes shape (see the Stretch)
COMPONENTS: dict[str, type[BaseModel]] = {
    m.__name__: m for m in (IssueCard, CitationList, DuplicateCandidates, ConfigDiff, ApprovalRequest)}

# script_id: day_080_generative_ui/components
def tool_name(component: str) -> str:
    """IssueCard -> show_issue_card. Derived, so tools and registry can't drift apart."""
    return "show_" + re.sub(r"(?<!^)(?=[A-Z])", "_", component).lower()

TOOL_TO_COMPONENT = {tool_name(n): n for n in COMPONENTS}

def component_tools() -> list[dict]:
    """One strict function tool per registered component, built from its props model."""
    return [pydantic_responses_function_tool(m, name=tool_name(n), description=f"Display a {n}. {m.__doc__}")
            for n, m in COMPONENTS.items()]

MAX_ROWS, MAX_CHARS = 10, 800           # a card is not a data dump

def _cap(value):
    """Truncate long strings and lists before validation, so one runaway call can't flood the page."""
    if isinstance(value, str):
        return value if len(value) <= MAX_CHARS else value[:MAX_CHARS] + "…"
    if isinstance(value, list):
        return [_cap(v) for v in value[:MAX_ROWS]]
    if isinstance(value, dict):
        return {k: _cap(v) for k, v in value.items()}
    return value

def fallback_markdown(component: str, arguments: str) -> str:
    """Plain text any client can show. The client still renders it through its safe sink (Day 79)."""
    try:
        fields = json.loads(arguments)
    except json.JSONDecodeError:
        fields = None
    lines = [f"- {k}: {str(v)[:200]}" for k, v in fields.items()] if isinstance(fields, dict) else []
    return f"(Couldn't draw the {component} card, showing plain text.)\n" + "\n".join(lines[:MAX_ROWS])

# script_id: day_080_generative_ui/components
def to_ui_event(name: str, arguments: str, seq: int = 0) -> AgentEvent:
    """Tool call -> validated `component` event, or a `token` event carrying a text fallback."""
    component = TOOL_TO_COMPONENT.get(name, name)        # accepts show_issue_card or IssueCard
    model = COMPONENTS.get(component)
    try:
        if model is None:
            raise KeyError(f"unknown component {component!r}")
        props = model.model_validate(_cap(json.loads(arguments)))
    except (KeyError, json.JSONDecodeError, ValidationError) as exc:
        log.warning("component miss %s: %s", component, exc)   # count these: it's an eval signal
        return AgentEvent(seq=seq, type="token", data={
            "text": fallback_markdown(component, arguments), "fallback": True,
            "reason": type(exc).__name__})
    return AgentEvent(seq=seq, type="component",
                      data={"name": component, "version": VERSION, "props": props.model_dump()})

# script_id: day_080_generative_ui/strict_lint
# Install: uv add pydantic "openai>=3.46"   No key needed. Needs components.py next to this file.
from openai import pydantic_responses_function_tool
from pydantic import BaseModel

from components import component_tools

def open_objects(schema: dict, path: str = "$") -> list[str]:
    """Paths of object schemas without additionalProperties: false (strict mode refuses them)."""
    found = [path] if schema.get("type") == "object" and schema.get("additionalProperties") is not False else []
    for key, sub in schema.get("properties", {}).items():
        found += open_objects(sub, f"{path}.{key}")
    for key, sub in schema.get("$defs", {}).items():
        found += open_objects(sub, f"$defs.{key}")
    if isinstance(schema.get("items"), dict):
        found += open_objects(schema["items"], f"{path}[]")
    for sub in schema.get("anyOf", []):
        found += open_objects(sub, path)
    return found

class LooseApproval(BaseModel):         # the mistake ApprovalArgs avoids
    action: str
    args: dict

loose = pydantic_responses_function_tool(LooseApproval, name="show_loose_approval")
for tool in [*component_tools(), loose]:
    problems = open_objects(tool["parameters"])
    print(f"{tool['name']:<28} {'OPEN: ' + ', '.join(problems) if problems else 'ok'}")

show_issue_card              ok
show_citation_list           ok
show_duplicate_candidates    ok
show_config_diff             ok
show_approval_request        ok
show_loose_approval          OPEN: $.args

# script_id: day_080_generative_ui/stream_app
# Install: uv add fastapi uvicorn "openai>=3.46" pydantic   Env: OPENAI_API_KEY, ATLAS_MODEL, ORBIT_DATA
# No key? It replays a fixture (ATLAS_OFFLINE=1 forces that). Needs components.py next to it.
# Run: uv run python stream_app.py          (serves on :8000)
#      uv run python stream_app.py --print  (prints one SSE transcript and exits)
import json
import os
import re
import sys
from collections.abc import AsyncGenerator, AsyncIterable, Callable
from functools import cache
from pathlib import Path
from typing import Literal

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent
from openai import AsyncOpenAI
from pydantic import BaseModel

from components import TOOL_TO_COMPONENT, AgentEvent, ApprovalArgs, component_tools, to_ui_event

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
RULES = ("You are Atlas, assistant to Orbit's maintainers. Investigate the issue with the tools. "
         "When the answer has structure (duplicates, a config change, a proposed write action), "
         "show it with the matching show_* tool, then add one or two plain sentences.")

@cache
def load_issues() -> dict[int, dict]:
    lines = (DATA_DIR / "issues" / "issues.jsonl").read_text(encoding="utf-8").splitlines()
    return {d["id"]: d for d in map(json.loads, lines)}

def search_issues(query: str, k: int = 5) -> list[dict]:
    """Keyword-overlap stand-in for R2's hybrid search (Day 27), over issue titles."""
    terms = set(re.findall(r"[a-z0-9-]{3,}", query.lower()))
    ranked = sorted(load_issues().values(), key=lambda i: -sum(t in i["title"].lower() for t in terms))
    return [{"id": i["id"], "title": i["title"], "state": i["state"]} for i in ranked[:k]]

# script_id: day_080_generative_ui/stream_app
def _tool(name: str, description: str, prop: str, typ: str) -> dict:
    return {"type": "function", "name": name, "description": description, "strict": True,
            "parameters": {"type": "object", "properties": {prop: {"type": typ}},
                           "required": [prop], "additionalProperties": False}}

READ_TOOLS = [_tool("get_issue", "Fetch an Orbit issue by number.", "issue_id", "integer"),
              _tool("search_issues", "Find Orbit issues whose titles match a query.", "query", "string")]
TOOLS = READ_TOOLS + component_tools()          # the registry generates the UI tools

def run_tool(name: str, args: dict) -> str:     # read-only, so safe to run on untrusted issue text
    if name == "get_issue":
        issue = load_issues().get(args["issue_id"], {})
        return json.dumps({k: issue.get(k) for k in ("id", "title", "state", "labels", "body")})[:4000]
    return json.dumps(search_issues(args["query"]))

# script_id: day_080_generative_ui/stream_app
async def investigate(client, issue_id: int, max_steps: int = 6) -> AsyncGenerator[AgentEvent, None]:
    """Compact Day 48 loop: text streams as `token` events, show_* calls become `component` events."""
    items: list = [{"role": "user", "content": f"Investigate Orbit issue #{issue_id}."}]
    for _ in range(max_steps):
        output = None
        stream = await client.responses.create(model=MODEL, instructions=RULES, input=items,
                                               tools=TOOLS, stream=True)
        async for ev in stream:
            if ev.type == "response.output_text.delta":
                yield AgentEvent(type="token", data={"text": ev.delta})
            elif ev.type == "response.completed":
                output = ev.response.output
        if output is None:                      # failed or incomplete: say so, don't hang
            yield AgentEvent(type="error", data={"code": "model_stream_ended"})
            return
        items += output
        calls = [i for i in output if i.type == "function_call"]
        if not calls:
            return
        for call in calls:                      # answer EVERY call, or the next request is rejected
            if call.name in TOOL_TO_COMPONENT:
                event = to_ui_event(call.name, call.arguments)
                yield event
                result = "rendered" if event.type == "component" else "not rendered: props invalid"
            else:
                yield AgentEvent(type="tool_call", data={"name": call.name})
                result = run_tool(call.name, json.loads(call.arguments))
            items.append({"type": "function_call_output", "call_id": call.call_id, "output": result})
    yield AgentEvent(type="error", data={"code": "max_steps"})

# script_id: day_080_generative_ui/stream_app
from types import SimpleNamespace as NS

FIXTURE = [   # (text the model streams, [(tool name, arguments)]) per model turn
    ("", [("search_issues", {"query": "redis failover reconnect storm ORB-1001"})]),
    ("", [("show_duplicate_candidates", {"issue_id": 161, "candidates": [
        {"issue_id": 143, "title": "3.1.0: reconnect storm after Redis failover", "confidence": "high",
         "reason": "Same 3.1.0 failover and ORB-1001 flood"},
        {"issue_id": 150, "title": "ORB-1001 spam after Redis primary switch", "confidence": "medium",
         "reason": "Same error code, less detail"}]}),
          ("show_approval_request", {"action": "mark_duplicate", "args": {"issue_id": 161, "value": "143"},
                                     "risk": "low"})]),
    ("#161 matches #143, fixed in 3.1.2. Upgrading stops the storm.", []),
]

class ReplayClient:
    """Offline stand-in for AsyncOpenAI: `client.responses.create(...)` replays FIXTURE turns."""
    def __init__(self, turns: list) -> None:
        self.responses, self._turns = self, iter(turns)   # so client.responses.create works

    async def create(self, **_: object):
        text, calls = next(self._turns)
        output = [NS(type="function_call", name=n, arguments=json.dumps(a), call_id=f"call_{n}")
                  for n, a in calls]

        async def events():
            for word in text.split():
                yield NS(type="response.output_text.delta", delta=word + " ")
            yield NS(type="response.completed", response=NS(output=output))
        return events()

# script_id: day_080_generative_ui/stream_app
class InvestigateRequest(BaseModel):
    issue_id: int

class Decision(BaseModel):
    decision: Literal["approve", "reject"]
    action: str
    args: ApprovalArgs

DECISIONS: list[Decision] = []                  # recorded only; executing is Day 55 / Day 76's job

def create_app(get_client: Callable[[], object]) -> FastAPI:
    app = FastAPI()

    @app.post("/v1/investigate/stream", response_class=EventSourceResponse)
    async def investigate_stream(req: InvestigateRequest) -> AsyncIterable[ServerSentEvent]:
        seq = 0
        async for e in investigate(get_client(), req.issue_id):
            seq += 1
            yield ServerSentEvent(id=str(seq), event=e.type, data=e.data)
        yield ServerSentEvent(id=str(seq + 1), event="done", data={})

    @app.post("/v1/approvals", status_code=202)
    async def approvals(d: Decision) -> dict:
        DECISIONS.append(d)
        return {"status": "recorded"}
    return app

if __name__ == "__main__":
    offline = os.environ.get("ATLAS_OFFLINE") == "1" or "OPENAI_API_KEY" not in os.environ
    shared = None if offline else AsyncOpenAI()  # one client for the process, reused per request
    app = create_app(lambda: ReplayClient(FIXTURE) if offline else shared)
    if "--print" in sys.argv:
        from fastapi.testclient import TestClient
        with TestClient(app).stream("POST", "/v1/investigate/stream", json={"issue_id": 161}) as r:
            print("\n".join(line for line in r.iter_lines() if line.startswith(("event", "data"))))
    else:
        import uvicorn
        uvicorn.run(app, port=8000)

event: tool_call
data: {"name": "search_issues"}
event: component
data: {"name": "DuplicateCandidates", "version": 1, "props": {"issue_id": 161, "candidates": [...]}}
event: component
data: {"name": "ApprovalRequest", "version": 1, "props": {"action": "mark_duplicate", ...}}
event: token
data: {"text": "#161 "}
...
event: done
data: {}

# script_id: day_080_generative_ui/components
ALLOWED_HOSTS = frozenset({"docs.orbit.example", "issues.orbit.example"})
ISSUES_URL, DOCS_URL = "https://issues.orbit.example/", "https://docs.orbit.example/"

def is_allowed_url(url: str) -> bool:
    """Compact Day 71 rule: https only, exact host match, no user@host tricks."""
    try:
        parts = urlsplit(url.strip())
        parts.port                                    # raises ValueError on a malformed port
    except ValueError:
        return False
    return parts.scheme == "https" and parts.hostname in ALLOWED_HOSTS and parts.username is None

def esc(value: object) -> str:
    return html.escape(str(value), quote=True)        # quote=True makes it safe inside attributes

def link(url: str, text: str) -> str:
    """A link only if the URL passes the allow-list; otherwise the words, as plain text."""
    if not is_allowed_url(url):
        return esc(text)
    return f'<a href="{esc(url)}" rel="noopener noreferrer nofollow">{esc(text)}</a>'

def _issue(n: int) -> str:
    return link(f"{ISSUES_URL}{n}", f"#{n}")          # n is a validated int, not model text

def _doc(doc_id: str) -> str:
    return link(DOCS_URL + doc_id, doc_id) if re.fullmatch(r"[a-z0-9-]+", doc_id) else esc(doc_id)

# script_id: day_080_generative_ui/components
def _issue_card(p: IssueCard) -> str:
    labels = " ".join(f'<span class="label">{esc(x)}</span>' for x in p.labels)
    return f"<h3>{_issue(p.issue_id)} {link(p.url, p.title)}</h3><p>{esc(p.state)} {labels}</p>"

def _citations(p: CitationList) -> str:
    return "<ol>" + "".join(f"<li>{_doc(c.doc_id)}: {esc(c.quote)}</li>" for c in p.citations) + "</ol>"

def _duplicates(p: DuplicateCandidates) -> str:
    rows = "".join(f"<tr><td>{_issue(c.issue_id)}</td><td>{esc(c.title)}</td>"
                   f"<td>{esc(c.confidence)}</td><td>{esc(c.reason)}</td></tr>" for c in p.candidates)
    return (f"<h3>Possible duplicates of {_issue(p.issue_id)}</h3><table><tr><th>Issue</th>"
            f"<th>Title</th><th>Confidence</th><th>Why</th></tr>{rows}</table>")

def _config_diff(p: ConfigDiff) -> str:
    notes = "".join(f"<li>{esc(n)}</li>" for n in p.notes)
    return f"<pre>{esc(p.before)}</pre><p>becomes</p><pre>{esc(p.after)}</pre><ul>{notes}</ul>"

def _approval(p: ApprovalRequest) -> str:
    # Inert buttons: no inline handlers. The host wires them to POST /v1/approvals.
    return (f"<p>Atlas asks to <b>{esc(p.action)}</b> on {_issue(p.args.issue_id)}: "
            f"{esc(p.args.value)} (risk: {esc(p.risk)})</p>"
            '<button type="button" data-decision="approve">Approve</button>'
            '<button type="button" data-decision="reject">Reject</button>')

TEMPLATES: dict[str, Callable[..., str]] = {
    "IssueCard": _issue_card, "CitationList": _citations, "DuplicateCandidates": _duplicates,
    "ConfigDiff": _config_diff, "ApprovalRequest": _approval}

# script_id: day_080_generative_ui/components
def render_html(name: str, props: dict) -> str:
    """Validate again (never trust the wire), then fill a fixed template. Failures render as text."""
    model, template = COMPONENTS.get(name), TEMPLATES.get(name)
    try:
        if model is None or template is None:
            raise KeyError(name)
        body = template(model.model_validate(_cap(props)))
    except (KeyError, ValidationError):
        return f'<div class="atlas-fallback">{esc(f"[{name}: this card cannot be displayed]")}</div>'
    return f'<div class="atlas-card" data-component="{esc(name)}">{body}</div>'

if __name__ == "__main__":
    tool = component_tools()[2]
    print(tool["name"], tool["strict"], sorted(tool["parameters"]["properties"]))
    args = json.dumps({"issue_id": 161, "candidates": [
        {"issue_id": 143, "title": "3.1.0: reconnect storm after Redis failover",
         "confidence": "high", "reason": "Same ORB-1001 flood after failover"}]})
    event = to_ui_event("show_duplicate_candidates", args, seq=1)
    print(event.type, event.data["name"])
    print(render_html(event.data["name"], event.data["props"])[:160], "...")
    print(to_ui_event("show_issue_card", '{"issue_id": 1, "labels": "bug"}').data["text"])

show_duplicate_candidates True ['candidates', 'issue_id']
component DuplicateCandidates
<div class="atlas-card" data-component="DuplicateCandidates"><h3>Possible duplicates of <a href="https://issues.orbit.example/161" rel="noopener noreferrer nofo ...
(Couldn't draw the IssueCard card, showing plain text.)
- issue_id: 1
- labels: bug

# script_id: day_080_generative_ui/test_renderer
# Install: uv add pytest pydantic "openai>=3.46"   Run: uv run pytest test_renderer.py -q
import json

import pytest

from components import render_html, to_ui_event

HOSTILE = "<img src=x onerror=alert(1)>"
CARD = {"issue_id": 999, "title": HOSTILE, "state": "open", "labels": ["type:bug"],
        "url": "https://issues.orbit.example/999"}

def test_strings_are_escaped() -> None:
    page = render_html("IssueCard", CARD)
    assert "<img" not in page and "&lt;img src=x onerror=alert(1)&gt;" in page

def test_javascript_url_is_not_a_link() -> None:
    page = render_html("IssueCard", {**CARD, "url": "javascript:alert(document.cookie)"})
    assert "javascript:" not in page

def test_unknown_component_falls_back() -> None:
    assert "atlas-fallback" in render_html("ShellCommand", {"cmd": "rm -rf /"})
    assert to_ui_event("show_shell_command", "{}").type == "token"

def test_bad_types_fall_back() -> None:
    event = to_ui_event("show_issue_card", json.dumps({**CARD, "labels": "bug"}))
    assert event.type == "token" and event.data["fallback"] is True

def test_oversize_props_are_capped() -> None:
    rows = [{"issue_id": i, "title": "x" * 5000, "confidence": "low", "reason": "r"} for i in range(50)]
    props = to_ui_event("show_duplicate_candidates", json.dumps({"issue_id": 1, "candidates": rows})).data["props"]
    assert len(props["candidates"]) == 10 and len(props["candidates"][0]["title"]) == 801

def test_valid_card_renders() -> None:
    page = render_html("IssueCard", {**CARD, "title": "Reconnect storm"})
    assert '<a href="https://issues.orbit.example/999"' in page and "Reconnect storm" in page

if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-q"]))

......                                                                   [100%]
6 passed in 0.05s

# script_id: day_080_generative_ui/gradio_view
# Install: uv add gradio pydantic "openai>=3.46"   Env: API_URL (default http://127.0.0.1:8000)
# Needs: stream_app.py running in another terminal, components.py next to this file.
import html
import json
import os
import urllib.request

import gradio as gr

from components import render_html

API_URL = os.environ.get("API_URL", "http://127.0.0.1:8000")

def post(path: str, body: dict) -> urllib.request.Request:
    return urllib.request.Request(API_URL + path, data=json.dumps(body).encode(),
                                  headers={"Content-Type": "application/json"})

def iter_sse(path: str, body: dict):
    """Compact Day 79 SSE reader: yields (event, data) pairs."""
    with urllib.request.urlopen(post(path, body), timeout=120) as resp:
        event, data = "message", []
        for raw in resp:
            line = raw.decode("utf-8").rstrip("\r\n")
            if line.startswith("event:"):
                event = line[6:].strip()
            elif line.startswith("data:"):
                data.append(line[5:].strip())
            elif not line and data:
                yield event, json.loads("\n".join(data))
                event, data = "message", []

# script_id: day_080_generative_ui/gradio_view
def investigate(issue_id: float):
    text, cards, pending = "", [], None
    for event, data in iter_sse("/v1/investigate/stream", {"issue_id": int(issue_id)}):
        if event == "token":
            text += data["text"]
        elif event == "component":
            cards.append(render_html(data["name"], data["props"]))   # re-validates and escapes
            pending = data["props"] if data["name"] == "ApprovalRequest" else pending
        page = "".join(cards) + f'<p style="white-space: pre-wrap">{html.escape(text)}</p>'
        yield page, pending, gr.update(visible=pending is not None)

def decide(decision: str, pending: dict | None) -> str:
    if not pending:
        return "Nothing to approve."
    body = {"decision": decision, "action": pending["action"], "args": pending["args"]}
    try:
        urllib.request.urlopen(post("/v1/approvals", body), timeout=10)
    except OSError as exc:
        return f"Couldn't reach the approvals API: {exc}"
    return f"Recorded: {decision} {pending['action']} (the API executes it, not this page)."

with gr.Blocks(title="Atlas: generative UI") as demo:
    issue = gr.Number(value=161, label="Issue #", precision=0)
    go, view, pending = gr.Button("Investigate"), gr.HTML(), gr.State(None)
    with gr.Row(visible=False) as approval_row:
        approve, reject = gr.Button("Approve", variant="primary"), gr.Button("Reject")
    status = gr.Markdown()
    go.click(investigate, inputs=issue, outputs=[view, pending, approval_row])
    approve.click(lambda p: decide("approve", p), inputs=pending, outputs=status)
    reject.click(lambda p: decide("reject", p), inputs=pending, outputs=status)

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