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.

backend/fluksio/ai/agent.py

@@ -1,9 +1,11 @@
 """The turn: the model, the tools, and the rounds between them.
 
-The tools are the MCP server's own, called in-process rather than over the
-endpoint: same names, same schemas, same REST calls underneath. A tool added
-for external agents is there for the assistant the moment it exists, and the
-assistant can do nothing an external agent could not.
+The turn takes its tools, its system prompt and the function that calls a
+tool from whoever runs it. The chat passes the MCP server's own, called
+in-process rather than over the endpoint: same names, same schemas, same REST
+calls underneath. A tool added for external agents is there for the chat the
+moment it exists, and it can do nothing an external agent could not. The
+assistant node passes the targets wired to it, and nothing else.
 
 The transcript is a list of chat-completion messages, mutated in place by
 :func:`run_turn` and persisted by the caller. It is also what the browser
@@ -14,7 +16,7 @@ from __future__ import annotations
 
 import json
 import logging
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Awaitable, Callable
 from typing import Any
 
 import httpx
@@ -83,6 +85,23 @@ def _arguments(raw: Any) -> dict[str, Any]:
     return parsed if isinstance(parsed, dict) else {}
 
 
+#: Runs one tool call and answers with what the model reads back.
+type ToolCaller = Callable[[str, dict[str, Any]], Awaitable[str]]
+
+
+def mcp_caller(authorization: str) -> ToolCaller:
+    """The MCP tools, each call made as the person chatting."""
+
+    async def call(name: str, arguments: dict[str, Any]) -> str:
+        token = caller_authorization.set(authorization)
+        try:
+            return await _call_tool(name, arguments)
+        finally:
+            caller_authorization.reset(token)
+
+    return call
+
+
 async def _call_tool(name: str, arguments: dict[str, Any]) -> str:
     """Run one tool; whatever goes wrong is text the model can read."""
     try:
@@ -114,8 +133,11 @@ async def run_turn(
     messages: list[dict[str, Any]],
     text: str,
     *,
-    flow: str | None,
-    authorization: str,
+    tools: list[dict[str, Any]],
+    system: str,
+    call_tool: ToolCaller,
+    extra: dict[str, Any] | None = None,
+    timeout: float | httpx.Timeout = 120.0,
 ) -> AsyncIterator[dict[str, Any]]:
     """One user turn, yielding progress events as it goes.
 
