DRU-660 -- The calls server: Pipecat carries the call

.github/workflows/build-calls-image.yml

@@ -0,0 +1,54 @@
+name: Build Calls Image
+
+# The calls server has its own release cycle: its image publishes only when the
+# server changes, so a Druks release does not restart it.
+on:
+  workflow_dispatch:
+  push:
+    branches: [main]
+    paths:
+      - "druks-calls/**"
+      - "deploy/calls/**"
+      - ".github/workflows/build-calls-image.yml"
+
+concurrency:
+  group: build-calls-image
+  cancel-in-progress: false
+
+permissions:
+  contents: read
+
+jobs:
+  calls:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: docker/setup-qemu-action@99012661954931238ded8c8b007157a8430204e1 # v4.4.0
+      - uses: docker/setup-buildx-action@f87e5991a6d7451dcb8d9637bfbc97413f497069 # v4.4.1
+      - uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+      - id: metadata
+        uses: docker/metadata-action@dc802804100637a589fabce1cb79ff13a1411302 # v6.2.0
+        with:
+          images: ghcr.io/${{ github.repository_owner }}/druks-calls
+          tags: |
+            type=raw,value=latest
+            type=sha,prefix=sha-,format=long
+      - uses: docker/build-push-action@c3c9e263c25d99ce0380d002d59b67737d91b0dc # v7.4.0
+        with:
+          context: druks-calls
+          file: deploy/calls/Dockerfile
+          platforms: linux/amd64,linux/arm64
+          push: true
+          tags: ${{ steps.metadata.outputs.tags }}
+          labels: ${{ steps.metadata.outputs.labels }}
+          cache-from: type=gha,scope=calls
+          cache-to: type=gha,scope=calls,mode=max

.github/workflows/on-pull-request-backend.yml

@@ -62,7 +62,7 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
       - run: uv sync --locked --dev
-      - run: uv run ruff check backend
+      - run: uv run ruff check backend druks-calls
       # The proof app is a standalone distribution that depends on druks;
       # it installs the way an author's would, so druks never depends on it.
       - run: uv pip install -e backend/tests/druks-field_notes

.github/workflows/on-pull-request-calls.yml

