DRU-651 -- Turn chat voice notes into text for the agent

Author: czpython

backend/druks/chat/constants.py

@@ -18,3 +18,9 @@
     "[Internal: Run {run} failed: {failure}. Tell the person in one short line, with no "
     "error details, and do not retry it.]"
 )
+# The line between what a person typed and what they said in the same message.
+VOICE_NOTE_MARKER = "[Voice note]"
+TRANSCRIPTION_FAILED_MESSAGE = (
+    "[Internal: Druks could not turn the person's voice note into text. Tell the person "
+    "in one short line to write it instead.]"
+)

backend/druks/chat/models.py

@@ -37,6 +37,8 @@ class Message(Base, Uuid7Pk):
     # reply, so the source's copy of it is known as Druks's own.
     source_id: Mapped[str | None] = mapped_column(unique=True)
     file: Mapped[File | None] = FileField()
+    # What the person said in the message's voice note.
+    transcript: Mapped[str] = mapped_column(default="", server_default=text("''"))
     created_at: Mapped[datetime] = mapped_column(default=Base.utc_now)
     delivered_at: Mapped[datetime | None]
 

backend/druks/chat/schemas.py

@@ -34,6 +34,7 @@ class MessageResponse(Schema):
     tool_calls: list[dict]
     is_internal: bool
     file: FileSummary | None
+    transcript: str
     created_at: datetime
     delivered_at: datetime | None
 

backend/druks/chat/service.py

@@ -15,6 +15,8 @@
 from druks.accounts.enums import AccountKind
 from druks.apps.loader import get_app
 from druks.apps.registry import channels
+from druks.core.exceptions import TranscriptionError
+from druks.core.services import SpeechToText
 from druks.durable.engine import step_session
 from druks.durable.models import Run
 from druks.files.constants import MAX_UPLOAD_BYTES
@@ -38,6 +40,7 @@
 from druks.sandbox.layout import get_remote_home, get_work_root
 from druks.sandbox.models import SandboxIdentity, SecretRef
 from druks.sandbox.templates import get_template_id
+from druks.services.exceptions import ServiceNotConnectedError
 from druks.workspaces import Workspace
 
 from .bots.constants import ADMIN_PROMPT, ADMIN_TOOLS
@@ -48,6 +51,8 @@
     FAILURE_MESSAGE,
     INTERNAL_MESSAGES_PROMPT,
     RESULT_MESSAGE,
+    TRANSCRIPTION_FAILED_MESSAGE,
+    VOICE_NOTE_MARKER,
 )
 from .enums import MessageRole, MessageState
 from .exceptions import ChatBridgeError, ChatHarnessError, ChatSandboxGone
@@ -235,8 +240,8 @@ async def send_turn(
     config: AgentConfig,
     prompt: str,
 ) -> Message | None:
-    """Start the agent and send the pending messages.
-    Return the turn's message, unless a Stop or pause came first."""
+    """Start the agent, transcribe the pending voice notes, and send the pending
+    messages. Return the turn's message, unless a Stop or pause came first."""
     host = bridge.host
     status = await bridge.request("status", conversationId=conversation.id)
     if status["status"] == "running":
