A Coding Deep Dive into Agentic UI, Generative UI, State Synchronization, and Interrupt-Driven Approval Flows

import subprocess, sys
for pkg in ["openai", "rich", "pydantic"]:
   subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", pkg])


import os, getpass


if os.environ.get("OPENAI_API_KEY"):
   API_KEY = os.environ["OPENAI_API_KEY"]
   print("✅ Using OPENAI_API_KEY from environment.")
else:
   try:
       from google.colab import userdata
       API_KEY = userdata.get("OPENAI_API_KEY")
       print("✅ Using OPENAI_API_KEY from Colab Secrets.")
   except Exception:
       API_KEY = getpass.getpass("🔑 Enter your OpenAI API key (hidden): ")
       print("✅ API key received.")


BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1")
MODEL    = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")


import json, re, time, uuid, copy, textwrap
from enum import Enum
from dataclasses import dataclass, field, asdict
from typing import Any, Optional, Generator
from pydantic import BaseModel, Field


from openai import OpenAI
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich.tree import Tree
from rich.text import Text
from rich.markdown import Markdown
from rich import box


console = Console(width=105)
client  = OpenAI(api_key=API_KEY, base_url=BASE_URL)


def llm(messages, **kw):
   try:
       return client.chat.completions.create(model=MODEL, messages=messages, temperature=0.2, **kw)
   except Exception as e:
       console.print(f"[red]LLM error: {e}[/]")
       return None


def hdr(n, title, sub=""):
   console.print()
   console.rule(f"[bold cyan]SECTION {n}", style="cyan")
   body = f"[bold white]{title}[/]\n[dim]{sub}[/]" if sub else f"[bold white]{title}[/]"
   console.print(Panel(body, border_style="cyan", padding=(1, 2)))




hdr(1, "AG-UI Protocol — Event System",
   "The real AG-UI protocol uses ~16 event types streamed via SSE.\n"
   "We implement all core event types and a streaming emitter in pure Python.")


class AGUIEventType(str, Enum):
   RUN_STARTED          = "RUN_STARTED"
   RUN_FINISHED         = "RUN_FINISHED"
   RUN_ERROR            = "RUN_ERROR"
   TEXT_MESSAGE_START   = "TEXT_MESSAGE_START"
   TEXT_MESSAGE_CONTENT = "TEXT_MESSAGE_CONTENT"
   TEXT_MESSAGE_END     = "TEXT_MESSAGE_END"
   TOOL_CALL_START      = "TOOL_CALL_START"
   TOOL_CALL_ARGS       = "TOOL_CALL_ARGS"
   TOOL_CALL_RESULT     = "TOOL_CALL_RESULT"
   TOOL_CALL_END        = "TOOL_CALL_END"
   STATE_SNAPSHOT       = "STATE_SNAPSHOT"
   STATE_DELTA          = "STATE_DELTA"
   INTERRUPT            = "INTERRUPT"
   CUSTOM               = "CUSTOM"
   STEP_STARTED         = "STEP_STARTED"
   STEP_FINISHED        = "STEP_FINISHED"




@dataclass
class AGUIEvent:
   type: AGUIEventType
   data: dict = field(default_factory=dict)
   event_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
   timestamp: float = field(default_factory=time.time)


   def to_sse(self) -> str:
       payload = {"type": self.type.value, "id": self.event_id, **self.data}
       return f"event: ag-ui\ndata: {json.dumps(payload)}\n\n"


   def to_json(self) -> dict:
       return {"type": self.type.value, "id": self.event_id, "ts": self.timestamp, **self.data}




class AGUIEventStream:
   def __init__(self):
       self.events: list[AGUIEvent] = []
       self.listeners: list = []


   def emit(self, event: AGUIEvent):
       self.events.append(event)
       for listener in self.listeners:
           listener(event)


   def on(self, callback):
       self.listeners.append(callback)


   def replay(self) -> list[dict]:
       return [e.to_json() for e in self.events]




