925fff731f
- Memorizer tracks user_expectation (conversational/delegated/waiting_input/observing) - Output node adjusts phrasing per expectation - PA retry loop: reformulates job on expert failure (all retries exhausted or tool skip) - Machine state in PA context: get_machine_summary includes current state, buttons, stored data - Expert writes to machine state via update_machine + transition_machine - Expanded baked schema coverage - Awareness panel shows color-coded expectation state - Dashboard and workspace component updates Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
567 lines
24 KiB
Python
567 lines
24 KiB
Python
"""UI Node: stateful renderer — manages UI state, handles local actions, renders controls."""
|
|
|
|
import json
|
|
import logging
|
|
import uuid
|
|
|
|
from .base import Node
|
|
from ..types import ThoughtResult, Artifact
|
|
|
|
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.thinker_controls: list[dict] = [] # buttons, labels, tables from Thinker
|
|
self.artifacts: list[dict] = [] # typed workspace artifacts
|
|
self.state: dict = {} # {"count": 0, "theme": "dark", ...}
|
|
self.bindings: dict = {} # {"increment": {"op": "inc", "var": "count"}, ...}
|
|
self.machines: dict = {} # {"nav": {initial, states, current}, ...}
|
|
|
|
@property
|
|
def current_controls(self) -> list[dict]:
|
|
"""Merged view: thinker controls + machine controls."""
|
|
return self.thinker_controls + self.get_machine_controls()
|
|
|
|
@current_controls.setter
|
|
def current_controls(self, value: list[dict]):
|
|
"""When set directly (e.g. after machine transition), split into layers."""
|
|
# Machine controls have machine_id — keep those in machines, rest in thinker
|
|
self.thinker_controls = [c for c in value if not c.get("machine_id")]
|
|
|
|
# --- 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 — handles both dict and list formats from Director
|
|
raw_states = op_data.get("states", {})
|
|
states = {}
|
|
if isinstance(raw_states, dict):
|
|
# Dict format: {main: {actions: [...], display: [...], content: [...]}}
|
|
for name, sdef in raw_states.items():
|
|
if not isinstance(sdef, dict):
|
|
states[name] = {"buttons": [], "content": []}
|
|
continue
|
|
buttons = sdef.get("buttons", []) or sdef.get("actions", [])
|
|
content = sdef.get("content", []) or sdef.get("display", [])
|
|
# Normalize display items to strings
|
|
if content and isinstance(content[0], dict):
|
|
content = [c.get("value", c.get("label", "")) for c in content]
|
|
# Normalize button format: ensure "go" field for navigation
|
|
for btn in buttons:
|
|
if isinstance(btn, dict) and not btn.get("go") and btn.get("payload"):
|
|
btn["go"] = btn["payload"]
|
|
states[name] = {"buttons": buttons, "content": content}
|
|
elif isinstance(raw_states, list):
|
|
# List format: [{name, buttons/actions, content/display}]
|
|
for s in raw_states:
|
|
if isinstance(s, str):
|
|
s = {"name": s}
|
|
name = s.get("name", "")
|
|
if name:
|
|
buttons = s.get("buttons", []) or s.get("actions", [])
|
|
content = s.get("content", []) or s.get("display", [])
|
|
if content and isinstance(content[0], dict):
|
|
content = [c.get("value", c.get("label", "")) for c in content]
|
|
for btn in buttons:
|
|
if isinstance(btn, dict) and not btn.get("go") and btn.get("payload"):
|
|
btn["go"] = btn["payload"]
|
|
states[name] = {"buttons": buttons, "content": content}
|
|
self.machines[mid] = {
|
|
"initial": initial,
|
|
"current": initial,
|
|
"states": states,
|
|
"data": {}, # wizard field storage (e.g. {"bundesland": "Bayern"})
|
|
}
|
|
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 == "update_data":
|
|
if mid not in self.machines:
|
|
log.warning(f"[ui] update_data: machine '{mid}' not found")
|
|
continue
|
|
data_update = op_data.get("data", {})
|
|
self.machines[mid]["data"].update(data_update)
|
|
log.info(f"[ui] machine data updated: {mid} += {data_update}")
|
|
await self.hud("machine_data_updated", id=mid, data=data_update)
|
|
|
|
elif op == "transition":
|
|
if mid not in self.machines:
|
|
log.warning(f"[ui] transition: machine '{mid}' not found")
|
|
continue
|
|
target = op_data.get("target", "")
|
|
if target in self.machines[mid]["states"]:
|
|
old = self.machines[mid]["current"]
|
|
self.machines[mid]["current"] = target
|
|
log.info(f"[ui] machine transition (expert): {mid} {old} -> {target}")
|
|
await self.hud("machine_transitioned", id=mid, old=old, target=target)
|
|
else:
|
|
log.warning(f"[ui] transition target '{target}' not found in {mid}")
|
|
|
|
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():
|
|
log.info(f"[ui] machine_controls: {mid} current={machine['current']} states={list(machine['states'].keys())} buttons={[b.get('label','?') for b in machine['states'].get(machine['current'],{}).get('buttons',[])]}")
|
|
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:
|
|
"""Rich summary for PA/Thinker context — includes current state details and stored data."""
|
|
if not self.machines:
|
|
return ""
|
|
parts = []
|
|
for mid, m in self.machines.items():
|
|
current = m["current"]
|
|
state_names = list(m["states"].keys())
|
|
state_def = m["states"].get(current, {})
|
|
line = f" machine '{mid}': state={current}, states={state_names}"
|
|
# Current state content
|
|
content = state_def.get("content", [])
|
|
if content:
|
|
line += f", content={content}"
|
|
# Current state buttons
|
|
buttons = state_def.get("buttons", [])
|
|
if buttons:
|
|
btn_labels = [b.get("label", b.get("action", "?")) for b in buttons if isinstance(b, dict)]
|
|
if btn_labels:
|
|
line += f", buttons={btn_labels}"
|
|
# Stored wizard data
|
|
data = m.get("data", {})
|
|
if data:
|
|
line += f", data={data}"
|
|
parts.append(line)
|
|
return "Active machines (interactive wizard/workflow state):\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]:
|
|
"""Build thinker controls only. Machine controls are added by the property."""
|
|
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 thinker buttons
|
|
if thought.actions:
|
|
controls.extend(self._parse_thinker_actions(thought.actions))
|
|
else:
|
|
# Retain existing thinker buttons (not machine buttons — those persist via property)
|
|
for ctrl in self.thinker_controls:
|
|
if ctrl.get("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 (cards, lists, or simple display)
|
|
if thought.display_items:
|
|
for item in thought.display_items:
|
|
item_type = item.get("type", "text")
|
|
if item_type in ("card", "list"):
|
|
# Pass through structured components as-is
|
|
controls.append(item)
|
|
else:
|
|
controls.append({
|
|
"type": "display",
|
|
"display_type": item_type,
|
|
"label": item.get("label", ""),
|
|
"value": item.get("value", ""),
|
|
"style": item.get("style", ""),
|
|
})
|
|
|
|
# 5. Extract tables from tool output
|
|
if thought.tool_output:
|
|
table = self._extract_table(thought.tool_output)
|
|
if table:
|
|
controls.append(table)
|
|
|
|
# 6. 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.get("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,
|
|
})
|
|
|
|
# Machine controls are NOT added here — the current_controls property merges them
|
|
|
|
return controls
|
|
|
|
def _build_artifacts(self, thought: ThoughtResult) -> list[dict]:
|
|
"""Convert ThoughtResult into typed artifacts."""
|
|
arts = []
|
|
|
|
# 1. Direct artifacts from expert's emit_artifact calls
|
|
if thought.artifacts:
|
|
for a in thought.artifacts:
|
|
if not a.get("id"):
|
|
a["id"] = str(uuid.uuid4())[:8]
|
|
arts.append(a)
|
|
|
|
# 2. Convert display_items (cards, lists) → entity_detail artifacts
|
|
if thought.display_items:
|
|
for item in thought.display_items:
|
|
item_type = item.get("type", "text")
|
|
if item_type == "card":
|
|
arts.append({
|
|
"id": str(uuid.uuid4())[:8],
|
|
"type": "entity_detail",
|
|
"data": {
|
|
"title": item.get("title", ""),
|
|
"subtitle": item.get("subtitle", ""),
|
|
"fields": item.get("fields", []),
|
|
},
|
|
"actions": item.get("actions", []),
|
|
"meta": {},
|
|
})
|
|
elif item_type == "list":
|
|
arts.append({
|
|
"id": str(uuid.uuid4())[:8],
|
|
"type": "entity_detail",
|
|
"data": {
|
|
"title": item.get("title", ""),
|
|
"items": item.get("items", []),
|
|
},
|
|
"actions": [],
|
|
"meta": {"list": True},
|
|
})
|
|
else:
|
|
arts.append({
|
|
"id": str(uuid.uuid4())[:8],
|
|
"type": "status",
|
|
"data": {
|
|
"display_type": item_type,
|
|
"label": item.get("label", ""),
|
|
"value": item.get("value", ""),
|
|
"style": item.get("style", ""),
|
|
},
|
|
"actions": [],
|
|
"meta": {},
|
|
})
|
|
|
|
# 3. Convert actions → action_bar artifact
|
|
if thought.actions:
|
|
btns = self._parse_thinker_actions(thought.actions)
|
|
arts.append({
|
|
"id": "action_bar",
|
|
"type": "action_bar",
|
|
"data": {},
|
|
"actions": [{"label": b["label"], "action": b["action"],
|
|
"payload": b.get("payload", {})} for b in btns],
|
|
"meta": {},
|
|
})
|
|
elif self.thinker_controls:
|
|
# Preserve existing buttons as action_bar
|
|
existing_btns = [c for c in self.thinker_controls if c.get("type") == "button"]
|
|
if existing_btns:
|
|
arts.append({
|
|
"id": "action_bar",
|
|
"type": "action_bar",
|
|
"data": {},
|
|
"actions": [{"label": b["label"], "action": b["action"],
|
|
"payload": b.get("payload", {})} for b in existing_btns],
|
|
"meta": {},
|
|
})
|
|
|
|
# 4. Convert tool_output table → data_table artifact
|
|
if thought.tool_output:
|
|
table = self._extract_table(thought.tool_output)
|
|
if table:
|
|
arts.append({
|
|
"id": str(uuid.uuid4())[:8],
|
|
"type": "data_table",
|
|
"data": {
|
|
"columns": table["columns"],
|
|
"rows": table["data"],
|
|
},
|
|
"actions": [],
|
|
"meta": {"source": thought.tool_used or "query_db"},
|
|
})
|
|
|
|
# 5. State variables → status artifacts
|
|
if thought.state_updates:
|
|
for key, value in thought.state_updates.items():
|
|
self.set_var(key, value)
|
|
for var, value in self.state.items():
|
|
arts.append({
|
|
"id": f"state_{var}",
|
|
"type": "status",
|
|
"data": {"label": var, "value": str(value), "display_type": "text"},
|
|
"actions": [],
|
|
"meta": {"state_var": True},
|
|
})
|
|
|
|
# 6. Machines → machine artifacts
|
|
for mid, machine in self.machines.items():
|
|
current = machine["current"]
|
|
state_def = machine["states"].get(current, {})
|
|
arts.append({
|
|
"id": f"machine_{mid}",
|
|
"type": "machine",
|
|
"data": {
|
|
"machine_id": mid,
|
|
"current": current,
|
|
"states": list(machine["states"].keys()),
|
|
"content": state_def.get("content", []),
|
|
"stored_data": machine.get("data", {}),
|
|
},
|
|
"actions": [{"label": b.get("label", ""), "action": b.get("action", ""),
|
|
"go": b.get("go", "")}
|
|
for b in state_def.get("buttons", []) if isinstance(b, dict)],
|
|
"meta": {"live": True},
|
|
})
|
|
|
|
return arts
|
|
|
|
def get_artifacts(self) -> list[dict]:
|
|
"""Return current artifact list."""
|
|
return self.artifacts
|
|
|
|
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)
|
|
|
|
# Build artifacts (new system)
|
|
self.artifacts = self._build_artifacts(thought)
|
|
|
|
# Build legacy controls (backward compat)
|
|
thinker_ctrls = self._build_controls(thought)
|
|
|
|
if thinker_ctrls:
|
|
self.thinker_controls = thinker_ctrls
|
|
# Always emit the merged view (thinker + machine)
|
|
merged = self.current_controls
|
|
if merged or self.artifacts:
|
|
await self.hud("controls", controls=merged)
|
|
log.info(f"[ui] emitting {len(merged)} controls + {len(self.artifacts)} artifacts")
|
|
else:
|
|
await self.hud("decided", instruction="no new controls")
|
|
|
|
return merged
|
|
|
|
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
|