feat: add push-to-talk voice dictation with Parakeet

app_server/dispatcher.py

@@ -321,6 +321,8 @@ def __init__(
             rpc_methods.TERMINAL_CLOSE: self._terminal_close,
             rpc_methods.TEST_DISCOVER: self._test_discover,
             rpc_methods.TEST_RUN: self._test_run,
+            rpc_methods.DICTATION_STATUS: self._dictation_status,
+            rpc_methods.DICTATION_TRANSCRIBE: self._dictation_transcribe,
         }
 
     @property
@@ -1058,6 +1060,24 @@ def _turn_list(self, params: Params) -> dict[str, Any]:
             "hasMore": len(turns) > limit,
         }
 
+    def _dictation_status(self, params: Params) -> dict[str, Any]:
+        params.only("projectId")
+        return self.application.dictation.status(
+            project_id=params.string("projectId", required=False)
+        )
+
+    def _dictation_transcribe(self, params: Params) -> dict[str, Any]:
+        params.only("audio", "mimeType", "language", "projectId")
+        language = (
+            params.nullable_string("language") if "language" in params.values else None
+        )
+        return self.application.dictation.transcribe(
+            audio=str(params.string("audio")),
+            mime_type=str(params.string("mimeType")),
+            language=language,
+            project_id=params.string("projectId", required=False),
+        )
+
     def _model_reasoning(self, params: Params) -> dict[str, Any]:
         params.only("projectId", "connectionId", "model")
         capabilities = self.application.llm.model_reasoning(

app_server/protocol/methods.py

@@ -105,3 +105,5 @@
 TERMINAL_CLOSE = "terminal/close"
 TEST_DISCOVER = "test/discover"
 TEST_RUN = "test/run"
+DICTATION_STATUS = "dictation/status"
+DICTATION_TRANSCRIBE = "dictation/transcribe"

app_server/protocol/retry.py

@@ -17,6 +17,9 @@
         "mcp/list",
         "mcp/presets",
         "diagnostics/read",
+        # Status is a pure config read; transcription is deliberately absent —
+        # it is a long, expensive call that must be re-issued by the user.
+        "dictation/status",
         "automation/list",
         "automation/runs",
         "thread/list",

core/application/application.py

@@ -17,6 +17,7 @@
 from core.application.automation_scheduler import AutomationScheduler
 from core.application.automation_service import AutomationService
 from core.application.diagnostics_service import DiagnosticsService
+from core.application.dictation_service import DictationService
 from core.application.errors import UpgradeRequiresExclusiveAccessError
 from core.application.event_service import (
     DEFAULT_RELAY_BATCH_SIZE,
@@ -100,6 +101,7 @@ def __init__(
             self.projects,
             credential_store=self.credentials,
         )
+        self.dictation = DictationService(self.projects)
         self.skill_hosts = SkillWorkspaceRegistry()
         self.plugins = PluginService(LocalPluginHost(self.skill_hosts))
         effective_session_factory = session_factory or DefaultAgentSessionFactory()

core/application/dictation_service.py

@@ -0,0 +1,236 @@
+"""Application service for voice dictation (prompt-box microphone input).
+
+Why this is a service and not a tool call: a transcript is not agent input by
+itself. The user dictates into the composer, sees the text, edits it, and sends
+it — so the audio is transcribed *before* a Turn exists and the result never
+reaches the model on its own. The service therefore lives on the input path,
+not in the agent loop.
+
+Responsibilities, in order:
+
+1. **Is dictation configured at all?** An absent ``dictation`` block means the
+   feature is off and the app server reports it as unavailable, so the prompt
+   box never offers a microphone that cannot work.
+2. **Is the endpoint allowed?** A recording is a verbatim copy of what the user
+   said, so the endpoint receiving it is a trust boundary, evaluated against the
+   same ``providers.egress`` policy as model traffic.
+3. **What are the bytes?** Format and size checks stay in
+   :mod:`core.dictation.audio`, which is pure and testable on its own.
+4. **Translate failures.** Every outcome becomes either a transcript or a stable
+   application error; no bare transport exception escapes this module.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any
+
+from loguru import logger
+
+from core.application.config_store import ConfigStore
+from core.application.errors import (
+    DictationNotConfiguredError,
+    DictationUnavailableError,
+    InvalidArgumentError,
+)
+from core.application.project_service import ProjectService
+from core.config import (
+    DeepCodeConfig,
+    DictationConfig,
+    load_config,
+    load_config_for_workspace,
+)
+from core.dictation.audio import (
+    UnsupportedAudioError,
+    canonical_mime_type,
+    decode_audio,
+)
+from core.dictation.client import SpeechToTextClient, TranscriptionFailed
+from core.dictation.local_runner import LocalParakeetClient
+from core.providers.egress import (
+    WARN,
+    evaluate_provider_egress,
+    resolve_egress_policy,
+)
+
+#: Builds the client for one request from the resolved endpoint config, the API
+#: key (or ``None``), and the language hint to send. Injectable so tests can
+#: substitute a client without patching process-global state.
+ClientFactory = Callable[[DictationConfig, str | None, str | None], Any]
+
+
+def _default_client_factory(
+    config: DictationConfig,
+    api_key: str | None,
+    language: str | None,
+) -> Any:
+    if config.endpoint == "local":
+        return LocalParakeetClient(
+            config.model,
+            language=language,
+            timeout_seconds=config.timeout_seconds,
+        )
+    return SpeechToTextClient(
+        config.endpoint,
+        config.model,
+        language=language,
+        api_key=api_key,
+        timeout_seconds=config.timeout_seconds,
+    )
+
+
+class DictationService:
+    """Resolve dictation config and transcribe one clip per request."""
+
+    def __init__(
+        self,
+        projects: ProjectService | None = None,
+        *,
+        config_store: ConfigStore | None = None,
+        client_factory: ClientFactory | None = None,
+    ) -> None:
+        self.projects = projects
+        self.config_store = config_store or ConfigStore()
+        self._client_factory = client_factory or _default_client_factory
+
+    def status(self, project_id: str | None = None) -> dict[str, Any]:
+        """Report the capability without revealing whether a secret resolves.
+
+        ``model`` and ``maxAudioSeconds`` describe the *configured* endpoint, so
+        they are ``None`` exactly when ``available`` is ``False``. The status
+        deliberately says nothing about credentials: whether a bearer token
+        happens to be set in the environment is not a fact the renderer needs,
+        and the failure would surface on the first clip anyway.
+        """
+
+        config = self._config(project_id).dictation
+        if config is None:
+            return {"available": False, "model": None, "maxAudioSeconds": None}
+        return {
+            "available": True,
+            "model": config.model,
+            "maxAudioSeconds": config.max_audio_seconds,
+        }
+
+    def transcribe(
+        self,
+        *,
+        audio: str,
+        mime_type: str,
+        language: str | None = None,
+        project_id: str | None = None,
+    ) -> dict[str, Any]:
+        """Transcribe one base64 clip into ``{"text", "model"}``.
+
+        ``language`` is a per-clip hint used only when ``dictation.language`` is
+        unset: a value in the config file is an explicit user choice and wins
+        over whatever a client offers for a single request.
+        """
+
+        loaded = self._config(project_id)
+        config = loaded.dictation
+        if config is None:
+            raise DictationNotConfiguredError()
+
+        denial = self._egress_denial(config, loaded)
+        if denial is not None:
+            raise DictationUnavailableError(denial, retryable=False)
+
+        try:
+            data, filename = decode_audio(audio, mime_type)
+        except UnsupportedAudioError as exc:
+            raise InvalidArgumentError(str(exc)) from exc
+
+        client = self._client_factory(
+            config, self._api_key(config), config.language or (language or None)
+        )
+        try:
+            text = client.transcribe(
+                data,
+                filename=filename,
+                mime_type=canonical_mime_type(mime_type),
+            )
+        except TranscriptionFailed as exc:
+            raise DictationUnavailableError(str(exc), retryable=exc.retryable) from exc
+        logger.trace(
+            "Dictation transcribed: model={} bytes={} chars={}",
+            config.model,
+            len(data),
+            len(text),
+        )
+        return {"text": text, "model": config.model}
+
+    # -- config and secrets -------------------------------------------------
+
+    def _config(self, project_id: str | None) -> DeepCodeConfig:
+        """Load the config that applies to this request.
+
+        Same precedence as ``LLMConfigurationService._config``: an unscoped
+        request reads the user config, a project-scoped one reads the layered
+        user + project config. The project layer cannot carry a ``dictation``
+        block (``_project_runtime_layer`` drops it), so a project-scoped read
+        never changes *whether* dictation is on, only honours the workspace.
+        """
+
+        if project_id is None:
+            return load_config(config_path=self.config_store.path)
+        if self.projects is None:
+            raise InvalidArgumentError(
+                "project-scoped dictation settings are unavailable"
+            )
+        project = self.projects.read(project_id)
+        workspace = Path(project.canonical_path).resolve(strict=False)
+        return load_config_for_workspace(workspace)
+
+    def _api_key(self, config: DictationConfig) -> str | None:
+        """Read the bearer token from the environment variable the user named."""
+
+        if not config.api_key_env:
+            return None
+        key = os.environ.get(config.api_key_env)
+        if not key:
+            raise DictationNotConfiguredError(
+                f"dictation.apiKeyEnv names {config.api_key_env!r}, which is not "
+                "set in the environment"
+            )
+        return key
+
+    # -- egress -------------------------------------------------------------
+
+    def _egress_denial(
+        self, config: DictationConfig, loaded: DeepCodeConfig
+    ) -> str | None:
+        """Return a denial message, or ``None`` when the endpoint may be used.
+
+        Mirrors ``make_llm_provider``: the decision is a hostname comparison, so
+        it costs nothing per request, and ``warn`` mode records the same message
+        without blocking. The message is written here rather than taken from
+        ``EgressDecision.reason`` because that text talks about
+        ``providers.<name>.apiBase``, and the operator needs to be pointed at
+        the key this endpoint actually lives under.
+        """
+        if config.endpoint == "local":
+            return None
+
+        policy = resolve_egress_policy(loaded)
+        decision = evaluate_provider_egress(
+            config.endpoint,
+            endpoint_class="dictation",
+            allowed_domains=policy.allowed_domains,
+            blocked_domains=policy.blocked_domains,
+        )
+        if decision.allowed:
+            logger.trace(
+                "Dictation egress ok: host={} model={}", decision.host, config.model
+            )
+            return None
+        message = (
+            f"dictation endpoint host {decision.host or config.endpoint!r} is not "
+            "allowed by the model egress policy (providers.egress)"
+        )
+        if policy.mode == WARN:
+            logger.warning(message)
+            return None
+        return message

core/application/errors.py

@@ -269,3 +269,49 @@ class TerminalNotFoundError(ApplicationError):
 
 class NotSupportedApplicationError(ApplicationError):
     code = "NOT_SUPPORTED"
+
+
+class DictationNotConfiguredError(ApplicationError):
+    """Voice input was requested but no ``dictation`` block exists.
+
+    Permanent as long as the config is: the endpoint is user-owned and cannot
+    be supplied per request, so this is not worth retrying from the UI.
+    """
+
+    code = "DICTATION_NOT_CONFIGURED"
+
+    def __init__(self, message: str = "dictation is not configured") -> None:
+        super().__init__(
+            message,
+            user_message=(
+                "Voice input is not configured. Add a dictation block with the "
+                "Parakeet endpoint to your DeepCode config."
+            ),
+        )
+
+
+class DictationUnavailableError(ApplicationError):
+    """The configured endpoint could not produce a transcript.
+
+    Covers every runtime failure on the remote side: unreachable, timed out,
+    refused, or answered with an error. ``retryable`` mirrors the client's
+    judgement (5xx and timeouts yes, 4xx no) rather than being hardcoded, so
+    the UI can offer "try again" only when it would be honest.
+    """
+
+    code = "DICTATION_UNAVAILABLE"
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        retryable: bool = True,
+    ) -> None:
+        super().__init__(
+            message,
+            user_message=(
+                "Voice input could not be transcribed. Check that the dictation "
+                "endpoint is running, then try again."
+            ),
+        )
+        self.retryable = retryable