def demo_agui_lifecycle():
   stream = AGUIEventStream()


   event_colors = {
       "RUN_": "bold green", "TEXT_": "cyan", "TOOL_": "magenta",
       "STATE_": "yellow", "INTERRUPT": "bold red", "STEP_": "dim",
   }


   def frontend_listener(event: AGUIEvent):
       color = "white"
       for prefix, c in event_colors.items():
           if event.type.value.startswith(prefix):
               color = c
               break
       detail = json.dumps(event.data)[:80] if event.data else ""
       console.print(f"  [{color}]⚡ {event.type.value:.<28}[/] {detail}")


   stream.on(frontend_listener)


   run_id = str(uuid.uuid4())[:8]
   console.print("[bold]Simulating full AG-UI agent run...[/]\n")


   stream.emit(AGUIEvent(AGUIEventType.RUN_STARTED, {"run_id": run_id}))


   stream.emit(AGUIEvent(AGUIEventType.STEP_STARTED, {"step": "analyzing_query", "label": "Understanding request"}))
   stream.emit(AGUIEvent(AGUIEventType.STEP_FINISHED, {"step": "analyzing_query"}))


   msg_id = str(uuid.uuid4())[:8]
   stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_START, {"message_id": msg_id, "role": "assistant"}))
   for chunk in ["I'll ", "look up ", "the data ", "and build ", "a dashboard ", "for you."]:
       stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_CONTENT, {"message_id": msg_id, "delta": chunk}))
   stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_END, {"message_id": msg_id}))


   tool_id = str(uuid.uuid4())[:8]
   stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_START, {"tool_call_id": tool_id, "name": "query_database"}))
   stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_ARGS, {"tool_call_id": tool_id, "args_delta": '{"query": "SELECT revenue FROM sales"}'}))
   stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_RESULT, {"tool_call_id": tool_id, "result": [{"month": "Jan", "revenue": 42000}, {"month": "Feb", "revenue": 58000}]}))
   stream.emit(AGUIEvent(AGUIEventType.TOOL_CALL_END, {"tool_call_id": tool_id}))


   stream.emit(AGUIEvent(AGUIEventType.STATE_SNAPSHOT, {
       "state": {"active_agent": "DataAnalyst", "stage": "rendering", "progress": 0.75}
   }))
   stream.emit(AGUIEvent(AGUIEventType.STATE_DELTA, {
       "delta": [{"op": "replace", "path": "/progress", "value": 1.0}]
   }))


   stream.emit(AGUIEvent(AGUIEventType.INTERRUPT, {
       "reason": "high_risk_action",
       "description": "Agent wants to send an email to all 5,000 customers.",
       "options": ["approve", "reject", "modify"],
   }))


   stream.emit(AGUIEvent(AGUIEventType.RUN_FINISHED, {"run_id": run_id, "status": "completed"}))


   console.print(Panel(
       stream.events[3].to_sse(),
       title="[bold]Example SSE wire format (how it looks on the network)",
       border_style="dim",
   ))


   table = Table(title="AG-UI Event Stream Summary", box=box.ROUNDED)
   table.add_column("Category", style="cyan", width=15)
   table.add_column("Events", justify="center", style="green")
   counts = {}
   for e in stream.events:
       cat = e.type.value.rsplit("_", 1)[0] if "_" in e.type.value else e.type.value
       counts[cat] = counts.get(cat, 0) + 1
   for cat, n in counts.items():
       table.add_row(cat, str(n))
   table.add_row("[bold]TOTAL", f"[bold]{len(stream.events)}")
   console.print(table)


demo_agui_lifecycle()

hdr(2, "A2UI — Declarative Component Trees",
   "Google's A2UI spec: agents emit flat JSON component lists with ID refs.\n"
   "The client's widget registry maps types → native widgets.\n"
   "Safe like data, expressive like code. No executable code sent.")


class A2UIMessageType(str, Enum):
   CREATE_SURFACE      = "createSurface"
   UPDATE_COMPONENTS   = "updateComponents"
   UPDATE_DATA_MODEL   = "updateDataModel"
   DELETE_SURFACE       = "deleteSurface"




@dataclass
class A2UIComponent:
   id: str
   type: str
   properties: dict = field(default_factory=dict)
   children: list[str] = field(default_factory=list)


   def to_dict(self) -> dict:
       d = {"id": self.id, "type": self.type, **self.properties}
       if self.children:
           d["children"] = self.children
       return d




@dataclass
class A2UIDataModel:
   data: dict = field(default_factory=dict)


   def get_binding(self, path: str) -> Any:
       parts = [p for p in path.split("/") if p]
       val = self.data
       for p in parts:
           if isinstance(val, dict):
               val = val.get(p)
           elif isinstance(val, list) and p.isdigit():
               val = val[int(p)]
           else:
               return None
       return val




@dataclass
class A2UISurface:
   surface_id: str
   components: list[A2UIComponent] = field(default_factory=list)
   data_model: A2UIDataModel = field(default_factory=A2UIDataModel)


   def to_messages(self) -> list[dict]:
       msgs = []
       msgs.append({
           "type": A2UIMessageType.CREATE_SURFACE.value,
           "surfaceId": self.surface_id,
       })
       msgs.append({
           "type": A2UIMessageType.UPDATE_COMPONENTS.value,
           "surfaceId": self.surface_id,
           "components": [c.to_dict() for c in self.components],
       })
       if self.data_model.data:
           msgs.append({
               "type": A2UIMessageType.UPDATE_DATA_MODEL.value,
               "surfaceId": self.surface_id,
               "dataModel": self.data_model.data,
           })
       return msgs




class WidgetRegistry:
   def __init__(self):
       self._renderers = {}


   def register(self, component_type: str, render_fn):
       self._renderers[component_type] = render_fn


   def render(self, component: A2UIComponent, surface: A2UISurface, indent: int = 0):
       fn = self._renderers.get(component.type)
       if fn:
           fn(component, surface, indent)
       else:
           pad = "  " * indent
           console.print(f"{pad}[dim]⟨{component.type} id={component.id}⟩ (no renderer)[/]")


   def render_tree(self, surface: A2UISurface):
       comp_map = {c.id: c for c in surface.components}
       all_children = set()
       for c in surface.components:
           all_children.update(c.children)
       roots = [c for c in surface.components if c.id not in all_children]


       def _render(comp_id: str, indent: int):
           comp = comp_map.get(comp_id)
           if not comp:
               return
           self.render(comp, surface, indent)
           for child_id in comp.children:
               _render(child_id, indent + 1)


       for root in roots:
           _render(root.id, 0)




registry = WidgetRegistry()


