Voice nodes (openpanel V4): stt, tts, wren reply/unhandled, socket publish

- `stt`: one realtime transcription session per utterance on the speech
  server's url, opened at `listening` true or the first chunk, committed at
  false; a whole clip goes to /v1/audio/transcriptions. Returns at once and
  publishes `partial` and `text` when the server answers.
- `tts`: one /v1/audio/speech call per sentence, cached by voice and text in
  node state; a reference is marked `append` while earlier ones still play.
- wren: `reply` says what was applied, `unhandled` carries the prompt when
  nothing was. The install hint names the git pin (PyPI's `wren` is another
  package).
- ai/client: transcribe, speak and transcribe_stream beside stream_completion.
- A device socket publishes scalar `in` ports as {"type": "publish"} text,
  in order with its binary frames.
- Pipeline.publish walks only what its values reach: a node downstream of a
  consumer that published nothing no longer runs again (wren re-acted on
  every microphone chunk).

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

backend/fluksio/ai/client.py

@@ -6,16 +6,22 @@ OpenRouter, Ollama, vLLM — so one client covers them without a provider SDK.
 The admin types the URL; the model never chooses where a request goes.
 
 The call is streamed, so the answer reaches the browser as it is written.
+
+Speech is the same dialect's audio half: a realtime transcription session, a
+whole clip transcribed, and a sentence spoken, as `parrot serve` answers them.
 """
 
 from __future__ import annotations
 
+import asyncio
+import base64
 import json
 from collections.abc import AsyncIterator
 from typing import Any
 from urllib.parse import urlparse
 
 import httpx
+from websockets.asyncio.client import connect
 
 
 class ProviderError(Exception):
@@ -152,3 +158,113 @@ async def list_models(api_base: str, api_key: str, timeout: float = 15.0) -> lis
         raise ProviderError(_detail(response))
     data = response.json().get("data") or []
     return sorted(str(m["id"]) for m in data if isinstance(m, dict) and "id" in m)
+
+
+async def transcribe(
+    api_base: str,
+    audio: bytes,
+    filename: str,
+    media_type: str,
+    *,
+    model: str = "whisper-1",
+    timeout: float = 30.0,
+) -> str:
+    """``POST {api_base}/audio/transcriptions``: a whole clip, and what it says."""
+    async with httpx.AsyncClient(timeout=timeout) as http:
+        response = await http.post(
+            api_base.rstrip("/") + "/audio/transcriptions",
+            files={"file": (filename, audio, media_type)},
+            data={"model": model, "response_format": "json"},
+        )
+    if response.is_error:
+        raise ProviderError(_detail(response))
+    return str(response.json().get("text") or "")
+
+
+async def speak(
+    api_base: str,
+    text: str,
+    voice: str,
+    *,
+    model: str = "tts-1",
+    timeout: float = 30.0,
+) -> bytes:
+    """``POST {api_base}/audio/speech``: ``text`` spoken, as WAV."""
+    async with httpx.AsyncClient(timeout=timeout) as http:
+        response = await http.post(
+            api_base.rstrip("/") + "/audio/speech",
+            json={
+                "model": model,
+                "input": text,
+                "voice": voice,
+                "response_format": "wav",
+            },
+        )
+    if response.is_error:
+        raise ProviderError(_detail(response))
+    return response.content
+
+
+async def transcribe_stream(
+    api_base: str,
+    audio: asyncio.Queue[bytes | None],
+    *,
+    rate: int = 16000,
+    idle_s: float = 10.0,
+) -> AsyncIterator[tuple[bool, str]]:
+    """``WS {api_base}/realtime``: one utterance, transcribed while it is spoken.
+
+    Sends each chunk of PCM16 taken off ``audio`` and commits at ``None``.
+    Yields ``(False, delta)`` as words are agreed on, then ``(True,
+    transcript)`` once. A chunk that takes longer than ``idle_s`` to come
+    raises ``TimeoutError``, so an utterance nobody commits ends.
+    """
+    url = "ws" + api_base.rstrip("/").removeprefix("http") + "/realtime"
+    async with connect(url, open_timeout=idle_s) as ws:
+        update = {"type": "audio/pcm", "rate": rate}
+        await ws.send(
+            json.dumps(
+                {
+                    "type": "session.update",
+                    "session": {
+                        "type": "transcription",
+                        "audio": {"input": {"format": update}},
+                    },
+                }
+            )
+        )
+
+        async def append() -> None:
+            try:
+                while (
+                    chunk := await asyncio.wait_for(audio.get(), idle_s)
+                ) is not None:
+                    data = base64.b64encode(chunk).decode()
+                    await ws.send(
+                        json.dumps({"type": "input_audio_buffer.append", "audio": data})
+                    )
+                await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
+            except BaseException:
+                # Ends the reading below, which then says why.
+                await ws.close()
+                raise
+
+        sender = asyncio.create_task(append())
+        sender.add_done_callback(lambda done: done.cancelled() or done.exception())
+        try:
+            async for raw in ws:
+                event = json.loads(raw)
+                kind = str(event.get("type") or "")
+                if kind == "error":
+                    raise ProviderError(str((event.get("error") or {}).get("message")))
+                if kind.endswith("input_audio_transcription.delta"):
+                    yield False, str(event.get("delta") or "")
+                elif kind.endswith("input_audio_transcription.completed"):
+                    yield True, str(event.get("transcript") or "")
+                    return
+        finally:
+            sender.cancel()
+        failure = (
+            sender.exception() if sender.done() and not sender.cancelled() else None
+        )
+        raise failure or ProviderError("the session closed before its transcript")

backend/fluksio/api/routes/flows.py

@@ -46,6 +46,7 @@ from fluksio.flow.controller import NODE_TYPES, FlowController, follow_header
 from fluksio.flow.dashboards import DashboardStore
 from fluksio.flow.demand import demand
 from fluksio.flow.devices import (
+    Binding,
     DeviceDef,
     DeviceStore,
     DeviceUnavailable,
@@ -1362,6 +1363,7 @@ FRAME_REFUSALS = {
     "not_media": "the port it names is not wired as an image or audio",
     "media_type": "its media type does not match its port",
     "too_large": "it is larger than MAX_FRAME_BYTES",
+    "not_scalar": "the port it names is not wired as a bool, int, float or str",
     "undeclared": "no flow declares its message",
     "refused": "its message refused the value",
 }
@@ -1393,14 +1395,7 @@ def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str:
     controller: FlowController | None = getattr(app.state, "flow_controller", None)
     if devices is None or store is None or controller is None:
         return "no_engine"
-    binding = next(
-        (
-            b
-            for b in devices.bound(device.id)
-            if b.direction == "in" and b.message == name
-        ),
-        None,
-    )
+    binding = _bound_in(devices, device, name)
     if binding is None:
         return "not_wired"
     if binding.dtype not in (DType.IMAGE, DType.AUDIO):
@@ -1420,6 +1415,59 @@ def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str:
     return ""
 
 
+def _bound_in(devices: DeviceStore, device: DeviceDef, name: str) -> Binding | None:
+    """The device's own ``in`` port wired to ``name``, if one is."""
+    return next(
+        (
+            b
+            for b in devices.bound(device.id)
+            if b.direction == "in" and b.message == name
+        ),
+        None,
+    )
+
+
+def published(frame: str) -> dict[str, Any] | None:
+    """A value a device publishes on its socket, or None for anything else."""
+    try:
+        payload = orjson.loads(frame)
+    except orjson.JSONDecodeError:
+        return None
+    if isinstance(payload, dict) and payload.get("type") == "publish":
+        return payload
+    return None
+
+
+def ingest_publish(app: Any, device: DeviceDef, payload: dict[str, Any]) -> str:
+    """Publish a value a device sent on its socket, as ``POST /messages`` would.
+
+    Returns the key in ``FRAME_REFUSALS`` it was refused for, or "" once it
+    is published. Only a message one of the device's own ``in`` ports is wired
+    to as a scalar, credited to the device node holding that port. Taken in
+    turn with the device's binary frames, so a ``listening`` false lands after
+    the last chunk of what was heard.
+    """
+    name = str(payload.get("name") or "")
+    devices: DeviceStore | None = getattr(app.state, "device_store", None)
+    controller: FlowController | None = getattr(app.state, "flow_controller", None)
+    if devices is None or controller is None:
+        return "no_engine"
+    binding = _bound_in(devices, device, name)
+    if binding is None:
+        return "not_wired"
+    if binding.dtype not in (DType.BOOL, DType.INT, DType.FLOAT, DType.STR):
+        return "not_scalar"
+    try:
+        controller.publish_message(
+            name, payload.get("value"), port_source(devices, device, name)
+        )
+    except KeyError:
+        return "undeclared"
+    except TypeError:
+        return "refused"
+    return ""
+
+
 async def _receive(websocket: WebSocket) -> str | bytes:
     """The client's next frame, text or binary."""
     message = await websocket.receive()
