Assistant node, and the Control tab as one conversation (openpanel V5)

run_turn takes its tools, system prompt and tool caller; the chat passes the
MCP tools as before. The assistant node offers one tool per wired target,
reads its other inputs as context, streams its reply a sentence at a time,
warms the provider on `listening` and says so when the provider is not there.
Completions share the per-server HTTP client. wren replies when nothing reads
`unhandled`, stt shows a note when nothing was heard. /ai/control lists one
conversation per entry node with the messages to watch.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ljdk7nDaguh9Uhi3zhcSjn

frontend/src/components/Chat/ChatPanel.tsx

@@ -34,9 +34,9 @@ export const aiStatusQueryOptions = () =>
   })
 
 /**
- * The wren nodes the Control tab can talk to. Under the flows prefix: the
- * socket invalidates it on every rebuild, so a node published anywhere shows
- * up without a reload.
+ * The conversations the Control tab can hold, one per wren or assistant node a
+ * command reaches first. Under the flows prefix: the socket invalidates it on
+ * every rebuild, so a node published anywhere shows up without a reload.
  */
 export const controlNodesQueryOptions = () =>
   queryOptions({
@@ -45,8 +45,8 @@ export const controlNodesQueryOptions = () =>
   })
 
 /**
- * What the panel is for: building flows with the assistant, or commanding a
- * wren node in one. A tab is there when there is something behind it, and the
+ * What the panel is for: building flows with the assistant, or commanding the
+ * wren and assistant nodes in one. A tab is there when there is something behind it, and the
  * strip only draws once there is a choice to make.
  */
 const TABS: readonly (readonly [ChatTab, string])[] = [
@@ -103,7 +103,7 @@ function fromTranscript(transcript: ChatItem[]): Item[] {
 }
 
 /**
- * The way into the Control tab from Home, there only while a wren node is.
+ * The way into the Control tab from Home, there only while a conversation is.
  * Shaped like the sidebar trigger, which it mirrors on a phone.
  */
 export function ControlTrigger({ className }: { className?: string }) {
@@ -371,7 +371,7 @@ export function ChatPanel() {
         {shown.length === 0 && !waiting ? (
           <p className="py-8 text-center text-sm text-muted-foreground">
             {controlling
-              ? "Type a command, like turn on the lights in the kitchen. What it did comes back here."
+              ? "Type a command, like turn on the kitchen light. What was said to the flow and what it did show up here."
               : "Describe a flow to build, a change to make, or ask what a failure means. Drafts land in the editor for you to publish."}
           </p>
         ) : null}
@@ -407,6 +407,18 @@ function Message({ item }: { item: Item }) {
           </Suspense>
         </div>
       )
+    case "heard":
+      return (
+        <div className="max-w-[85%] self-end whitespace-pre-wrap break-words rounded-lg bg-muted px-3 py-1.5 text-sm text-muted-foreground">
+          {item.text}
+        </div>
+      )
+    case "reply":
+      return (
+        <div className="min-w-0 max-w-full self-start whitespace-pre-wrap break-words px-1 text-sm">
+          {item.text}
+        </div>
+      )
     case "error":
       return <div className="px-1 text-xs text-destructive">{item.text}</div>
     case "actions":
@@ -431,7 +443,7 @@ function Message({ item }: { item: Item }) {
   }
 }
 
-/** A wren node's answer: each call, where it went, and whether it was applied. */
+/** What a node set: each call, where it went, and whether it was applied. */
 function Actions({ actions }: { actions: Action[] }) {
   if (actions.length === 0) {
     return (
@@ -462,9 +474,11 @@ function Actions({ actions }: { actions: Action[] }) {
               : " (nothing wired for it)"}
         </div>
       ))}
-      <div className="text-muted-foreground">
-        confidence {actions[0].confidence.toFixed(2)}
-      </div>
+      {typeof actions[0].confidence === "number" ? (
+        <div className="text-muted-foreground">
+          confidence {actions[0].confidence.toFixed(2)}
+        </div>
+      ) : null}
     </div>
   )
 }

frontend/src/components/Chat/useControl.ts

@@ -1,61 +1,96 @@
 import { useEffect, useState } from "react"
 
 import { ApiError, type ControlNode, MessagesService } from "@/client"
-import {
-  liveStore,
-  useLiveValue,
-  useNodeFailure,
-} from "@/components/Flow/liveStore"
+import { liveStore } from "@/components/Flow/liveStore"
 
-/** One call the node answered with, as its `actions` output records it. */
+/** One value a node set, as its `actions` output records it. */
 export type Action = {
   message: string
   value: unknown
   applied: boolean
   delay_s: number | null
   call: string
-  confidence: number
+  /** Wren's certainty; the assistant has none. */
+  confidence?: number
 }
 
 export type ControlItem =
-  | { kind: "user" | "error"; text: string }
+  | { kind: "user" | "heard" | "reply" | "error"; text: string }
   | { kind: "actions"; actions: Action[] }
 