def _resolve(comp, surface, key, default=None):
   val = comp.properties.get(key, default)
   binding = comp.properties.get("dataBinding")
   if binding and isinstance(binding, str) and binding.startswith("/"):
       resolved = surface.data_model.get_binding(binding)
       if resolved is not None:
           return resolved
   if isinstance(val, str) and val.startswith("/") and "/" in val[1:]:
       resolved = surface.data_model.get_binding(val)
       if resolved is not None:
           return resolved
   return val


def _to_float(val, default=0.0):
   if isinstance(val, (int, float)):
       return float(val)
   if isinstance(val, str):
       cleaned = val.strip().rstrip("%")
       try:
           f = float(cleaned)
           if "%" in val or f > 1:
               return f / 100.0
           return f
       except ValueError:
           return default
   return default


def render_card(comp, surface, indent):
   pad = "  " * indent
   title = str(_resolve(comp, surface, "title", "Card"))
   console.print(f"{pad}┌─{'─' * 50}─┐")
   console.print(f"{pad}│ [bold]{title:^50}[/] │")
   console.print(f"{pad}├─{'─' * 50}─┤")
   if not comp.children:
       subtitle = str(_resolve(comp, surface, "subtitle", ""))
       if subtitle:
           console.print(f"{pad}│  {subtitle:<49}│")
       console.print(f"{pad}└─{'─' * 50}─┘")


def render_text(comp, surface, indent):
   pad = "  " * indent
   text = _resolve(comp, surface, "text", "")
   style = comp.properties.get("style", "body")
   styles = {"headline": "bold white", "body": "white", "caption": "dim", "label": "bold cyan"}
   console.print(f"{pad}[{styles.get(style, 'white')}]{text}[/]")


def render_button(comp, surface, indent):
   pad = "  " * indent
   label = str(_resolve(comp, surface, "label", "Button"))
   variant = comp.properties.get("variant", "primary")
   colors = {"primary": "bold white on blue", "secondary": "white on grey30", "danger": "bold white on red"}
   console.print(f"{pad}  [{colors.get(variant, 'white')}]  {label}  [/]")


def render_text_field(comp, surface, indent):
   pad = "  " * indent
   label = comp.properties.get("label", "Input")
   placeholder = comp.properties.get("placeholder", "")
   console.print(f"{pad}  {label}: [dim]┌──────────────────────────┐[/]")
   console.print(f"{pad}          [dim]│ {placeholder:<25}│[/]")
   console.print(f"{pad}          [dim]└──────────────────────────┘[/]")


def render_row(comp, surface, indent):
   pass


def render_column(comp, surface, indent):
   pass


def render_image(comp, surface, indent):
   pad = "  " * indent
   alt = comp.properties.get("alt", "image")
   console.print(f"{pad}  [dim]🖼  [{alt}][/]")


def render_divider(comp, surface, indent):
   pad = "  " * indent
   console.print(f"{pad}  {'─' * 50}")


def render_chip(comp, surface, indent):
   pad = "  " * indent
   label = str(_resolve(comp, surface, "label", ""))
   console.print(f"{pad}  [on grey23] {label} [/]")


def render_progress(comp, surface, indent):
   pad = "  " * indent
   raw_value = _resolve(comp, surface, "value", 0)
   value = max(0.0, min(1.0, _to_float(raw_value, 0.0)))
   label = str(_resolve(comp, surface, "label", ""))
   bar_len = int(value * 40)
   bar = f"[green]{'█' * bar_len}[/][dim]{'░' * (40 - bar_len)}[/]"
   console.print(f"{pad}  {label}: {bar} {value*100:.0f}%")


for name, fn in [
   ("card", render_card), ("text", render_text), ("button", render_button),
   ("text-field", render_text_field), ("row", render_row), ("column", render_column),
   ("image", render_image), ("divider", render_divider), ("chip", render_chip),
   ("progress-bar", render_progress),
]:
   registry.register(name, fn)




console.print("\n[bold]Demo: A2UI booking form — agent generates a restaurant reservation UI[/]\n")


booking_surface = A2UISurface(
   surface_id="booking-form-1",
   components=[
       A2UIComponent("root", "card", {"title": "🍽️  Reserve a Table"}, children=["c1", "c2", "c3", "c4", "c5", "c6"]),
       A2UIComponent("c1", "text", {"text": "", "dataBinding": "/restaurant/name", "style": "headline"}),
       A2UIComponent("c2", "text", {"text": "", "dataBinding": "/restaurant/cuisine", "style": "caption"}),
       A2UIComponent("c3", "divider", {}),
       A2UIComponent("c4", "text-field", {"label": "Date", "placeholder": "YYYY-MM-DD"}),
       A2UIComponent("c5", "text-field", {"label": "Guests", "placeholder": "1-12"}),
       A2UIComponent("c6", "button", {"label": "Reserve Now", "variant": "primary", "action": "submit_booking"}),
   ],
   data_model=A2UIDataModel({"restaurant": {"name": "Chez Laurent", "cuisine": "French Contemporary • $$$$"}})
)


console.print(Panel(
   "\n".join(json.dumps(m, indent=2)[:200] for m in booking_surface.to_messages()),
   title="[bold]A2UI JSONL stream (what goes over the wire)",
   border_style="yellow",
))


console.print("[bold]Rendered by client widget registry:[/]\n")
registry.render_tree(booking_surface)