@@ -1465,8 +1513,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
 
     A client says which media it is drawing (``{"type": "media", "names":
     [...]}``) and is pushed those frames. A device also sends its camera's
-    frames here, as binary frames in the layout it is pushed them in, and is
-    told when its on-demand ports have demand.
+    frames here, as binary frames in the layout it is pushed them in, and
+    the values of its scalar ports as ``publish`` messages, and is told when
+    its on-demand ports have demand.
 
     A live view's signalling rides here too (``fluksio.flow.live``): ``live``
     messages on a person's socket, ``signal`` messages on a device's.
@@ -1628,6 +1677,23 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
                         if asked is not None:
                             wanted = asked
                             demand.want(viewer, wanted)
+                        elif (
+                            device_def is not None
+                            and (value := published(frame)) is not None
+                        ):
+                            if credential is not None and await credential.revoked():
+                                await websocket.close(code=1008)
+                                return
+                            reason = await run_in_threadpool(
+                                ingest_publish, websocket.app, device_def, value
+                            )
+                            if reason and reason not in refused:
+                                refused.add(reason)
+                                logger.warning(
+                                    "Dropped a value from %s: %s",
+                                    device,
+                                    FRAME_REFUSALS[reason],
+                                )
                         elif (signal := live_view.parse(frame)) is not None:
                             live_view.sessions.heard(
                                 signal,

backend/fluksio/flow/controller.py

@@ -57,7 +57,9 @@ from fluksio.flow.nodes import (
     NtfyNode,
     RbeNode,
     ServiceNode,
+    SttNode,
     SwitchNode,
+    TtsNode,
     WebSocketNode,
     WrenNode,
 )
@@ -383,6 +385,18 @@ NODE_TYPES: dict[str, NodeType] = {
         cls=WrenNode,
         params_schema=_schema_of(WrenNode),
     ),
+    "stt": NodeType(
+        title="Speech to text",
+        description="Turn a microphone's audio into text, through a speech server.",
+        cls=SttNode,
+        params_schema=_schema_of(SttNode),
+    ),
+    "tts": NodeType(
+        title="Text to speech",
+        description="Speak text as audio for a speaker, through a speech server.",
+        cls=TtsNode,
+        params_schema=_schema_of(TtsNode),
+    ),
     "inject": NodeType(
         title="Inject",
         description="Emit a value on request, on a timer, or when the flow starts.",

backend/fluksio/flow/nodes/__init__.py

@@ -18,6 +18,7 @@ from fluksio.flow.nodes.mlp import MLPNode
 from fluksio.flow.nodes.mqtt import MqttNode
 from fluksio.flow.nodes.ntfy import NtfyNode
 from fluksio.flow.nodes.service import ServiceNode
+from fluksio.flow.nodes.speech import SttNode, TtsNode
 from fluksio.flow.nodes.websocket import WebSocketNode
 from fluksio.flow.nodes.wren import WrenNode
 
@@ -39,7 +40,9 @@ __all__ = [
     "RESERVED_SETTINGS",
     "RbeNode",
     "ServiceNode",
+    "SttNode",
     "SwitchNode",
+    "TtsNode",
     "WebSocketNode",
     "WrenNode",
 ]

backend/fluksio/flow/nodes/speech.py

@@ -0,0 +1,353 @@
+"""Speech: a microphone's audio to text, and text to audio a speaker plays.
+
+Both talk to an OpenAI-compatible speech server, such as `parrot serve` kept
+running by a Service node, whose address arrives on a ``url`` input. An empty
+url is a server that is not answering: the node reads degraded and drops what
+it gets.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import io
+import logging
+import re
+import threading
+import time
+import wave
+from collections.abc import Coroutine, Iterable
+from concurrent.futures import Future
+from typing import TYPE_CHECKING, Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from fluksio.ai import client
+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__)
+
+#: Seconds a session waits for its next chunk, and the most one utterance may
+#: take from its first chunk to its text. Past either it is given up.
+IDLE_S = 10.0
+UTTERANCE_S = 60.0
+#: The rate of a microphone's PCM chunks, the panels' (openpanel V3).
+RATE = 16000
+#: How many spoken sentences a tts node keeps.
+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
+
+
+def _check(specs: list[MessageSpec], ports: dict[str, DType], node: str) -> None:
+    """Refuse a port the node does not have, or one of another type."""
+    for spec in specs:
+        dtype = ports.get(spec.port)
+        if dtype is None:
+            raise ValueError(
+                f"'{spec.port}' is not a port of {node}. Its ports are "
+                f"{', '.join(ports)}."
+            )
+        if spec.dtype is not dtype:
+            raise ValueError(f"{node}'s '{spec.port}' is {dtype.value}.")
+
+
+def _is_pcm(audio: dict[str, Any]) -> bool:
+    """A microphone chunk rather than a whole clip."""
+    media_type = str(audio.get("media_type") or "").lower()
+    return media_type.startswith(("audio/l16", "audio/pcm"))
+
+
+def sentences(text: str) -> list[str]:
+    """``text`` cut after each ``.``, ``!`` and ``?``, one call to speak each."""
+    return [s for s in (p.strip() for p in re.split(r"(?<=[.!?])\s+", text)) if s]
+
+
+class _Speech(Node):
+    """What both nodes share: the url, the artifact store and their health."""
+
+    __slots__ = ("_artifacts", "_loop")
+
+    def __init__(
+        self,
+        f: Any,
+        requires: Iterable[MessageSpec],
+        provides: Iterable[MessageSpec],
+        params: dict[str, Any] | None,
+        name: str,
+        inputs: dict[str, DType],
+        outputs: dict[str, DType],
+    ):
+        # Set on start; a test sets them directly.
+        self._artifacts: Any = None
+        self._loop: asyncio.AbstractEventLoop | None = None
+        super().__init__(
+            f, requires=requires, provides=provides, params=params, name=name
+        )
+        _check(self.input_ports, inputs, name)
+        _check(self.output_ports, outputs, name)
+        if not any(spec.port == "url" for spec in self.input_ports):
+            raise ValueError(
+                f"{name} needs a 'url' input: the speech server's address."
+            )
+
+    async def start(self, app: FastAPI | None = None) -> None:
+        self._loop = asyncio.get_running_loop()
+        self._artifacts = getattr(app.state, "artifact_store", None) if app else None
+
+    def _reachable(self, url: str) -> bool:
+        """Whether there is a server to ask, said as the node's health."""
+        if not url:
+            if self._health != "degraded":
+                self.report_health("degraded", "no speech server: its url is empty")
+            return False
+        if self._health != "ok":
+            self.report_health("ok")
+        return True
+
+    def _load(self, reference: dict[str, Any]) -> bytes:
+        return b"".join(self._artifacts.read(str(reference["digest"])))
+
+    def _say(self, **values: Any) -> None:
+        """Publish on the ports that are wired."""
+        wired = {spec.port for spec in self.output_ports if spec.name}
+        self.emit({port: value for port, value in values.items() if port in wired})
+
+
+class SttNode(_Speech):
+    """Speech to text: a microphone's audio in, what was said out.
+
+    Inputs: ``audio``, a stream of 16 kHz PCM16 chunks or a whole clip,
+    ``listening``, true while the device listens, and ``url``.
+    Outputs: ``partial``, the words agreed on so far, and ``text``.
+
+    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.
+    """
+
+    __slots__ = ("_heard", "_listening", "_lock", "_session", "_tasks")
+
+    def __init__(
+        self,
+        requires: MessageSpec | Iterable[MessageSpec] = (),
+        provides: MessageSpec | Iterable[MessageSpec] = (),
+        params: dict[str, Any] | None = None,
+        name: str | None = None,
+    ):
+        self._lock = threading.Lock()
+        #: The chunks of the utterance being spoken, None between utterances.
+        self._session: asyncio.Queue[bytes | None] | None = None
+        #: The digest last heard, and whether the device was listening.
+        self._heard = ""
+        self._listening: bool | None = None
+        self._tasks: set[Future[None]] = set()
+        inputs = [
+            # The url is read, not waited for: its changing is nothing to hear.
+            spec.model_copy(update={"trigger": spec.port != "url"})
+            for spec in self._normalize_ports(requires)
+        ]
+        if not any(spec.port == "audio" for spec in inputs):
+            raise ValueError("stt needs an 'audio' input: the microphone.")
+        super().__init__(
+            self._hear,
+            inputs,
+            self._normalize_ports(provides),
+            params,
+            name or "stt",
+            {"audio": DType.AUDIO, "listening": DType.BOOL, "url": DType.STR},
+            {"partial": DType.STR, "text": DType.STR},
+        )
+
+    async def start(self, app: FastAPI | None = None) -> None:
+        await super().start(app)
+        if self._pipeline is None:
+            return
+        # What the inputs hold already was heard before this start.
+        held = await asyncio.to_thread(
+            self._pipeline.state.get_present, list(self.requires)
+        )
+        for spec in self.input_ports:
+            value = held.get(spec.name)
+            if spec.port == "audio" and isinstance(value, dict):
+                self._heard = str(value.get("digest") or "")
+            elif spec.port == "listening" and isinstance(value, bool):
+                self._listening = value
+
+    async def stop(self, app: FastAPI | None = None) -> None:
+        for task in list(self._tasks):
+            task.cancel()
+        self._session = None
+
+    def _hear(
+        self,
+        params: dict[str, Any],
+        audio: dict[str, Any] | None = None,
+        listening: bool | None = None,
+        url: str = "",
+    ) -> None:
+        with self._lock:
+            digest = str((audio or {}).get("digest") or "")
+            fresh = bool(digest) and digest != self._heard
+            self._heard = digest or self._heard
+            was, self._listening = self._listening, listening
+            if not self._reachable(url):
+                self._session = None
+                return None
+            if fresh and audio is not None and not _is_pcm(audio):
+                self._spawn(self._clip(url, audio, self._load(audio)))
+            elif fresh and audio is not None:
+                if self._session is None:
+                    self._session = self._open(url)
+                self._feed(self._session, self._load(audio))
+            elif listening and not was and self._session is None:
+                self._session = self._open(url)
+            elif was and not listening and self._session is not None:
+                self._feed(self._session, None)
+                self._session = None
+        return None
+
+    def _spawn(self, work: Coroutine[Any, Any, None]) -> None:
+        """Run on the engine's loop, so this run can return at once."""
+        if self._loop is None:
+            work.close()
+            raise RuntimeError("stt only hears in a live flow")
+        task = asyncio.run_coroutine_threadsafe(work, self._loop)
+        self._tasks.add(task)
+        task.add_done_callback(self._tasks.discard)
+
+    def _open(self, url: str) -> asyncio.Queue[bytes | None]:
+        chunks: asyncio.Queue[bytes | None] = asyncio.Queue()
+        self._spawn(self._transcribe(url, chunks))
+        return chunks
+
+    def _feed(self, chunks: asyncio.Queue[bytes | None], chunk: bytes | None) -> None:
+        assert self._loop is not None
+        self._loop.call_soon_threadsafe(chunks.put_nowait, chunk)
+
+    async def _transcribe(self, url: str, chunks: asyncio.Queue[bytes | None]) -> None:
+        words = ""
+        try:
+            async with asyncio.timeout(UTTERANCE_S):
+                async for final, text in client.transcribe_stream(
+                    url + "/v1", chunks, rate=RATE, idle_s=IDLE_S
+                ):
+                    words = text.strip() if final else words + text
+                    if final and words:
+                        await asyncio.to_thread(self._say, partial=words, text=words)
+                    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
+            )
+        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.
+            with self._lock:
+                if self._session is chunks:
+                    self._session = None
+
+    async def _clip(self, url: str, audio: dict[str, Any], data: bytes) -> None:
+        try:
+            text = await client.transcribe(
+                url + "/v1", data, "speech.wav", str(audio.get("media_type"))
+            )
+        except Exception as exc:
+            logger.warning("'%s' lost a clip: %s: %s", self.id, type(exc).__name__, exc)
+            return
+        if text.strip():
+            await asyncio.to_thread(self._say, partial=text.strip(), text=text.strip())
+
+
+class TtsNode(_Speech):
+    """Text to speech: each sentence of ``text`` spoken, out on ``speech``.
+
+    One call to the server per sentence, each sentence published as an
+    ``audio`` reference as soon as it is spoken, so a reply plays before its
+    last sentence is. A sentence spoken before in the same voice is not spoken
+    again.
+
+    A reference is marked ``"append": true`` while the ones before it are
+    still playing, which queues it on the device. The first sentence of a
+    reply carries no mark and cuts off whatever was playing, however its
+    sentences arrive: as one text, or one value each while they are written.
+    """
+
+    class Params(BaseModel):
+        model_config = ConfigDict(extra="allow")
+
+        voice: str = Field(
+            default="af_heart",
+            description="The voice, one of Kokoro's such as af_heart or bm_george.",
+        )
+
+    __slots__ = ("cfg", "_lock", "_until")
+
+    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 {})
+        # One reply at a time, so its sentences go out in order.
+        self._lock = threading.Lock()
+        #: When what was published so far stops playing, on the monotonic clock.
+        self._until = 0.0
+        inputs = [
+            spec.model_copy(update={"trigger": spec.port != "url"})
+            for spec in self._normalize_ports(requires)
+        ]
+        if not any(spec.port == "text" for spec in inputs):
+            raise ValueError("tts needs a 'text' input: what to say.")
+        super().__init__(
+            self._speak,
+            inputs,
+            self._normalize_ports(provides),
+            params,
+            name or "tts",
+            {"text": DType.STR, "url": DType.STR},
+            {"speech": DType.AUDIO},
+        )
+
+    def _speak(self, params: dict[str, Any], text: str = "", url: str = "") -> None:
+        if not text.strip() or not self._reachable(url):
+            return None
+        with self._lock:
+            for sentence in sentences(text):
+                reference = self._spoken(url, sentence)
+                now = time.monotonic()
+                if now < self._until + PLAYING_SLACK_S:
+                    reference = {**reference, "append": True}
+                self._until = max(now, self._until) + reference["meta"]["seconds"]
+                self._say(speech=reference)
+        return None
+
+    def _spoken(self, url: str, sentence: str) -> dict[str, Any]:
+        """The reference to ``sentence`` spoken, from the cache or the server."""
+        key = hashlib.sha256(f"{self.cfg.voice}\n{sentence}".encode()).hexdigest()
+        # Kept in state, so the sweep keeps the bytes it refers to.
+        spoken = dict(self.recall("spoken") or {})
+        reference = spoken.get(key)
+        if reference is not None and self._artifacts.has(reference["digest"]):
+            return dict(reference)
+        data = asyncio.run(client.speak(url + "/v1", sentence, self.cfg.voice))
+        with wave.open(io.BytesIO(data)) as clip:
+            seconds = round(clip.getnframes() / clip.getframerate(), 3)
+        stored: dict[str, Any] = self._artifacts.put(
+            [data], name="speech.wav", media_type="audio/wav"
+        )
+        reference = stored | {"meta": {"seconds": seconds}}
+        spoken.pop(key, None)
+        spoken[key] = reference
+        self.remember("spoken", dict(list(spoken.items())[-SPOKEN:]))
+        return reference

