"""UI Node: stateful renderer — manages UI state, handles local actions, renders controls.""" import json import logging from .base import Node from ..types import ThoughtResult log = logging.getLogger("runtime") class UINode(Node): name = "ui" # No model — pure code, no LLM calls def __init__(self, send_hud): super().__init__(send_hud) self.current_controls: list[dict] = [] self.state: dict = {} # {"count": 0, "theme": "dark", ...} self.bindings: dict = {} # {"increment": {"op": "inc", "var": "count"}, ...} self.machines: dict = {} # {"nav": {initial, states, current}, ...} # --- Machine operations --- async def apply_machine_ops(self, ops: list[dict]) -> None: """Apply machine operations from Thinker tool calls.""" for op_data in ops: op = op_data.get("op") mid = op_data.get("id", "") if op == "create": initial = op_data.get("initial", "") # Parse states from array format [{name, buttons, content}] states_list = op_data.get("states", []) states = {} for s in states_list: name = s.get("name", "") if name: states[name] = { "buttons": s.get("buttons", []), "content": s.get("content", []), } self.machines[mid] = { "initial": initial, "current": initial, "states": states, } log.info(f"[ui] machine created: {mid} (initial={initial}, {len(states)} states)") await self.hud("machine_created", id=mid, initial=initial, state_count=len(states)) elif op == "add_state": if mid not in self.machines: log.warning(f"[ui] add_state: machine '{mid}' not found") continue state_name = op_data.get("state", "") self.machines[mid]["states"][state_name] = { "buttons": op_data.get("buttons", []), "content": op_data.get("content", []), } log.info(f"[ui] state added: {mid}.{state_name}") await self.hud("machine_state_added", id=mid, state=state_name) elif op == "reset": if mid not in self.machines: log.warning(f"[ui] reset: machine '{mid}' not found") continue initial = self.machines[mid]["initial"] self.machines[mid]["current"] = initial log.info(f"[ui] machine reset: {mid} -> {initial}") await self.hud("machine_reset", id=mid, state=initial) elif op == "destroy": if mid in self.machines: del self.machines[mid] log.info(f"[ui] machine destroyed: {mid}") await self.hud("machine_destroyed", id=mid) def try_machine_transition(self, action: str) -> tuple[bool, str | None]: """Check if action triggers a machine transition. Returns (handled, result_text).""" for mid, machine in self.machines.items(): current = machine["current"] state_def = machine["states"].get(current, {}) for btn in state_def.get("buttons", []): if btn.get("action") == action and btn.get("go"): target = btn["go"] if target in machine["states"]: machine["current"] = target log.info(f"[ui] machine transition: {mid} {current} -> {target}") return True, f"Navigated to {target}" else: log.warning(f"[ui] machine transition target '{target}' not found in {mid}") return True, f"State '{target}' not found" return False, None def get_machine_controls(self) -> list[dict]: """Render all machines' current states as controls.""" controls = [] for mid, machine in self.machines.items(): current = machine["current"] state_def = machine["states"].get(current, {}) # Add content as display items for text in state_def.get("content", []): controls.append({ "type": "display", "display_type": "text", "label": f"{mid}", "value": text, "machine_id": mid, }) # Add buttons for btn in state_def.get("buttons", []): controls.append({ "type": "button", "label": btn.get("label", ""), "action": btn.get("action", ""), "machine_id": mid, }) return controls def get_machine_summary(self) -> str: """Summary for Thinker context — shape only, not full data.""" if not self.machines: return "" parts = [] for mid, m in self.machines.items(): current = m["current"] state_names = list(m["states"].keys()) parts.append(f" machine '{mid}': state={current}, states={state_names}") return "Machines:\n" + "\n".join(parts) # --- State operations --- def set_var(self, name: str, value) -> str: self.state[name] = value log.info(f"[ui] state: {name} = {value}") return f"{name} = {value}" def get_var(self, name: str): return self.state.get(name) def handle_local_action(self, action: str, payload: dict = None) -> str | None: """Try to handle an action locally. Returns result text or None if not local.""" binding = self.bindings.get(action) if not binding: return None op = binding.get("op") var = binding.get("var") step = binding.get("step", 1) if var not in self.state: return None if op == "inc": self.state[var] += step elif op == "dec": self.state[var] -= step elif op == "set" and payload and "value" in payload: self.state[var] = payload["value"] elif op == "toggle": self.state[var] = not self.state[var] else: return None log.info(f"[ui] local action {action}: {var} = {self.state[var]}") return f"{var} is now {self.state[var]}" # --- Action parsing from Thinker --- def _parse_thinker_actions(self, actions: list[dict]) -> list[dict]: """Parse Thinker's actions, register bindings, return controls.""" controls = [] for action in actions: act = action.get("action", "") label = action.get("label", "Action") payload = action.get("payload", {}) # Detect state-binding hints in payload if "var" in payload and "op" in payload: self.bindings[act] = { "op": payload["op"], "var": payload["var"], "step": payload.get("step", 1), } # Auto-create state var if it doesn't exist if payload["var"] not in self.state: self.set_var(payload["var"], payload.get("initial", 0)) log.info(f"[ui] bound action '{act}' -> {payload['op']} {payload['var']}") controls.append({ "type": "button", "label": label, "action": act, "payload": payload, }) return controls # --- Table extraction --- def _extract_table(self, tool_output: str) -> dict | None: if not tool_output: return None lines = [l for l in tool_output.strip().split("\n") if l.strip()] if len(lines) < 2: return None # Detect separator: tab or pipe sep = None if "\t" in lines[0]: sep = "\t" elif " | " in lines[0]: sep = " | " if sep: columns = [c.strip() for c in lines[0].split(sep)] data = [] for line in lines[1:]: if line.startswith("-") or line.startswith("=") or line.startswith("..."): continue vals = [v.strip() for v in line.split(sep)] if len(vals) == len(columns): data.append(dict(zip(columns, vals))) if data: return {"type": "table", "columns": columns, "data": data} # Legacy "Table:" prefix format if lines[0].startswith("Table:"): if len(lines) >= 2 and " | " in lines[1]: columns = [c.strip() for c in lines[1].split(" | ")] data = [] for line in lines[2:]: vals = [v.strip() for v in line.split(" | ")] if len(vals) == len(columns): data.append(dict(zip(columns, vals))) if data: return {"type": "table", "columns": columns, "data": data} return None # --- Render controls --- def _build_controls(self, thought: ThoughtResult) -> list[dict]: controls = [] # 1. Apply state_updates from Thinker's set_state() calls if thought.state_updates: for key, value in thought.state_updates.items(): self.set_var(key, value) # 2. Parse actions from Thinker (registers bindings) OR preserve existing buttons if thought.actions: controls.extend(self._parse_thinker_actions(thought.actions)) else: # Retain existing buttons when Thinker doesn't emit new ones for ctrl in self.current_controls: if ctrl["type"] == "button": controls.append(ctrl) # 3. Add labels for all state variables (bound + set_state) for var, value in self.state.items(): controls.append({ "type": "label", "id": f"var_{var}", "text": var, "value": str(value), }) # 4. Add display items from Thinker's emit_display() calls if thought.display_items: for item in thought.display_items: controls.append({ "type": "display", "display_type": item.get("type", "text"), "label": item.get("label", ""), "value": item.get("value", ""), "style": item.get("style", ""), }) # 3. Extract tables from tool output if thought.tool_output: table = self._extract_table(thought.tool_output) if table: controls.append(table) # 4. Add label for short tool results (if no table and no state vars) if thought.tool_used and thought.tool_output and not any(c["type"] == "table" for c in controls): output = thought.tool_output.strip() if "\n" not in output and len(output) < 100 and not self.state: controls.append({ "type": "label", "id": "tool_result", "text": thought.tool_used, "value": output, }) # 5. Add machine controls controls.extend(self.get_machine_controls()) return controls async def process(self, thought: ThoughtResult, history: list[dict], memory_context: str = "") -> list[dict]: # Apply machine ops first (create/add_state/reset/destroy) if thought.machine_ops: await self.apply_machine_ops(thought.machine_ops) controls = self._build_controls(thought) if controls: self.current_controls = controls await self.hud("controls", controls=controls) log.info(f"[ui] emitting {len(controls)} controls") else: if self.current_controls: controls = self.current_controls await self.hud("decided", instruction="no new controls") return controls async def process_local_action(self, action: str, payload: dict = None) -> tuple[str | None, list[dict]]: """Handle a local action. Returns (result_text, updated_controls) or (None, []) if not local.""" result = self.handle_local_action(action, payload) if result is None: return None, [] # Re-render controls with updated state controls = [] # Re-add existing buttons for ctrl in self.current_controls: if ctrl["type"] == "button": controls.append(ctrl) # Update labels with current state for var, value in self.state.items(): controls.append({ "type": "label", "id": f"var_{var}", "text": var, "value": str(value), }) self.current_controls = controls await self.hud("controls", controls=controls) log.info(f"[ui] re-rendered after local action: {action}") return result, controls