console.print()
t = Table(title="A2UI Flat Component List (Adjacency Model)", box=box.ROUNDED)
t.add_column("ID", style="cyan", width=8)
t.add_column("Type", style="green", width=14)
t.add_column("Children", style="yellow", width=20)
t.add_column("Bindings", style="magenta", width=25)
for c in booking_surface.components:
   binding = c.properties.get("dataBinding", "")
   t.add_row(c.id, c.type, ", ".join(c.children) if c.children else "—", binding or "—")
console.print(t)

hdr(3, "Generative UI — LLM Produces Live Interfaces",
   "The agent generates A2UI component trees dynamically based on the query.\n"
   "This is the core of 'Generative UI' — context-adaptive interfaces\n"
   "that go far beyond text-only chat responses.")


A2UI_GENERATION_PROMPT = """\
You are an A2UI Generative UI agent. Given a user query, you generate a rich \
interactive interface — NOT text. You output an A2UI component tree as JSON.


RULES:
1. Output a flat list of components using the adjacency model (children = list of IDs).
2. Available component types: card, text, button, text-field, row, column, divider, chip, image, progress-bar, select, date-picker, data-table
3. Include a separate "dataModel" object for dynamic values. Use "/path/to/value" bindings.
4. The ROOT component should be a "card" with all others as descendants.
5. Think about what UI BEST serves the user — forms for input, tables for data, \
  progress bars for status, chips for tags, buttons for actions.


OUTPUT FORMAT (strict JSON, nothing else):
{
 "surfaceId": "unique-id",
 "components": [
   {"id": "root", "type": "card", "title": "...", "children": ["c1", "c2"]},
   {"id": "c1", "type": "text", "text": "...", "style": "headline"},
   ...
 ],
 "dataModel": { ... }
}
"""




def generate_ui(user_query: str) -> Optional[A2UISurface]:
   console.print(f"  [dim]Generating UI for:[/] [bold]{user_query}[/]")
   response = llm([
       {"role": "system", "content": A2UI_GENERATION_PROMPT},
       {"role": "user", "content": user_query},
   ], max_tokens=1200)


   if not response:
       return None


   raw = response.choices[0].message.content
   try:
       cleaned = re.sub(r'```json\s*|\s*```', '', raw).strip()
       spec = json.loads(cleaned)
   except json.JSONDecodeError:
       console.print(f"[red]Failed to parse generated UI: {raw[:200]}[/]")
       return None


   components = []
   for c in spec.get("components", []):
       components.append(A2UIComponent(
           id=c.get("id", str(uuid.uuid4())[:6]),
           type=c.get("type", "text"),
           properties={k: v for k, v in c.items() if k not in ("id", "type", "children")},
           children=c.get("children", []),
       ))


   surface = A2UISurface(
       surface_id=spec.get("surfaceId", f"gen-{uuid.uuid4().hex[:6]}"),
       components=components,
       data_model=A2UIDataModel(spec.get("dataModel", {})),
   )


   return surface




def demo_generative_ui(query: str):
   surface = generate_ui(query)
   if surface:
       console.print(f"\n[bold green]Generated {len(surface.components)} components:[/]")
       registry.render_tree(surface)


       console.print()
       types = {}
       for c in surface.components:
           types[c.type] = types.get(c.type, 0) + 1
       console.print("  [dim]Component types used:[/] " + ", ".join(f"[cyan]{t}[/]×{n}" for t, n in types.items()))


       if surface.data_model.data:
           console.print(f"  [dim]Data model keys:[/] {list(surface.data_model.data.keys())}")
       console.print()




console.print("\n[bold]Demo 1: Agent generates an onboarding form[/]")
demo_generative_ui(
   "Create a user onboarding flow: collect name, email, role (dropdown), "
   "preferred notification method (chips), and a 'Get Started' button."
)


console.print("\n[bold]Demo 2: Agent generates a data dashboard[/]")
demo_generative_ui(
   "Show a project status dashboard with: project name 'Atlas v2', "
   "4 team members, sprint progress at 68%, 3 blockers flagged as critical, "
   "and action buttons for 'View Backlog' and 'Schedule Standup'."
)


console.print("\n[bold]Demo 3: Agent generates a confirmation dialog[/]")
demo_generative_ui(
   "Show a payment confirmation: $2,450 charge to Visa ending 4242, "
   "order #ORD-8891, with Approve and Decline buttons."
)

hdr(4, "State Synchronization — Shared State Between Agent & UI",
   "AG-UI syncs state bidirectionally using STATE_SNAPSHOT and STATE_DELTA.\n"
   "The agent IS the state machine; the UI IS the renderer.\n"
   "JSON Patch diffs keep updates minimal and efficient.")




class SharedState:
   def __init__(self, initial: dict = None):
       self.state: dict = initial or {}
       self.history: list[dict] = []
       self.version: int = 0


   def snapshot(self) -> AGUIEvent:
       return AGUIEvent(AGUIEventType.STATE_SNAPSHOT, {"state": copy.deepcopy(self.state), "version": self.version})


   def apply_delta(self, operations: list[dict]) -> AGUIEvent:
       for op in operations:
           path_parts = [p for p in op["path"].split("/") if p]
           target = self.state
           for part in path_parts[:-1]:
               if isinstance(target, dict):
                   target = target.setdefault(part, {})
               elif isinstance(target, list) and part.isdigit():
                   target = target[int(part)]


           key = path_parts[-1] if path_parts else None
           if key is None:
               continue


           if op["op"] == "replace":
               target[key] = op["value"]
           elif op["op"] == "add":
               if isinstance(target, list) and key.isdigit():
                   target.insert(int(key), op["value"])
               else:
                   target[key] = op["value"]
           elif op["op"] == "remove":
               if isinstance(target, dict):
                   target.pop(key, None)


       self.version += 1
       self.history.append({"version": self.version, "ops": operations})
       return AGUIEvent(AGUIEventType.STATE_DELTA, {"delta": operations, "version": self.version})