backend/fluksio/flow/nodes/wren.py

@@ -30,7 +30,11 @@ def _wren() -> Any:
     try:
         import wren
     except ImportError:
-        raise RuntimeError("the wren node needs the model: pip install wren") from None
+        # Not `pip install wren`: that name on PyPI is another package.
+        raise RuntimeError(
+            "the wren node needs the model: pip install git+https://github.com/"
+            "stroblme/wren@eae4d6ff9ded68e36624972fd3f4a8f8d210b0c2"
+        ) from None
     return wren
 
 
@@ -62,8 +66,11 @@ INPUTS = {
     "dark": {DType.BOOL},
 }
 
-#: The output the calls are reported on. Every other output is a device.
+#: The output the calls are reported on, and the two that say what came of
+#: the command. Every other output is a device.
 ACTIONS = "actions"
+REPLY = "reply"
+UNHANDLED = "unhandled"
 
 _TARGET_TYPES = {DType.BOOL, DType.INT, DType.FLOAT, DType.STR}
 
@@ -93,6 +100,21 @@ _VALUES: dict[str, tuple[bool | None, float | None, str | None]] = {
     "watering.stop": (False, None, "off"),
 }
 
+#: An applied call, said: "kitchen light on". `heating.set` says its number.
+_SAID = {
+    "lights.on": "light on",
+    "lights.off": "light off",
+    "media.play": "media playing",
+    "media.stop": "media stopped",
+    "heating.boost": "heating boosted",
+    "heating.off": "heating off",
+    "shutters.up": "shutters up",
+    "shutters.down": "shutters down",
+    "shutters.stop": "shutters stopped",
+    "watering.start": "watering on",
+    "watering.stop": "watering off",
+}
+
 #: What every area held in training: a light, and in the living room, which
 #: always existed, media and heating besides. A house missing them reads as
 #: unfamiliar and the confidence drops with it, so they are filled in; a call