@@ -125,56 +147,53 @@ async def run_turn(
     part-way leaves a transcript the provider still accepts.
     """
     messages.append({"role": "user", "content": text})
-    tools = await tool_specs()
-    system = {"role": "system", "content": system_prompt(flow)}
-    token = caller_authorization.set(authorization)
-    try:
-        for _ in range(config.max_rounds):
-            choice: dict[str, Any] = {}
-            try:
-                async for part in client.stream_completion(
-                    api_base=config.api_base,
-                    api_key=config.api_key,
-                    model=config.model,
-                    messages=[system, *messages],
-                    tools=tools,
-                ):
-                    if isinstance(part, str):
-                        yield {"type": "delta", "content": part}
-                    else:
-                        choice = part
-            except (client.ProviderError, httpx.HTTPError) as exc:
-                yield {"type": "error", "detail": f"The provider answered {exc}"}
-                return
-            content = choice.get("content") or ""
-            calls = choice.get("tool_calls") or []
-            assistant: dict[str, Any] = {"role": "assistant", "content": content}
-            if content:
-                yield {"type": "message", "content": content}
-            if not calls:
-                messages.append(assistant)
-                return
-            assistant["tool_calls"] = calls
-            answers: list[dict[str, Any]] = []
-            for call in calls:
-                function = call.get("function") or {}
-                name = str(function.get("name", ""))
-                arguments = _arguments(function.get("arguments"))
-                yield {"type": "tool_call", "name": name, "arguments": arguments}
-                result = await _call_tool(name, arguments)
-                answers.append(
-                    {
-                        "role": "tool",
-                        "tool_call_id": call.get("id", ""),
-                        "content": result,
-                    }
-                )
-                yield {"type": "tool_result", "name": name, **summarize(result)}
+    head = {"role": "system", "content": system}
+    for _ in range(config.max_rounds):
+        choice: dict[str, Any] = {}
+        try:
+            async for part in client.stream_completion(
+                api_base=config.api_base,
+                api_key=config.api_key,
+                model=config.model,
+                messages=[head, *messages],
+                tools=tools,
+                timeout=timeout,
+                extra=extra,
+            ):
+                if isinstance(part, str):
+                    yield {"type": "delta", "content": part}
+                else:
+                    choice = part
+        except (client.ProviderError, httpx.HTTPError) as exc:
+            yield {"type": "error", "detail": f"The provider answered {exc}"}
+            return
+        content = choice.get("content") or ""
+        calls = choice.get("tool_calls") or []
+        assistant: dict[str, Any] = {"role": "assistant", "content": content}
+        if content:
+            yield {"type": "message", "content": content}
+        if not calls:
             messages.append(assistant)
-            messages.extend(answers)
-        yield {
-            "type": "error",
-            "detail": f"Stopped after {config.max_rounds} rounds of tool calls.",
-        }
-    finally:
-        caller_authorization.reset(token)
+            return
+        assistant["tool_calls"] = calls
+        answers: list[dict[str, Any]] = []
+        for call in calls:
+            function = call.get("function") or {}
+            name = str(function.get("name", ""))
+            arguments = _arguments(function.get("arguments"))
+            yield {"type": "tool_call", "name": name, "arguments": arguments}
+            result = await call_tool(name, arguments)
+            answers.append(
+                {
+                    "role": "tool",
+                    "tool_call_id": call.get("id", ""),
+                    "content": result,
+                }
+            )
+            yield {"type": "tool_result", "name": name, **summarize(result)}
+        messages.append(assistant)
+        messages.extend(answers)
+    yield {
+        "type": "error",
+        "detail": f"Stopped after {config.max_rounds} rounds of tool calls.",
+    }

backend/fluksio/ai/client.py

@@ -118,26 +118,32 @@ async def stream_completion(
     model: str,
     messages: list[dict[str, Any]],
     tools: list[dict[str, Any]],
-    timeout: float = 120.0,
+    timeout: float | httpx.Timeout = 120.0,
+    extra: dict[str, Any] | None = None,
 ) -> AsyncIterator[str | dict[str, Any]]:
     """``POST {api_base}/chat/completions`` streamed.
 
     Yields each piece of the answer's text as it arrives, then the whole
     message. The timeout is per chunk, so a long answer is not cut off.
+    ``extra`` is added to the request as it is, for what one provider
+    understands and another does not, such as Ollama's ``keep_alive``.
     """
-    body: dict[str, Any] = {"model": model, "messages": messages, "stream": True}
+    body: dict[str, Any] = {
+        **(extra or {}),
+        "model": model,
+        "messages": messages,
+        "stream": True,
+    }
     if tools:
         body["tools"] = tools
     assembler = Assembler()
-    async with (
-        httpx.AsyncClient(timeout=timeout) as http,
-        http.stream(
-            "POST",
-            api_base.rstrip("/") + "/chat/completions",
-            json=body,
-            headers=_headers(api_key),
-        ) as response,
-    ):
+    async with _client(api_base).stream(
+        "POST",
+        api_base.rstrip("/") + "/chat/completions",
+        json=body,
+        headers=_headers(api_key),
+        timeout=timeout,
+    ) as response:
         if response.is_error:
             await response.aread()
             raise ProviderError(_detail(response))
@@ -160,24 +166,24 @@ async def list_models(api_base: str, api_key: str, timeout: float = 15.0) -> lis
     return sorted(str(m["id"]) for m in data if isinstance(m, dict) and "id" in m)
 
 
-#: One client per speech server, so its connection stays open between calls
+#: One client per server, so its connection stays open between calls
 #: (`parrot serve` keeps an idle one 600 s). A new connection over a lossy
 #: link can cost a second, which a spoken reply cannot spare.
-_speech: dict[str, httpx.AsyncClient] = {}
+_clients: dict[str, httpx.AsyncClient] = {}
 
 
-def _speech_client(api_base: str) -> httpx.AsyncClient:
-    """The client for this speech server. Used from the engine's loop only."""
-    http = _speech.get(api_base)
+def _client(api_base: str) -> httpx.AsyncClient:
+    """The client for this server. Used from the engine's loop only."""
+    http = _clients.get(api_base)
     if http is None:
         limits = httpx.Limits(keepalive_expiry=300)
-        http = _speech[api_base] = httpx.AsyncClient(limits=limits)
+        http = _clients[api_base] = httpx.AsyncClient(limits=limits)
     return http
 
 
 async def warm(api_base: str) -> None:
     """Open the connection to a speech server ahead of the call that needs it."""
-    await _speech_client(api_base).get(api_base.rstrip("/") + "/models", timeout=10)
+    await _client(api_base).get(api_base.rstrip("/") + "/models", timeout=10)
 
 
 async def transcribe(
@@ -190,7 +196,7 @@ async def transcribe(
     timeout: float = 30.0,
 ) -> str:
     """``POST {api_base}/audio/transcriptions``: a whole clip, and what it says."""
-    response = await _speech_client(api_base).post(
+    response = await _client(api_base).post(
         api_base.rstrip("/") + "/audio/transcriptions",
         files={"file": (filename, audio, media_type)},
         data={"model": model, "response_format": "json"},
@@ -214,7 +220,7 @@ async def speak(
     MP3 rather than WAV: a seventh of the bytes, which over a slow link to
     the server is most of the wait for a sentence.
     """
-    response = await _speech_client(api_base).post(
+    response = await _client(api_base).post(
         api_base.rstrip("/") + "/audio/speech",
         json={"model": model, "input": text, "voice": voice, "response_format": "mp3"},
         timeout=timeout,

backend/fluksio/api/routes/ai.py

@@ -12,7 +12,7 @@ import json
 import logging
 import uuid
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, Literal
 
 import httpx
 from fastapi import APIRouter, Depends, HTTPException, Request
@@ -29,7 +29,7 @@ from fluksio.api.deps import (
 )
 from fluksio.core.config import settings
 from fluksio.flow.messages import qualify
-from fluksio.flow.schemas import NAME_PATTERN
+from fluksio.flow.schemas import NAME_PATTERN, NodeDef
 from fluksio.mcp import server as tools
 
 logger = logging.getLogger(__name__)
@@ -77,15 +77,28 @@ class ChatItem(BaseModel):
     error: str | None = None
 
 
+type WatchKind = Literal["partial", "text", "reply", "actions"]
+
+
+class ControlWatch(BaseModel):
+    """A message the conversation shows, as one kind of line."""
+
+    msg: str
+    kind: WatchKind
+    #: The node whose values on it count. Empty counts every value, as the
+    #: prompt does: a command typed here or spoken to a device.
+    node: str = ""
+
+
 class ControlNode(BaseModel):
-    """A wren node the chat can talk to: where it sends and where it reads."""
+    """A conversation the chat can hold: where a command goes, and what to show."""
 
     #: Qualified, ``flow.node``.
     id: str
     flow: str
     title: str
     prompt_msg: str
-    actions_msg: str
+    watch: list[ControlWatch]
 
 
 class TurnBody(BaseModel):
@@ -173,33 +186,75 @@ def read_ai_status(_: CurrentUser) -> Any:
     return AIStatus(available=True, model=config.model)
 
 
+#: The node types a command can be typed to.
+_TALKERS = ("wren", "assistant")
+
+
+def _wired(flow: str, node: NodeDef, port: str) -> str:
+    """The message a node's port is wired to, qualified, or empty."""
+    for spec in [*node.requires, *node.provides]:
+        if spec.port == port and spec.name:
+            return qualify(flow, spec.name)
+    return ""
+
+
 @router.get("/control", response_model=list[ControlNode])
 def read_control_nodes(_: CurrentUser, controller: FlowControllerDep) -> Any:
-    """The wren nodes of the running live flows, for the chat's Control tab.
+    """The conversations of the running live flows, for the chat's Control tab.
 
-    Only those with both ends wired: the chat publishes the command on the
-    prompt's message and reads the answer off the actions'.
+    One per wren or assistant node a command reaches first. A node whose
+    prompt is another's ``unhandled`` answers in that one's conversation,
+    after it. The conversation shows the prompt, the words heard so far (the
+    ``partial`` of a ``stt`` node writing the prompt), and each node's
+    ``reply`` and ``actions``, so typed and spoken commands are one.
     """
     found = []
     for flow in controller.store.read_all():
         if flow.mode != "live" or flow.name in controller.disabled:
             continue
-        for node in flow.nodes:
-            if node.type != "wren":
+        talkers = [
+            node
+            for node in flow.nodes
+            if node.type in _TALKERS and _wired(flow.name, node, "prompt")
+        ]
+        passed = {_wired(flow.name, node, "unhandled") for node in talkers}
+        for node in talkers:
+            prompt = _wired(flow.name, node, "prompt")
+            if prompt in passed:
                 continue
-            ends = {
-                spec.port: qualify(flow.name, spec.name)
-                for spec in [*node.requires, *node.provides]
-                if spec.port in ("prompt", "actions") and spec.name
-            }
-            if len(ends) == 2:
+            watch = [ControlWatch(msg=prompt, kind="text")]
+            watch += [
+                ControlWatch(msg=partial, kind="partial", node=f"{flow.name}.{stt.id}")
+                for stt in flow.nodes
+                if stt.type == "stt"
+                and _wired(flow.name, stt, "text") == prompt
+                and (partial := _wired(flow.name, stt, "partial"))
+            ]
+            chain = [node]
+            for talker in chain:
+                unhandled = _wired(flow.name, talker, "unhandled")
+                chain += [
+                    other
+                    for other in talkers
+                    if unhandled
+                    and _wired(flow.name, other, "prompt") == unhandled
+                    and other not in chain
+                ]
+            answers: tuple[WatchKind, ...] = ("reply", "actions")
+            watch += [
+                ControlWatch(msg=msg, kind=kind, node=f"{flow.name}.{talker.id}")
+                for talker in chain
+                for kind in answers
+                if (msg := _wired(flow.name, talker, kind))
+            ]
+            if any(w.kind in answers for w in watch):
                 found.append(
                     ControlNode(
                         id=f"{flow.name}.{node.id}",
                         flow=flow.name,
-                        title=node.title or node.id,
-                        prompt_msg=ends["prompt"],
-                        actions_msg=ends["actions"],
+                        title=f"{flow.title or flow.name} · {node.title or node.id}",
+                        prompt_msg=prompt,
+                        watch=watch,
                     )
                 )
     return found
@@ -280,8 +335,9 @@ def chat(
                 config,
                 messages,
                 body.message,
-                flow=body.flow,
-                authorization=f"Bearer {token}",
+                tools=await agent.tool_specs(),
+                system=agent.system_prompt(body.flow),
+                call_tool=agent.mcp_caller(f"Bearer {token}"),
             ):
                 yield _sse(event)
         except Exception as exc:  # noqa: BLE001 - the stream is the error channel

backend/fluksio/flow/controller.py

@@ -41,6 +41,7 @@ from fluksio.flow.executor import ExecutionService
 from fluksio.flow.messages import MessageSpec, flow_of, qualify
 from fluksio.flow.nodes import (
     RESERVED_SETTINGS,
+    AssistantNode,
     ChangeNode,
     DebounceNode,
     DelayNode,
@@ -385,6 +386,12 @@ NODE_TYPES: dict[str, NodeType] = {
         cls=WrenNode,
         params_schema=_schema_of(WrenNode),
     ),
+    "assistant": NodeType(
+        title="Assistant",
+        description="Answer a request with the AI provider, acting on what is wired to it.",
+        cls=AssistantNode,
+        params_schema=_schema_of(AssistantNode),
+    ),
     "stt": NodeType(
         title="Speech to text",
         description="Turn a microphone's audio into text, through a speech server.",

backend/fluksio/flow/nodes/__init__.py

@@ -4,6 +4,7 @@ Split by the outside world each one talks to. Importing from
 ``fluksio.flow.nodes`` keeps working, which is what every caller does.
 """
 
+from fluksio.flow.nodes.assistant import AssistantNode
 from fluksio.flow.nodes.base import RESERVED_SETTINGS, Node
 from fluksio.flow.nodes.debounce import DebounceNode
 from fluksio.flow.nodes.delay import DelayNode
@@ -23,6 +24,7 @@ from fluksio.flow.nodes.websocket import WebSocketNode
 from fluksio.flow.nodes.wren import WrenNode
 
 __all__ = [
+    "AssistantNode",
     "ChangeNode",
     "DebounceNode",
     "DelayNode",

backend/fluksio/flow/nodes/assistant.py

@@ -0,0 +1,350 @@
+"""Assistant: a request answered by the instance's language model.
+
+The request arrives on ``prompt``, typically what a wren node could not do.
+Every output but ``reply`` and ``actions`` is a target the model may set,
+offered as one tool named after its port and typed by its message, so the
+wiring is the allowlist, as for wren: the MCP authoring tools are never
+offered. Every other input is read as context ("outside_temperature: 7.5").
+
+The answer goes out on ``reply`` a sentence at a time while the model writes
+it, so a ``tts`` node speaks the first before the last is written.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import dataclasses
+import json
+import logging
+import re
+import threading
+from collections.abc import Coroutine, Iterable
+from concurrent.futures import Future
+from typing import TYPE_CHECKING, Any
+
+import httpx
+from pydantic import BaseModel, ConfigDict, Field
+
+from fluksio.ai import agent, client
+from fluksio.ai import config as ai_config
+from fluksio.flow.messages import DType, MessageSpec
+from fluksio.flow.nodes.base import Node
+
+if TYPE_CHECKING:
+    from fastapi import FastAPI
+
+logger = logging.getLogger(__name__)
+
+REPLY = "reply"
+ACTIONS = "actions"
+#: An optional input: true warms the provider while the request is spoken.
+LISTENING = "listening"
+
+#: Said when there is no provider, or it does not answer.
+UNAVAILABLE = "The assistant isn't available right now."
+#: Said when the model ends a turn without a word.
+DONE, CANT = "Done.", "Sorry, I can't do that."
+
+#: A server that is not there is given up on after a few seconds, while a
+#: model that has to be loaded first takes about ten.
+TIMEOUT = httpx.Timeout(15.0, connect=3.0)
+#: Rounds of tool calls one request may take.
+ROUNDS = 3
+
+#: A target's type, as its tool's parameter.
+_SCHEMAS: dict[DType, dict[str, str]] = {
+    DType.BOOL: {"type": "boolean"},
+    DType.INT: {"type": "integer"},
+    DType.FLOAT: {"type": "number"},
+    DType.STR: {"type": "string"},
+}
+
+_SYSTEM = """You are the voice of a home: someone spoke to a panel on the wall, \
+and you answer out loud.
+
+- To change something, call its tool. The tools are all you can change.
+- After a change, say what you did in one short sentence.
+- If no tool can do what was asked, or you did not understand it, say so in a \
+few words, such as "Sorry, I can't do that."
+- Answer questions from the readings below.
+- Speak plainly: one or two short sentences, no lists, no markdown, no emoji.
+"""
+
+_END = re.compile(r"(?<=[.!?])\s+")
+
+
+def cut(text: str) -> tuple[list[str], str]:
+    """The sentences ``text`` has finished, and the one still being written."""
+    *done, rest = _END.split(text)
+    return [s.strip() for s in done if s.strip()], rest
+
+
+class AssistantNode(Node):
+    """Answer the request on ``prompt``, acting only on the targets wired to it.
+
+    Inputs: ``prompt`` (the request, which wakes the node), ``listening``
+    (optional; turning true opens the connection and loads the model while
+    the request is still spoken) and any other, read as context.
+
+    Outputs: ``reply``, the answer a sentence at a time, ``actions``, a list
+    of records, one per value set, and any number of targets, each a tool.
+    When the provider is not set up or does not answer, the reply says so.
+    """
+
+    class Params(BaseModel):
+        model_config = ConfigDict(extra="allow")
+
+        model: str = Field(
+            default="", description="The model to ask. Empty asks the instance's."
+        )
+        extra_body: dict[str, Any] = Field(
+            default_factory=lambda: {
+                "temperature": 0,
+                "reasoning_effort": "none",
+                "keep_alive": "30m",
+            },
+            description=(
+                "Added to every request as it is. The default answers a request "
+                "the same way each time, turns thinking off and keeps the model "
+                "loaded for 30 minutes on Ollama; drop what a provider refuses."
+            ),
+        )
+
+    __slots__ = ("cfg", "_listening", "_lock", "_loop")
+
+    # A tool call is a command to a device, as wren's are.
+    idempotent = False
+
+    def __init__(
+        self,
+        requires: MessageSpec | Iterable[MessageSpec] = (),
+        provides: MessageSpec | Iterable[MessageSpec] = (),
+        params: dict[str, Any] | None = None,
+        name: str | None = None,
+    ):
+        self.cfg = self.Params.model_validate(params or {})
+        self._lock = threading.Lock()
+        #: What ``listening`` held at the last run, to tell its change apart.
+        self._listening: bool | None = None
+        self._loop: asyncio.AbstractEventLoop | None = None
+        inputs = self._normalize_ports(requires)
+        outputs = self._normalize_ports(provides)
+        typed = {"prompt": DType.STR, LISTENING: DType.BOOL}
+        for spec in inputs:
+            if spec.port in typed and spec.dtype is not typed[spec.port]:
+                raise ValueError(f"'{spec.port}' is a {typed[spec.port].value}.")
+        if not any(spec.port == "prompt" for spec in inputs):
+            raise ValueError("assistant needs a 'prompt' input: the request.")
+        for spec in outputs:
+            if spec.port == REPLY and spec.dtype is not DType.STR:
+                raise ValueError(f"'{REPLY}' is a str.")
+            if spec.port == ACTIONS and (
+                spec.dtype is not DType.LIST or spec.item_dtype is not DType.RECORD
+            ):
+                raise ValueError(f"'{ACTIONS}' is a list of records.")
+            if spec.port not in (REPLY, ACTIONS) and spec.dtype not in _SCHEMAS:
+                raise ValueError(
+                    f"'{spec.port}' is {spec.dtype.value}; a target is bool, int, "
+                    "float or str."
+                )
+        super().__init__(
+            self._answer,
+            requires=[
+                spec.model_copy(update={"trigger": spec.port in ("prompt", LISTENING)})
+                for spec in inputs
+            ],
+            provides=outputs,
+            params=params,
+            name=name or "assistant",
+        )
+
+    async def start(self, app: FastAPI | None = None) -> None:
+        self._loop = asyncio.get_running_loop()
+        names = [spec.name for spec in self.input_ports if spec.port == LISTENING]
+        if names and self._pipeline is not None:
+            held = await asyncio.to_thread(self._pipeline.state.get_present, names)
+            self._listening = held.get(names[0])
+
+    def _targets(self) -> list[MessageSpec]:
+        return [
+            spec
+            for spec in self.output_ports
+            if spec.name and spec.port not in (REPLY, ACTIONS)
+        ]
+
+    def _tools(self) -> list[dict[str, Any]]:
+        return [
+            {
+                "type": "function",
+                "function": {
+                    "name": spec.port,
+                    "description": f"Set the {spec.port.replace('_', ' ')}.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {"value": _SCHEMAS[spec.dtype]},
+                        "required": ["value"],
+                    },
+                },
+            }
+            for spec in self._targets()
+        ]
+
+    def _system(self, context: dict[str, Any]) -> str:
+        """The instructions, then what the inputs read and the targets hold."""
+        targets = self._targets()
+        held = (
+            self._pipeline.state.get_present([spec.name for spec in targets])
+            if self._pipeline is not None
+            else {}
+        )
+        lines = [_SYSTEM, "Readings:"]
+        lines += [f"{port}: {json.dumps(value)}" for port, value in context.items()]
+        lines += ["", "What the tools set, now:"]
+        lines += [
+            f"{spec.port}: {json.dumps(held[spec.name])}"
+            for spec in targets
+            if spec.name in held
+        ]
+        return "\n".join(lines)
+
+    def _config(self) -> ai_config.AIConfig | None:
+        config = ai_config.load()
+        if config is None or not config.ready:
+            return None
+        return dataclasses.replace(
+            config, model=self.cfg.model or config.model, max_rounds=ROUNDS
+        )
+
+    def _call(self, work: Coroutine[Any, Any, Any]) -> Future[Any]:
+        """Run on the engine's loop, where the provider's connection is kept."""
+        if self._loop is None:
+            work.close()
+            raise RuntimeError(f"{self.name} only runs in a live flow")
+        return asyncio.run_coroutine_threadsafe(work, self._loop)
+
+    def _answer(
+        self,
+        params: dict[str, Any],
+        prompt: str = "",
+        listening: bool | None = None,
+        **context: Any,
+    ) -> dict[str, Any] | None:
+        with self._lock:
+            was, self._listening = self._listening, listening
+        config = self._config()
+        if listening != was:
+            # Not a request: the device started or stopped listening.
+            if listening and config is not None:
+                self._call(self._warm(config))
+            return None
+        if not prompt.strip():
+            return None
+        actions: list[dict[str, Any]] = []
+        problem = self._call(
+            self._turn(config, prompt, self._system(context), actions)
+        ).result()
+        if problem:
+            self.report_health("degraded", problem)
+        elif self._health != "ok":
+            self.report_health("ok")
+        wired = {spec.port for spec in self.output_ports if spec.name}
+        return {ACTIONS: actions} if ACTIONS in wired else None
+
+    async def _warm(self, config: ai_config.AIConfig) -> None:
+        """A request for one token: the connection opens and the model loads.
+
+        Nothing more is worth sending ahead: a model with a sliding window,
+        such as Gemma, reads the whole prompt again on every request anyway.
+        """
+        with contextlib.suppress(Exception):
+            async for _ in client.stream_completion(
+                api_base=config.api_base,
+                api_key=config.api_key,
+                model=config.model,
+                messages=[{"role": "user", "content": "Hello."}],
+                tools=[],
+                timeout=60.0,
+                extra={**self.cfg.extra_body, "max_tokens": 1},
+            ):
+                pass
+
+    async def _turn(
+        self,
+        config: ai_config.AIConfig | None,
+        prompt: str,
+        system: str,
+        actions: list[dict[str, Any]],
+    ) -> str:
+        """Ask, act and say. Returns what went wrong, or an empty string."""
+        said = False
+
+        async def say(text: str) -> None:
+            nonlocal said
+            said = True
+            await asyncio.to_thread(self._say, text)
+
+        if config is None:
+            await say(UNAVAILABLE)
+            return "no AI provider is set up"
+        rest, problem = "", ""
+        try:
+            async for event in agent.run_turn(
+                config,
+                [],
+                prompt,
+                tools=self._tools(),
+                system=system,
+                call_tool=self._caller(actions),
+                extra=self.cfg.extra_body,
+                timeout=TIMEOUT,
+            ):
+                if event["type"] == "delta":
+                    done, rest = cut(rest + event["content"])
+                    for sentence in done:
+                        await say(sentence)
+                elif event["type"] == "message" and rest.strip():
+                    await say(rest.strip())
+                    rest = ""
+                elif event["type"] == "error":
+                    problem = str(event["detail"])
+        except Exception as exc:  # noqa: BLE001 - said, then reported as the health
+            problem = f"{type(exc).__name__}: {exc}"
+        if problem:
+            logger.warning("'%s' got no answer: %s", self.id, problem)
+        if not said:
+            await say(UNAVAILABLE if problem else DONE if actions else CANT)
+        return problem
+
+    def _caller(self, actions: list[dict[str, Any]]) -> agent.ToolCaller:
+        """Set the target a call names, and record it in ``actions``."""
+        specs = {spec.port: spec for spec in self._targets()}
+
+        async def call(name: str, arguments: dict[str, Any]) -> str:
+            spec = specs.get(name)
+            if spec is None:
+                return json.dumps({"error": f"There is no tool named {name}."})
+            value = arguments.get("value")
+            if spec.dtype is DType.FLOAT and type(value) is int:
+                value = float(value)
+            try:
+                spec.check(value)
+            except TypeError as exc:
+                return json.dumps({"error": str(exc)})
+            await asyncio.to_thread(self.emit, {name: value})
+            actions.append(
+                {
+                    "message": spec.name,
+                    "value": value,
+                    "applied": True,
+                    "delay_s": None,
+                    "call": f"{name}({json.dumps(value)})",
+                }
+            )
+            return json.dumps({name: value})
+
+        return call
+
+    def _say(self, text: str) -> None:
+        if any(spec.port == REPLY and spec.name for spec in self.output_ports):
+            self.emit({REPLY: text})

backend/fluksio/flow/nodes/speech.py

@@ -41,6 +41,9 @@ SPOKEN = 256
 #: How long after its estimated end a sentence still counts as playing: the
 #: device fetches it before it starts.
 PLAYING_SLACK_S = 1.0
+#: The partial for an utterance that came back empty, often a wake word that
+#: was not one, or was lost: shown on the caption, never spoken.
+NOTHING_HEARD = "Didn't catch that."
 
 
 def _check(specs: list[MessageSpec], ports: dict[str, DType], node: str) -> None:
@@ -152,8 +155,9 @@ class SttNode(_Speech):
     One realtime session per utterance, opened when ``listening`` turns true
     or at the first chunk, and committed when it turns false. A clip goes to
     the server whole. Either way the node returns at once, since a device's
-    next frame waits for it, and publishes once the server answers. An
-    utterance that fails publishes nothing and is logged.
+    next frame waits for it, and publishes once the server answers. Nothing
+    heard, or an utterance lost and logged, is a short note on ``partial``
+    alone.
     """
 
     __slots__ = ("_heard", "_listening", "_lock", "_session", "_tasks")
@@ -268,12 +272,15 @@ class SttNode(_Speech):
                     words = text.strip() if final else words + text
                     if final and words:
                         await asyncio.to_thread(self._say, partial=words, text=words)
+                    elif final:
+                        await asyncio.to_thread(self._say, partial=NOTHING_HEARD)
                     elif words:
                         await asyncio.to_thread(self._say, partial=words)
         except Exception as exc:
             logger.warning(
                 "'%s' lost an utterance: %s: %s", self.id, type(exc).__name__, exc
             )
+            await asyncio.to_thread(self._say, partial=NOTHING_HEARD)
         finally:
             # Ended without a commit, by a timeout or a failure: the next chunk
             # opens a session of its own rather than feeding this dead one.
@@ -288,9 +295,11 @@ class SttNode(_Speech):
             )
         except Exception as exc:
             logger.warning("'%s' lost a clip: %s: %s", self.id, type(exc).__name__, exc)
-            return
+            text = ""
         if text.strip():
             await asyncio.to_thread(self._say, partial=text.strip(), text=text.strip())
+        else:
+            await asyncio.to_thread(self._say, partial=NOTHING_HEARD)
 
 
 class TtsNode(_Speech):

backend/fluksio/flow/nodes/wren.py

@@ -71,6 +71,8 @@ INPUTS = {
 ACTIONS = "actions"
 REPLY = "reply"
 UNHANDLED = "unhandled"
+#: The reply when nothing was done and nothing reads ``unhandled``.
+CANT = "Sorry, I can't do that yet."
 
 _TARGET_TYPES = {DType.BOOL, DType.INT, DType.FLOAT, DType.STR}
 
@@ -198,8 +200,9 @@ class WrenNode(Node):
     (otherwise read off the clock: before 07:00 or from 18:00).
 
     Outputs: ``actions``, a list of records, one per call and target,
-    ``reply``, a sentence saying what was applied ("Kitchen light on."),
-    ``unhandled``, the prompt when nothing was, and any number of targets
+    ``reply``, a sentence saying what was applied ("Kitchen light on."), or
+    that nothing was when no node reads ``unhandled``, ``unhandled``, the
+    prompt when nothing was, and any number of targets
     named ``<area>_<kind>``. Targets of one area and kind are one device to
     the model; each takes what the call means for its type.
     A delayed call is held by the engine's work queue, like the Trigger
@@ -401,4 +404,11 @@ class WrenNode(Node):
             writes[REPLY] = reply[0].upper() + reply[1:] + "."
         if not said and UNHANDLED in wired:
             writes[UNHANDLED] = prompt
+        if not said and REPLY in wired and not self._passed_on():
+            writes[REPLY] = CANT
         return writes or None
+
+    def _passed_on(self) -> bool:
+        """Whether a node reads ``unhandled``, such as an assistant."""
+        name = next((s.name for s in self.output_ports if s.port == UNHANDLED), "")
+        return bool(name and self._pipeline and self._pipeline.consumes.get(name))

backend/tests/flow/test_assistant.py

@@ -0,0 +1,133 @@
+"""The assistant node: the wiring is its allowlist, and it speaks as it writes."""
+
+import asyncio
+import json
+from collections.abc import AsyncIterator
+from typing import Any
+
+import httpx
+import pytest
+
+from fluksio.ai import client
+from fluksio.ai import config as ai_config
+from fluksio.flow.messages import DType, MessageSpec
+from fluksio.flow.nodes.assistant import UNAVAILABLE, AssistantNode, cut
+from fluksio.flow.pipeline import Pipeline
+
+PROVIDER = ai_config.AIConfig(api_base="http://llm:11434/v1", api_key="k", model="m")
+
+
+def _call(name: str, value: Any) -> dict[str, Any]:
+    arguments = json.dumps({"value": value})
+    return {"id": name, "function": {"name": name, "arguments": arguments}}
+
+
+def _node() -> AssistantNode:
+    node = AssistantNode(
+        requires=[
+            MessageSpec(name="unhandled", port="prompt", dtype=DType.STR),
+            MessageSpec(name="outdoor", port="outside_temperature", dtype=DType.FLOAT),
+        ],
+        provides=[
+            MessageSpec(name="kitchen_light", dtype=DType.BOOL),
+            MessageSpec(
+                name="setpoint", port="living_room_setpoint", dtype=DType.FLOAT
+            ),
+            MessageSpec(name="reply", dtype=DType.STR),
+            MessageSpec(name="actions", dtype=DType.LIST),
+        ],
+    )
+    node.assign_flow("voice", "assistant")
+    Pipeline([node])
+    return node
+
+
+def _ask(node: AssistantNode, prompt: str) -> dict[str, Any] | None:
+    async def scenario() -> dict[str, Any] | None:
+        await node.start()
+        inputs = {"voice.unhandled": prompt, "voice.outdoor": 7.5}
+        return await asyncio.to_thread(node.execute, inputs)
+
+    return asyncio.run(scenario())
+
+
+def test_a_sentence_is_cut_once_it_ends():
+    assert cut("Done. It is 7.5 degr") == (["Done."], "It is 7.5 degr")
+    assert cut("Lights off! Anything else? ") == (["Lights off!", "Anything else?"], "")
+
+
+def test_only_wired_targets_are_tools_and_each_sentence_goes_out_as_written(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    log: list[Any] = []
+    seen: list[dict[str, Any]] = []
+    rounds = iter(
+        [
+            # The model reaches for a tool nothing is wired to, as well.
+            [
+                _call("kitchen_light", False),
+                _call("delete_flow", "demo"),
+                _call("living_room_setpoint", 22),
+            ],
+            ["Kitchen light off", " and heating to 22. It is ", "7.5 degrees outside."],
+        ]
+    )
+
+    async def completion(**kwargs: Any) -> AsyncIterator[str | dict[str, Any]]:
+        seen.append(kwargs)
+        step = next(rounds)
+        if isinstance(step[0], dict):
+            yield {"role": "assistant", "content": None, "tool_calls": step}
+            return
+        for piece in step:
+            log.append(piece)
+            yield piece
+        yield {"role": "assistant", "content": "".join(step)}
+
+    monkeypatch.setattr(ai_config, "load", lambda: PROVIDER)
+    monkeypatch.setattr(client, "stream_completion", completion)
+    monkeypatch.setattr(AssistantNode, "emit", lambda self, values: log.append(values))
+
+    out = _ask(_node(), "turn off the kitchen light and make it warmer")
+
+    tools = {tool["function"]["name"] for tool in seen[0]["tools"]}
+    assert tools == {"kitchen_light", "living_room_setpoint"}
+    system = seen[0]["messages"][0]["content"]
+    assert "outside_temperature: 7.5" in system
+    assert seen[0]["extra"]["reasoning_effort"] == "none"
+    answers = [m["content"] for m in seen[1]["messages"] if m["role"] == "tool"]
+    assert "error" in json.loads(answers[1])
+    # Set as called, and the first sentence said before the second is written.
+    assert log == [
+        {"kitchen_light": False},
+        {"living_room_setpoint": 22.0},
+        "Kitchen light off",
+        " and heating to 22. It is ",
+        {"reply": "Kitchen light off and heating to 22."},
+        "7.5 degrees outside.",
+        {"reply": "It is 7.5 degrees outside."},
+    ]
+    assert out is not None
+    assert [a["message"] for a in out["voice.actions"]] == [
+        "voice.kitchen_light",
+        "voice.setpoint",
+    ]
+
+
+def test_no_provider_or_one_that_does_not_answer_is_said(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    said: list[dict[str, Any]] = []
+    monkeypatch.setattr(AssistantNode, "emit", lambda self, values: said.append(values))
+
+    async def refused(**kwargs: Any) -> AsyncIterator[str | dict[str, Any]]:
+        raise httpx.ConnectError("refused")
+        yield ""
+
+    monkeypatch.setattr(client, "stream_completion", refused)
+    for provider in (None, PROVIDER):
+        monkeypatch.setattr(ai_config, "load", lambda provider=provider: provider)
+        node = _node()
+        _ask(node, "what is the weather tomorrow")
+        assert said.pop() == {"reply": UNAVAILABLE}
+        assert node._health == "degraded"

backend/tests/flow/test_node_types.py

@@ -105,6 +105,11 @@ FIXTURES: dict[str, dict] = {
         "requires": [MessageSpec(name="prompt", dtype=DType.STR)],
         "provides": [MessageSpec(name="kitchen_light", dtype=DType.BOOL)],
     },
+    "assistant": {
+        "params": {"model": "gemma4:12b"},
+        "requires": [MessageSpec(name="unhandled", port="prompt", dtype=DType.STR)],
+        "provides": [MessageSpec(name="kitchen_light", dtype=DType.BOOL)],
+    },
     "stt": {
         "params": {},
         "requires": [

backend/tests/flow/test_wren.py

@@ -10,7 +10,8 @@ from typing import Any
 import pytest
 
 from fluksio.flow.messages import DType, MessageSpec
-from fluksio.flow.nodes.wren import WrenNode, _value
+from fluksio.flow.nodes.base import Node
+from fluksio.flow.nodes.wren import CANT, WrenNode, _value
 from fluksio.flow.pipeline import Pipeline
 
 
@@ -184,8 +185,15 @@ def test_a_reply_says_what_was_applied_and_the_rest_is_unhandled(fake_wren):
 
     assert out == {"house.kitchen_light": True, "house.reply": "Kitchen light on."}
 
-    # Nothing wired for the call, or nothing to call at all.
+    # Nothing wired for the call, or nothing to call at all. Nothing reads
+    # what is left, so the node says so itself.
+    asked = "what is the weather tomorrow"
     for calls in ([FakeCall("lights.on", {"id": "garage:light"})], []):
         fake_wren.answer = (calls, 0.99)
-        out = node.execute({"house.command": "what is the weather tomorrow"})
-        assert out == {"house.unhandled": "what is the weather tomorrow"}
+        out = node.execute({"house.command": asked})
+        assert out == {"house.unhandled": asked, "house.reply": CANT}
+
+    # An assistant reading it answers instead.
+    unhandled = MessageSpec(name="house.unhandled", port="unhandled", dtype=DType.STR)
+    Pipeline([node, Node(lambda params, unhandled: None, requires=[unhandled])])
+    assert node.execute({"house.command": asked}) == {"house.unhandled": asked}