console.print("\n[bold]Demo: Document review pipeline — 3 agents, shared state[/]\n")


stream = AGUIEventStream()
state = SharedState({
   "document": {"title": "Q4 Strategy Report", "status": "draft", "word_count": 2840},
   "pipeline": {"stage": "research", "progress": 0.0},
   "agents": {"active": "Researcher", "queue": ["Editor", "Reviewer"]},
   "feedback": [],
})


def log_event(event: AGUIEvent):
   if event.type in (AGUIEventType.STATE_SNAPSHOT, AGUIEventType.STATE_DELTA):
       if event.type == AGUIEventType.STATE_DELTA:
           ops = event.data.get("delta", [])
           for op in ops:
               console.print(f"  [yellow]STATE_DELTA[/] v{event.data.get('version')}: "
                             f"[cyan]{op['op']}[/] {op['path']} → {op.get('value', '∅')}")
       else:
           console.print(f"  [yellow]STATE_SNAPSHOT[/] v{event.data.get('version')}: {list(event.data['state'].keys())}")


stream.on(log_event)


stream.emit(state.snapshot())


console.print("\n[bold green]▸ Researcher agent working...[/]")
stream.emit(state.apply_delta([
   {"op": "replace", "path": "/pipeline/stage", "value": "research_complete"},
   {"op": "replace", "path": "/pipeline/progress", "value": 0.33},
   {"op": "add",     "path": "/feedback/0", "value": {"agent": "Researcher", "note": "Added 4 new data sources"}},
]))


console.print("\n[bold green]▸ Editor agent working...[/]")
stream.emit(state.apply_delta([
   {"op": "replace", "path": "/agents/active", "value": "Editor"},
   {"op": "replace", "path": "/pipeline/stage", "value": "editing"},
   {"op": "replace", "path": "/pipeline/progress", "value": 0.66},
   {"op": "replace", "path": "/document/word_count", "value": 3150},
]))


console.print("\n[bold green]▸ Reviewer agent working...[/]")
stream.emit(state.apply_delta([
   {"op": "replace", "path": "/agents/active", "value": "Reviewer"},
   {"op": "replace", "path": "/pipeline/stage", "value": "review_complete"},
   {"op": "replace", "path": "/pipeline/progress", "value": 1.0},
   {"op": "replace", "path": "/document/status", "value": "approved"},
]))


console.print(Panel(
   json.dumps(state.state, indent=2),
   title="[bold]Final shared state after pipeline",
   border_style="green",
))


t = Table(title="State History (versions)", box=box.ROUNDED)
t.add_column("Version", style="cyan", justify="center")
t.add_column("Operations", style="yellow")
for h in state.history:
   ops_summary = "; ".join(f"{o['op']} {o['path']}" for o in h["ops"])
   t.add_row(str(h["version"]), ops_summary[:70])
console.print(t)




hdr(5, "Human-in-the-Loop — AG-UI INTERRUPT Events",
   "When an agent hits a high-stakes action, it emits an INTERRUPT event.\n"
   "The frontend renders an approval UI. Execution pauses until the human\n"
   "approves, rejects, or modifies. State is preserved throughout.")




@dataclass
class InterruptRequest:
   interrupt_id: str
   action_description: str
   risk_level: str
   affected_resources: list[str]
   proposed_changes: dict
   options: list[str] = field(default_factory=lambda: ["approve", "reject", "modify"])




@dataclass
class InterruptResponse:
   interrupt_id: str
   decision: str
   modifications: Optional[dict] = None




class InterruptableAgent:
   RISK_RULES = {
       "delete":     "critical",
       "payment":    "critical",
       "email_all":  "high",
       "publish":    "high",
       "update":     "medium",
       "read":       "low",
   }


   def __init__(self):
       self.stream = AGUIEventStream()
       self.pending_interrupts: dict[str, InterruptRequest] = {}


   def assess_and_maybe_interrupt(self, action: str, details: dict) -> Optional[InterruptRequest]:
       risk = "low"
       for keyword, level in self.RISK_RULES.items():
           if keyword in action.lower():
               risk = level
               break


       if risk in ("critical", "high"):
           interrupt = InterruptRequest(
               interrupt_id=str(uuid.uuid4())[:8],
               action_description=action,
               risk_level=risk,
               affected_resources=details.get("resources", []),
               proposed_changes=details.get("changes", {}),
           )
           self.pending_interrupts[interrupt.interrupt_id] = interrupt


           self.stream.emit(AGUIEvent(AGUIEventType.INTERRUPT, {
               "interrupt_id": interrupt.interrupt_id,
               "reason": risk,
               "description": interrupt.action_description,
               "affected_resources": interrupt.affected_resources,
               "proposed_changes": interrupt.proposed_changes,
               "options": interrupt.options,
           }))
           return interrupt
       return None


   def resolve_interrupt(self, response: InterruptResponse) -> str:
       interrupt = self.pending_interrupts.pop(response.interrupt_id, None)
       if not interrupt:
           return "No pending interrupt found."


       if response.decision == "approve":
           return f"✅ APPROVED: '{interrupt.action_description}' executing now."
       elif response.decision == "reject":
           return f"🚫 REJECTED: '{interrupt.action_description}' cancelled."
       elif response.decision == "modify":
           return f"✏️  MODIFIED: '{interrupt.action_description}' updated with: {response.modifications}"
       return f"Unknown decision: {response.decision}"