@@ -150,6 +172,18 @@ def _value(call: Any, dtype: DType) -> Any:
     return int(number) if dtype is DType.INT else float(number)
 
 
+def _said(call: Any, area: str) -> str:
+    """An applied call as the reply says it: "garage light on in 5 minutes"."""
+    if call.tool == "heating.set":
+        said = f"heating to {_value(call, DType.FLOAT):g} degrees"
+    else:
+        said = _SAID.get(call.tool, call.tool)
+    if call.delay_s:
+        minutes = round(call.delay_s / 60)
+        said += f" in {minutes} minute{'s' * (minutes != 1)}"
+    return f"{area} {said}"
+
+
 def _text(call: Any) -> str:
     """The call as the chat shows it, less the device it names."""
     args = ", ".join(f"{k}={v}" for k, v in call.args.items() if k != "id")
@@ -163,9 +197,11 @@ class WrenNode(Node):
     ``presence`` (the area the user is in), ``outdoor`` (°C) and ``dark``
     (otherwise read off the clock: before 07:00 or from 18:00).
 
-    Outputs: ``actions``, a list of records, one per call and target, 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.
+    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
+    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
     node's second value.
     """
@@ -224,6 +260,10 @@ class WrenNode(Node):
                 if spec.dtype is not DType.LIST or spec.item_dtype is not DType.RECORD:
                     raise ValueError(f"'{ACTIONS}' is a list of records.")
                 continue
+            if spec.port in (REPLY, UNHANDLED):
+                if spec.dtype is not DType.STR:
+                    raise ValueError(f"'{spec.port}' is a str.")
+                continue
             device = _device(spec.port, areas)
             if device is None:
                 raise ValueError(
@@ -316,6 +356,7 @@ class WrenNode(Node):
         write = self.cfg.apply and confidence >= self.cfg.min_confidence
         writes: dict[str, Any] = {}
         actions: list[dict[str, Any]] = []
+        said: list[str] = []
         for call in calls:
             area, _, kind = str(call.args.get("id", "")).partition(":")
             ports = self.devices.get((area, kind), [])
@@ -336,6 +377,7 @@ class WrenNode(Node):
             }
             if not hits:
                 actions.append(record)
+            done = False
             for spec, value in hits:
                 applied = write
                 if applied and call.delay_s:
@@ -347,7 +389,16 @@ class WrenNode(Node):
                 actions.append(
                     record | {"message": spec.name, "value": value, "applied": applied}
                 )
+                done = done or applied
+            if done:
+                said.append(_said(call, area))
 
-        if any(spec.port == ACTIONS for spec in self.output_ports):
+        wired = {spec.port for spec in self.output_ports}
+        if ACTIONS in wired:
             writes[ACTIONS] = actions
+        if said and REPLY in wired:
+            reply = ", ".join(said)
+            writes[REPLY] = reply[0].upper() + reply[1:] + "."
+        if not said and UNHANDLED in wired:
+            writes[UNHANDLED] = prompt
         return writes or None

backend/fluksio/flow/pipeline.py

@@ -1717,7 +1717,12 @@ class Pipeline:
                 targets.add(consumer)
                 targets.update(self._get_downstream(consumer))
         if targets:
-            self._execute_parallel(targets, self._state, check_ready=True)
+            # Only what these values reach, as in a cascade: a node further
+            # down whose producer published nothing would otherwise run again
+            # on what it already read, once per microphone chunk.
+            self._execute_parallel(
+                targets, self._state, check_ready=True, changed=set(values)
+            )
 
     def defer(
         self,

backend/tests/flow/test_node_types.py

@@ -105,6 +105,22 @@ FIXTURES: dict[str, dict] = {
         "requires": [MessageSpec(name="prompt", dtype=DType.STR)],
         "provides": [MessageSpec(name="kitchen_light", dtype=DType.BOOL)],
     },
+    "stt": {
+        "params": {},
+        "requires": [
+            MessageSpec(name="mic", port="audio", dtype=DType.AUDIO),
+            MessageSpec(name="url", dtype=DType.STR),
+        ],
+        "provides": [MessageSpec(name="text", dtype=DType.STR)],
+    },
+    "tts": {
+        "params": {"voice": "bm_george"},
+        "requires": [
+            MessageSpec(name="reply", port="text", dtype=DType.STR),
+            MessageSpec(name="url", dtype=DType.STR),
+        ],
+        "provides": [MessageSpec(name="speech", dtype=DType.AUDIO)],
+    },
     "device": {
         "params": {"device": "hall", "config": {"gpio.4": {"pull": "up"}}},
         "requires": [

backend/tests/flow/test_speech.py

@@ -0,0 +1,147 @@
+"""The speech nodes: one session per utterance, and each sentence spoken once."""
+
+import asyncio
+import hashlib
+import io
+import wave
+from typing import Any
+
+from fluksio.ai import client
+from fluksio.flow.messages import DType, MessageSpec
+from fluksio.flow.nodes.speech import SttNode, TtsNode, sentences
+from fluksio.flow.pipeline import Pipeline
+
+URL = "http://speech:8100"
+
+
+class Store:
+    """The artifact store's surface the nodes use."""
+
+    def __init__(self) -> None:
+        self.data: dict[str, bytes] = {}
+
+    def put(self, chunks: list[bytes], name: str, media_type: str) -> dict[str, Any]:
+        data = b"".join(chunks)
+        digest = "sha256:" + hashlib.sha256(data).hexdigest()
+        self.data[digest] = data
+        return {"digest": digest, "size": len(data), "media_type": media_type}
+
+    def has(self, digest: str) -> bool:
+        return digest in self.data
+
+    def read(self, digest: str) -> list[bytes]:
+        return [self.data[digest]]
+
+
+def test_sentences_are_cut_after_their_ending():
+    assert sentences(" Kitchen light on.  Heating off! Is it?\nYes") == [
+        "Kitchen light on.",
+        "Heating off!",
+        "Is it?",
+        "Yes",
+    ]
+
+
+def test_an_utterance_is_one_session_committed_when_listening_ends(monkeypatch):
+    heard: list[bytes] = []
+
+    async def session(api_base, audio, *, rate, idle_s):
+        assert (api_base, rate) == (f"{URL}/v1", 16000)
+        while (chunk := await audio.get()) is not None:
+            heard.append(chunk)
+        yield False, "turn on"
+        yield False, " the"
+        yield True, " Turn on the light."
+
+    monkeypatch.setattr(client, "transcribe_stream", session)
+
+    async def scenario() -> None:
+        node = SttNode(
+            requires=[
+                MessageSpec(name="mic", port="audio", dtype=DType.AUDIO),
+                MessageSpec(name="listening", dtype=DType.BOOL),
+                MessageSpec(name="url", dtype=DType.STR),
+            ],
+            provides=[
+                MessageSpec(name="partial", dtype=DType.STR),
+                MessageSpec(name="text", dtype=DType.STR),
+            ],
+        )
+        node.assign_flow("voice", "stt")
+        pipeline = Pipeline([node])
+        store = Store()
+        chunks = [
+            store.put([bytes([n]) * 3200], "voice.mic", "audio/L16;rate=16000")
+            for n in range(3)
+        ]
+        # Left over from before the engine started: not part of what is said.
+        pipeline.state["voice.mic"] = chunks[0]
+        await node.start()
+        node._artifacts = store
+
+        async def hear(audio: dict, listening: bool, url: str = URL) -> None:
+            inputs = {"voice.mic": audio, "voice.listening": listening}
+            await asyncio.to_thread(node.execute, inputs | {"voice.url": url})
+
+        await hear(chunks[0], True)
+        await hear(chunks[1], True)
+        await hear(chunks[2], True)
+        await hear(chunks[2], False)
+        for _ in range(200):
+            if "voice.text" in pipeline.state:
+                break
+            await asyncio.sleep(0.01)
+
+        assert heard == [bytes([1]) * 3200, bytes([2]) * 3200]
+        assert pipeline.state["voice.text"] == "Turn on the light."
+        assert pipeline.state["voice.partial"] == "Turn on the light."
+        assert node._session is None
+
+        # No server, no session: the node says so instead.
+        await hear(chunks[1], True, url="")
+        assert node._session is None
+        assert node._health == "degraded"
+
+    asyncio.run(scenario())
+
+
+def _wav(seconds: float) -> bytes:
+    out = io.BytesIO()
+    with wave.open(out, "wb") as clip:
+        clip.setnchannels(1)
+        clip.setsampwidth(2)
+        clip.setframerate(24000)
+        clip.writeframes(b"\0\0" * int(24000 * seconds))
+    return out.getvalue()
+
+
+def test_a_sentence_is_spoken_once_and_the_rest_of_a_reply_queues(monkeypatch):
+    spoken: list[str] = []
+    published: list[dict] = []
+
+    async def speak(api_base, text, voice):
+        spoken.append(text)
+        return _wav(len(spoken))
+
+    monkeypatch.setattr(client, "speak", speak)
+    monkeypatch.setattr(TtsNode, "_say", lambda self, speech: published.append(speech))
+    node = TtsNode(
+        requires=[
+            MessageSpec(name="reply", port="text", dtype=DType.STR),
+            MessageSpec(name="url", dtype=DType.STR),
+        ],
+        provides=[MessageSpec(name="speech", dtype=DType.AUDIO)],
+    )
+    node.assign_flow("voice", "tts")
+    Pipeline([node])
+    node._artifacts = Store()
+
+    node.execute({"voice.reply": "Kitchen light on. Heating off.", "voice.url": URL})
+    # Long after both stopped playing.
+    node._until = -10.0
+    node.execute({"voice.reply": "Kitchen light on.", "voice.url": URL})
+
+    assert spoken == ["Kitchen light on.", "Heating off."]
+    assert ["append" in speech for speech in published] == [False, True, False]
+    assert published[2] == published[0]
+    assert published[1]["meta"] == {"seconds": 2.0}