@@ -0,0 +1,55 @@
+name: On Pull Request Calls
+
+# The calls server is its own package, with its own lock, Python, and image. These
+# checks run only when it changes.
+on:
+  pull_request:
+    branches: [main, "codex/**"]
+    paths:
+      - "druks-calls/**"
+      - "deploy/calls/**"
+      - ".github/workflows/on-pull-request-calls.yml"
+
+concurrency:
+  group: on-pull-request-calls-${{ github.event.pull_request.number }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  tests:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+        with:
+          python-version: "3.11"
+      - run: uv run --locked pytest
+        working-directory: druks-calls
+
+  # amd64 only and without pushing, so a broken Dockerfile fails the PR instead of
+  # the deploy. arm64 stays covered by the publish build.
+  image:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: docker/setup-buildx-action@f87e5991a6d7451dcb8d9637bfbc97413f497069 # v4.4.1
+      - uses: docker/build-push-action@c3c9e263c25d99ce0380d002d59b67737d91b0dc # v7.4.0
+        with:
+          context: druks-calls
+          file: deploy/calls/Dockerfile
+          platforms: linux/amd64
+          push: false
+          load: true
+          tags: druks-calls:pr
+          cache-from: type=gha,scope=calls
+      - uses: ./.github/actions/dgoss
+        with:
+          image: druks-calls:pr
+          files-path: deploy/calls

AGENTS.md

@@ -84,6 +84,9 @@ For app-surface changes, inspect the proof app at
   packaged proof app.
 - **Frontend:** `frontend/` contains the React 19 and Vite shared SPA. The
   backend image includes its repository-root `dist/` output.
+- **Calls server:** `druks-calls/` contains `calls`, the Pipecat program that
+  carries phone calls, with its own `pyproject.toml` and lock. Its image builds
+  from `deploy/calls/` and has its own release workflow.
 - **Deployment:** `deploy/` contains Compose files, Caddy configuration, and
   sandbox image inputs. The public runbook is `docs/deployment.md`.
 - **Documentation:** `docs/` contains the public and contributor guides.
@@ -104,8 +107,8 @@ docker compose -f deploy/compose.dev.yaml up -d
 Run the backend gates:
 
 ```bash
-uv run ruff check backend
-uv run ruff format --check backend
+uv run ruff check backend druks-calls
+uv run ruff format --check backend druks-calls
 uv pip install -e backend/tests/druks-field_notes
 uv run pytest backend/
 ```
@@ -124,6 +127,12 @@ npm --prefix frontend test
 npm --prefix frontend run build
 ```
 
+Run the calls server's tests from `druks-calls/`:
+
+```bash
+uv run --locked pytest
+```
+
 The PR workflows in `.github/workflows/on-pull-request-*.yml` are the source of
 truth for CI, including the proof-app install phase.
 

backend/druks/chat/channels/base.py

@@ -15,7 +15,7 @@ class Channel:
     name: ClassVar[str]
     service: ClassVar[type[Service]]
     # Whether chat runs agent turns on the channel's conversations. False when something
-    # else answers the people on the channel, such as a voice server.
+    # outside chat answers the people on the channel.
     has_turns: ClassVar[bool] = True
 
     def __init_subclass__(cls) -> None:

backend/druks/chat/channels/calls/constants.py

@@ -5,7 +5,7 @@
 MAX_CALL_SECONDS = 300
 # The call ends when the caller says nothing for this long.
 NO_SPEECH_SECONDS = 15
-# A call token outlives the longest call by a minute: the voice server reports the end
+# A call token outlives the longest call by a minute: the calls server reports the end
 # after the call.
 CALL_TOKEN_SECONDS = MAX_CALL_SECONDS + 60
 # How many lines of the caller's earlier calls the voice model gets.

backend/druks/chat/channels/calls/routes.py

@@ -56,8 +56,8 @@ async def link_number(
     if sid in await Twilio.list_linked_sids(session):
         raise CallsLinkError("This number is already linked. Remove it first.")
     owner = await Account.create_for_bot(session, AccountKind.BOT)
-    voice_url = f"{webhook_base}/_external/twilio/calls/"
-    return await Twilio.link(session, owner, app=app, sid=sid, voice_url=voice_url)
+    calls_url = f"{webhook_base}/_external/twilio/calls/"
+    return await Twilio.link(session, owner, app=app, sid=sid, calls_url=calls_url)
 
 
 @router.delete("/numbers/{number_id}", status_code=204)

backend/druks/chat/channels/calls/services.py

@@ -1,6 +1,6 @@
 from pydantic import BaseModel, Field, SecretStr
 
-from druks.services import Service
+from druks.services import Service, ServiceConnectError
 
 
 class Voice(Service):
@@ -21,3 +21,20 @@ class Settings(BaseModel):
             title="Voice name",
             description="The vendor's voice, for example marin or Kore. Empty uses the default.",
         )
+        transcription_model: str = Field(
+            "gpt-4o-mini-transcribe",
+            title="Transcription model",
+            description=(
+                "Only an OpenAI model uses it. It writes the caller's words as text. Empty "
+                "uses gpt-4o-mini-transcribe."
+            ),
+        )
+
+    @classmethod
+    async def verify(cls, settings: Settings) -> dict:
+        if settings.model.partition("/")[0] in ("openai", "google"):
+            return {}
+        raise ServiceConnectError(
+            "The calls server runs OpenAI and Google models. Write the model as "
+            "openai/<model> or google/<model>."
+        )

backend/druks/chat/channels/calls/webhooks.py

@@ -37,8 +37,8 @@
 
 
 def get_stream_url(token: str) -> str:
-    """Where Twilio streams a call's audio: the voice server, behind the webhook host."""
-    return f"wss://{urlsplit(load_settings().urls.webhook_base).netloc}/_voice/calls/{token}"
+    """Where Twilio streams a call's audio: the calls server, behind the webhook host."""
+    return f"wss://{urlsplit(load_settings().urls.webhook_base).netloc}/_calls/{token}"
 
 
 def sign_claim(connection: VaultSecret, claim: str) -> str:
@@ -75,7 +75,7 @@ async def get_call(session: AsyncSession, token: str) -> Conversation | None:
 
 class TwilioCalls(TwilioWebhook):
     """Twilio's call to a linked number. Druks answers with TwiML that streams the audio to
-    the voice server."""
+    the calls server."""
 
     category = "calls"
 
@@ -84,7 +84,7 @@ def get_action(self) -> str:
 
     async def on_call(self) -> Response:
         """Start the caller's conversation for the call, and tell Twilio to stream its audio
-        to the voice server. Twilio strips a query string from a stream URL, so the call
+        to the calls server. Twilio strips a query string from a stream URL, so the call
         token goes in the path."""
         session = db_session()
         connection = await Twilio.get_for_number(session, self.data["To"])
@@ -109,10 +109,10 @@ async def on_call(self) -> Response:
         return Response(content="<Response><Reject/></Response>", media_type="text/xml")
 
 
-class VoiceEvents(Webhook):
-    """The voice server's requests for a call. Each one carries the call's token."""
+class CallsEvents(Webhook):
+    """The calls server's requests for a call. Each one carries the call's token."""
 
-    provider = "voice"
+    provider = "calls"
     category = "events"
 
     def get_action(self) -> str:
@@ -122,7 +122,7 @@ async def request_is_authentic(self) -> bool:
         session = db_session()
         self.conversation = await get_call(session, self.data["token"])
         if self.conversation and self.get_action() == "pickup":
-            # The voice server forwards the signature of Twilio's stream handshake.
+            # The calls server forwards the signature of Twilio's stream handshake.
             card = await Twilio.get()
             verify_twilio_signature(
                 url=get_stream_url(self.data["token"]),
@@ -133,7 +133,7 @@ async def request_is_authentic(self) -> bool:
         return bool(self.conversation)
 
     async def on_pickup(self) -> Response:
-        """Give the voice server what the call needs, once: the prompt, the caller's facts,
+        """Give the calls server what the call needs, once: the prompt, the caller's facts,
         the Bot's key, the Voice card, and the timers."""
         session = db_session()
         conversation = self.conversation
@@ -142,7 +142,7 @@ async def on_pickup(self) -> Response:
             name=f"calls:{conversation.id}:pickup", value="1", nx=True, ex=CALL_TOKEN_SECONDS
         )
         if not is_first_pickup:
-            raise HTTPException(409, "The voice server picked up this call already.")
+            raise HTTPException(409, "The calls server picked up this call already.")
         _, prompt, tools = await get_agent(session, conversation)
         key = await get_druks_account_token(
             session, account_id=conversation.account_id, allowed_tools=tools, name=CALLS_KEY_NAME
@@ -216,7 +216,7 @@ async def end_run_outcomes(self, session: AsyncSession) -> list[Message]:
         return sorted(outcomes, key=attrgetter("created_at", "id"))
 
     async def on_line(self) -> Response:
-        """Save a line of the call under its sequence. The voice server numbers the lines
+        """Save a line of the call under its sequence. The calls server numbers the lines
         in the order they were said, so the call reads in that order when a line arrives
         late. A caller line waits for the assistant's next line."""
         session = db_session()

backend/druks/core/services.py

@@ -335,9 +335,9 @@ async def get_client(cls, session: AsyncSession) -> TwilioClient:
 
     @classmethod
     async def link(
-        cls, session: AsyncSession, owner: Account, *, app: str, sid: str, voice_url: str
+        cls, session: AsyncSession, owner: Account, *, app: str, sid: str, calls_url: str
     ) -> VaultSecret:
-        """Save the number with a new signing secret, then point its calls at ``voice_url``.
+        """Save the number with a new signing secret, then point its calls at ``calls_url``.
         If Twilio refuses the URL, the caller's rollback removes the row."""
         client = await cls.get_client(session)
         number = await client.get_number(sid)
@@ -351,7 +351,7 @@ async def link(
         )
         session.add(connection)
         await session.flush()
-        await client.set_voice_url(sid, voice_url)
+        await client.set_voice_url(sid=sid, url=calls_url)
         return connection
 
     @classmethod
@@ -360,7 +360,7 @@ async def unlink(cls, session: AsyncSession, connection: VaultSecret, reason: st
         client = await cls.get_client(session)
         # The account may have released the number. Then there is no URL to clear.
         with suppress(TwilioNotFoundError):
-            await client.set_voice_url(connection.identity["sid"], "")
+            await client.set_voice_url(sid=connection.identity["sid"], url="")
         await connection.revoke(reason)
 
     @classmethod