console.print("\n[bold]Demo: Agent encounters actions of varying risk levels[/]\n")
agent = InterruptableAgent()


actions = [
   ("Read user profile",      {"resources": ["user:123"]}),
   ("Update user preferences", {"resources": ["user:123"], "changes": {"theme": "dark"}}),
   ("Delete user account",     {"resources": ["user:123", "data:all"], "changes": {"action": "permanent_delete"}}),
   ("Email all 12,000 users",  {"resources": ["email:newsletter"], "changes": {"subject": "Big Announcement"}}),
   ("Publish blog post",       {"resources": ["post:draft-42"], "changes": {"status": "public"}}),
]


def event_logger(event):
   if event.type == AGUIEventType.INTERRUPT:
       d = event.data
       risk_style = {"critical": "bold red", "high": "bold yellow"}.get(d["reason"], "white")
       console.print(f"\n  [bold]🚨 INTERRUPT EVENT[/]")
       console.print(f"    Risk: [{risk_style}]{d['reason'].upper()}[/]")
       console.print(f"    Action: {d['description']}")
       console.print(f"    Affected: {d['affected_resources']}")
       console.print(f"    Options: {d['options']}")


agent.stream.on(event_logger)


results = []
for action_desc, details in actions:
   interrupt = agent.assess_and_maybe_interrupt(action_desc, details)
   if interrupt:
       decision = "reject" if interrupt.risk_level == "critical" else "approve"
       result = agent.resolve_interrupt(InterruptResponse(interrupt.interrupt_id, decision))
   else:
       result = f"⚡ AUTO-EXECUTED: '{action_desc}' (low risk, no approval needed)"


   results.append((action_desc, result))


console.print()
t = Table(title="Execution Results", box=box.ROUNDED, show_lines=True)
t.add_column("Action", style="white", width=28)
t.add_column("Outcome", style="dim", width=55)
for action_desc, result in results:
   t.add_row(action_desc, result)
console.print(t)

hdr(6, "Full Pipeline — LLM-Driven Adaptive UI",
   "The complete Agentic UI architecture in one pipeline:\n"
   "  User query → Intent analysis → UI pattern selection →\n"
   "  A2UI generation → AG-UI event streaming → State sync → Render")




UI_ROUTER_PROMPT = """\
You are a UI routing agent. Given a user query, decide what type of UI to generate.


RESPOND IN JSON ONLY:
{
 "intent": "form | dashboard | confirmation | list | detail | wizard | error",
 "reasoning": "why this UI pattern fits",
 "ui_complexity": "simple | moderate | complex",
 "needs_approval": true/false,
 "data_requirements": ["what data the UI needs"]
}
"""




class AgenticUIPipeline:
   def __init__(self):
       self.stream = AGUIEventStream()
       self.state = SharedState({"pipeline": {"stage": "idle"}, "renders": 0})
       self.interrupt_agent = InterruptableAgent()


   def route(self, query: str) -> dict:
       resp = llm([
           {"role": "system", "content": UI_ROUTER_PROMPT},
           {"role": "user", "content": query},
       ], max_tokens=300)


       if not resp:
           return {"intent": "dashboard", "reasoning": "fallback", "ui_complexity": "simple",
                   "needs_approval": False, "data_requirements": []}
       raw = resp.choices[0].message.content
       try:
           return json.loads(re.sub(r'```json\s*|\s*```', '', raw).strip())
       except json.JSONDecodeError:
           return {"intent": "dashboard", "reasoning": "parse_fallback", "ui_complexity": "simple",
                   "needs_approval": False, "data_requirements": []}


   def run(self, user_query: str):
       run_id = str(uuid.uuid4())[:8]
       console.print(Panel(f"[bold]{user_query}[/]", title="🧑 User Query", border_style="white"))


       self.stream.emit(AGUIEvent(AGUIEventType.RUN_STARTED, {"run_id": run_id}))
       self.stream.emit(AGUIEvent(AGUIEventType.STEP_STARTED, {"step": "routing"}))


       routing = self.route(user_query)
       console.print(f"\n  [bold cyan]📡 Router Decision:[/]")
       console.print(f"    Intent: [green]{routing.get('intent')}[/] | "
                     f"Complexity: [yellow]{routing.get('ui_complexity')}[/] | "
                     f"Approval: {'🔒' if routing.get('needs_approval') else '✅'}")
       console.print(f"    Reasoning: [dim]{routing.get('reasoning', '')}[/]")


       self.stream.emit(AGUIEvent(AGUIEventType.STEP_FINISHED, {"step": "routing", "result": routing}))


       self.state.apply_delta([
           {"op": "replace", "path": "/pipeline/stage", "value": "generating"},
       ])


       self.stream.emit(AGUIEvent(AGUIEventType.STEP_STARTED, {"step": "generating_ui"}))


       msg_id = str(uuid.uuid4())[:8]
       self.stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_START, {"message_id": msg_id}))
       self.stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_CONTENT, {
           "message_id": msg_id,
           "delta": f"Building a {routing.get('intent')} interface for you..."
       }))
       self.stream.emit(AGUIEvent(AGUIEventType.TEXT_MESSAGE_END, {"message_id": msg_id}))


       surface = generate_ui(user_query)
       self.stream.emit(AGUIEvent(AGUIEventType.STEP_FINISHED, {"step": "generating_ui"}))


       if routing.get("needs_approval") and surface:
           self.stream.emit(AGUIEvent(AGUIEventType.INTERRUPT, {
               "reason": "ui_confirmation",
               "description": f"Generated {len(surface.components)} component UI. Render it?",
               "options": ["render", "regenerate", "cancel"],
           }))
           console.print("\n  [bold yellow]⏸  INTERRUPT:[/] UI generated, awaiting human approval...")
           console.print("  [green]→ Auto-approving for demo...[/]")


       if surface:
           self.state.apply_delta([
               {"op": "replace", "path": "/pipeline/stage", "value": "rendering"},
               {"op": "replace", "path": "/renders", "value": self.state.state.get("renders", 0) + 1},
           ])


           console.print(f"\n[bold green]🖥  Rendered Interface ({len(surface.components)} components):[/]\n")
           registry.render_tree(surface)


           self.stream.emit(AGUIEvent(AGUIEventType.CUSTOM, {
               "subtype": "a2ui_surface",
               "surface": surface.to_messages(),
           }))


       self.state.apply_delta([{"op": "replace", "path": "/pipeline/stage", "value": "complete"}])
       self.stream.emit(AGUIEvent(AGUIEventType.RUN_FINISHED, {"run_id": run_id, "status": "success"}))


       console.print()
       event_counts = {}
       for e in self.stream.events:
           event_counts[e.type.value] = event_counts.get(e.type.value, 0) + 1
       t = Table(title="Pipeline Event Summary", box=box.ROUNDED)
       t.add_column("Event Type", style="cyan")
       t.add_column("Count", justify="center", style="green")
       for etype, count in sorted(event_counts.items()):
           t.add_row(etype, str(count))
       console.print(t)