backend/tests/flow/test_wren.py

@@ -165,3 +165,27 @@ def test_the_model_runs_on_one_blas_thread(fake_wren, monkeypatch):
 
     assert threads
     assert set(threads) == {1}
+
+
+def test_a_reply_says_what_was_applied_and_the_rest_is_unhandled(fake_wren):
+    node = WrenNode(
+        requires=[MessageSpec(name="command", port="prompt", dtype=DType.STR)],
+        provides=[
+            MessageSpec(name="kitchen_light", dtype=DType.BOOL),
+            MessageSpec(name="reply", dtype=DType.STR),
+            MessageSpec(name="unhandled", dtype=DType.STR),
+        ],
+        params={"apply": True},
+    )
+    node.assign_flow("house", "control")
+    fake_wren.answer = ([FakeCall("lights.on", {"id": "kitchen:light"})], 0.99)
+
+    out = node.execute({"house.command": "turn on the kitchen light"})
+
+    assert out == {"house.kitchen_light": True, "house.reply": "Kitchen light on."}
+
+    # Nothing wired for the call, or nothing to call at all.
+    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"}

docs/code/api.md

@@ -286,8 +286,12 @@ that family, and the whole message at most `MAX_FRAME_BYTES`. The frame is
 held in memory and published in one step, named after the message and credited
 to its device node. A frame that does not
 qualify is dropped, the reason logged once per socket, and the socket stays