-/** How long a command may go unanswered before the chat says so. */
-const TIMEOUT_MS = 15_000
+/** The line each kind of message makes. */
+const LINE = { partial: "heard", text: "user", reply: "reply" } as const
+
+/** How long a typed command may go unanswered before the chat says so. */
+const TIMEOUT_MS = 20_000
+
+/** The next item list with `item` added, merged into the line it continues. */
+function added(items: ControlItem[], item: ControlItem): ControlItem[] {
+  const last = items[items.length - 1]
+  // The words heard so far grow in one line, which the final text replaces.
+  if (last?.kind === "heard" && (item.kind === "heard" || item.kind === "user"))
+    return [...items.slice(0, -1), item]
+  // A reply arrives a sentence at a time.
+  if (last?.kind === "reply" && item.kind === "reply")
+    return [
+      ...items.slice(0, -1),
+      { ...last, text: `${last.text} ${item.text}` },
+    ]
+  return [...items, item]
+}
 
 /**
- * Talking to a wren node: a command goes out on its prompt's message, and the
- * answer is the next value of its actions' message — or its next failure.
- *
- * The answer is told apart by the engine's own timestamps, taken before the
- * command leaves, so no clock on this side is compared with the server's. The
- * history lives here only, and a reload starts it over.
+ * One conversation with a flow's wren and assistant nodes: what was heard or
+ * typed, what each answered and what it set, as the engine publishes it. A
+ * typed command goes out on the prompt's message, which is where a spoken one
+ * arrives too, so both run through the same flow and show up the same way.
+ * The history lives here only, and a reload starts it over.
  */
 export function useControl(node: ControlNode | undefined) {
   const [items, setItems] = useState<ControlItem[]>([])
-  // What "newer" means for this command: the last answer and the last failure
-  // before it was sent.
-  const [pending, setPending] = useState<{
-    answer: number
-    failure: number
-  } | null>(null)
-  const answer = useLiveValue(node?.actions_msg)
-  const failure = useNodeFailure(node?.id ?? "")
+  const [pending, setPending] = useState(false)
 
   useEffect(() => {
-    if (!pending) return
-    if ((answer?.ts ?? 0) > pending.answer) {
-      const actions = Array.isArray(answer?.value)
-        ? (answer.value as Action[])
-        : []
-      setItems((prev) => [...prev, { kind: "actions", actions }])
-      setPending(null)
-    } else if (failure && failure.ts > pending.failure) {
-      setItems((prev) => [...prev, { kind: "error", text: failure.error }])
-      setPending(null)
+    if (!node) return
+    const replies = node.watch.some((w) => w.kind === "reply")
+    const add = (item: ControlItem) => setItems((prev) => added(prev, item))
+    const stops = node.watch.map((watch) => {
+      let last = liveStore.getValue(watch.msg)
+      return liveStore.subscribe(`value:${watch.msg}`, () => {
+        const value = liveStore.getValue(watch.msg)
+        if (!value || value === last || value.prior) return
+        last = value
+        if (watch.node && value.source?.id !== watch.node) return
+        if (watch.kind === "actions") {
+          const actions = Array.isArray(value.value)
+            ? (value.value as Action[])
+            : []
+          // With a reply to come, an empty list says nothing.
+          if (actions.length > 0 || !replies) add({ kind: "actions", actions })
+          if (!replies) setPending(false)
+        } else if (typeof value.value === "string" && value.value) {
+          add({ kind: LINE[watch.kind], text: value.value })
+          if (watch.kind === "reply") setPending(false)
+        }
+      })
+    })
+    const answering = node.watch.filter((w) => w.kind !== "partial" && w.node)
+    for (const id of new Set(answering.map((w) => w.node ?? ""))) {
+      let since = liveStore.getFailure(id)?.ts ?? 0
+      stops.push(
+        liveStore.subscribe(`failure:${id}`, () => {
+          const failure = liveStore.getFailure(id)
+          if (!failure || failure.ts <= since) return
+          since = failure.ts
+          add({ kind: "error", text: failure.error })
+          setPending(false)
+        }),
+      )
     }
-  }, [pending, answer, failure])
+    return () => {
+      for (const stop of stops) stop()
+    }
+  }, [node])
 
   useEffect(() => {
     if (!pending) return
@@ -67,18 +102,14 @@ export function useControl(node: ControlNode | undefined) {
           text: "No answer yet. Check that its flow is running.",
         },
       ])
-      setPending(null)
+      setPending(false)
     }, TIMEOUT_MS)
     return () => clearTimeout(timer)
   }, [pending])
 
   const send = async (text: string) => {
     if (!node) return
-    setItems((prev) => [...prev, { kind: "user", text }])
-    setPending({
-      answer: liveStore.getValue(node.actions_msg)?.ts ?? 0,
-      failure: liveStore.getFailure(node.id)?.ts ?? 0,
-    })
+    setPending(true)
     try {
       await MessagesService.publishMessage({
         name: node.prompt_msg,
@@ -94,9 +125,9 @@ export function useControl(node: ControlNode | undefined) {
           text: typeof detail === "string" ? detail : String(error),
         },
       ])
-      setPending(null)
+      setPending(false)
     }
   }
 
-  return { items, busy: pending !== null, send, clear: () => setItems([]) }
+  return { items, busy: pending, send, clear: () => setItems([]) }
 }

frontend/src/components/Flow/FlowNode.tsx

@@ -2,6 +2,7 @@ import { Handle, type NodeProps, Position } from "@xyflow/react"
 import {
   Bell,
   Bird,
+  Bot,
   Braces,
   Bug,
   Cable,
@@ -65,6 +66,7 @@ const NODE_ICONS = {
   file: FileText,
   ntfy: Bell,
   wren: Bird,
+  assistant: Bot,
   stt: Mic,
   tts: Volume2,
   // The glyph the Devices tab goes by.

frontend/src/components/Flow/liveStore.ts

@@ -126,6 +126,8 @@ function subscribeKey(key: string, listener: Listener) {
 }
 
 export const liveStore = {
+  /** Told of every change to one key, such as `value:<name>`, as it happens. */
+  subscribe: subscribeKey,
   setValue(name: string, value: LiveValue) {
     values.set(name, value)
     notify(`value:${name}`)