pipeline = AgenticUIPipeline()


console.print("\n[bold]Demo 1: Agent builds a settings form[/]")
pipeline.run(
   "Create a notification settings panel where I can toggle email/SMS/push, "
   "set quiet hours, and pick a notification sound."
)


pipeline.stream = AGUIEventStream()
pipeline.state = SharedState({"pipeline": {"stage": "idle"}, "renders": 0})


console.print("\n[bold]Demo 2: Agent builds an order tracking dashboard[/]")
pipeline.run(
   "Show order #ORD-7742 status: shipped via FedEx, tracking 789456123, "
   "estimated delivery March 24, 2 of 3 items delivered. Show a progress bar "
   "and action buttons for 'Contact Support' and 'Request Refund'."
)




hdr(7, "Incremental UI Updates — Live Surface Modification",
   "A2UI surfaces are incrementally updateable. The agent can add, remove,\n"
   "or modify components and data bindings on a live surface without\n"
   "regenerating the whole tree. Essential for real-time collaboration.")




class LiveSurface:
   def __init__(self, surface: A2UISurface):
       self.surface = surface
       self.update_log: list[str] = []


   def add_component(self, component: A2UIComponent, parent_id: Optional[str] = None):
       self.surface.components.append(component)
       if parent_id:
           for c in self.surface.components:
               if c.id == parent_id:
                   c.children.append(component.id)
                   break
       self.update_log.append(f"ADD  {component.type}#{component.id} → parent:{parent_id or 'root'}")


   def update_component(self, component_id: str, new_props: dict):
       for c in self.surface.components:
           if c.id == component_id:
               c.properties.update(new_props)
               self.update_log.append(f"UPD  #{component_id} props: {list(new_props.keys())}")
               return
       self.update_log.append(f"ERR  #{component_id} not found")


   def remove_component(self, component_id: str):
       self.surface.components = [c for c in self.surface.components if c.id != component_id]
       for c in self.surface.components:
           if component_id in c.children:
               c.children.remove(component_id)
       self.update_log.append(f"DEL  #{component_id}")


   def update_data(self, path: str, value: Any):
       self.surface.data_model.data = _set_nested(self.surface.data_model.data, path, value)
       self.update_log.append(f"DATA {path} = {value}")




def _set_nested(d: dict, path: str, value: Any) -> dict:
   parts = [p for p in path.split("/") if p]
   d = copy.deepcopy(d)
   current = d
   for p in parts[:-1]:
       current = current.setdefault(p, {})
   if parts:
       current[parts[-1]] = value
   return d




console.print("\n[bold]Demo: Live collaborative editing — agent modifies UI in real-time[/]\n")


initial = A2UISurface(
   surface_id="task-board",
   components=[
       A2UIComponent("board", "card", {"title": "🗂  Sprint Board"}, children=["t1", "t2", "t3"]),
       A2UIComponent("t1", "chip", {"label": "AUTH-101: Login flow", "variant": "in_progress"}),
       A2UIComponent("t2", "chip", {"label": "AUTH-102: OAuth setup", "variant": "todo"}),
       A2UIComponent("t3", "chip", {"label": "AUTH-103: 2FA", "variant": "todo"}),
   ],
   data_model=A2UIDataModel({"sprint": {"name": "Sprint 14", "velocity": 21}}),
)


live = LiveSurface(initial)


console.print("[bold]Initial board:[/]")
registry.render_tree(live.surface)