-open. With `http` it uploads with `PUT /artifacts?volatile=true&name=<message>`
-and publishes with `POST /messages/<message>`. A device that is unpaired or
+open. A value of one of its `in` ports wired as `bool`, `int`, `float` or `str`
+goes as the text message `{"type": "publish", "name": "<message>", "value":
+...}`, taken in turn with the frames, so a microphone's `listening` false
+arrives after its last chunk. With `http` it uploads with
+`PUT /artifacts?volatile=true&name=<message>` and publishes with
+`POST /messages/<message>`. A device that is unpaired or
 deleted while connected is closed with 1008, at the latest with its next frame.
 
 ### Live view

docs/code/nodes.md

@@ -179,7 +179,10 @@ and ARM, so nothing has to be installed into the image.
 
 openpanel publishes a 16 kHz mono WAV on its `mic` port after its wake word,
 and plays any `audio` reference sent to its `speaker` port (WAV, MP3 or AAC).
-Two Function nodes close that loop with local models.
+The [Speech to text](../reference/node-types.md#speech-to-text) and
+[Text to speech](../reference/node-types.md#text-to-speech) nodes close that
+loop through a speech server on a GPU. Without one, two Function nodes close it
+with local models.
 [faster-whisper](https://github.com/SYSTRAN/faster-whisper) turns speech into
 text and [piper](https://github.com/OHF-Voice/piper1-gpl) turns text into
 speech.

docs/interface/dashboards.md

@@ -514,8 +514,10 @@ to 640×360. Re-encode other clips with the [clip recipe](../code/nodes.md#makin
 The device decides what its [port](#the-devices-own-ports) names
 mean. openpanel [reports](#what-a-device-reports) the ones it serves, such as
 `gpio.<n>`, `backlight`, `touch`, `mic` and `speaker`, with the settings each
-one reads, and ignores any other name. A browser screen serves none of them. The [speech recipe](../code/nodes.md#speech-to-and-from-a-panel)
-turns `mic` into text and text into `speaker`.
+one reads, and ignores any other name. A browser screen serves none of them. The
+[Speech to text](../reference/node-types.md#speech-to-text) and
+[Text to speech](../reference/node-types.md#text-to-speech) nodes turn `mic`
+into text and text into `speaker`, and `caption` shows what is being heard.
 
 ### The device's own ports
 

docs/reference/node-types.md

@@ -344,7 +344,9 @@ use a [batch flow](../concepts/runs.md) and a Function node.
 
 [Wren](https://github.com/stroblme/wren) is an optional local language model
 for device commands. Install it in the engine's environment with
-`pip install wren`; the node reports a missing dependency if it is absent.
+`pip install git+https://github.com/stroblme/wren@eae4d6ff9ded68e36624972fd3f4a8f8d210b0c2`.
+The `wren` package on PyPI is an unrelated one. Without the model the node
+reports a missing dependency.
 
 | Input | Type | Notes |
 |---|---|---|
@@ -353,7 +355,8 @@ for device commands. Install it in the engine's environment with
 | `outdoor` | `float` or `int` | °C; 20 without it, clamped to −5…30 |
 | `dark` | `bool` | without it, dark before 07:00 and from 18:00 |
 
-The outputs are `actions`, a list of records, and any number of **targets**. A
+The outputs are `actions`, a list of records, `reply`, `unhandled`, and any
+number of **targets**. A
 target is an output named `<area>_<kind>` and bound to the message its device
 listens on: `living_room_setpoint` bound to `home.setpoint` is the living
 room's heating. The wiring is the allowlist, so the model can write nothing
@@ -393,6 +396,10 @@ Each record in `actions` contains `message`, `value`, `applied`, `delay_s`,
 `call`, and `confidence`. Unwired calls have an empty `message` and are not
 applied. Keep `apply=false` while checking how the model interprets commands.
 
+`reply` is one sentence saying what was applied, such as "Kitchen light on."
+or "Garage light on in 5 minutes.". When nothing was applied, `unhandled`
+carries the prompt instead, for something else to answer.
+
 A call with a delay ("in 10m") is held by the engine's work queue, like the
 Debounce node's second value: it survives a restart where the queue is in
 Redis, and not otherwise.
@@ -408,6 +415,57 @@ They are reported but cannot receive writes unless wired.
 
 The chat's [Control tab](../code/agents.md#control) sends commands by hand.
 
+### Speech to text
+
+**`stt`** — turn a microphone's audio into text, through a speech server.
+
+| Input | Type | Notes |
+|---|---|---|
+| `audio` | `audio` | 16 kHz mono PCM16 chunks (`audio/L16`), or a whole clip such as a WAV |
+| `listening` | `bool` | true while the device listens; false ends the utterance |
+| `url` | `str` | the speech server, from a [Service](#service) node; read, never waited for |
+
+| Output | Type | Notes |
+|---|---|---|
+| `partial` | `str` | the words agreed on so far, as they come, then the final text |
+| `text` | `str` | what was said, once per utterance |
+
+The server is any that speaks OpenAI's audio API, such as `parrot serve`. Each
+utterance is one realtime transcription session (`{url}/v1/realtime`), opened
+when `listening` turns true or at the first chunk and committed when it turns
+false. A clip is sent whole to `{url}/v1/audio/transcriptions`. The node
+returns at once and publishes when the server answers, since a device's next
+chunk waits for it. An utterance that fails, or gets no chunk for 10 seconds,
+publishes nothing and is logged.
+
+`listening` has to hold a value before a chunk wakes the node. Declare it as a
+flow input with the initial value `false` when a device may never send it,
+such as one that sends whole clips.
+
+### Text to speech
+
+**`tts`** — speak text as audio for a speaker, through a speech server.
+
+| Input | Type | Notes |
+|---|---|---|
+| `text` | `str` | what to say, a whole reply or one sentence at a time |
+| `url` | `str` | the speech server, as for Speech to text |
+
+| Output | Type | Notes |
+|---|---|---|
+| `speech` | `audio` | one WAV reference per sentence |
+
+| Setting | Default | Notes |
+|---|---|---|
+| `voice` | `af_heart` | one of Kokoro's voices, such as `bm_george` |
+
+Each sentence is one call to `{url}/v1/audio/speech` and published as soon as
+it is spoken. A sentence spoken before in the same voice is reused rather than
+spoken again, for the last 256 sentences. A reference carries `"append": true`
+while the sentences before it are still playing, which queues it on a panel's
+`speaker`. The first sentence of a reply carries none and cuts off whatever
+was playing.
+
 ## Devices
 
 ### Device

frontend/src/components/Flow/FlowNode.tsx

@@ -12,6 +12,7 @@ import {
   Filter,
   Globe,
   Merge,
+  Mic,
   MonitorSmartphone,
   Play,
   Plug,
@@ -22,6 +23,7 @@ import {
   Square,
   Terminal,
   Timer,
+  Volume2,
 } from "lucide-react"
 import { memo, useEffect, useRef, useState } from "react"
 
@@ -63,6 +65,8 @@ const NODE_ICONS = {
   file: FileText,
   ntfy: Bell,
   wren: Bird,
+  stt: Mic,
+  tts: Volume2,
   // The glyph the Devices tab goes by.
   device: MonitorSmartphone,
 } as const