@@ -274,13 +279,32 @@ async def send_turn(
     expires_at = Base.utc_now() + timedelta(seconds=SANDBOX_HOST_LEASE_SECONDS)
     await sandbox_client.set_expiry(host_id=host.id, expires_at=expires_at)
     identity.expires_at = expires_at
+    # The account's other conversations write this row too. Release it before the
+    # transcription calls.
+    await session.commit()
     messages = [message]
     if conversation.connection:
         # The sandbox can take seconds to start, and a person can take the chat over meanwhile.
         if await conversation.is_held(session):
             return
         messages = await conversation.list_pending_messages(session)
-    delivered_messages = [pending for pending in messages if await pending.mark_delivered(session)]
+    notes = []
+    for pending in messages:
+        file = pending.file
+        if file and file.content_type.startswith("audio/"):
+            try:
+                pending.transcript = await get_transcript(session, file)
+            except (TranscriptionError, ServiceNotConnectedError) as error:
+                logger.warning("Chat message %s has no transcript: %s", pending.id, error)
+                notes.append(
+                    await conversation.create_message(
+                        session, TRANSCRIPTION_FAILED_MESSAGE, is_internal=True
+                    )
+                )
+    # The notes are the newest messages, so they close the turn.
+    delivered_messages = [
+        pending for pending in (*messages, *notes) if await pending.mark_delivered(session)
+    ]
     await session.commit()
     if delivered_messages:
         await bridge.request(
@@ -295,16 +319,33 @@ async def send_turn(
     return
 
 
+async def get_transcript(session: AsyncSession, file: File) -> str:
+    """The words in a voice note, from the Speech To Text card. Druks refuses a note
+    over the upload cap before the call."""
+    if file.size > MAX_UPLOAD_BYTES:
+        raise TranscriptionError(
+            f"The voice note is {file.size} bytes. The cap is {MAX_UPLOAD_BYTES} bytes."
+        )
+    content = await asyncio.to_thread(get_file_storage().open, file.id)
+    return await SpeechToText.transcribe(
+        session, name=file.name, content_type=file.content_type, content=content
+    )
+
+
 async def get_turn_content(
     host: Host, conversation_root: str, messages: list[Message]
 ) -> list[dict]:
-    """The ACP content blocks the agent reads: each message's text, then its file. An
-    image travels in the prompt. Audio adds nothing. Any other file goes to the
-    conversation's folder in the sandbox, and the agent gets a link to it."""
+    """The ACP content blocks the agent reads: each message's text, with the words of
+    its voice note under a marker, then its file. An image travels in the prompt. Audio
+    adds nothing more. Any other file goes to the conversation's folder in the sandbox,
+    and the agent gets a link to it."""
     content = []
     for message in messages:
-        if message.body:
-            content.append({"type": "text", "text": message.body})
+        parts = [message.body]
+        if message.transcript:
+            parts += [VOICE_NOTE_MARKER, message.transcript]
+        if text := "\n".join(part for part in parts if part):
+            content.append({"type": "text", "text": text})
         if file := message.file:
             if file.content_type.startswith("image/") and file.size <= MAX_UPLOAD_BYTES:
                 image = await asyncio.to_thread(get_file_storage().open, file.id)

backend/druks/core/exceptions.py

@@ -0,0 +1,5 @@
+from druks.exceptions import DruksError
+
+
+class TranscriptionError(DruksError):
+    """Druks got no transcript for a voice note."""

backend/druks/core/services.py

@@ -6,17 +6,23 @@
 from githubkit import GitHub
 from pydantic import BaseModel, Field, SecretStr
 from slack_sdk.errors import SlackApiError
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from druks.core.apis.github import GITHUB_AUTHORITY, GitHubClient
 from druks.core.apis.linear import LINEAR_GRAPHQL_URL
 from druks.core.apis.slack import SLACK_AUTHORITY, SLACK_BOT_SCOPES, SlackClient
+from druks.core.exceptions import TranscriptionError
+from druks.secrets.datastructures import Audience
 from druks.secrets.enums import SecretKind
+from druks.secrets.models import VaultSecret
 from druks.services import Service, ServiceConnectError
+from druks.services.exceptions import ServiceNotConnectedError
 from druks.settings import load_settings
 
 logger = logging.getLogger(__name__)
 
 _VERIFY_TIMEOUT = 10.0
+_TRANSCRIBE_TIMEOUT = 120.0
 
 
 class Github(Service):
@@ -265,3 +271,45 @@ def get_manifest(cls) -> dict[str, Any]:
                 "token_rotation_enabled": False,
             },
         }
+
+
+class SpeechToText(Service):
+    """The server Druks sends voice notes to: any server that speaks the OpenAI audio
+    API, such as OpenAI, Groq, or a local one."""
+
+    description = (
+        "The service Druks sends voice notes to. Any server that speaks the OpenAI audio API."
+    )
+    required = False
+
+    class Settings(BaseModel):
+        url: str = Field(
+            title="Address", description="Base URL, for example https://api.openai.com/v1."
+        )
+        key: SecretStr = Field(title="Key")
+        model: str = Field(title="Model", description="For example whisper-1.")
+
+    @classmethod
+    async def transcribe(
+        cls, session: AsyncSession, *, name: str, content_type: str, content: bytes
+    ) -> str:
+        """The words in an audio file. An empty answer is a failure."""
+        card = await VaultSecret.lookup(session, cls.secret_kind, Audience.service(cls.slug))
+        if not card:
+            raise ServiceNotConnectedError(cls.slug)
+        try:
+            async with httpx.AsyncClient(timeout=_TRANSCRIBE_TIMEOUT) as client:
+                response = await client.post(
+                    f"{card.identity['url'].rstrip('/')}/audio/transcriptions",
+                    # A key pasted with a space would echo through the transport's error.
+                    headers={"Authorization": f"Bearer {card.secrets['key'].strip()}"},
+                    data={"model": card.identity["model"]},
+                    files={"file": (name, content, content_type)},
+                )
+            response.raise_for_status()
+            text = response.json()["text"].strip()
+        except Exception as error:  # noqa: BLE001 — any transport or shape failure is a failed note
+            raise TranscriptionError(f"{cls.title} gave no transcript: {error}") from error
+        if not text:
+            raise TranscriptionError(f"{cls.title} gave an empty transcript.")
+        return text

backend/migrations/versions/e613b1a81427_chat_message_transcript.py

@@ -0,0 +1,27 @@
+"""Keep the words of a chat message's voice note.
+
+Revision ID: e613b1a81427
+Revises: 2f9d7f55982a
+Create Date: 2026-09-26
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "e613b1a81427"
+down_revision: str | Sequence[str] | None = "2f9d7f55982a"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "chat_messages",
+        sa.Column("transcript", sa.String(), nullable=False, server_default=""),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("chat_messages", "transcript")

backend/tests/test_whatsapp.py

@@ -29,7 +29,12 @@
 from druks.chat.channels.whatsapp.constants import WAHA_AUDIENCE
 from druks.chat.channels.whatsapp.services import Waha
 from druks.chat.channels.whatsapp.webhooks import WahaEvents
-from druks.chat.constants import CONVERSATION_HEADER, INTERNAL_MESSAGES_PROMPT
+from druks.chat.constants import (
+    CONVERSATION_HEADER,
+    INTERNAL_MESSAGES_PROMPT,
+    TRANSCRIPTION_FAILED_MESSAGE,
+    VOICE_NOTE_MARKER,
+)
 from druks.chat.enums import (
     BotAccess,
     ConversationSource,
@@ -38,6 +43,7 @@
     PauseSignal,
 )
 from druks.chat.models import Conversation
+from druks.core.exceptions import TranscriptionError
 from druks.harnesses.claude import ClaudeHarness
 from druks.mcp.enums import Toolkit
 from druks.mcp.server import _is_visible, _validate_agent_tools
@@ -348,13 +354,19 @@ async def test_one_turn_answers_every_pending_message_and_knows_its_own_reply(
     photo["payload"].update(hasMedia=True, media=waha_media("photo.png", "image/png"))
     form = message_event(ANA, "Here is the form", key="M4")
     form["payload"].update(hasMedia=True, media=waha_media("form.pdf", "application/pdf"))
-    note = message_event(ANA, "", key="M5")
+    note = message_event(ANA, "Call me", key="M5")
     note["payload"].update(hasMedia=True, media=waha_media("note.oga", "audio/ogg"))
+    failed_note = message_event(ANA, "", key="M6")
+    failed_note["payload"].update(hasMedia=True, media=waha_media("again.oga", "audio/ogg"))
     await receive(connection, message_event(ANA, "Hello", key="M1"))
     await receive(connection, photo)
     await receive(connection, message_event(ANA, "And the second one?", key="M3"))
     await receive(connection, form)
     await receive(connection, note)
+    await receive(connection, failed_note)
+    await receive(connection, message_event(ANA, "Anyone there?", key="M7"))
+    transcribe = AsyncMock(side_effect=["at six", TranscriptionError("The provider is down.")])
+    monkeypatch.setattr(service.SpeechToText, "transcribe", transcribe)
     [conversation] = await Conversation.list_for_connection(druks_db, connection.id)
     config = SimpleNamespace(
         harness_class=ClaudeHarness,
@@ -418,14 +430,24 @@ async def copy_arrives_first() -> None:
             "name": "form.pdf",
             "mimeType": "application/pdf",
         },
+        {"type": "text", "text": f"Call me\n{VOICE_NOTE_MARKER}\nat six"},
+        {"type": "text", "text": "Anyone there?"},
+        {"type": "text", "text": TRANSCRIPTION_FAILED_MESSAGE},
+    ]
+    assert (prompt["timeout"], prompt["messageId"]) == (60, asked[-1].id)
+    assert [(message.body, message.transcript) for message in asked[4:]] == [
+        ("Call me", "at six"),
+        ("", ""),
+        ("Anyone there?", ""),
+        (TRANSCRIPTION_FAILED_MESSAGE, ""),
     ]
-    assert prompt["timeout"] == 60
+    assert asked[-1].is_internal
     [upload] = host.upload_file.await_args_list
     assert (upload.kwargs["local"].is_file(), upload.kwargs["remote"]) == (True, copy)
     [start] = [values for method, values in requests if method == "start"]
     assert start["headers"] == [{"name": CONVERSATION_HEADER, "value": conversation.id}]
     assert start["meta"]["claudeCode"]["options"]["systemPrompt"] == "Be kind."
-    assert [message.state for message in asked] == [MessageState.REPLIED] * 5
+    assert [message.state for message in asked] == [MessageState.REPLIED] * 8
     assert reply.source_id == "REPLY1"
     sends = [body for method, path, body in waha.calls if path == "/api/sendText"]
     assert sends == [

docs/chat.md

@@ -235,6 +235,15 @@ bulk. Before it sends a reply, Druks takes a new message ID from WAHA and
 records it. WAHA's copy of the sent message then carries a known ID, even when
 the copy arrives before the send returns.
 
+A voice note becomes text before its turn. Druks sends the audio to the
+[Speech To Text card](configuration.md#speech-to-text) and saves the words on
+the message. The agent reads them under what the person typed, marked as a
+voice note, and answers in text. Nothing goes to the person before the reply.
+When Druks cannot transcribe a note, when no card is connected, or when the
+note is over 25 MiB, the agent gets an internal message instead. It then tells
+the person in one line to write instead. The web page shows the typed text, the
+words, and a link to the audio.
+
 Druks also adds **internal messages** to a conversation. Each one comes from a
 fixed template. An internal message starts a turn like any message, and it
 never goes to WhatsApp. Druks talks to agents, and agents talk to people.
@@ -320,9 +329,10 @@ A file you send to the bot, in a direct message or in a thread you joined,
 becomes a Druks file on its message. A message with only a file starts a turn
 like any other. An image reaches the agent with its message. Any other file
 except audio reaches the agent as a link to a copy in the sandbox. The agent
-opens the copy with its tools. An image over 25 MiB goes as a link too. Each
-further file in one Slack message gets a message of its own. The agent sends no
-files back.
+opens the copy with its tools. An image over 25 MiB goes as a link too. An audio
+clip becomes text the way a WhatsApp voice note does: see
+[WhatsApp](#whatsapp). Each further file in one Slack message gets a message of
+its own. The agent sends no files back.
 
 ### Rooms
 

docs/configuration.md

@@ -484,6 +484,17 @@ event older than five minutes. With rotation off, the pasted bot token and each
 person's Slack token live until someone revokes them. A person connects their
 own Slack account through the same app: see [Chat](chat.md#slack).
 
+## Speech to text
+
+**Speech To Text** is the service that Druks sends voice notes to: any server
+that speaks the OpenAI audio API, such as OpenAI, Groq, or a local server.
+Connect it from **Settings → Connections → Services** with the server's base
+URL, for example `https://api.openai.com/v1`, a key, and a model, for example
+`whisper-1`. Druks checks none of the values when you save the card, so a wrong
+key shows up on the first voice note. Druks sends a note of at most 25 MiB and
+refuses a bigger one without a call. Without the card, the agent tells the
+person to write instead. See [Chat](chat.md#whatsapp) for what the agent gets.
+
 ## Harnesses
 
 Druks registers two subscription providers, `anthropic` and `openai`. Each