console.print("\n[bold yellow]Agent updating board in real-time...[/]\n")


live.update_component("t1", {"variant": "done", "label": "✅ AUTH-101: Login flow"})
live.update_component("t2", {"variant": "in_progress", "label": "🔄 AUTH-102: OAuth setup"})
live.add_component(
   A2UIComponent("t4", "chip", {"label": "AUTH-104: Password reset", "variant": "todo"}),
   parent_id="board"
)
live.update_data("/sprint/velocity", 25)
live.remove_component("t3")


console.print("[bold]Updated board:[/]")
registry.render_tree(live.surface)


console.print()
t = Table(title="Incremental Update Log", box=box.ROUNDED)
t.add_column("#", style="cyan", width=4, justify="center")
t.add_column("Operation", style="yellow")
for i, entry in enumerate(live.update_log, 1):
   t.add_row(str(i), entry)
console.print(t)

hdr(8, "Reference — The Agentic UI Protocol Stack",
   "How AG-UI, A2UI, MCP, and A2A fit together in the modern agent architecture.")


console.print(Panel("""\
[bold white]THE AGENTIC UI STACK (2026)[/]


 ┌──────────────────────────────────────────────────────┐
 │  [bold cyan]USER INTERFACE[/]  (React, Flutter, SwiftUI, Terminal)  │
 │  Renders native widgets from component specs          │
 └────────────────────────┬─────────────────────────────┘
                          │  A2UI component trees (JSON)
                          │  AG-UI events (SSE / WebSocket)
 ┌────────────────────────┴─────────────────────────────┐
 │  [bold yellow]AG-UI PROTOCOL[/]  (Agent ↔ User Interaction)          │
 │  • Event streaming (TEXT, TOOL_CALL, STATE, INTERRUPT)│
 │  • Bidirectional state sync (SNAPSHOT + DELTA)        │
 │  • Human-in-the-loop (INTERRUPT → approval flow)      │
 └────────────────────────┬─────────────────────────────┘
                          │
 ┌────────────────────────┴─────────────────────────────┐
 │  [bold magenta]AGENT RUNTIME[/]  (LangGraph, CrewAI, custom, etc.)    │
 │  • Generates A2UI surfaces (Generative UI)            │
 │  • Manages shared state                               │
 │  • Orchestrates sub-agents via A2A protocol           │
 │  • Accesses tools via MCP protocol                    │
 └────────────────────────┬─────────────────────────────┘
                          │
 ┌────────────────────────┴─────────────────────────────┐
 │  [bold green]LLM BACKBONE[/]  (GPT, Claude, Gemini, etc.)           │
 │  • Generates component trees as structured output     │
 │  • Reasons about UI patterns per context              │
 │  • Streams tokens for real-time rendering              │
 └──────────────────────────────────────────────────────┘


[dim]Protocol roles:
 AG-UI  = Agent ↔ User (streaming events, state, HITL)
 A2UI   = Agent → UI (declarative component specs)
 A2A    = Agent → Agent (delegation, sub-agents)
 MCP    = Agent → Tools (function calling, context)[/]
""", title="[bold]Architecture Reference", border_style="cyan"))




ref = Table(title="Agentic UI Concepts — Quick Reference", box=box.DOUBLE_EDGE, show_lines=True)
ref.add_column("Concept", style="bold cyan", width=22)
ref.add_column("What It Does", style="white", width=35)
ref.add_column("Key Mechanism", style="yellow", width=28)


ref.add_row("AG-UI Events",     "Stream agent actions to frontend in real-time",  "SSE/WebSocket + ~16 event types")
ref.add_row("A2UI Components",  "Declarative UI trees — safe, portable, native",  "Flat JSON + widget registry")
ref.add_row("State Sync",       "Keep agent & UI state in lockstep",              "STATE_SNAPSHOT + STATE_DELTA")
ref.add_row("Generative UI",    "LLM generates UI at runtime, not just text",     "A2UI JSON as structured output")
ref.add_row("INTERRUPT (HITL)", "Pause execution for human approval",             "INTERRUPT event → approval flow")
ref.add_row("Incremental Update","Modify live surfaces without full regeneration", "A2UI updateComponents message")
ref.add_row("Data Binding",     "UI reads from a shared data model",              "JSON Pointer paths (/path/to/val)")
ref.add_row("Widget Registry",  "Client maps abstract types to native widgets",   "Catalog of trusted components")


console.print(ref)




console.print(Panel(
   "[bold green]Tutorial complete![/]\n\n"
   "[dim]What you built:[/]\n"
   "  • A full AG-UI event system with all 16 event types\n"
   "  • An A2UI renderer with flat adjacency-list components + data binding\n"
   "  • LLM-powered Generative UI that creates interfaces from natural language\n"
   "  • Bidirectional state sync with JSON Patch deltas\n"
   "  • Human-in-the-loop interrupt and approval flows\n"
   "  • Incremental live surface updates\n\n"
   "[dim]To go further:[/]\n"
   "  • Serve AG-UI events over real SSE with FastAPI\n"
   "  • Connect to CopilotKit React components for a real frontend\n"
   "  • Use Pydantic AI's AGUIAdapter for production agent hosting\n"
   "  • Add A2A protocol for multi-agent delegation\n"
   "  • Deploy on AWS Bedrock AgentCore with native AG-UI support",
   title="[bold]🎓 What's Next?",
   border_style="green",
   padding=(1, 2),
))