From 7d9bc5fb4cb293139086a4cff88332a25a63fea1 Mon Sep 17 00:00:00 2001 From: Paul Dittrich Date: Thu, 22 Oct 2026 17:44:41 +0200 Subject: [PATCH 1/3] docs(fork): plan the voice call feature Implementation plan for a GPT-Live voice call on the ChatGPT login: the server mints the realtime call and drives it over the sideband, every delegation steers the thread open on screen, results come back as speech. Covers protocol, server and client modules, persistence, settings, pause and quota, four phases, testing, risks and later work. Co-Authored-By: Claude Opus 5.5 --- docs/internals/voice-call.md | 553 +++++++++++++++++++++++++++++++++++ 1 file changed, 553 insertions(+) create mode 100644 docs/internals/voice-call.md diff --git a/docs/internals/voice-call.md b/docs/internals/voice-call.md new file mode 100644 index 0000000000000..4e1ba228dc674 --- /dev/null +++ b/docs/internals/voice-call.md @@ -0,0 +1,553 @@ +# Voice call (plan) + +Status: implementation plan for the fork, not yet built. PAD's decisions of +2026-10-21/22 are the source; the v1 scope cut and the gpt-live-1.1 findings +override older notes. When the feature lands, rewrite this page into the +durable decisions and traps and drop the phase list. + +A voice call lets the user talk to esveo code. A realtime voice model +(OpenAI's GPT-Live, on the user's ChatGPT subscription) holds the +conversation, hands real work to the thread that is open on screen, and +reads the result back. It is separate from [voice input](voice-input.md), +which only dictates into the composer. + +## Goal and non-goals + +Goal for v1: + +- Start a call from the open thread on desktop (Electron) or web. Everything + the voice delegates goes to the thread on screen: it steers the active run, + or starts a normal turn when the thread is idle. +- The backend is the thread's own provider and model. The voice is only the + front end; no extra model hop sits between them. +- Results, questions and approvals of that thread come back as speech. +- The call survives thread switches (the target follows the screen), pauses + itself when it is quiet because open time is billed, and wakes on speech + or on a result. +- A voice panel with both transcripts and a card per delegated message, a + global call bar, and past calls stored on the server. +- Behind a feature flag, off by default. + +Non-goals for v1: + +- Routing: no "send this to thread 2", no new threads by voice, no queueing, + no promote-to-steer, no request header format. See [Later](#later). +- An overview of all threads for the voice. It sees only the open thread. +- Mobile, multi-device handover, other voice vendors, sound effects. +- The classic Realtime protocol with function tools (works in the spike, but + its quota pool and terms are unclear). + +## Relation to upstream + +Upstream PR [pingdotgg/t3code#14865](https://github.com/pingdotgg/t3code/pull/14865) +(open, "Codex voice conversations") shares the browser-audio / server-SDP split +and has desktop microphone permission code worth copying. It differs in the +parts that matter here: + +- It runs the call through the Codex app-server (`thread/realtime/start`), so + every handoff starts a Codex turn and only Codex threads can be the backend. + We call the realtime endpoint directly and dispatch into any provider. +- Its conversation lives as long as one client stream and is not persisted. + We pause, resume across views, and keep call history. + +If upstream merges it, rebase onto its desktop permission and client-runtime +peer pieces instead of keeping our own copies. + +## Architecture + +```mermaid +flowchart LR + subgraph Client["Browser / Electron renderer"] + Mic[Mic + speaker] + Peer[RTCPeerConnection
audio + oai-events DC] + UI[Call button, call bar,
voice panel] + end + subgraph Server["esveo code server"] + RPC[voiceCall.* RPCs] + Svc[VoiceCallService] + Auth[VoiceCallAuth
Codex ChatGPT token] + Side[Sideband WS client] + TM[ThreadManagementService
sendToThread] + Orch[Orchestrator V2
domain events] + DB[(fork_voice_call_* tables)] + end + OAI[(chatgpt.com realtime/calls
api.openai.com/v1/live/id)] + + Mic --> Peer + Peer <-- "audio (WebRTC, direct)" --> OAI + UI <-- "SDP offer/answer, state, transcript" --> RPC + RPC --> Svc + Svc --> Auth + Svc -- "POST offer + session" --> OAI + Side <-- "events, context appends" --> OAI + Svc --- Side + Svc -- "delegation text" --> TM --> Orch + Orch -- "run / turn item / request events" --> Svc + Svc --> DB +``` + +- Audio never touches the esveo code socket. The client owns the mic, the + speaker and the peer; audio flows between the client and OpenAI, so remote + and tunnelled clients cost the same as local ones. +- The access token never reaches the client. The server POSTs the client's + SDP offer and returns the answer. +- The server is the only writer of realtime events and reads them from the + sideband. The client keeps the `oai-events` data channel open because the + session needs it, but sends nothing on it and ignores what arrives, so no + event is sent or handled twice. + +## Protocol + +Everything here comes from PAD's two local spikes (`/private/tmp/voice-spike/`, +not in the repo; 2026-10-22, codex-cli 0.163) and from Codex `main` @ `94350a9`. It is an +undocumented endpoint; re-check it when Codex changes its realtime code. + +**Creating a call** + +- `POST https://chatgpt.com/backend-api/codex/realtime/calls?intent=quicksilver&architecture=avas` +- Headers: `Authorization: Bearer `, `ChatGPT-Account-ID`, + `originator: codex_cli_rs`, `openai-alpha: quicksilver=v2`, + `User-Agent: codex_cli_rs/ …`, `Content-Type: application/json`. +- Body `{ sdp, session }` with `session`: + - `model: "gpt-live-1-codex"`. The string selects nothing; the account + decides the revision (1.1 on Plus and up). Never send `gpt-live-1.1-codex`. + - `audio.output.voice`: one of the 25 v3 voices. An unknown voice ends the + session with `error {code:"forbidden", message:"Voice session access denied."}`. + - `delegation: { type: "client" }`. No `tools`: the v3 session rejects them + and dies. + - `instructions`: front-end prompt, language line, persona, recap (see + [Prompts](#prompts)). +- Response: `201`, the SDP answer as text, `Location: …/rtc_`. The call id + is the last path segment. The POST took about 1.2 s. + +**Sideband** + +- `wss://api.openai.com/v1/live/{call_id}` with the same bearer and account + headers. It carries the same event stream as the data channel and accepts + our events. `session.update` with `delegation:{type:"client"}` works there + and returns `session.updated`. +- Effect's `Socket.makeWebSocket` cannot send request headers, so the server + adds `ws` as a direct dependency (it is already in the tree through + `@effect/platform-node-shared`; `@types/ws` is in the catalog) or wraps it + in a custom `Socket.WebSocketConstructor`. + +**Events we read** + +| Event | Use | +|---|---| +| `session.started {session:{id, status, expires_at}}` | segment is live; `expires_at` is about 2 h out | +| `input_transcript.added {item.text, start_ms, end_ms}` | user transcript deltas, silence timer | +| `output_transcript.added` | voice transcript deltas, silence timer | +| `turn.created` / `turn.done {turn:{id, role, transcript}}` | speaking gate; persisted transcript on `turn.done` | +| `delegation.created {item:{id, content[].text}, handoff_id, user_bidi_turn_id}` | the request to dispatch | +| `delegation.context.appended`, `session.context.appended` | delivery acks | +| `session.usage.updated {usage.audio_duration_ms, usage_limit{status, reset_seconds}}` | quota, every 15 s | +| `session.closed`, `error` | segment ended; `forbidden` means a bad voice | + +`session.input_audio.append` and `session.output_audio.delta` also arrive and +are dropped unread. No event names the model revision. + +**Events we send (sideband only)** + +| Event | Use | +|---|---| +| `delegation.context.append {delegation_item_id, channel:"speakable", content:[{type:"input_text", text}]}` | the answer to one delegation | +| `session.context.append {channel:"speakable", content:[…]}` | unprompted events: results, questions, approvals | +| `session.close` | pause, hang-up, expiry, failed connect | + +Appends are limited to 500 tokens each. Like Codex, cap a result at about +1,000 tokens (`REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET`) and send it in +500-byte chunks. + +**Proven:** a delegation held for 180 s is still spoken when answered; an +unsolicited `session.context.append` is spoken; the data channel alone works +end to end. **Not tested:** holds over 180 s, several open delegations at +once, barge-in during injected speech, a non-spoken context channel +(`session.thinking.append`, `initial_items`). Phase 2 tests the open ones. + +## Server modules + +Everything new lives in `apps/server/src/voiceCall/` (the name avoids the +existing `voiceInput/` dictation module), with the contract in +`packages/contracts/src/voiceCall.ts`. + +- **`VoiceCallAuth.ts`** returns `{ accessToken, accountId }` for the Codex + provider instance (v1: the default `codex` instance). + - *Existing mode* (PAD's setup: no `providerInstances.codex`, so the default + mode, with `~/.codex/auth.json`): read `/auth.json` on every + mint (resolve the home as + [`threadRoots.ts`](../../apps/server/src/forkInstructions/threadRoots.ts) + does). `tokens.access_token` and `tokens.account_id` are what the spike + used. **Never write the file**: the Codex CLI owns it and a concurrent + write would race its refresh. If the JWT `exp` is near, ask Codex to + refresh (to verify in phase 1: the app-server account read with token + refresh); if that is not possible, fail with "renew the Codex login". + - *Managed mode*: `makeCodexChatGptAuth(...).access` in + [`CodexChatGptAuth.ts`](../../apps/server/src/provider/CodexChatGptAuth.ts) + already refreshes under the shared session lock. The account id is not a + field there; read it from the id token's ChatGPT claim. Its token is + scoped to `api.openai.com/v1` with `chatgpt.tokens.use.direct`; verify + that chatgpt.com accepts it. + - An API-key Codex login cannot make this call; report it as unavailable. +- **`RealtimeCallClient.ts`**: the POST (via `HttpClient`, as in + [`cliproxyApi.ts`](../../apps/server/src/usage/cliproxyApi.ts)), the sideband + socket, typed decode of the events above, `send`, `close`. No call logic. +- **`VoiceCallSegment.ts`**: one realtime session. Owns the speaking gate, + the injection queue, the silence timer and usage tracking. +- **`VoiceCallService.ts`**: one call per environment. Owns the target + thread, open delegations, the segment lifecycle (connect, pause, resume, + expiry, retry), persistence and the change stream. A second client that + connects takes the call over; the first gets a `takenOver` state and closes + its peer. +- **`VoiceCallReadBack.ts`**: subscribes to `streamDomainEvents` (pattern of + [`DelegatedTaskQuestionReactor.ts`](../../apps/server/src/orchestration-v2/DelegatedTaskQuestionReactor.ts) + with `forkParked`) and turns run, turn-item and runtime-request events into + answers and injections. +- **`voiceCallPrompt.ts`**: front-end instructions, backend hint, recap. + +**RPCs** (`VOICE_CALL_WS_METHODS`, all `AuthOrchestrationOperateScope`): + +```ts +voiceCall.subscribe: {} -> stream VoiceCallEvent // snapshot, then deltas +voiceCall.connect: { offerSdp, target: ThreadRef | null, retryOf?: SegmentId } + -> { answerSdp, segmentId } // start, resume and retry +voiceCall.setTarget: { target: ThreadRef | null } -> void +voiceCall.pause: {} -> void // manual "lazy hang-up" +voiceCall.hangUp: {} -> void +voiceCall.listCalls / voiceCall.readCall // history, phase 4 +``` + +The subscription sends deltas, not full snapshots, so a long call does not +resend its whole transcript on every word. When the owning client's +subscription ends while a segment is live, the server closes the segment +at once, so a dropped client cannot leave a billed session open. + +Wiring follows the existing fork RPCs (commit `d1de6178f0` lists the files): +`forkWsRpc.ts` (`ForkWsRpcGroup`, `FORK_WS_RPC_TAGS`), contracts `index.ts`, +`forkHandlers` in `ws.ts`, `RPC_REQUIRED_SCOPES` in +`auth/RpcAuthorization.ts`, `observability/forkRpcAggregates.ts`, the layer in +`server.ts`, and **`EnvironmentSubscriptionRpcTag` in +`packages/client-runtime/src/rpc/client.ts`**, without which the client types +the stream as unary. [`ForkNotes.ts`](../../apps/server/src/forkNotes/ForkNotes.ts) +is the template for the service, its table and its PubSub stream. + +### The steer path + +On `delegation.created`: + +1. No target (draft, settings page, no thread open) or an archived thread: + answer the delegation at once with one sentence saying so. +2. Otherwise call `ThreadManagementService.sendToThread` with `mode: "auto"`. + It already is the v1 rule: it steers a steerable run and otherwise starts + one. Ids follow + [`ScheduledTaskService`](../../apps/server/src/scheduledTasks/ScheduledTaskService.ts): + `CommandId` `voice-call::`, `MessageId` + `voice-call-message::`, so a retried event cannot + dispatch twice. `createdBy: "user"`, `creationSource: "server"`. +3. Record the delegation `{ delegationItemId, segmentId, threadId, messageId, + runId, status: open }` and push a card to the panel. + +The voice has already said a filler ("On it.") and stays conversational; the +answer follows when the run settles. A dispatch failure is answered at once +with a short reason. + +**Marking the message as voice.** Do not add `"voice"` to `creationSource`; +it is an upstream `Schema.Literals` and every sync would conflict. Use a +context record instead: the text starts with a reference +`[Voice](t3-context://voice-call/)` and `context.records` holds +`{ kind: "voice-call", payload: { hint } }`. `UnknownContextRecord` accepts +the custom kind, and +[`projectComposerContextForProvider`](../../packages/shared/src/composerContextReferences.ts) +puts the payload into the `` envelope of the provider prompt for +every provider. The reference renders as a chip in the user bubble, which +doubles as the "via voice" badge; check how the web renders an unknown kind +and add a small fork chip if it looks wrong. + +## Read-back + +- **Answer to a delegation.** When a run linked to open delegations reaches a + terminal status, take its last assistant message, cap it at about 1,000 + tokens (cut at a sentence; replace code blocks with "code is in the + thread"), prefix the thread title, and send it with + `delegation.context.append` on the newest open delegation of that run. + Older open delegations of the same run (steered twice) are marked + superseded and get nothing. A failed or interrupted run answers with one + sentence and the error. +- **Unprompted events** in the thread that is the target *when the event + fires*: a run that settles with no open delegation, a new user-input + question, a new approval request. They go out as `session.context.append` + ("Update from : …"). A run that answered a delegation never also + triggers an unprompted update. +- **Speaking gate.** Injections wait in a queue until no user turn is open + (every `turn.created role:user` has its `turn.done`), no assistant turn is + open, and the last input transcript delta is at least 1 s old. Answers to + delegations pass the same gate. Codex has no such gate and relies on the + model; we add it because PAD wants events held until he is done talking. +- **Segment gone.** A delegation id belongs to its segment. If that segment + has closed (pause, expiry), the answer goes out as `session.context.append` + on the next segment. While the call is auto-paused, a result wakes it (see + [Pause](#pause-resume-expiry-and-quota)); while manually paused, results + queue and are spoken after resume. +- **Server restart.** Open delegations are persisted. On start, each one is + checked against its run in `ProjectionStoreV2`; a run that settled while + the server was down is delivered as an unprompted result on the next + segment. `streamDomainEvents` is live-tail only, so there is no replay to + rely on. + +### Questions and approvals (decision for PAD) + +With no routing, "yes, allow it" would become a steer and the request would +stay pending. v1 needs exactly one rule for this. Recommendation: + +- While the target thread has exactly one pending `user_input` request, the + next delegation is its free-text answer: `runtime-request.respond` with + `answers` (the path MCP `t3_pending_request_respond` uses), not a steer. +- Approvals: PAD runs with full access, so they are rare. The voice announces + them and the front-end prompt tells it to delegate exactly `approve` or + `decline` as the answer. A delegation that normalises to one of those words + while one approval is pending is answered with `decision: accept|decline`; + anything else is a normal steer. Unlike MCP, this server path may send + `decision`, so the rule stays this narrow. +- More than one pending request: announce how many, answer nothing by voice, + point to the screen. + +Alternatives: approvals only announced, never answered by voice; or a +request header format (moved to Later). + +## Prompts + +`voiceCallPrompt.ts` builds three texts; `prompt.py` in Bloo's `bloo-voice` +plugin is the reference for tone and for the doze notes. + +- **Front-end instructions** (POST `instructions`): you are the voice of + esveo code talking to the user; chat on your own, but hand every request + about code, files, the thread or its agent to the client as a cleaned-up + instruction in your own words; the client sends it to the thread open on + screen; say a short filler and stay responsive while it runs; read results + back short and without Markdown; when told about a question or approval, + ask the user and pass the answer on (exact words for approvals); do not + invent results. Plus the language line (`auto`: answer in the user's + language; `de`: always German, from the first sentence), the persona text, + the current target thread's title, the time, and on resume a recap. +- **Backend hint** (the `voice-call` context record), modelled on Codex's + `realtime_start.md`: this message comes from a live voice call, written by + the voice assistant from speech, so it may contain recognition errors; your + final answer is read aloud by that assistant; keep it short and + action-oriented; leave long output, code and lists in the thread and say + that they are there. +- **Recap** for a new segment: the last 8 transcript turns (Bloo's + `RECAP_TURNS`) and any open delegations, as part of `instructions`. That is + the one carry-over mechanism we control and know works; `initial_items` and + a silent context channel are unverified. + +## Client modules + +New folder `apps/web/src/components/voiceCall/`. Mobile is out of scope, so +the controller stays in the web app instead of `packages/client-runtime`. + +- **`voiceCallState.ts`**: atoms built with + `createEnvironmentRpcSubscriptionAtomFamily` and `createEnvironmentRpcCommand`, + copied from [`forkNotesState.ts`](../../apps/web/src/components/forkNotes/forkNotesState.ts). +- **`voiceCallPeer.ts`**: `getUserMedia`, the peer, the `oai-events` channel, + remote audio playback. Needs a secure context, like dictation. +- **`useVoiceCallController.ts`**: connect, retry, pause and wake (see + below), takeover, hang-up. +- **`useVoiceCallTarget.ts`**: the thread on screen, sent with `setTarget` on + change. In split view that is the focused pane (`activeLeafId` in + `splitThreadStore`, `"route"` meaning the route thread); otherwise + `useThreadRouteRef()`. A draft has no target. +- **`VoiceCallButton.tsx`**: start, and stop while live. Mounted with one + `// Fork:` line in the `panelToggleControls` span of `ChatView.tsx`, next to + `ThreadFindHeaderButton`. Also a command palette action. +- **`VoiceCallBar.tsx`**: a small fixed pill on every route: state + (connecting, live, paused, auto-paused, error), minutes, quota warning, + pause/resume, hang up, open panel. Mounted in the coordinator row of + `RootRouteView` in `routes/__root.tsx`. Popout windows do not get it. +- **`VoiceCallPanel.tsx`**: a `voice-call` right-panel surface registered + through `useForkRightPanelSurfaces` (model: the Instructions tab, commit + `f0373c7ffc`). The right-panel store is per thread, so the panel opens on + the focused thread and follows to the new thread when the target changes + while it is open. The content reads the call store, so it is the same + everywhere. It shows the user and voice transcripts and a card per + delegation (thread, text, status, answer excerpt), plus the past calls list. +- **`voiceCallJump.ts`**: clicking a card navigates to + `/$environmentId/$threadId` with the message id in history state; a small + hook mounted in `MessagesTimeline` finds the row by `data-message-id`, + pages in earlier history like `useAssistantCitationTarget`, and calls + `scrollToIndex`. Citations stay as they are; they only target assistant + messages. +- **`VoiceCallSettingRows.tsx`**: see [Settings](#settings-and-feature-flag). + +**Desktop.** Dictation already records in the desktop app (see "Dictation" in +`docs/fork/setup.md`), so check first whether WebRTC needs more. If it does, +copy `MainWindowPermissions.ts` from #14865 into a fork file mounted in +`DesktopWindow.ts`, and add `NSMicrophoneUsageDescription` in all three +places: `scripts/build-desktop-artifact.ts` (`extendInfo`), +`apps/desktop/scripts/electron-launcher.mjs`, and `scripts/fork-app.sh`, +which re-signs ad hoc and so needs its own plist patch. + +## Persistence + +Plain fork tables, created in `ensureForkSchema` +([`forkSchema.ts`](../../apps/server/src/threadOrchestration/forkSchema.ts)), +not orchestration events: a call is not thread state. + +- `fork_voice_calls`: id, started/ended at, state, voice, last target, + total audio ms. +- `fork_voice_call_segments`: id, call id, realtime call id, started/ended + at, end reason (pause, auto-pause, expiry, retry, hang-up, error), + audio ms, expires at. +- `fork_voice_call_entries`: id, call id, segment id, sequence, kind + (`user`, `voice`, `delegation`, `injection`), text, thread id, message id, + run id, delegation status, timestamps. + +Transcript turns are written on `turn.done`, never per delta. Live deltas only +go over the stream. Delegations are written when they arrive, because the +read-back needs them after a restart. + +## Settings and feature flag + +Server settings, because the server builds the session. They follow the +`enable*` fork pattern in `packages/contracts/src/settings.ts` (commit +`57befeadb1` has the touch list): a field with a decoding default plus the +matching optional key in `ServerSettingsPatch`. + +- `enableVoiceCall: boolean`, default `false`. Off: the RPCs fail with + `VoiceCallDisabledError`, the button, bar and panel are hidden. +- `voiceCallVoice`: one of the 25 v3 voices (a constant list in the contract: + juniper, maple, spruce, ember, vale, breeze, arbor, sol, cove, aube, + bubbie, haetsal, hanul, himari, leher, miko, neer, porto, rindo, rio, + rivage, selva, sonna, tinta, viola). Default `sonna` (German worked; + PAD may prefer `miko`). Only the list is offered, because an off-list voice + kills the session. +- `voiceCallLanguage: "auto" | "de"`, default `"auto"`. +- `voiceCallPersona: string`, default empty, capped in length. + +Rows sit in Settings next to `VoiceInputSettingRow`, use +`esveoSearchableSetting`, and their search items set `fork: true`. The voice, +language and persona rows show only when the flag is on. The subscription +reports availability, so the UI can explain why a call cannot start: flag +off, no Codex instance, Codex logged in with an API key, login expired. + +## Pause, resume, expiry and quota + +`audio_duration_ms` is wall-clock session time, so an idle open session uses +quota. Pausing therefore closes the realtime session; the call itself stays. + +- **Auto-pause.** The server pauses after 30 s without any transcript delta + in either direction, when no turn is open and the voice is not speaking. + An open delegation does not hold the session open; its result wakes it. + The server sends `session.close`, persists the segment, and tells the + client, which closes its peer but keeps the mic. +- **Wake on speech.** While auto-paused the client runs a level gate (Bloo: + peak about 500/32767 for 300 ms of speech) and keeps a 12 s ring buffer. + On speech it connects a new segment and replays the buffer first, so the + model hears the whole sentence that woke it. Bloo drops a replay shorter + than 1.5 s, because the model makes things up from a cough. **Trap:** + Discord sends no packets in silence, so Bloo's backlog drains by itself; a + browser mic produces frames all the time, so a replayed buffer never + catches up. The input track (an `AudioWorklet` into a + `MediaStreamAudioDestinationNode`) must skip quiet frames while a backlog + remains. +- **Wake on a result.** The server asks the owning client to connect; the + queued answer is spoken once the new segment has started. With no client + subscribed, it stays queued. +- **Manual pause ("lazy hang-up").** Closes the session and releases the mic. + No auto-wake. Results queue and are spoken after resume. Resume works from + the call bar on any route of the same app; no cross-device resume. +- **Expiry.** `expires_at` is about 2 h. From 5 minutes before, the next quiet + moment (the speaking gate) triggers a close and an immediate new segment + with a recap; at 1 minute before it is forced. +- **Retry.** The data channel sometimes never opens (one of 17 v3 calls in + the 1.1 spike; worse on the classic route). If the peer is not connected + and the channel not open within 15 s, or the sideband fails to connect, the + client calls `connect` with `retryOf`; the server closes the old call and + mints a new one. Three attempts, then an error state. Each attempt is + logged. +- **Quota.** `session.usage.updated` gives segment minutes; the bar shows this + call's minutes and the last 24 h from the segments table. The plan limit is + not known reliably (help-center figures conflict), so show minutes used, + not minutes left. When `usage_limit.status` is not null, show it and + `reset_seconds` as a warning. + +## Phases + +Each phase is one reviewable PR on the feature branch, behind the flag. + +1. **Server call and steer path.** Contract and RPC wiring, `VoiceCallAuth` + (existing mode first), `RealtimeCallClient` with the sideband, + `VoiceCallService` with `connect` / `setTarget` / `hangUp` / `subscribe`, + the steer path with the voice context record, the front-end prompt, the + flag and voice/language/persona settings (no rows yet), connect retry. A + bare `VoiceCallButton` behind the flag (start and stop, transcript in the + browser console) is the end-to-end check; no panel, bar or pause yet. + Desktop mic only if the check shows it is needed. +2. **Read-back.** `VoiceCallReadBack`, delegation answers with cap and + chunking, unprompted events for the target, the speaking gate, + questions and approvals per PAD's decision, the delegations table and + restart re-arm. Settles the untested protocol points listed above. +3. **Pause, resume, expiry, quota.** Auto-pause, client level gate and + replay buffer, wake on result, manual pause, recap, 2 h rollover, usage + tracking. +4. **UI.** Voice panel with transcripts and cards, jump to message, call + bar, command palette action, settings rows, history tables and + `listCalls` / `readCall`. Rewrite this page into the durable internals + doc, add a user doc section and the changelog entry. + +## Testing + +- Unit tests with a fake `RealtimeCallClient` that replays recorded spike + event sequences (`/private/tmp/voice-spike/logs/a1_v3_dc.txt`, + `c1_v3_ws_hold60.txt`, `l11_v_marin.txt`): delegation to dispatch, ids + stable on duplicates, no-target and archived answers, auth failures. +- Read-back with the orchestrator test harness: drain + `OrchestrationEffectWorkerV2.drain` or await the persisted run event, never + sleep. Cover steer twice then one answer, unprompted versus delegated + dedupe, the speaking gate holding an injection until `turn.done`, segment + gone, restart re-arm. +- Pause logic with a test clock: silence timer, expiry window, retry limit. +- `VoiceCallAuth`: existing mode never writes `auth.json`; expired token + path; managed mode account id from the id token. +- Manual, per phase: a real call on PAD's login in a worktree dev server + seeded with `vp run migrate-dev-db`, never against `~/.t3`. The spike + scripts (`t_a_direct.py`, `t_live11.py`, `cleanup_calls.py`) stay the + reference for protocol questions; close every test call. +- Run only the touched specs and a scoped typecheck, per `FORK.md`. + +## Risks + +- **Undocumented endpoint.** `/backend-api/codex/realtime/calls` can change + or close at any time. We send `originator: codex_cli_rs` and a Codex user + agent, which is a gray area under OpenAI's terms for a third-party app on a + subscription login. `X-OAI-Attestation` is optional today and could become + mandatory, which would end this path. Keep the code behind + `RealtimeCallClient` so a switch to the metered `gpt-live-1` API (API key, + per minute) touches one module. +- **No model signal.** No event reports the model revision. 1.1 is inferred + from the account plan and from 1.1-only voices rendering. If voices and + model ever decouple, we cannot tell. +- **Idle billing.** Open time counts, whether anyone talks or not. A bug in + pause, takeover or disconnect handling burns the user's voice allowance; + the server must close a segment whenever its owning client goes away. +- **Token handling.** Existing mode depends on the Codex CLI keeping + `auth.json` fresh. The token must never be logged or sent to the client. +- **Unclear quota pool.** Whether voice minutes draw on the ChatGPT voice + allowance or the Codex limits is unknown. +- **Recognition errors.** Without routing, every delegation steers the thread + on screen. A misheard request lands in whatever thread PAD is looking at. +- **Untested protocol points** listed under [Protocol](#protocol). + +## Later + +- Routing: send to another thread (and switch the UI there), open a new + thread with project and model inference, queue instead of steer, promote + the last queued message to a steer. Options: a fixed request header the + voice writes and the server parses, a cheap router model as fallback, or + the classic Realtime protocol with real function tools once its quota and + terms are clear. +- An overview of all active and recent threads for the voice. +- A silent context channel for target-change notes, once one is verified. +- Mobile (react-native-webrtc, as in #14865). +- A choice of Codex instance when there are several. +- A health check that starts a 1.1-only voice to confirm the revision. +- Keybinding for start/stop and mute. From 621e2b23faf90fd4c16656e75ef0a3a23344441c Mon Sep 17 00:00:00 2001 From: Paul Dittrich <paul.dittrich@esvo.com> Date: Fri, 23 Oct 2026 16:59:27 +0200 Subject: [PATCH 2/3] feat(fork): voice call server, steer path and call button A voice call runs on the user's ChatGPT login: the server creates the realtime call, holds the sideband, and steers every delegation into the thread on screen, marked with a voice-call context record. The web shows a bare call button behind the enableVoiceCall flag. The plan doc records PAD's decisions of 2026-10-23 and the deviations from the plan. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --- apps/server/src/auth/RpcAuthorization.ts | 6 + .../src/observability/forkRpcAggregates.ts | 5 + .../ThreadManagementService.ts | 4 + apps/server/src/server.ts | 4 + .../src/voiceCall/RealtimeCallClient.ts | 217 ++++++++ apps/server/src/voiceCall/VoiceCallAuth.ts | 137 +++++ .../src/voiceCall/VoiceCallService.test.ts | 238 +++++++++ apps/server/src/voiceCall/VoiceCallService.ts | 489 ++++++++++++++++++ .../src/voiceCall/__fixtures__/a1_v3_dc.jsonl | 93 ++++ .../voiceCall/__fixtures__/l11_v_marin.jsonl | 1 + .../src/voiceCall/realtimeEvents.test.ts | 72 +++ apps/server/src/voiceCall/realtimeEvents.ts | 217 ++++++++ apps/server/src/voiceCall/voiceCallPrompt.ts | 52 ++ apps/server/src/ws.ts | 7 + apps/web/src/components/ChatView.tsx | 5 + .../components/voiceCall/VoiceCallButton.tsx | 110 ++++ .../src/components/voiceCall/voiceCallPeer.ts | 112 ++++ .../components/voiceCall/voiceCallState.ts | 81 +++ docs/internals/voice-call.md | 134 +++-- packages/client-runtime/src/rpc/client.ts | 3 + packages/contracts/src/forkWsRpc.ts | 16 + packages/contracts/src/index.ts | 1 + packages/contracts/src/settings.ts | 20 + packages/contracts/src/voiceCall.ts | 223 ++++++++ 24 files changed, 2195 insertions(+), 52 deletions(-) create mode 100644 apps/server/src/voiceCall/RealtimeCallClient.ts create mode 100644 apps/server/src/voiceCall/VoiceCallAuth.ts create mode 100644 apps/server/src/voiceCall/VoiceCallService.test.ts create mode 100644 apps/server/src/voiceCall/VoiceCallService.ts create mode 100644 apps/server/src/voiceCall/__fixtures__/a1_v3_dc.jsonl create mode 100644 apps/server/src/voiceCall/__fixtures__/l11_v_marin.jsonl create mode 100644 apps/server/src/voiceCall/realtimeEvents.test.ts create mode 100644 apps/server/src/voiceCall/realtimeEvents.ts create mode 100644 apps/server/src/voiceCall/voiceCallPrompt.ts create mode 100644 apps/web/src/components/voiceCall/VoiceCallButton.tsx create mode 100644 apps/web/src/components/voiceCall/voiceCallPeer.ts create mode 100644 apps/web/src/components/voiceCall/voiceCallState.ts create mode 100644 packages/contracts/src/voiceCall.ts diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index fc9ea6b662d1f..bf0f8b4dcb6f1 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -28,6 +28,7 @@ import { THREAD_COORDINATORS_WS_METHODS, THREAD_DECISIONS_WS_METHODS, FORK_NOTES_WS_METHODS, + VOICE_CALL_WS_METHODS, FORK_INSTRUCTIONS_WS_METHODS, VOICE_INPUT_WS_METHODS, AuthTerminalReadScope, @@ -228,6 +229,11 @@ export const RPC_REQUIRED_SCOPES = { // Fork: dictation in the composer. [VOICE_INPUT_WS_METHODS.prepare]: AuthOrchestrationOperateScope, [VOICE_INPUT_WS_METHODS.transcribe]: AuthOrchestrationOperateScope, + // Fork: voice call. Even watching it needs operate: the subscription keeps a billed session alive. + [VOICE_CALL_WS_METHODS.subscribe]: AuthOrchestrationOperateScope, + [VOICE_CALL_WS_METHODS.connect]: AuthOrchestrationOperateScope, + [VOICE_CALL_WS_METHODS.setTarget]: AuthOrchestrationOperateScope, + [VOICE_CALL_WS_METHODS.hangUp]: AuthOrchestrationOperateScope, // Fork: provider subagent controls. [PROVIDER_SUBAGENT_CONTROL_WS_METHODS.target]: AuthOrchestrationReadScope, [PROVIDER_SUBAGENT_CONTROL_WS_METHODS.stop]: AuthOrchestrationOperateScope, diff --git a/apps/server/src/observability/forkRpcAggregates.ts b/apps/server/src/observability/forkRpcAggregates.ts index ff181988e5cad..db0416587db4b 100644 --- a/apps/server/src/observability/forkRpcAggregates.ts +++ b/apps/server/src/observability/forkRpcAggregates.ts @@ -5,6 +5,7 @@ import { PROVIDER_SUBAGENT_CONTROL_WS_METHODS, THREAD_COORDINATORS_WS_METHODS, THREAD_DECISIONS_WS_METHODS, + VOICE_CALL_WS_METHODS, VOICE_INPUT_WS_METHODS, WS_METHODS, } from "@t3tools/contracts"; @@ -23,6 +24,10 @@ export const FORK_RPC_AGGREGATES = { [THREAD_COORDINATORS_WS_METHODS.set]: "orchestration", [VOICE_INPUT_WS_METHODS.prepare]: "server", [VOICE_INPUT_WS_METHODS.transcribe]: "server", + [VOICE_CALL_WS_METHODS.subscribe]: "orchestration", + [VOICE_CALL_WS_METHODS.connect]: "orchestration", + [VOICE_CALL_WS_METHODS.setTarget]: "orchestration", + [VOICE_CALL_WS_METHODS.hangUp]: "orchestration", [PROVIDER_SUBAGENT_CONTROL_WS_METHODS.target]: "orchestration", [PROVIDER_SUBAGENT_CONTROL_WS_METHODS.stop]: "orchestration", [AGENT_TIME_WS_METHODS.summary]: "server", diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index d83dd9578c279..3fb12ffa4d6c8 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1,4 +1,5 @@ import type { + OrchestrationMessageContext, // Fork: voice call OrchestrationV2SearchThreadInput, OrchestrationV2SearchThreadResult, } from "@t3tools/contracts"; @@ -126,6 +127,8 @@ export interface ThreadManagementSendInput { readonly scheduledTaskId?: ScheduledTaskId; readonly senderThreadId?: ThreadId; readonly text: string; + // Fork: the voice call marks its messages with a context record. + readonly context?: OrchestrationMessageContext; readonly attachments: ReadonlyArray<ChatAttachment>; readonly modelSelection?: ModelSelection; readonly mode: ThreadManagementSendMode; @@ -788,6 +791,7 @@ const make = Effect.gen(function* () { ...(input.scheduledTaskId === undefined ? {} : { scheduledTaskId: input.scheduledTaskId }), ...(input.senderThreadId === undefined ? {} : { senderThreadId: input.senderThreadId }), text: input.text, + ...(input.context === undefined ? {} : { context: input.context }), // Fork: voice call attachments: input.attachments, ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }), dispatchMode, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index c7baf3157ac11..daade17708420 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -90,6 +90,8 @@ import * as DelegatedAttachments from "./mcp/toolkits/threads/attachments.ts"; import * as ThreadCoordinators from "./threadOrchestration/ThreadCoordinators.ts"; import * as ThreadDecisions from "./threadDecisions/ThreadDecisions.ts"; import * as ForkNotes from "./forkNotes/ForkNotes.ts"; +import * as VoiceCall from "./voiceCall/VoiceCallService.ts"; +import * as RealtimeCallClient from "./voiceCall/RealtimeCallClient.ts"; import * as ForkInstructions from "./forkInstructions/ForkInstructions.ts"; import * as AgentTime from "./agentTime/AgentTime.ts"; import { hasCloudPublicConfig } from "./cloud/publicConfig.ts"; @@ -613,6 +615,8 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( ForkInstructions.layer.pipe(Layer.provide(ProjectionStoreV2.layer)), // Fork: agent time on the Usage page. AgentTime.layer, + // Fork: voice call with GPT-Live. + VoiceCall.layer.pipe(Layer.provide(RealtimeCallClient.layer)), Layer.effectDiscard(ProjectSettingsOverrideSweep.sweep), ).pipe( Layer.provideMerge(layerOrchestrationApplication), diff --git a/apps/server/src/voiceCall/RealtimeCallClient.ts b/apps/server/src/voiceCall/RealtimeCallClient.ts new file mode 100644 index 0000000000000..eed107dee86c6 --- /dev/null +++ b/apps/server/src/voiceCall/RealtimeCallClient.ts @@ -0,0 +1,217 @@ +/** + * Fork: creates GPT-Live calls on the user's ChatGPT login and holds their + * event sideband. The client's WebRTC peer carries the audio; this side only + * reads and writes events. + * + * Errors never carry the request: its headers hold the access token. + */ +import type { VoiceCallVoice } from "@t3tools/contracts"; +import { VoiceCallError } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Schema from "effect/Schema"; +import type * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import { HttpClient, HttpClientRequest } from "effect/http"; + +import { + decodeRealtimeEvent, + type RealtimeEvent, + type RealtimeOutbound, +} from "./realtimeEvents.ts"; + +const CALLS_URL = + "https://chatgpt.com/backend-api/codex/realtime/calls?intent=quicksilver&architecture=avas"; +const sidebandUrl = (realtimeCallId: string) => + `wss://api.openai.com/v1/live/${encodeURIComponent(realtimeCallId)}`; +const REALTIME_MODEL = "gpt-live-1-codex"; + +export interface RealtimeAuth { + readonly accessToken: string; + readonly accountId: string; +} + +export interface RealtimeSessionConfig { + readonly instructions: string; + readonly voice: VoiceCallVoice; +} + +export interface RealtimeSideband { + /** Ends when the socket closes. */ + readonly events: Stream.Stream<RealtimeEvent>; + readonly send: (event: RealtimeOutbound) => Effect.Effect<void, VoiceCallError>; +} + +export class RealtimeCallClient extends Context.Service< + RealtimeCallClient, + { + readonly createCall: (input: { + readonly auth: RealtimeAuth; + readonly offerSdp: string; + readonly session: RealtimeSessionConfig; + }) => Effect.Effect< + { readonly answerSdp: string; readonly realtimeCallId: string }, + VoiceCallError + >; + /** Closing the scope ends the realtime session, so nothing keeps billing after a crash. */ + readonly openSideband: (input: { + readonly auth: RealtimeAuth; + readonly realtimeCallId: string; + }) => Effect.Effect<RealtimeSideband, VoiceCallError, Scope.Scope>; + } +>()("t3/voiceCall/RealtimeCallClient") {} + +/** What Codex sends; the endpoint rejects calls without the quicksilver headers. */ +const codexHeaders = (auth: RealtimeAuth) => ({ + Authorization: `Bearer ${auth.accessToken}`, + "ChatGPT-Account-ID": auth.accountId, + originator: "codex_cli_rs", + "openai-alpha": "quicksilver=v2", + "User-Agent": "codex_cli_rs/0.163.0 (esveo code)", +}); + +const isVoiceCallError = Schema.is(VoiceCallError); +const JsonString = Schema.fromJsonString(Schema.Unknown); +const decodeJson = Schema.decodeUnknownOption(JsonString); +const encodeJson = Schema.encodeSync(JsonString); + +const excerpt = (text: string) => text.replace(/\s+/g, " ").slice(0, 200); + +const createCall = ( + http: HttpClient.HttpClient, + { auth, offerSdp, session }: Parameters<RealtimeCallClient["Service"]["createCall"]>[0], +) => + Effect.gen(function* () { + const response = yield* http.execute( + HttpClientRequest.post(CALLS_URL).pipe( + HttpClientRequest.setHeaders(codexHeaders(auth)), + HttpClientRequest.bodyJsonUnsafe({ + sdp: offerSdp, + session: { + model: REALTIME_MODEL, + instructions: session.instructions, + audio: { output: { voice: session.voice } }, + // The voice hands work to us instead of calling tools itself. + delegation: { type: "client" }, + }, + }), + ), + ); + const body = yield* response.text; + if (response.status === 401 || response.status === 403) { + return yield* new VoiceCallError({ + message: `ChatGPT refused the call (${response.status}). Renew the Codex login.`, + reason: "codex-login-expired", + }); + } + if (response.status !== 200 && response.status !== 201) { + return yield* new VoiceCallError({ + message: `ChatGPT refused the call (${response.status}): ${excerpt(body)}`, + }); + } + const realtimeCallId = response.headers["location"]?.split("/").pop(); + if (!realtimeCallId || !body.startsWith("v=")) { + return yield* new VoiceCallError({ message: "ChatGPT answered the call without a session." }); + } + return { answerSdp: body, realtimeCallId }; + }).pipe( + Effect.timeoutOrElse({ + duration: "20 seconds", + orElse: () => + Effect.fail(new VoiceCallError({ message: "ChatGPT did not answer the call." })), + }), + Effect.mapError((error) => + isVoiceCallError(error) + ? error + : new VoiceCallError({ message: `Could not reach ChatGPT (${errorTag(error)}).` }), + ), + Effect.withSpan("RealtimeCallClient.createCall"), + ); + +const errorTag = (error: unknown) => + typeof error === "object" && error !== null && "_tag" in error ? String(error._tag) : "unknown"; + +/** + * Node's WebSocket (undici) takes request headers as a second-argument + * option; the DOM typing does not know about it. + */ +type HeaderWebSocketConstructor = new ( + url: string, + options: { readonly headers: Record<string, string> }, +) => WebSocket; + +const openSideband: RealtimeCallClient["Service"]["openSideband"] = ({ auth, realtimeCallId }) => + Effect.gen(function* () { + const events = yield* Queue.unbounded<RealtimeEvent, Cause.Done>(); + const socket = yield* Effect.acquireRelease( + Effect.callback<WebSocket, VoiceCallError>((resume) => { + const socket = new (WebSocket as unknown as HeaderWebSocketConstructor)( + sidebandUrl(realtimeCallId), + { headers: codexHeaders(auth) }, + ); + socket.addEventListener("message", (message) => { + if (typeof message.data !== "string") return; + const event = Option.flatMap(decodeJson(message.data), decodeRealtimeEvent); + if (Option.isSome(event)) Queue.offerUnsafe(events, event.value); + }); + socket.addEventListener("close", () => { + Queue.endUnsafe(events); + }); + socket.addEventListener("open", () => resume(Effect.succeed(socket)), { once: true }); + socket.addEventListener( + "error", + () => { + if (socket.readyState !== WebSocket.OPEN) { + resume( + Effect.fail( + new VoiceCallError({ message: "The voice session sideband did not open." }), + ), + ); + } + }, + { once: true }, + ); + return Effect.sync(() => socket.close()); + }).pipe( + Effect.timeoutOrElse({ + duration: "15 seconds", + orElse: () => + Effect.fail(new VoiceCallError({ message: "The voice session sideband timed out." })), + }), + ), + (socket) => + Effect.sync(() => { + if (socket.readyState === WebSocket.OPEN) { + socket.send(encodeJson({ type: "session.close" } satisfies RealtimeOutbound)); + } + socket.close(); + Queue.endUnsafe(events); + }), + ); + return { + events: Stream.fromQueue(events), + send: (event) => + Effect.suspend(() => { + if (socket.readyState !== WebSocket.OPEN) { + return Effect.fail(new VoiceCallError({ message: "The voice session has ended." })); + } + socket.send(encodeJson(event)); + return Effect.void; + }), + } satisfies RealtimeSideband; + }); + +export const layer = Layer.effect( + RealtimeCallClient, + Effect.gen(function* () { + const http = yield* HttpClient.HttpClient; + return RealtimeCallClient.of({ + createCall: (input) => createCall(http, input), + openSideband, + }); + }), +); diff --git a/apps/server/src/voiceCall/VoiceCallAuth.ts b/apps/server/src/voiceCall/VoiceCallAuth.ts new file mode 100644 index 0000000000000..78a1bc43dde38 --- /dev/null +++ b/apps/server/src/voiceCall/VoiceCallAuth.ts @@ -0,0 +1,137 @@ +/** + * Fork: the ChatGPT login a voice call runs on is the Codex login of the + * default Codex instance. It is read fresh for every call and never written: + * Codex owns refreshing it. + */ +import { + CodexSettings, + ProviderInstanceId, + type ServerSettings, + type VoiceCallUnavailableReason, +} from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; + +import { resolveCodexHomeLayout } from "../provider/Drivers/CodexHomeLayout.ts"; +import type { RealtimeAuth } from "./RealtimeCallClient.ts"; + +const CodexAuthFile = Schema.Struct({ + auth_mode: Schema.optional(Schema.NullOr(Schema.String)), + tokens: Schema.optional( + Schema.NullOr( + Schema.Struct({ + access_token: Schema.optional(Schema.String), + account_id: Schema.optional(Schema.NullOr(Schema.String)), + }), + ), + ), +}); +const decodeCodexAuthFile = Schema.decodeUnknownOption(Schema.fromJsonString(CodexAuthFile)); +const decodeCodexSettings = Schema.decodeUnknownOption(CodexSettings); +const decodeCodexSettingsSync = Schema.decodeSync(CodexSettings); +const defaultCodexSettings = () => decodeCodexSettingsSync({}); + +/** A token about to expire would end the call mid-sentence; ask for a fresh login instead. */ +const EXPIRY_MARGIN_MS = 5 * 60_000; + +export type VoiceCallAuthResult = + | { readonly _tag: "ready"; readonly auth: RealtimeAuth } + | { + readonly _tag: "unavailable"; + readonly availability: { + readonly available: false; + readonly reason: VoiceCallUnavailableReason; + readonly message: string; + }; + }; + +const unavailable = (reason: VoiceCallUnavailableReason, message: string): VoiceCallAuthResult => ({ + _tag: "unavailable", + availability: { available: false, reason, message }, +}); + +export const jwtExpiresAtMs = (token: string): number | null => { + const payload = token.split(".")[1]; + if (!payload) return null; + try { + const claims: unknown = JSON.parse(Buffer.from(payload, "base64url").toString("utf8")); + return typeof claims === "object" && + claims !== null && + "exp" in claims && + typeof claims.exp === "number" + ? claims.exp * 1000 + : null; + } catch { + return null; + } +}; + +export const codexAuthFromFile = (contents: string, nowMs: number): VoiceCallAuthResult => { + const file = decodeCodexAuthFile(contents); + if (Option.isNone(file)) { + return unavailable( + "codex-not-logged-in", + "Log in to Codex with ChatGPT to start a voice call.", + ); + } + const accessToken = file.value.tokens?.access_token; + const accountId = file.value.tokens?.account_id; + if (file.value.auth_mode !== "chatgpt" || !accessToken || !accountId) { + return unavailable( + "codex-api-key", + "Voice calls need Codex logged in with ChatGPT, not an API key.", + ); + } + const expiresAtMs = jwtExpiresAtMs(accessToken); + if (expiresAtMs !== null && expiresAtMs - nowMs < EXPIRY_MARGIN_MS) { + return unavailable( + "codex-login-expired", + "The Codex login has expired. Run a Codex turn or `codex login` to renew it.", + ); + } + return { _tag: "ready", auth: { accessToken, accountId } }; +}; + +/** Follows how the Codex driver finds its home: instance setting, then CODEX_HOME, then ~/.codex. */ +const codexAuthPath = Effect.fn("VoiceCallAuth.codexAuthPath")(function* ( + settings: ServerSettings, +) { + const path = yield* Path.Path; + const instance = settings.providerInstances[ProviderInstanceId.make("codex")]; + const config = Option.getOrElse( + decodeCodexSettings(instance?.config ?? {}), + defaultCodexSettings, + ); + const environmentHome = ( + instance?.environment?.findLast((variable) => variable.name === "CODEX_HOME")?.value ?? + process.env.CODEX_HOME + )?.trim(); + const layout = yield* resolveCodexHomeLayout( + !config.homePath.trim() && !config.shadowHomePath.trim() && environmentHome + ? { ...config, homePath: environmentHome } + : config, + ); + return path.join(layout.effectiveHomePath ?? layout.sharedHomePath, "auth.json"); +}); + +export const readVoiceCallAuth = Effect.fn("VoiceCallAuth.read")(function* ( + settings: ServerSettings, +) { + if (!settings.enableVoiceCall) { + return unavailable("disabled", "Voice calls are turned off in Settings."); + } + const fs = yield* FileSystem.FileSystem; + const authPath = yield* codexAuthPath(settings); + const contents = yield* fs.readFileString(authPath).pipe(Effect.option); + if (Option.isNone(contents)) { + return unavailable( + "codex-not-logged-in", + "Log in to Codex with ChatGPT to start a voice call.", + ); + } + return codexAuthFromFile(contents.value, yield* Clock.currentTimeMillis); +}); diff --git a/apps/server/src/voiceCall/VoiceCallService.test.ts b/apps/server/src/voiceCall/VoiceCallService.test.ts new file mode 100644 index 0000000000000..8ab15b1b870b9 --- /dev/null +++ b/apps/server/src/voiceCall/VoiceCallService.test.ts @@ -0,0 +1,238 @@ +// @effect-diagnostics nodeBuiltinImport:off - fixtures load synchronously at module scope. +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import { + type OrchestrationV2ThreadShell, + ProjectId, + type ServerSettings, + ThreadId, + type VoiceCallEvent, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import * as NodeFS from "node:fs"; + +import { + ThreadManagementService, + type ThreadManagementSendInput, +} from "../orchestration-v2/ThreadManagementService.ts"; +import { ServerSettingsService } from "../serverSettings.ts"; +import { RealtimeCallClient } from "./RealtimeCallClient.ts"; +import { + decodeRealtimeEvent, + type RealtimeEvent, + type RealtimeOutbound, +} from "./realtimeEvents.ts"; +import { noTargetAnswer } from "./voiceCallPrompt.ts"; +import * as VoiceCall from "./VoiceCallService.ts"; + +const fixture = (name: string): ReadonlyArray<RealtimeEvent> => + NodeFS.readFileSync(new URL(`./__fixtures__/${name}.jsonl`, import.meta.url), "utf8") + .split("\n") + .filter((line) => line.trim().length > 0) + .flatMap((line) => Option.toArray(decodeRealtimeEvent(JSON.parse(line)))); + +const TARGET = ThreadId.make("thread-alpha"); +const ARCHIVED = ThreadId.make("thread-archived"); +const shell = (id: ThreadId, archived: boolean) => + ({ + id, + title: id === TARGET ? "Alpha" : "Old work", + projectId: ProjectId.make("project-1"), + archivedAt: archived ? DateTime.makeUnsafe(0) : null, + deletedAt: null, + }) as OrchestrationV2ThreadShell; + +const fakeJwt = (expSeconds: number) => + `e30.${Buffer.from(JSON.stringify({ exp: expSeconds })).toString("base64url")}.sig`; + +/** A realtime session the test feeds events into and reads sent events from. */ +const makeHarness = Effect.gen(function* () { + const inbound = yield* Queue.unbounded<RealtimeEvent, Cause.Done>(); + const outbound = yield* Queue.unbounded<RealtimeOutbound | "closed">(); + const sent: Array<ThreadManagementSendInput> = []; + const realtime = Layer.succeed(RealtimeCallClient)({ + createCall: () => Effect.succeed({ answerSdp: "v=0 answer", realtimeCallId: "rtc_fixture" }), + openSideband: () => + Effect.acquireRelease( + Effect.succeed({ + events: Stream.fromQueue(inbound), + send: (event: RealtimeOutbound) => Queue.offer(outbound, event).pipe(Effect.asVoid), + }), + () => Queue.offer(outbound, "closed"), + ), + }); + const management = Layer.mock(ThreadManagementService)({ + getThreadShell: (threadId) => + Effect.succeed( + threadId === TARGET + ? shell(TARGET, false) + : threadId === ARCHIVED + ? shell(ARCHIVED, true) + : null, + ), + sendToThread: (input) => + Effect.sync(() => void sent.push(input)).pipe( + Effect.as({ message: { id: input.messageId }, run: { id: "run-1" } } as never), + ), + }); + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const codexHome = yield* fs.makeTempDirectoryScoped(); + yield* fs.writeFileString( + path.join(codexHome, "auth.json"), + JSON.stringify({ + auth_mode: "chatgpt", + tokens: { access_token: fakeJwt(4_000_000_000), account_id: "account-1" }, + }), + ); + const settings = ServerSettingsService.layerTest({ + enableVoiceCall: true, + providerInstances: { codex: { driver: "codex", config: { homePath: codexHome } } }, + } as Partial<ServerSettings> as never); + const layer = VoiceCall.layer.pipe( + Layer.provide(Layer.mergeAll(realtime, management, settings, NodeServices.layer)), + ); + return { inbound, outbound, sent, layer }; +}); + +/** Subscribes as `clientId` and waits for the snapshot, so the server counts the subscription. */ +const subscribeAs = (clientId: string) => + Effect.gen(function* () { + const service = yield* VoiceCall.VoiceCallService; + const events = yield* Queue.unbounded<VoiceCallEvent>(); + const fiber = yield* service.subscribe(clientId).pipe( + Stream.runForEach((event) => Queue.offer(events, event)), + Effect.forkScoped, + ); + const snapshot = yield* Queue.take(events); + assert.equal(snapshot.type, "snapshot"); + return { events, fiber }; + }); + +const nextOf = <T extends VoiceCallEvent["type"]>( + events: Queue.Queue<VoiceCallEvent>, + type: T, +): Effect.Effect<Extract<VoiceCallEvent, { readonly type: T }>> => + Effect.flatMap(Queue.take(events), (event) => + event.type === type + ? Effect.succeed(event as Extract<VoiceCallEvent, { readonly type: T }>) + : nextOf(events, type), + ); + +const callStates = (events: Queue.Queue<VoiceCallEvent>) => + Effect.map(Queue.takeAll(events), (taken) => + taken.flatMap((event) => (event.type === "call" && event.call ? [event.call.state] : [])), + ); + +describe("VoiceCallService", () => { + it.effect("steers the thread on screen once per delegation, marked as voice", () => + Effect.gen(function* () { + const harness = yield* makeHarness; + yield* Effect.gen(function* () { + const service = yield* VoiceCall.VoiceCallService; + const client = yield* subscribeAs("tab-1"); + const result = yield* service.connect({ + clientId: "tab-1", + offerSdp: "v=0 offer", + target: TARGET, + }); + assert.equal(result.answerSdp, "v=0 answer"); + assert.equal(result.segmentId, "rtc_fixture"); + + const events = fixture("a1_v3_dc").filter((event) => event.type !== "session.closed"); + const delegation = events.find((event) => event.type === "delegation.created"); + assert.isDefined(delegation); + // A redelivered event must not steer twice. + yield* Queue.offerAll(harness.inbound, [...events, delegation!]); + const pushed = yield* nextOf(client.events, "delegation"); + assert.equal(pushed.delegation.status, "open"); + + assert.equal(harness.sent.length, 1); + const message = harness.sent[0]!; + assert.equal(message.threadId, TARGET); + assert.equal(message.mode, "auto"); + assert.equal(message.commandId, `voice-call:${result.callId}:item_Eblngey9L7Mxg91SDSMZ6`); + assert.match(message.text, /^\[Voice\]\(t3-context:\/\/v1\/voice-call\/voice-item_/); + assert.include(message.text, "run the unit tests"); + assert.equal(message.context?.records[0]?.kind, VoiceCall.VOICE_CALL_CONTEXT_KIND); + + const [recorded] = yield* service.delegations; + assert.equal(recorded?.status, "open"); + assert.equal(recorded?.runId, "run-1"); + }).pipe(Effect.provide(harness.layer)); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + + it.effect("answers at once when no thread or an archived one is on screen", () => + Effect.gen(function* () { + const harness = yield* makeHarness; + yield* Effect.gen(function* () { + const service = yield* VoiceCall.VoiceCallService; + yield* subscribeAs("tab-1"); + yield* service.connect({ clientId: "tab-1", offerSdp: "v=0 offer", target: null }); + yield* Queue.offer(harness.inbound, { + type: "delegation.created", + itemId: "item_one", + text: "Run the tests", + }); + const first = yield* Queue.take(harness.outbound); + assert.deepEqual(first, { + type: "delegation.context.append", + delegation_item_id: "item_one", + channel: "speakable", + content: [{ type: "input_text", text: noTargetAnswer }], + }); + + yield* service.setTarget(ARCHIVED); + yield* Queue.offer(harness.inbound, { + type: "delegation.created", + itemId: "item_two", + text: "Run the tests", + }); + const second = yield* Queue.take(harness.outbound); + assert.include(JSON.stringify(second), "archived"); + assert.equal(harness.sent.length, 0); + const statuses = (yield* service.delegations).map((delegation) => delegation.status); + assert.deepEqual(statuses, ["failed", "failed"]); + }).pipe(Effect.provide(harness.layer)); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + + it.effect("closes the session when the owning client goes away", () => + Effect.gen(function* () { + const harness = yield* makeHarness; + yield* Effect.gen(function* () { + const service = yield* VoiceCall.VoiceCallService; + const watcher = yield* subscribeAs("tab-2"); + const owner = yield* subscribeAs("tab-1"); + yield* service.connect({ clientId: "tab-1", offerSdp: "v=0 offer", target: TARGET }); + yield* Fiber.interrupt(owner.fiber); + assert.equal(yield* Queue.take(harness.outbound), "closed"); + const states = yield* callStates(watcher.events); + assert.deepEqual(states, ["connecting", "live", "error"]); + }).pipe(Effect.provide(harness.layer)); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); + + it.effect("refuses to connect a client that does not watch the call", () => + Effect.gen(function* () { + const harness = yield* makeHarness; + yield* Effect.gen(function* () { + const service = yield* VoiceCall.VoiceCallService; + const exit = yield* service + .connect({ clientId: "tab-1", offerSdp: "v=0 offer", target: TARGET }) + .pipe(Effect.exit); + assert.isTrue(exit._tag === "Failure"); + }).pipe(Effect.provide(harness.layer)); + }).pipe(Effect.provide(NodeServices.layer), Effect.scoped), + ); +}); diff --git a/apps/server/src/voiceCall/VoiceCallService.ts b/apps/server/src/voiceCall/VoiceCallService.ts new file mode 100644 index 0000000000000..fb1f814af9dde --- /dev/null +++ b/apps/server/src/voiceCall/VoiceCallService.ts @@ -0,0 +1,489 @@ +/** + * Fork: one voice call per environment. The client's WebRTC peer carries the + * audio; this service creates each realtime session (a segment), holds its + * sideband, hands the voice's delegations to the thread on screen and tells + * every client how the call stands. See docs/internals/voice-call.md. + */ +import { + CommandId, + DEFAULT_VOICE_CALL_VOICE, + MessageId, + type OrchestrationMessageContext, + type ThreadId, + type VoiceCallAvailability, + type VoiceCallConnectInput, + type VoiceCallConnectResult, + type VoiceCallDelegation, + VoiceCallDisabledError, + VoiceCallError, + type VoiceCallEvent, + type VoiceCallStatus, +} from "@t3tools/contracts"; +import { formatComposerContextReference } from "@t3tools/shared/composerContextReferences"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as PubSub from "effect/PubSub"; +import * as Scope from "effect/Scope"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; + +import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts"; +import { ServerSettingsService } from "../serverSettings.ts"; +import { RealtimeCallClient, type RealtimeSideband } from "./RealtimeCallClient.ts"; +import { + delegationAnswerEvents, + type RealtimeEvent, + type RealtimeOutbound, + type RealtimeSpeaker, + sessionUpdateEvents, +} from "./realtimeEvents.ts"; +import { readVoiceCallAuth } from "./VoiceCallAuth.ts"; +import { + archivedTargetAnswer, + BACKEND_HINT, + dispatchFailedAnswer, + frontEndInstructions, + noTargetAnswer, +} from "./voiceCallPrompt.ts"; + +export const VOICE_CALL_CONTEXT_KIND = "voice-call"; + +export interface VoiceCallSegmentHandle { + readonly id: string; + readonly send: (event: RealtimeOutbound) => Effect.Effect<void>; +} + +export class VoiceCallService extends Context.Service< + VoiceCallService, + { + readonly subscribe: (clientId: string) => Stream.Stream<VoiceCallEvent>; + readonly connect: ( + input: VoiceCallConnectInput, + ) => Effect.Effect<VoiceCallConnectResult, VoiceCallError | VoiceCallDisabledError>; + readonly setTarget: (target: ThreadId | null) => Effect.Effect<void>; + readonly hangUp: Effect.Effect<void>; + /** The delegation the voice is waiting on gets this text as its answer. */ + readonly answerDelegation: (delegationId: string, text: string) => Effect.Effect<void>; + /** Something the voice should say without being asked. */ + readonly speak: (text: string) => Effect.Effect<void>; + readonly delegations: Effect.Effect<ReadonlyArray<VoiceCallDelegation>>; + readonly targetThreadId: Effect.Effect<ThreadId | null>; + /** Every realtime event of the live segment, for the read-back. */ + readonly realtimeEvents: Stream.Stream<{ + readonly segmentId: string; + readonly event: RealtimeEvent; + }>; + readonly updateDelegation: ( + delegationId: string, + patch: Partial<Pick<VoiceCallDelegation, "status" | "answer">>, + ) => Effect.Effect<void>; + } +>()("t3/voiceCall/VoiceCallService") {} + +interface Segment { + readonly id: string; + readonly scope: Scope.Closeable; + readonly sideband: RealtimeSideband; + readonly turnSpeakers: Map<string, RealtimeSpeaker>; +} + +const contextIdFor = (delegationId: string) => + `voice-${delegationId.replace(/[^a-z0-9_-]/gi, "").slice(-100)}`; + +/** The steered message: a voice chip, then the voice's instruction. */ +export const voiceDelegationMessage = (input: { + readonly callId: string; + readonly delegationId: string; + readonly text: string; +}): { readonly text: string; readonly context: OrchestrationMessageContext } => { + const contextId = contextIdFor(input.delegationId); + const reference = formatComposerContextReference({ + kind: VOICE_CALL_CONTEXT_KIND, + contextId: contextId as never, + label: "Voice", + }); + return { + text: `${reference} ${input.text}`, + context: { + version: 1, + records: [ + { + version: 1, + contextId: contextId as never, + label: "Voice", + kind: VOICE_CALL_CONTEXT_KIND, + payload: { hint: BACKEND_HINT, callId: input.callId, delegationId: input.delegationId }, + }, + ], + }, + }; +}; + +const make = Effect.gen(function* () { + const settingsService = yield* ServerSettingsService; + const threads = yield* ThreadManagementService; + const realtime = yield* RealtimeCallClient; + const crypto = yield* Crypto.Crypto; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const lifecycle = yield* Semaphore.make(1); + const changes = yield* Effect.acquireRelease(PubSub.unbounded<VoiceCallEvent>(), (pubsub) => + PubSub.shutdown(pubsub), + ); + const realtimeEvents = yield* Effect.acquireRelease( + PubSub.unbounded<{ readonly segmentId: string; readonly event: RealtimeEvent }>(), + (pubsub) => PubSub.shutdown(pubsub), + ); + + let call: VoiceCallStatus | null = null; + let segment: Segment | null = null; + const delegations = new Map<string, VoiceCallDelegation>(); + const subscriptions = new Map<string, number>(); + + const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + const uuid = crypto.randomUUIDv4().pipe(Effect.orDie); + const publish = (event: VoiceCallEvent) => PubSub.publish(changes, event).pipe(Effect.asVoid); + const publishCall = Effect.suspend(() => publish({ type: "call", call })); + const setCall = (patch: Partial<VoiceCallStatus>) => + Effect.suspend(() => { + if (call === null) return Effect.void; + call = { ...call, ...patch }; + return publishCall; + }); + const recordDelegation = (delegation: VoiceCallDelegation) => + Effect.suspend(() => { + delegations.set(delegation.id, delegation); + return publish({ type: "delegation", delegation }); + }); + + const readAuth = Effect.gen(function* () { + const settings = yield* settingsService.getSettings.pipe( + Effect.mapError(() => new VoiceCallError({ message: "Could not read the settings." })), + ); + const auth = yield* readVoiceCallAuth(settings).pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(Path.Path, path), + ); + return { settings, auth }; + }); + + const availability: Effect.Effect<VoiceCallAvailability> = readAuth.pipe( + Effect.map(({ auth }) => + auth._tag === "ready" ? { available: true, reason: null, message: null } : auth.availability, + ), + Effect.catch((error) => + Effect.succeed({ available: false, reason: null, message: error.message }), + ), + ); + + const sendAll = (events: ReadonlyArray<RealtimeOutbound>) => + Effect.suspend(() => { + const current = segment; + if (current === null) return Effect.void; + return Effect.forEach(events, current.sideband.send, { discard: true }).pipe( + Effect.catch((error) => Effect.logWarning("voice call: send failed", error.message)), + ); + }); + + /** Closing the scope sends `session.close`; the realtime session stops billing. */ + const closeSegment = Effect.suspend(() => { + const current = segment; + segment = null; + return current === null ? Effect.void : Scope.close(current.scope, Exit.void); + }); + + const endCall = (error: string | null) => + Effect.gen(function* () { + yield* closeSegment; + if (call !== null && (call.state === "connecting" || call.state === "live")) { + yield* setCall({ state: error === null ? "ended" : "error", segmentId: null, error }); + } + }); + + const threadTitle = (threadId: ThreadId) => + threads.getThreadShell(threadId).pipe( + Effect.map((shell) => shell?.title ?? null), + Effect.orElseSucceed(() => null), + ); + + const answerDelegation = (delegationId: string, text: string) => + sendAll(delegationAnswerEvents(delegationId, text)); + + const steer = (callId: string, segmentId: string, delegationId: string, text: string) => + Effect.gen(function* () { + if (delegations.has(delegationId)) return; + const targetThreadId = call?.targetThreadId ?? null; + const base: VoiceCallDelegation = { + id: delegationId, + segmentId, + text, + threadId: targetThreadId, + messageId: null, + runId: null, + status: "open", + answer: null, + createdAt: yield* nowIso, + }; + const fail = (answer: string) => + Effect.andThen( + recordDelegation({ ...base, status: "failed", answer }), + answerDelegation(delegationId, answer), + ); + if (targetThreadId === null) return yield* fail(noTargetAnswer); + const shell = yield* threads + .getThreadShell(targetThreadId) + .pipe(Effect.orElseSucceed(() => null)); + if (shell === null) return yield* fail(noTargetAnswer); + if (shell.archivedAt !== null) return yield* fail(archivedTargetAnswer(shell.title)); + const message = voiceDelegationMessage({ callId, delegationId, text }); + const sent = yield* threads + .sendToThread({ + projectId: shell.projectId, + commandId: CommandId.make(`voice-call:${callId}:${delegationId}`), + threadId: targetThreadId, + messageId: MessageId.make(`voice-call-message:${callId}:${delegationId}`), + text: message.text, + context: message.context, + attachments: [], + mode: "auto", + createdBy: "user", + creationSource: "server", + }) + .pipe(Effect.option); + if (Option.isNone(sent)) return yield* fail(dispatchFailedAnswer(shell.title)); + yield* recordDelegation({ + ...base, + messageId: sent.value.message.id, + runId: sent.value.run.id, + }); + }); + + const handleEvent = (callId: string, current: Segment, event: RealtimeEvent) => + Effect.gen(function* () { + yield* PubSub.publish(realtimeEvents, { segmentId: current.id, event }); + switch (event.type) { + case "turn.created": + current.turnSpeakers.set(event.turnId, event.speaker); + return; + case "turn.delta": { + const speaker = current.turnSpeakers.get(event.turnId); + if (speaker === undefined) return; + return yield* publish({ + type: "transcript", + speaker, + turnId: event.turnId, + text: event.text, + done: false, + }); + } + case "turn.done": + current.turnSpeakers.delete(event.turnId); + return yield* publish({ + type: "transcript", + speaker: event.speaker, + turnId: event.turnId, + text: event.text, + done: true, + }); + case "delegation.created": + return yield* steer(callId, current.id, event.itemId, event.text); + case "error": + yield* Effect.logWarning("voice call: realtime error", event.code, event.message); + if (event.code === "forbidden") yield* setCall({ error: event.message }); + return; + default: + return; + } + }); + + /** The sideband ended on its own: the session closed or the socket dropped. */ + const segmentEnded = (current: Segment, reason: string | null) => + lifecycle.withPermits(1)( + Effect.suspend(() => + segment !== current + ? Effect.void + : endCall( + reason === null || reason === "client_request" + ? null + : `The voice session ended (${reason}).`, + ), + ), + ); + + const connect: VoiceCallService["Service"]["connect"] = (input) => + lifecycle.withPermits(1)( + Effect.gen(function* () { + if ((subscriptions.get(input.clientId) ?? 0) === 0) { + return yield* new VoiceCallError({ message: "Subscribe to the call before connecting." }); + } + const { settings, auth } = yield* readAuth; + if (auth._tag === "unavailable") { + if (auth.availability.reason === "disabled") { + return yield* new VoiceCallDisabledError({ message: auth.availability.message }); + } + return yield* new VoiceCallError({ + message: auth.availability.message, + reason: auth.availability.reason, + }); + } + // A retry, a takeover by another client and a fresh start all replace + // the segment; only a call that already ended starts over. + yield* closeSegment; + const continuing = call !== null && (call.state === "connecting" || call.state === "live"); + if (!continuing) delegations.clear(); + const callId = continuing && call !== null ? call.callId : `vc-${yield* uuid}`; + call = { + callId, + state: "connecting", + ownerClientId: input.clientId, + segmentId: null, + targetThreadId: input.target, + startedAt: continuing && call !== null ? call.startedAt : yield* nowIso, + error: null, + }; + yield* publishCall; + + const targetTitle = input.target === null ? null : yield* threadTitle(input.target); + const created = yield* realtime + .createCall({ + auth: auth.auth, + offerSdp: input.offerSdp, + session: { + instructions: frontEndInstructions({ + language: settings.voiceCallLanguage, + persona: settings.voiceCallPersona, + targetTitle, + nowIso: yield* nowIso, + }), + voice: settings.voiceCallVoice ?? DEFAULT_VOICE_CALL_VOICE, + }, + }) + .pipe(Effect.tapError((error) => setCall({ state: "error", error: error.message }))); + + const scope = yield* Scope.make(); + const sideband = yield* realtime + .openSideband({ auth: auth.auth, realtimeCallId: created.realtimeCallId }) + .pipe( + Scope.provide(scope), + Effect.tapError((error) => + Effect.andThen( + Scope.close(scope, Exit.void), + setCall({ state: "error", error: error.message }), + ), + ), + ); + const current: Segment = { + id: created.realtimeCallId, + scope, + sideband, + turnSpeakers: new Map(), + }; + segment = current; + yield* sideband.events.pipe( + Stream.runForEach((event) => + event.type === "session.closed" + ? segmentEnded(current, event.reason) + : handleEvent(callId, current, event), + ), + Effect.andThen(segmentEnded(current, "disconnected")), + Effect.forkIn(scope), + ); + yield* setCall({ state: "live", segmentId: current.id }); + return { callId, segmentId: current.id, answerSdp: created.answerSdp }; + }), + ); + + const hangUp = lifecycle.withPermits(1)(endCall(null)); + + const setTarget = (target: ThreadId | null) => setCall({ targetThreadId: target }); + + const subscribe = (clientId: string): Stream.Stream<VoiceCallEvent> => + Stream.unwrap( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(changes); + subscriptions.set(clientId, (subscriptions.get(clientId) ?? 0) + 1); + const snapshot: VoiceCallEvent = { + type: "snapshot", + availability: yield* availability, + call, + delegations: [...delegations.values()], + }; + return Stream.concat(Stream.make(snapshot), Stream.fromSubscription(subscription)); + }), + ).pipe( + Stream.ensuring( + Effect.suspend(() => { + const remaining = (subscriptions.get(clientId) ?? 1) - 1; + if (remaining > 0) { + subscriptions.set(clientId, remaining); + return Effect.void; + } + subscriptions.delete(clientId); + // The owner's peer died with its last subscription: stop paying for the session. + return call?.ownerClientId === clientId && segment !== null + ? lifecycle.withPermits(1)(endCall("The client carrying the call went away.")) + : Effect.void; + }), + ), + ); + + // A call must not outlive the flag: turning it off hangs up and tells every client. + yield* settingsService.streamChanges.pipe( + Stream.map((settings) => settings.enableVoiceCall), + Stream.changes, + Stream.runForEach((enabled) => + Effect.gen(function* () { + if (!enabled) yield* hangUp; + yield* publish({ type: "availability", availability: yield* availability }); + }), + ), + Effect.forkScoped, + ); + yield* Effect.addFinalizer(() => closeSegment); + + return VoiceCallService.of({ + subscribe, + connect, + setTarget, + hangUp, + answerDelegation, + speak: (text) => sendAll(sessionUpdateEvents(text)), + delegations: Effect.sync(() => [...delegations.values()]), + targetThreadId: Effect.sync(() => call?.targetThreadId ?? null), + realtimeEvents: Stream.fromPubSub(realtimeEvents), + updateDelegation: (delegationId, patch) => + Effect.suspend(() => { + const existing = delegations.get(delegationId); + return existing === undefined ? Effect.void : recordDelegation({ ...existing, ...patch }); + }), + }); +}); + +export const layer = Layer.effect(VoiceCallService, make); + +const withService = <A, E>(use: (service: VoiceCallService["Service"]) => Effect.Effect<A, E>) => + Effect.gen(function* () { + const service = yield* Effect.serviceOption(VoiceCallService); + if (Option.isNone(service)) { + return yield* new VoiceCallError({ message: "This server does not offer voice calls." }); + } + return yield* use(service.value); + }); + +export const subscribeRpc = (input: { readonly clientId: string }) => + Stream.unwrap(withService((service) => Effect.succeed(service.subscribe(input.clientId)))); + +export const connectRpc = (input: VoiceCallConnectInput) => + withService((service) => service.connect(input)); + +export const setTargetRpc = (input: { readonly target: ThreadId | null }) => + withService((service) => service.setTarget(input.target)).pipe(Effect.as({})); + +export const hangUpRpc = () => withService((service) => service.hangUp).pipe(Effect.as({})); diff --git a/apps/server/src/voiceCall/__fixtures__/a1_v3_dc.jsonl b/apps/server/src/voiceCall/__fixtures__/a1_v3_dc.jsonl new file mode 100644 index 0000000000000..29afaa1b6a039 --- /dev/null +++ b/apps/server/src/voiceCall/__fixtures__/a1_v3_dc.jsonl @@ -0,0 +1,93 @@ +{"type": "session.started", "session": {"id": "rtc_fixture", "status": "active", "expires_at": 1792677779}} +{"type": "input_transcript.added", "end_ms": 2200, "item": {"id": "item_Eblnace3qgStyxjB4c6YO", "type": "input_transcript", "text": "Please"}, "start_ms": 2000} +{"type": "turn.created", "turn": {"id": "turn_EblnaoTCinA9dOtWSoHHk", "end_ms": 2200, "role": "user", "start_ms": 2000, "transcript": "Please"}} +{"type": "input_transcript.added", "end_ms": 2400, "item": {"id": "item_Eblnaf77tDKQi8Ag1xFTy", "type": "input_transcript", "text": " tell"}, "start_ms": 2200} +{"type": "turn.delta", "delta": " tell", "end_ms": 2400, "start_ms": 2200, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 2600, "item": {"id": "item_Eblnaz1R6lPX7qSf6GSmp", "type": "input_transcript", "text": " the"}, "start_ms": 2400} +{"type": "turn.delta", "delta": " the", "end_ms": 2600, "start_ms": 2400, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 3000, "item": {"id": "item_EblnbqgYaGE3kEDfGszXB", "type": "input_transcript", "text": " coding"}, "start_ms": 2800} +{"type": "turn.delta", "delta": " coding", "end_ms": 3000, "start_ms": 2800, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 3400, "item": {"id": "item_EblnbNOcK1OvEbRKsrU1g", "type": "input_transcript", "text": " agent in"}, "start_ms": 3200} +{"type": "turn.delta", "delta": " agent in", "end_ms": 3400, "start_ms": 3200, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 3800, "item": {"id": "item_Eblnc49T6QoQbA2KegjlU", "type": "input_transcript", "text": " thread"}, "start_ms": 3600} +{"type": "turn.delta", "delta": " thread", "end_ms": 3800, "start_ms": 3600, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 4000, "item": {"id": "item_Eblncb6Dg8KZPQJVzHh2E", "type": "input_transcript", "text": " alpha"}, "start_ms": 3800} +{"type": "turn.delta", "delta": " alpha", "end_ms": 4000, "start_ms": 3800, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 4200, "item": {"id": "item_EblncPZmhesmaNkMHuN3I", "type": "input_transcript", "text": " to"}, "start_ms": 4000} +{"type": "turn.delta", "delta": " to", "end_ms": 4200, "start_ms": 4000, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 4400, "item": {"id": "item_EblncvacIwEJtBMv4cWsA", "type": "input_transcript", "text": " run"}, "start_ms": 4200} +{"type": "turn.delta", "delta": " run", "end_ms": 4400, "start_ms": 4200, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 4600, "item": {"id": "item_EblnciIun6Wm57c5mbrxR", "type": "input_transcript", "text": " the"}, "start_ms": 4400} +{"type": "turn.delta", "delta": " the", "end_ms": 4600, "start_ms": 4400, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 4800, "item": {"id": "item_Eblnd7NzVBxp91jFI2vuO", "type": "input_transcript", "text": " unit"}, "start_ms": 4600} +{"type": "turn.delta", "delta": " unit", "end_ms": 4800, "start_ms": 4600, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 5400, "item": {"id": "item_EblndEHHkJ0oDPRFyttps", "type": "input_transcript", "text": " tests"}, "start_ms": 5200} +{"type": "turn.delta", "delta": " tests", "end_ms": 5400, "start_ms": 5200, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 6000, "item": {"id": "item_EblneXkocsfxUqwKma9CK", "type": "input_transcript", "text": "; send it"}, "start_ms": 5800} +{"type": "turn.delta", "delta": "; send it", "end_ms": 6000, "start_ms": 5800, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 6200, "item": {"id": "item_EblnerYJ6IP6OdMw4Ty0f", "type": "input_transcript", "text": " as a"}, "start_ms": 6000} +{"type": "turn.delta", "delta": " as a", "end_ms": 6200, "start_ms": 6000, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "input_transcript.added", "end_ms": 7200, "item": {"id": "item_Eblnfurs6nzDExqiaHZFY", "type": "input_transcript", "text": " queued message"}, "start_ms": 7000} +{"type": "turn.delta", "delta": " queued message", "end_ms": 7200, "start_ms": 7000, "turn_id": "turn_EblnaoTCinA9dOtWSoHHk"} +{"type": "delegation.created", "item": {"id": "item_Eblngey9L7Mxg91SDSMZ6", "type": "delegation", "content": [{"type": "input_text", "text": "Please tell the coding agent in thread alpha to run the unit tests; send it as a queued message"}], "handoff_id": "handoff_1", "target": "client", "user_bidi_turn_id": "turn_EblnaoTCinA9dOtWSoHHk"}, "offset_ms": 7600} +{"type": "output_transcript.added", "end_ms": 8000, "item": {"id": "item_Eblng31tAQBdTUSNLw9Ot", "type": "output_transcript", "text": "On"}, "start_ms": 7800} +{"type": "turn.done", "turn": {"id": "turn_EblnaoTCinA9dOtWSoHHk", "end_ms": 8000, "role": "user", "start_ms": 2000, "transcript": "Please tell the coding agent in thread alpha to run the unit tests; send it as a queued message"}} +{"type": "turn.created", "turn": {"id": "turn_EblngRap0jY0Kuim48t6t", "end_ms": 8000, "role": "assistant", "start_ms": 7800, "transcript": "On"}} +{"type": "output_transcript.added", "end_ms": 8400, "item": {"id": "item_EblngxdPFbp63KJLflbFR", "type": "output_transcript", "text": " that"}, "start_ms": 8200} +{"type": "turn.delta", "delta": " that", "end_ms": 8400, "start_ms": 8200, "turn_id": "turn_EblngRap0jY0Kuim48t6t"} +{"type": "output_transcript.added", "end_ms": 8600, "item": {"id": "item_EblnhZgh4gjNWj8c5oJVi", "type": "output_transcript", "text": ", checking"}, "start_ms": 8400} +{"type": "turn.delta", "delta": ", checking", "end_ms": 8600, "start_ms": 8400, "turn_id": "turn_EblngRap0jY0Kuim48t6t"} +{"type": "output_transcript.added", "end_ms": 9000, "item": {"id": "item_EblnhGWm2Q9HEX3hPeCPY", "type": "output_transcript", "text": "."}, "start_ms": 8800} +{"type": "turn.delta", "delta": ".", "end_ms": 9000, "start_ms": 8800, "turn_id": "turn_EblngRap0jY0Kuim48t6t"} +{"type": "delegation.context.appended", "delegation_item_id": "item_Eblngey9L7Mxg91SDSMZ6", "end_ms": 11200, "start_ms": 11000} +{"type": "turn.done", "turn": {"id": "turn_EblngRap0jY0Kuim48t6t", "end_ms": 11200, "role": "assistant", "start_ms": 7800, "transcript": "On that, checking."}} +{"type": "output_transcript.added", "end_ms": 11400, "item": {"id": "item_Eblnj7IoBdRine7Hmry9j", "type": "output_transcript", "text": "Done"}, "start_ms": 11200} +{"type": "turn.created", "turn": {"id": "turn_EblnjEfWpRmWfEi9y205n", "end_ms": 11400, "role": "assistant", "start_ms": 11200, "transcript": "Done"}} +{"type": "output_transcript.added", "end_ms": 11600, "item": {"id": "item_EblnktFA5MtnOwkITI6uZ", "type": "output_transcript", "text": "\u2014queued"}, "start_ms": 11400} +{"type": "turn.delta", "delta": "\u2014queued", "end_ms": 11600, "start_ms": 11400, "turn_id": "turn_EblnjEfWpRmWfEi9y205n"} +{"type": "output_transcript.added", "end_ms": 12200, "item": {"id": "item_Eblnkep6ON8uzfMy0L9OL", "type": "output_transcript", "text": " as"}, "start_ms": 12000} +{"type": "turn.delta", "delta": " as", "end_ms": 12200, "start_ms": 12000, "turn_id": "turn_EblnjEfWpRmWfEi9y205n"} +{"type": "output_transcript.added", "end_ms": 12600, "item": {"id": "item_EblnlNUJxtmqmnV5mQ3Tv", "type": "output_transcript", "text": " item"}, "start_ms": 12400} +{"type": "turn.delta", "delta": " item", "end_ms": 12600, "start_ms": 12400, "turn_id": "turn_EblnjEfWpRmWfEi9y205n"} +{"type": "output_transcript.added", "end_ms": 12800, "item": {"id": "item_Eblnld4q70wQu0hkGkZmK", "type": "output_transcript", "text": " "}, "start_ms": 12600} +{"type": "turn.delta", "delta": " ", "end_ms": 12800, "start_ms": 12600, "turn_id": "turn_EblnjEfWpRmWfEi9y205n"} +{"type": "output_transcript.added", "end_ms": 13000, "item": {"id": "item_EblnlxAC33tUl3JdImkMM", "type": "output_transcript", "text": "42"}, "start_ms": 12800} +{"type": "turn.delta", "delta": "42", "end_ms": 13000, "start_ms": 12800, "turn_id": "turn_EblnjEfWpRmWfEi9y205n"} +{"type": "output_transcript.added", "end_ms": 13600, "item": {"id": "item_EblnmRxOHH0Fi8sg7vJIM", "type": "output_transcript", "text": " in"}, "start_ms": 13400} +{"type": "turn.delta", "delta": " in", "end_ms": 13600, "start_ms": 13400, "turn_id": "turn_EblnjEfWpRmWfEi9y205n"} +{"type": "output_transcript.added", "end_ms": 13800, "item": {"id": "item_EblnmSRk3ZseBjKgSiJWV", "type": "output_transcript", "text": " thread"}, "start_ms": 13600} +{"type": "turn.delta", "delta": " thread", "end_ms": 13800, "start_ms": 13600, "turn_id": "turn_EblnjEfWpRmWfEi9y205n"} +{"type": "output_transcript.added", "end_ms": 14200, "item": {"id": "item_EblnmbUE94cMEji5WhtTj", "type": "output_transcript", "text": " alpha"}, "start_ms": 14000} +{"type": "turn.delta", "delta": " alpha", "end_ms": 14200, "start_ms": 14000, "turn_id": "turn_EblnjEfWpRmWfEi9y205n"} +{"type": "session.usage.updated", "usage": {"audio_duration_ms": 14200, "backend_model_usage": []}, "usage_limit": {"status": null, "reset_seconds": null}} +{"type": "output_transcript.added", "end_ms": 14600, "item": {"id": "item_Eblnn2n3ZK9Am7eaz1S9W", "type": "output_transcript", "text": "."}, "start_ms": 14400} +{"type": "turn.delta", "delta": ".", "end_ms": 14600, "start_ms": 14400, "turn_id": "turn_EblnjEfWpRmWfEi9y205n"} +{"type": "turn.done", "turn": {"id": "turn_EblnjEfWpRmWfEi9y205n", "end_ms": 16800, "role": "assistant", "start_ms": 11200, "transcript": "Done\u2014queued as item 42 in thread alpha."}} +{"type": "session.context.appended", "end_ms": 21400, "start_ms": 21200} +{"type": "output_transcript.added", "end_ms": 21800, "item": {"id": "item_EblnuSzXCJMLuRcQzooVI", "type": "output_transcript", "text": " Update"}, "start_ms": 21600} +{"type": "turn.created", "turn": {"id": "turn_Eblnub16LQlluTz6kEJQl", "end_ms": 21800, "role": "assistant", "start_ms": 21600, "transcript": " Update"}} +{"type": "output_transcript.added", "end_ms": 22000, "item": {"id": "item_EblnueiUhSxcXW9dVfFFy", "type": "output_transcript", "text": ":"}, "start_ms": 21800} +{"type": "turn.delta", "delta": ":", "end_ms": 22000, "start_ms": 21800, "turn_id": "turn_Eblnub16LQlluTz6kEJQl"} +{"type": "output_transcript.added", "end_ms": 22200, "item": {"id": "item_Eblnu2sDryaVsP9Q3deeb", "type": "output_transcript", "text": " unit"}, "start_ms": 22000} +{"type": "turn.delta", "delta": " unit", "end_ms": 22200, "start_ms": 22000, "turn_id": "turn_Eblnub16LQlluTz6kEJQl"} +{"type": "output_transcript.added", "end_ms": 22400, "item": {"id": "item_EblnuFcYeqrEgjodggyF7", "type": "output_transcript", "text": " tests"}, "start_ms": 22200} +{"type": "turn.delta", "delta": " tests", "end_ms": 22400, "start_ms": 22200, "turn_id": "turn_Eblnub16LQlluTz6kEJQl"} +{"type": "output_transcript.added", "end_ms": 22800, "item": {"id": "item_EblnvMzzyrETeV7QA7mL4", "type": "output_transcript", "text": " finished"}, "start_ms": 22600} +{"type": "turn.delta", "delta": " finished", "end_ms": 22800, "start_ms": 22600, "turn_id": "turn_Eblnub16LQlluTz6kEJQl"} +{"type": "output_transcript.added", "end_ms": 23200, "item": {"id": "item_EblnvpV6vhkjOpyUZOfbN", "type": "output_transcript", "text": "\u2014"}, "start_ms": 23000} +{"type": "turn.delta", "delta": "\u2014", "end_ms": 23200, "start_ms": 23000, "turn_id": "turn_Eblnub16LQlluTz6kEJQl"} +{"type": "output_transcript.added", "end_ms": 23400, "item": {"id": "item_EblnvYBjs5NKBuZKWCH8W", "type": "output_transcript", "text": "128"}, "start_ms": 23200} +{"type": "turn.delta", "delta": "128", "end_ms": 23400, "start_ms": 23200, "turn_id": "turn_Eblnub16LQlluTz6kEJQl"} +{"type": "output_transcript.added", "end_ms": 24200, "item": {"id": "item_EblnwumjciZzAXYzA7Pew", "type": "output_transcript", "text": " passed"}, "start_ms": 24000} +{"type": "turn.delta", "delta": " passed", "end_ms": 24200, "start_ms": 24000, "turn_id": "turn_Eblnub16LQlluTz6kEJQl"} +{"type": "output_transcript.added", "end_ms": 24800, "item": {"id": "item_EblnxDML2hhPJVSatOY2W", "type": "output_transcript", "text": ","}, "start_ms": 24600} +{"type": "turn.delta", "delta": ",", "end_ms": 24800, "start_ms": 24600, "turn_id": "turn_Eblnub16LQlluTz6kEJQl"} +{"type": "output_transcript.added", "end_ms": 25000, "item": {"id": "item_EblnxHFufC9sOU05eS50P", "type": "output_transcript", "text": " 2"}, "start_ms": 24800} +{"type": "turn.delta", "delta": " 2", "end_ms": 25000, "start_ms": 24800, "turn_id": "turn_Eblnub16LQlluTz6kEJQl"} +{"type": "output_transcript.added", "end_ms": 25200, "item": {"id": "item_EblnxbXtmtLYelST2yVJT", "type": "output_transcript", "text": " failed"}, "start_ms": 25000} +{"type": "turn.delta", "delta": " failed", "end_ms": 25200, "start_ms": 25000, "turn_id": "turn_Eblnub16LQlluTz6kEJQl"} +{"type": "output_transcript.added", "end_ms": 25800, "item": {"id": "item_EblnyrbhspTWHtBGkJV8i", "type": "output_transcript", "text": "."}, "start_ms": 25600} +{"type": "turn.delta", "delta": ".", "end_ms": 25800, "start_ms": 25600, "turn_id": "turn_Eblnub16LQlluTz6kEJQl"} +{"type": "turn.done", "turn": {"id": "turn_Eblnub16LQlluTz6kEJQl", "end_ms": 28000, "role": "assistant", "start_ms": 21600, "transcript": " Update: unit tests finished\u2014128 passed, 2 failed."}} +{"type": "session.usage.updated", "usage": {"audio_duration_ms": 29400, "backend_model_usage": []}, "usage_limit": {"status": null, "reset_seconds": null}} +{"type": "session.closed", "reason": "client_request", "usage": {"audio_duration_ms": 36400, "backend_model_usage": []}} diff --git a/apps/server/src/voiceCall/__fixtures__/l11_v_marin.jsonl b/apps/server/src/voiceCall/__fixtures__/l11_v_marin.jsonl new file mode 100644 index 0000000000000..e21757133a282 --- /dev/null +++ b/apps/server/src/voiceCall/__fixtures__/l11_v_marin.jsonl @@ -0,0 +1 @@ +{"type": "error", "event_id": "event_d672656488c646fd9ece4e045156e6d8", "error": {"message": "Voice session access denied.", "type": "", "code": "forbidden", "param": ""}} diff --git a/apps/server/src/voiceCall/realtimeEvents.test.ts b/apps/server/src/voiceCall/realtimeEvents.test.ts new file mode 100644 index 0000000000000..38136228579c6 --- /dev/null +++ b/apps/server/src/voiceCall/realtimeEvents.test.ts @@ -0,0 +1,72 @@ +import { assert, describe, it } from "@effect/vitest"; +import * as Option from "effect/Option"; + +import { codexAuthFromFile } from "./VoiceCallAuth.ts"; +import { + contextAppendChunks, + decodeRealtimeEvent, + delegationAnswerEvents, +} from "./realtimeEvents.ts"; + +const jwt = (expSeconds: number) => + `e30.${Buffer.from(JSON.stringify({ exp: expSeconds })).toString("base64url")}.sig`; + +describe("realtime events", () => { + it("reads a refused voice as an error the call can show", () => { + const event = decodeRealtimeEvent({ + type: "error", + event_id: "event_1", + error: { message: "Voice session access denied.", type: "", code: "forbidden", param: "" }, + }); + assert.deepEqual(Option.getOrNull(event), { + type: "error", + code: "forbidden", + message: "Voice session access denied.", + }); + }); + + it("ignores audio frames and unknown events", () => { + assert.isTrue(Option.isNone(decodeRealtimeEvent({ type: "session.output_audio.delta" }))); + assert.isTrue(Option.isNone(decodeRealtimeEvent("not an event"))); + }); + + it("splits answers into appends of at most 500 bytes without breaking a character", () => { + const text = "ä".repeat(400); + const chunks = contextAppendChunks(text); + assert.deepEqual( + chunks.map((chunk) => Buffer.byteLength(chunk)), + [500, 300], + ); + assert.equal(chunks.join(""), text); + const events = delegationAnswerEvents("item_1", text); + assert.equal(events.length, 2); + assert.equal(events[0]?.type, "delegation.context.append"); + }); +}); + +describe("codexAuthFromFile", () => { + const now = 1_800_000_000_000; + const file = (authMode: string, exp: number) => + JSON.stringify({ + auth_mode: authMode, + tokens: { access_token: jwt(exp), account_id: "account-1" }, + }); + + it("uses a ChatGPT login", () => { + const result = codexAuthFromFile(file("chatgpt", now / 1000 + 3600), now); + assert.equal(result._tag, "ready"); + }); + + it("refuses an API key login", () => { + const result = codexAuthFromFile(file("apikey", now / 1000 + 3600), now); + assert.equal(result._tag === "unavailable" && result.availability.reason, "codex-api-key"); + }); + + it("asks for a fresh login when the token is about to expire", () => { + const result = codexAuthFromFile(file("chatgpt", now / 1000 + 60), now); + assert.equal( + result._tag === "unavailable" && result.availability.reason, + "codex-login-expired", + ); + }); +}); diff --git a/apps/server/src/voiceCall/realtimeEvents.ts b/apps/server/src/voiceCall/realtimeEvents.ts new file mode 100644 index 0000000000000..0edfba0f65a2b --- /dev/null +++ b/apps/server/src/voiceCall/realtimeEvents.ts @@ -0,0 +1,217 @@ +/** + * Fork: the realtime v3 ("GPT-Live") events the voice call reads and sends. + * The protocol is undocumented; everything here comes from Codex's own client + * and the 2026-10-22 spike (docs/internals/voice-call.md, "Protocol"). + */ +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; + +export type RealtimeSpeaker = "user" | "voice"; + +export type RealtimeEvent = + | { readonly type: "session.started"; readonly sessionId: string; readonly expiresAt: number } + | { readonly type: "transcript"; readonly speaker: RealtimeSpeaker; readonly text: string } + | { + readonly type: "turn.created"; + readonly turnId: string; + readonly speaker: RealtimeSpeaker; + readonly text: string; + } + | { + readonly type: "turn.delta"; + readonly turnId: string; + readonly text: string; + } + | { + readonly type: "turn.done"; + readonly turnId: string; + readonly speaker: RealtimeSpeaker; + readonly text: string; + } + | { readonly type: "delegation.created"; readonly itemId: string; readonly text: string } + | { readonly type: "context.appended"; readonly delegationItemId: string | null } + | { + readonly type: "usage"; + readonly audioDurationMs: number; + readonly limitStatus: string | null; + readonly limitResetSeconds: number | null; + } + | { readonly type: "session.closed"; readonly reason: string | null } + | { readonly type: "error"; readonly code: string; readonly message: string }; + +const Role = Schema.Literals(["user", "assistant"]); +const speakerOf = (role: typeof Role.Type): RealtimeSpeaker => (role === "user" ? "user" : "voice"); +const Text = Schema.Struct({ text: Schema.String }); +const Turn = Schema.Struct({ + id: Schema.String, + role: Role, + transcript: Schema.optional(Schema.String), +}); + +const SessionStarted = Schema.Struct({ + session: Schema.Struct({ id: Schema.String, expires_at: Schema.Number }), +}); +const TranscriptAdded = Schema.Struct({ item: Text }); +const TurnEvent = Schema.Struct({ turn: Turn }); +const TurnDelta = Schema.Struct({ turn_id: Schema.String, delta: Schema.String }); +const DelegationCreated = Schema.Struct({ + item: Schema.Struct({ id: Schema.String, content: Schema.optional(Schema.Array(Text)) }), +}); +const DelegationAppended = Schema.Struct({ delegation_item_id: Schema.String }); +const UsageUpdated = Schema.Struct({ + usage: Schema.Struct({ audio_duration_ms: Schema.Number }), + usage_limit: Schema.optional( + Schema.NullOr( + Schema.Struct({ + status: Schema.optional(Schema.NullOr(Schema.String)), + reset_seconds: Schema.optional(Schema.NullOr(Schema.Number)), + }), + ), + ), +}); +const SessionClosed = Schema.Struct({ reason: Schema.optional(Schema.NullOr(Schema.String)) }); +const ErrorEvent = Schema.Struct({ + error: Schema.Struct({ code: Schema.optional(Schema.String), message: Schema.String }), +}); +const Envelope = Schema.Struct({ type: Schema.String }); +const decodeDelegationAppended = Schema.decodeUnknownOption(DelegationAppended); +const decodeDelegationCreated = Schema.decodeUnknownOption(DelegationCreated); +const decodeEnvelope = Schema.decodeUnknownOption(Envelope); +const decodeErrorEvent = Schema.decodeUnknownOption(ErrorEvent); +const decodeSessionClosed = Schema.decodeUnknownOption(SessionClosed); +const decodeSessionStarted = Schema.decodeUnknownOption(SessionStarted); +const decodeTranscriptAdded = Schema.decodeUnknownOption(TranscriptAdded); +const decodeTurnDelta = Schema.decodeUnknownOption(TurnDelta); +const decodeTurnEvent = Schema.decodeUnknownOption(TurnEvent); +const decodeUsageUpdated = Schema.decodeUnknownOption(UsageUpdated); + +/** The events the call acts on; everything else (audio frames among them) is `None`. */ +export const decodeRealtimeEvent = (raw: unknown): Option.Option<RealtimeEvent> => { + const envelope = decodeEnvelope(raw); + if (Option.isNone(envelope)) return Option.none(); + switch (envelope.value.type) { + case "session.started": + return Option.map(decodeSessionStarted(raw), (event) => ({ + type: "session.started" as const, + sessionId: event.session.id, + expiresAt: event.session.expires_at, + })); + case "input_transcript.added": + case "output_transcript.added": + return Option.map(decodeTranscriptAdded(raw), (event) => ({ + type: "transcript" as const, + speaker: envelope.value.type === "input_transcript.added" ? "user" : "voice", + text: event.item.text, + })); + case "turn.created": + case "turn.done": + return Option.map(decodeTurnEvent(raw), (event) => ({ + type: envelope.value.type as "turn.created" | "turn.done", + turnId: event.turn.id, + speaker: speakerOf(event.turn.role), + text: event.turn.transcript ?? "", + })); + case "turn.delta": + return Option.map(decodeTurnDelta(raw), (event) => ({ + type: "turn.delta" as const, + turnId: event.turn_id, + text: event.delta, + })); + case "delegation.created": + return Option.map(decodeDelegationCreated(raw), (event) => ({ + type: "delegation.created" as const, + itemId: event.item.id, + text: (event.item.content ?? []).map((part) => part.text).join(""), + })); + case "delegation.context.appended": + return Option.map(decodeDelegationAppended(raw), (event) => ({ + type: "context.appended" as const, + delegationItemId: event.delegation_item_id, + })); + case "session.context.appended": + return Option.some({ type: "context.appended", delegationItemId: null }); + case "session.usage.updated": + return Option.map(decodeUsageUpdated(raw), (event) => ({ + type: "usage" as const, + audioDurationMs: event.usage.audio_duration_ms, + limitStatus: event.usage_limit?.status ?? null, + limitResetSeconds: event.usage_limit?.reset_seconds ?? null, + })); + case "session.closed": + return Option.map(decodeSessionClosed(raw), (event) => ({ + type: "session.closed" as const, + reason: event.reason ?? null, + })); + case "error": + return Option.map(decodeErrorEvent(raw), (event) => ({ + type: "error" as const, + code: event.error.code ?? "", + message: event.error.message, + })); + default: + return Option.none(); + } +}; + +export type RealtimeOutbound = + | { + readonly type: "delegation.context.append"; + readonly delegation_item_id: string; + readonly channel: "speakable"; + readonly content: ReadonlyArray<{ readonly type: "input_text"; readonly text: string }>; + } + | { + readonly type: "session.context.append"; + readonly channel: "speakable"; + readonly content: ReadonlyArray<{ readonly type: "input_text"; readonly text: string }>; + } + | { readonly type: "session.close" }; + +/** The session rejects a context append over 500 tokens; Codex sends 500 bytes at a time. */ +export const CONTEXT_APPEND_MAX_BYTES = 500; + +const encoder = new TextEncoder(); + +/** Splits text into pieces of at most `maxBytes` UTF-8 bytes, never inside a character. */ +export const contextAppendChunks = ( + text: string, + maxBytes = CONTEXT_APPEND_MAX_BYTES, +): ReadonlyArray<string> => { + const chunks: Array<string> = []; + let current = ""; + let currentBytes = 0; + for (const character of text) { + const bytes = encoder.encode(character).length; + if (currentBytes + bytes > maxBytes && current.length > 0) { + chunks.push(current); + current = ""; + currentBytes = 0; + } + current += character; + currentBytes += bytes; + } + if (current.length > 0 || chunks.length === 0) chunks.push(current); + return chunks; +}; + +const content = (text: string) => [{ type: "input_text" as const, text }]; + +/** The answer to one delegation, in as many appends as the size limit needs. */ +export const delegationAnswerEvents = ( + delegationItemId: string, + text: string, +): ReadonlyArray<RealtimeOutbound> => + contextAppendChunks(text).map((chunk) => ({ + type: "delegation.context.append", + delegation_item_id: delegationItemId, + channel: "speakable", + content: content(chunk), + })); + +/** Something the voice should say without being asked. */ +export const sessionUpdateEvents = (text: string): ReadonlyArray<RealtimeOutbound> => + contextAppendChunks(text).map((chunk) => ({ + type: "session.context.append", + channel: "speakable", + content: content(chunk), + })); diff --git a/apps/server/src/voiceCall/voiceCallPrompt.ts b/apps/server/src/voiceCall/voiceCallPrompt.ts new file mode 100644 index 0000000000000..d2c9f9f324071 --- /dev/null +++ b/apps/server/src/voiceCall/voiceCallPrompt.ts @@ -0,0 +1,52 @@ +/** + * Fork: the texts the voice call sends. Tone follows Codex's realtime + * prompts (`codex-rs/prompts/templates/realtime/`) and Bloo's voice plugin. + */ +import type { VoiceCallLanguage } from "@t3tools/contracts"; + +const FRONT_END_INSTRUCTIONS = `You are the voice of esveo code, a desktop app in which the user directs coding agents. You are on a live voice call with the user. + +- Chat with the user on your own about anything that needs no tools. +- Hand every request about code, files, the project, the thread or its agent to the client as a delegation: one cleaned-up instruction in your own words, complete enough to act on without the conversation. The client sends it to the agent of the thread open on the user's screen. +- After delegating, say a short filler such as "On it." and stay responsive. The result arrives later as context; until then, do not guess or invent it. +- Read results back short and natural, without Markdown, lists or code. Say that long output, code or lists are in the thread instead of reading them. +- When you are told about something that happened in the thread without being asked, mention it briefly. +- Questions or approvals the agent asks for are answered on screen, not by voice. When you hear of one, tell the user to look at the screen.`; + +const languageLine = (language: VoiceCallLanguage) => + language === "de" + ? "Always speak German, from your very first sentence on, whatever language you hear." + : "Answer in the language the user speaks."; + +export const frontEndInstructions = (input: { + readonly language: VoiceCallLanguage; + readonly persona: string; + readonly targetTitle: string | null; + readonly nowIso: string; +}): string => + [ + FRONT_END_INSTRUCTIONS, + languageLine(input.language), + input.targetTitle === null + ? "No thread is open on the user's screen right now, so delegations cannot be carried out; say so if the user asks for work." + : `The thread open on the user's screen is "${input.targetTitle}".`, + `The time is ${input.nowIso}.`, + input.persona.trim().length > 0 + ? `About the user and how to talk:\n${input.persona.trim()}` + : null, + ] + .filter((line) => line !== null) + .join("\n\n"); + +/** Rides on every delegated message as the `voice-call` context record. */ +export const BACKEND_HINT = + "This message comes from a live voice call. A voice assistant wrote it from the user's speech, so it may contain recognition errors; read it for intent. Your final answer is read aloud by that assistant: keep it short and action-oriented, and leave long output, code and lists in the thread, saying that they are there."; + +export const noTargetAnswer = + "No thread is open on the user's screen, so nothing was sent. Ask the user to open the thread the request is for."; + +export const archivedTargetAnswer = (title: string) => + `The thread "${title}" is archived, so nothing was sent. Ask the user to open an active thread.`; + +export const dispatchFailedAnswer = (title: string) => + `The request could not be sent to the thread "${title}". Ask the user to check the screen.`; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 5bb6bfc269323..4188fe10e7500 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -67,6 +67,7 @@ import { THREAD_COORDINATORS_WS_METHODS, THREAD_DECISIONS_WS_METHODS, FORK_NOTES_WS_METHODS, + VOICE_CALL_WS_METHODS, FORK_INSTRUCTIONS_WS_METHODS, VOICE_INPUT_WS_METHODS, PROVIDER_SUBAGENT_CONTROL_WS_METHODS, @@ -209,6 +210,7 @@ import * as WorkspaceFileSystem from "./workspace/WorkspaceFileSystem.ts"; import { readWorkflowScript } from "./orchestration-v2/workflowScriptQuery.ts"; import * as ThreadDecisions from "./threadDecisions/ThreadDecisions.ts"; import * as ForkNotes from "./forkNotes/ForkNotes.ts"; +import * as VoiceCall from "./voiceCall/VoiceCallService.ts"; import * as ForkInstructions from "./forkInstructions/ForkInstructions.ts"; import * as ThreadCoordinators from "./threadOrchestration/ThreadCoordinators.ts"; import * as VoiceInput from "./voiceInput/VoiceInput.ts"; @@ -3122,6 +3124,11 @@ const layerWsRpc = ( // dictation in the composer. [VOICE_INPUT_WS_METHODS.prepare]: (input) => VoiceInput.prepareRpc(input), [VOICE_INPUT_WS_METHODS.transcribe]: (input) => VoiceInput.transcribeRpc(input), + // voice call with GPT-Live. + [VOICE_CALL_WS_METHODS.subscribe]: (input) => VoiceCall.subscribeRpc(input), + [VOICE_CALL_WS_METHODS.connect]: (input) => VoiceCall.connectRpc(input), + [VOICE_CALL_WS_METHODS.setTarget]: (input) => VoiceCall.setTargetRpc(input), + [VOICE_CALL_WS_METHODS.hangUp]: (_input) => VoiceCall.hangUpRpc(), // stop and message a provider subagent from its thread. [PROVIDER_SUBAGENT_CONTROL_WS_METHODS.target]: (input) => ProviderSubagentControl.target(providerSubagentControlDeps, input), diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index f0704dece682d..f2776d6ac4eab 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -235,6 +235,7 @@ import { } from "../rightPanelStore"; import { useIsActiveChatPane } from "./split/chatPane"; import { ThreadFindHeaderButton } from "./chatFind/ThreadFindHeaderButton"; +import { VoiceCallButton } from "./voiceCall/VoiceCallButton"; import { installDesktopClipboardPasteFallback } from "../lib/desktopClipboardPaste"; import { isPreviewSupportedInRuntime, @@ -7854,6 +7855,10 @@ export default function ChatView(props: ChatViewProps) { // Fork: find in thread sits in front of the panel toggles; a maximized // panel hides the timeline it searches. <span className="flex h-full shrink-0 items-center gap-1"> + {/* Fork: voice call with the thread on screen. */} + {isServerThread && canOperateThread ? ( + <VoiceCallButton environmentId={environmentId} threadId={threadId} active={isActivePane} /> + ) : null} {isServerThread && serverConfig?.threadFind === true && !paintOnlyDisplayedTimeline && diff --git a/apps/web/src/components/voiceCall/VoiceCallButton.tsx b/apps/web/src/components/voiceCall/VoiceCallButton.tsx new file mode 100644 index 0000000000000..0e30b918160ef --- /dev/null +++ b/apps/web/src/components/voiceCall/VoiceCallButton.tsx @@ -0,0 +1,110 @@ +import { squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; +import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { PhoneIcon } from "lucide-react"; +import { useEffect } from "react"; + +import { useEnvironmentQuery } from "~/state/query"; +import { useAtomCommand } from "~/state/use-atom-command"; +import { toastManager } from "../ui/toast"; +import { Toggle } from "../ui/toggle"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; +import { + isVoiceCallPeerActive, + startVoiceCallPeer, + stopVoiceCallPeer, + stopVoiceCallPeerUnlessStarting, +} from "./voiceCallPeer"; +import { voiceCallClientId, voiceCallEnvironment } from "./voiceCallState"; + +const failureMessage = (failure: Parameters<typeof squashAtomCommandFailure>[0]) => { + const error = squashAtomCommandFailure(failure); + return error instanceof Error ? error.message : "The voice call failed."; +}; + +/** + * Fork: starts and ends a voice call with the thread on screen as its target + * (Settings flag `enableVoiceCall`). Phase 1 has no call UI beyond this + * toggle; the transcript goes to the browser console. + */ +export function VoiceCallButton(props: { + environmentId: EnvironmentId; + threadId: ThreadId; + active: boolean; +}) { + const { environmentId, threadId, active } = props; + const view = useEnvironmentQuery( + voiceCallEnvironment.view({ environmentId, input: { clientId: voiceCallClientId } }), + ).data; + const connect = useAtomCommand(voiceCallEnvironment.connect, { reportFailure: false }); + const setTarget = useAtomCommand(voiceCallEnvironment.setTarget, { reportFailure: false }); + const hangUp = useAtomCommand(voiceCallEnvironment.hangUp, { reportFailure: false }); + + const call = view?.call ?? null; + const running = call !== null && (call.state === "connecting" || call.state === "live"); + const mine = running && call.ownerClientId === voiceCallClientId; + + // The server ended the call, or another client took it over. + useEffect(() => { + if (!mine && isVoiceCallPeerActive()) stopVoiceCallPeerUnlessStarting(); + }, [mine]); + + // The call follows the thread on screen. + useEffect(() => { + if (mine && active && call.targetThreadId !== threadId) { + void setTarget({ environmentId, input: { target: threadId } }); + } + }, [active, call?.targetThreadId, environmentId, mine, setTarget, threadId]); + + const end = () => { + stopVoiceCallPeer(); + void hangUp({ environmentId, input: {} }); + }; + + const start = () => { + void startVoiceCallPeer(async (input) => { + const result = await connect({ + environmentId, + input: { clientId: voiceCallClientId, target: threadId, ...input }, + }); + if (result._tag === "Failure") throw new Error(failureMessage(result)); + return result.value; + }, end).catch((error: unknown) => { + end(); + toastManager.add({ + type: "error", + title: "Voice call failed", + description: error instanceof Error ? error.message : "The voice call failed.", + }); + }); + }; + + const availability = view?.availability ?? null; + if (availability === null || availability.reason === "disabled") return null; + + const label = mine + ? "End voice call" + : running + ? "Take over the voice call from the other device" + : availability.available + ? "Start voice call" + : (availability.message ?? "Voice calls are unavailable"); + + return ( + <Tooltip> + <TooltipTrigger render={<span className="flex shrink-0" />}> + <Toggle + className="shrink-0 [-webkit-app-region:no-drag]" + pressed={mine} + disabled={!mine && !availability.available} + onPressedChange={(pressed) => (pressed ? start() : end())} + aria-label={label} + variant="ghost" + size="sm" + > + <PhoneIcon className="size-4" /> + </Toggle> + </TooltipTrigger> + <TooltipPopup side="bottom">{call?.error && !running ? call.error : label}</TooltipPopup> + </Tooltip> + ); +} diff --git a/apps/web/src/components/voiceCall/voiceCallPeer.ts b/apps/web/src/components/voiceCall/voiceCallPeer.ts new file mode 100644 index 0000000000000..3512c51493cec --- /dev/null +++ b/apps/web/src/components/voiceCall/voiceCallPeer.ts @@ -0,0 +1,112 @@ +import type { VoiceCallConnectResult } from "@t3tools/contracts"; + +/** + * Fork: the browser side of a voice call, one per tab. It owns the + * microphone, the speaker and the WebRTC peer; the server owns everything + * else, including the event channel. It lives outside React so switching + * threads does not drop the call. + */ +const CONNECT_TIMEOUT_MS = 15_000; +const CONNECT_ATTEMPTS = 3; + +export type VoiceCallConnect = (input: { + readonly offerSdp: string; + readonly retryOf?: string; +}) => Promise<VoiceCallConnectResult>; + +interface ActivePeer { + readonly peer: RTCPeerConnection; + readonly microphone: MediaStream; + readonly speaker: HTMLAudioElement; +} + +let active: ActivePeer | null = null; +let starting = false; + +export const isVoiceCallPeerActive = () => active !== null || starting; + +function closePeer() { + if (active === null) return; + active.peer.close(); + for (const track of active.microphone.getTracks()) track.stop(); + active.speaker.srcObject = null; + active = null; +} + +function waitForConnected(peer: RTCPeerConnection): Promise<boolean> { + return new Promise((resolve) => { + const done = (connected: boolean) => { + clearTimeout(timer); + peer.removeEventListener("connectionstatechange", onChange); + resolve(connected); + }; + const onChange = () => { + if (peer.connectionState === "connected") done(true); + else if (peer.connectionState === "failed" || peer.connectionState === "closed") done(false); + }; + const timer = setTimeout(() => done(false), CONNECT_TIMEOUT_MS); + peer.addEventListener("connectionstatechange", onChange); + onChange(); + }); +} + +/** + * Must run inside the click handler: Safari only allows the microphone and + * audio playback from a user gesture. + */ +export async function startVoiceCallPeer( + connect: VoiceCallConnect, + /** The connection broke after it was up; the server cannot see that on its own. */ + onDropped: () => void, +): Promise<void> { + if (isVoiceCallPeerActive()) return; + starting = true; + const speaker = document.createElement("audio"); + speaker.autoplay = true; + try { + const microphone = await navigator.mediaDevices.getUserMedia({ audio: true }); + let retryOf: string | undefined; + for (let attempt = 1; attempt <= CONNECT_ATTEMPTS; attempt++) { + const peer = new RTCPeerConnection(); + active = { peer, microphone, speaker }; + peer.addEventListener("track", (event) => { + speaker.srcObject = event.streams[0] ?? null; + }); + for (const track of microphone.getAudioTracks()) peer.addTrack(track, microphone); + // The session expects this channel; the server holds the events, so it stays unused. + peer.createDataChannel("oai-events"); + await peer.setLocalDescription(await peer.createOffer()); + const result = await connect({ + offerSdp: peer.localDescription?.sdp ?? "", + ...(retryOf === undefined ? {} : { retryOf }), + }); + await peer.setRemoteDescription({ type: "answer", sdp: result.answerSdp }); + if (await waitForConnected(peer)) { + peer.addEventListener("connectionstatechange", () => { + if (peer.connectionState === "failed" && active?.peer === peer) { + closePeer(); + onDropped(); + } + }); + return; + } + peer.close(); + retryOf = result.segmentId; + } + throw new Error("The voice call could not connect."); + } catch (error) { + closePeer(); + throw error; + } finally { + starting = false; + } +} + +export function stopVoiceCallPeer() { + closePeer(); +} + +/** The server ended the call or another client took it over. */ +export function stopVoiceCallPeerUnlessStarting() { + if (!starting) closePeer(); +} diff --git a/apps/web/src/components/voiceCall/voiceCallState.ts b/apps/web/src/components/voiceCall/voiceCallState.ts new file mode 100644 index 0000000000000..66cc95abaf49a --- /dev/null +++ b/apps/web/src/components/voiceCall/voiceCallState.ts @@ -0,0 +1,81 @@ +import { + VOICE_CALL_WS_METHODS, + type VoiceCallAvailability, + type VoiceCallDelegation, + type VoiceCallEvent, + type VoiceCallStatus, +} from "@t3tools/contracts"; +import { + createEnvironmentRpcCommand, + createEnvironmentRpcSubscriptionAtomFamily, +} from "@t3tools/client-runtime/state/runtime"; +import * as Effect from "effect/Effect"; +import * as Stream from "effect/Stream"; + +import { connectionAtomRuntime } from "~/connection/runtime"; +import { randomUUID } from "~/lib/utils"; + +/** Identifies this tab to the server, which closes the call once its owner's tab is gone. */ +export const voiceCallClientId = randomUUID(); + +export interface VoiceCallView { + readonly availability: VoiceCallAvailability | null; + readonly call: VoiceCallStatus | null; + readonly delegations: ReadonlyArray<VoiceCallDelegation>; +} + +const EMPTY_VIEW: VoiceCallView = { availability: null, call: null, delegations: [] }; + +/** The server sends a snapshot, then deltas; transcripts only go to the console for now. */ +export function reduceVoiceCallEvent(view: VoiceCallView, event: VoiceCallEvent): VoiceCallView { + switch (event.type) { + case "snapshot": + return { availability: event.availability, call: event.call, delegations: event.delegations }; + case "availability": + return { ...view, availability: event.availability }; + case "call": + return { ...view, call: event.call }; + case "delegation": { + const others = view.delegations.filter((entry) => entry.id !== event.delegation.id); + return { ...view, delegations: [...others, event.delegation] }; + } + case "transcript": + return view; + } +} + +const logTranscript = (event: VoiceCallEvent) => + event.type === "transcript" && event.done + ? Effect.logInfo(`[voice call] ${event.speaker}: ${event.text}`) + : event.type === "delegation" + ? Effect.logInfo( + `[voice call] delegation ${event.delegation.status}: ${event.delegation.text}`, + ) + : Effect.void; + +export const voiceCallEnvironment = { + view: createEnvironmentRpcSubscriptionAtomFamily(connectionAtomRuntime, { + label: "environment-data:voice-call:view", + tag: VOICE_CALL_WS_METHODS.subscribe, + // Long enough to survive switching threads; the server hangs up when it ends. + idleTtlMs: 30_000, + transform: (stream) => + stream.pipe( + Stream.tap(logTranscript), + Stream.scan(() => EMPTY_VIEW, reduceVoiceCallEvent), + Stream.drop(1), + ), + }), + connect: createEnvironmentRpcCommand(connectionAtomRuntime, { + label: "environment-data:voice-call:connect", + tag: VOICE_CALL_WS_METHODS.connect, + }), + setTarget: createEnvironmentRpcCommand(connectionAtomRuntime, { + label: "environment-data:voice-call:set-target", + tag: VOICE_CALL_WS_METHODS.setTarget, + }), + hangUp: createEnvironmentRpcCommand(connectionAtomRuntime, { + label: "environment-data:voice-call:hang-up", + tag: VOICE_CALL_WS_METHODS.hangUp, + }), +}; diff --git a/docs/internals/voice-call.md b/docs/internals/voice-call.md index 4e1ba228dc674..4d02b87ea6d35 100644 --- a/docs/internals/voice-call.md +++ b/docs/internals/voice-call.md @@ -1,9 +1,12 @@ # Voice call (plan) -Status: implementation plan for the fork, not yet built. PAD's decisions of -2026-10-21/22 are the source; the v1 scope cut and the gpt-live-1.1 findings -override older notes. When the feature lands, rewrite this page into the -durable decisions and traps and drop the phase list. +Status: implementation plan for the fork; phase 1 is built (see +[Deviations](#deviations-from-the-plan)). PAD's decisions of 2026-10-21/22/23 +are the source; the v1 scope cut and the gpt-live-1.1 findings override older +notes. The plan is a guide, not a contract: where the code or a test showed a +better way, the build deviates and says so below. When the feature lands, +rewrite this page into the durable decisions and traps and drop the phase +list. A voice call lets the user talk to esveo code. A realtime voice model (OpenAI's GPT-Live, on the user's ChatGPT subscription) holds the @@ -11,6 +14,39 @@ conversation, hands real work to the thread that is open on screen, and reads the result back. It is separate from [voice input](voice-input.md), which only dictates into the composer. +## Deviations from the plan + +Phase 1: + +- **Sideband socket.** Node's global `WebSocket` (undici) with the headers in + its options, not the `ws` package; no new dependency. +- **`sendToThread` gained an optional `context`** (`// Fork:` lines in + `ThreadManagementService.ts`), because it had no way to attach the + `voice-call` record. +- **The reference href is `t3-context://v1/voice-call/<id>`**, the form + `formatComposerContextReference` produces; the plan's form without `v1` + does not parse. +- **The client identifies itself.** `subscribe` and `connect` carry a + per-tab `clientId`. The server only connects a client that holds a + subscription and ends the call when the owner's last subscription closes, + which is the "dropped client" rule made concrete. A takeover is not a + separate state: any `connect` (retry, takeover, fresh) closes the current + segment, the status names the new owner, and the old tab closes its peer + when it sees that it is no longer the owner. +- **`target` is a `ThreadId`**, not a `ThreadRef`: the call lives in one + environment, so the environment part was redundant. +- **The segment lives inside `VoiceCallService`** for now; a separate + `VoiceCallSegment.ts` waits until pause and expiry give it enough to own. +- **Auth: existing mode only, no refresh.** `VoiceCallAuth` reads the + instance's `auth.json` (honouring `homePath`, shadow homes and + `CODEX_HOME`), never writes it, and fails fast with "renew the Codex login" + when the token expires within 5 minutes. Asking Codex to refresh and + managed mode are left for a later phase; PAD's setup does not need them. +- **The subscription needs the operate scope** as well, since it keeps a + billed session alive. +- **Transcripts** go to the browser console (`[voice call] …` via + `Effect.logInfo`) until the panel exists. + ## Goal and non-goals Goal for v1: @@ -20,7 +56,9 @@ Goal for v1: or starts a normal turn when the thread is idle. - The backend is the thread's own provider and model. The voice is only the front end; no extra model hop sits between them. -- Results, questions and approvals of that thread come back as speech. +- Results of that thread come back as speech. A new question or approval + in the target thread is announced, but answered on screen only (see + [Later](#later)). - The call survives thread switches (the target follows the screen), pauses itself when it is quiet because open time is billed, and wakes on speech or on a result. @@ -133,27 +171,27 @@ undocumented endpoint; re-check it when Codex changes its realtime code. **Events we read** -| Event | Use | -|---|---| -| `session.started {session:{id, status, expires_at}}` | segment is live; `expires_at` is about 2 h out | -| `input_transcript.added {item.text, start_ms, end_ms}` | user transcript deltas, silence timer | -| `output_transcript.added` | voice transcript deltas, silence timer | -| `turn.created` / `turn.done {turn:{id, role, transcript}}` | speaking gate; persisted transcript on `turn.done` | -| `delegation.created {item:{id, content[].text}, handoff_id, user_bidi_turn_id}` | the request to dispatch | -| `delegation.context.appended`, `session.context.appended` | delivery acks | -| `session.usage.updated {usage.audio_duration_ms, usage_limit{status, reset_seconds}}` | quota, every 15 s | -| `session.closed`, `error` | segment ended; `forbidden` means a bad voice | +| Event | Use | +| ------------------------------------------------------------------------------------- | -------------------------------------------------- | +| `session.started {session:{id, status, expires_at}}` | segment is live; `expires_at` is about 2 h out | +| `input_transcript.added {item.text, start_ms, end_ms}` | user transcript deltas, silence timer | +| `output_transcript.added` | voice transcript deltas, silence timer | +| `turn.created` / `turn.done {turn:{id, role, transcript}}` | speaking gate; persisted transcript on `turn.done` | +| `delegation.created {item:{id, content[].text}, handoff_id, user_bidi_turn_id}` | the request to dispatch | +| `delegation.context.appended`, `session.context.appended` | delivery acks | +| `session.usage.updated {usage.audio_duration_ms, usage_limit{status, reset_seconds}}` | quota, every 15 s | +| `session.closed`, `error` | segment ended; `forbidden` means a bad voice | `session.input_audio.append` and `session.output_audio.delta` also arrive and are dropped unread. No event names the model revision. **Events we send (sideband only)** -| Event | Use | -|---|---| -| `delegation.context.append {delegation_item_id, channel:"speakable", content:[{type:"input_text", text}]}` | the answer to one delegation | -| `session.context.append {channel:"speakable", content:[…]}` | unprompted events: results, questions, approvals | -| `session.close` | pause, hang-up, expiry, failed connect | +| Event | Use | +| ---------------------------------------------------------------------------------------------------------- | ------------------------------------------------ | +| `delegation.context.append {delegation_item_id, channel:"speakable", content:[{type:"input_text", text}]}` | the answer to one delegation | +| `session.context.append {channel:"speakable", content:[…]}` | unprompted events: results, questions, approvals | +| `session.close` | pause, hang-up, expiry, failed connect | Appends are limited to 500 tokens each. Like Codex, cap a result at about 1,000 tokens (`REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET`) and send it in @@ -173,7 +211,7 @@ existing `voiceInput/` dictation module), with the contract in - **`VoiceCallAuth.ts`** returns `{ accessToken, accountId }` for the Codex provider instance (v1: the default `codex` instance). - - *Existing mode* (PAD's setup: no `providerInstances.codex`, so the default + - _Existing mode_ (PAD's setup: no `providerInstances.codex`, so the default mode, with `~/.codex/auth.json`): read `<CODEX_HOME>/auth.json` on every mint (resolve the home as [`threadRoots.ts`](../../apps/server/src/forkInstructions/threadRoots.ts) @@ -182,7 +220,7 @@ existing `voiceInput/` dictation module), with the contract in write would race its refresh. If the JWT `exp` is near, ask Codex to refresh (to verify in phase 1: the app-server account read with token refresh); if that is not possible, fail with "renew the Codex login". - - *Managed mode*: `makeCodexChatGptAuth(...).access` in + - _Managed mode_: `makeCodexChatGptAuth(...).access` in [`CodexChatGptAuth.ts`](../../apps/server/src/provider/CodexChatGptAuth.ts) already refreshes under the shared session lock. The account id is not a field there; read it from the id token's ChatGPT claim. Its token is @@ -245,7 +283,7 @@ On `delegation.created`: `voice-call-message:<callId>:<delegationItemId>`, so a retried event cannot dispatch twice. `createdBy: "user"`, `creationSource: "server"`. 3. Record the delegation `{ delegationItemId, segmentId, threadId, messageId, - runId, status: open }` and push a card to the panel. +runId, status: open }` and push a card to the panel. The voice has already said a filler ("On it.") and stays conversational; the answer follows when the run settles. A dispatch failure is answered at once @@ -273,8 +311,8 @@ and add a small fork chip if it looks wrong. Older open delegations of the same run (steered twice) are marked superseded and get nothing. A failed or interrupted run answers with one sentence and the error. -- **Unprompted events** in the thread that is the target *when the event - fires*: a run that settles with no open delegation, a new user-input +- **Unprompted events** in the thread that is the target _when the event + fires_: a run that settles with no open delegation, a new user-input question, a new approval request. They go out as `session.context.append` ("Update from <title>: …"). A run that answered a delegation never also triggers an unprompted update. @@ -294,25 +332,10 @@ and add a small fork chip if it looks wrong. segment. `streamDomainEvents` is live-tail only, so there is no replay to rely on. -### Questions and approvals (decision for PAD) - -With no routing, "yes, allow it" would become a steer and the request would -stay pending. v1 needs exactly one rule for this. Recommendation: - -- While the target thread has exactly one pending `user_input` request, the - next delegation is its free-text answer: `runtime-request.respond` with - `answers` (the path MCP `t3_pending_request_respond` uses), not a steer. -- Approvals: PAD runs with full access, so they are rare. The voice announces - them and the front-end prompt tells it to delegate exactly `approve` or - `decline` as the answer. A delegation that normalises to one of those words - while one approval is pending is answered with `decision: accept|decline`; - anything else is a normal steer. Unlike MCP, this server path may send - `decision`, so the rule stays this narrow. -- More than one pending request: announce how many, answer nothing by voice, - point to the screen. - -Alternatives: approvals only announced, never answered by voice; or a -request header format (moved to Later). +**Questions and approvals** are not answered by voice in v1 (PAD, +2026-10-23). A new pending request in the target thread may be announced as +an unprompted update; the front-end prompt tells the voice to send the user +to the screen. Anything the user says in reply is a normal steer. ## Prompts @@ -325,8 +348,7 @@ plugin is the reference for tone and for the doze notes. instruction in your own words; the client sends it to the thread open on screen; say a short filler and stay responsive while it runs; read results back short and without Markdown; when told about a question or approval, - ask the user and pass the answer on (exact words for approvals); do not - invent results. Plus the language line (`auto`: answer in the user's + tell the user it waits on screen; do not invent results. Plus the language line (`auto`: answer in the user's language; `de`: always German, from the first sentence), the persona text, the current target thread's title, the time, and on resume a recap. - **Backend hint** (the `voice-call` context record), modelled on Codex's @@ -417,9 +439,8 @@ matching optional key in `ServerSettingsPatch`. - `voiceCallVoice`: one of the 25 v3 voices (a constant list in the contract: juniper, maple, spruce, ember, vale, breeze, arbor, sol, cove, aube, bubbie, haetsal, hanul, himari, leher, miko, neer, porto, rindo, rio, - rivage, selva, sonna, tinta, viola). Default `sonna` (German worked; - PAD may prefer `miko`). Only the list is offered, because an off-list voice - kills the session. + rivage, selva, sonna, tinta, viola). Default `miko` (PAD, 2026-10-23). + Only the list is offered, because an off-list voice kills the session. - `voiceCallLanguage: "auto" | "de"`, default `"auto"`. - `voiceCallPersona: string`, default empty, capped in length. @@ -483,9 +504,9 @@ Each phase is one reviewable PR on the feature branch, behind the flag. browser console) is the end-to-end check; no panel, bar or pause yet. Desktop mic only if the check shows it is needed. 2. **Read-back.** `VoiceCallReadBack`, delegation answers with cap and - chunking, unprompted events for the target, the speaking gate, - questions and approvals per PAD's decision, the delegations table and - restart re-arm. Settles the untested protocol points listed above. + chunking, unprompted events for the target, the speaking gate, the + delegations table and restart re-arm. Questions and approvals are only + announced. Settles the untested protocol points listed above. 3. **Pause, resume, expiry, quota.** Auto-pause, client level gate and replay buffer, wake on result, manual pause, recap, 2 h rollover, usage tracking. @@ -539,6 +560,15 @@ Each phase is one reviewable PR on the feature branch, behind the flag. ## Later +- Answering questions and approvals by voice. The rule PAD postponed on + 2026-10-23: while the target thread has exactly one pending `user_input` + request, the next delegation is its free-text answer + (`runtime-request.respond` with `answers`, the path MCP + `t3_pending_request_respond` uses), not a steer; a delegation of exactly + `approve` or `decline` while one approval is pending answers it with + `decision: accept|decline`; with more than one pending request, announce + how many and point to the screen. The alternative is a request header + format (see routing). - Routing: send to another thread (and switch the UI there), open a new thread with project and model inference, queue instead of steer, promote the last queued message to a steer. Options: a fixed request header the diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 0ca92c35e6241..2276f5e3856b9 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -1,6 +1,7 @@ import { EnvironmentAuthorizationError, FORK_NOTES_WS_METHODS, + VOICE_CALL_WS_METHODS, clientRpcRequiredScopes, authScopeRequiredResponse, type EnvironmentId, @@ -80,6 +81,8 @@ export type EnvironmentSubscriptionRpcTag = | typeof THREAD_COORDINATORS_WS_METHODS.subscribe // Fork: the Notes tab. | typeof FORK_NOTES_WS_METHODS.subscribe + // Fork: the voice call. + | typeof VOICE_CALL_WS_METHODS.subscribe | typeof VOICE_INPUT_WS_METHODS.prepare | typeof WS_METHODS.terminalObserve; diff --git a/packages/contracts/src/forkWsRpc.ts b/packages/contracts/src/forkWsRpc.ts index 870371f03c0dd..8bd658fc737a5 100644 --- a/packages/contracts/src/forkWsRpc.ts +++ b/packages/contracts/src/forkWsRpc.ts @@ -22,6 +22,13 @@ import { WsThreadDecisionsActRpc, WsThreadDecisionsSubscribeRpc, } from "./threadDecisions.ts"; +import { + VOICE_CALL_WS_METHODS, + WsVoiceCallConnectRpc, + WsVoiceCallHangUpRpc, + WsVoiceCallSetTargetRpc, + WsVoiceCallSubscribeRpc, +} from "./voiceCall.ts"; import { VOICE_INPUT_WS_METHODS, WsVoiceInputPrepareRpc, @@ -49,6 +56,11 @@ export const ForkWsRpcGroup = RpcGroup.make( // Dictation in the composer. WsVoiceInputPrepareRpc, WsVoiceInputTranscribeRpc, + // Voice call with GPT-Live. + WsVoiceCallSubscribeRpc, + WsVoiceCallConnectRpc, + WsVoiceCallSetTargetRpc, + WsVoiceCallHangUpRpc, // Stop and message a provider subagent from its thread. WsProviderSubagentTargetRpc, WsProviderSubagentStopRpc, @@ -68,6 +80,10 @@ export const FORK_WS_RPC_TAGS = [ THREAD_COORDINATORS_WS_METHODS.set, VOICE_INPUT_WS_METHODS.prepare, VOICE_INPUT_WS_METHODS.transcribe, + VOICE_CALL_WS_METHODS.subscribe, + VOICE_CALL_WS_METHODS.connect, + VOICE_CALL_WS_METHODS.setTarget, + VOICE_CALL_WS_METHODS.hangUp, PROVIDER_SUBAGENT_CONTROL_WS_METHODS.target, PROVIDER_SUBAGENT_CONTROL_WS_METHODS.stop, AGENT_TIME_WS_METHODS.summary, diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 7f17ca32def59..8230229b3ff9b 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -70,6 +70,7 @@ export * from "./forkNotes.ts"; export * from "./forkInstructions.ts"; export * from "./threadCoordinators.ts"; export * from "./voiceInput.ts"; +export * from "./voiceCall.ts"; export * from "./providerSubagentControl.ts"; export * from "./forkWsRpc.ts"; export * from "./agentTime.ts"; diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 4e712284a1fe9..9dd47f8ae7e75 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -5,6 +5,12 @@ import { IdleAutoCompactAfterMinutes, IdleAutoCompactMinContextTokens, } from "./idleAutoCompact.ts"; // Fork +import { + DEFAULT_VOICE_CALL_VOICE, + VoiceCallLanguage, + VoiceCallPersona, + VoiceCallVoice, +} from "./voiceCall.ts"; // Fork import { AuthOrchestrationOperateScope, AuthSettingsWriteScope, @@ -1139,6 +1145,15 @@ export const ServerSettings = Schema.Struct({ idleAutoCompactMinContextTokens: IdleAutoCompactMinContextTokens.pipe( Schema.withDecodingDefault(Effect.succeed(DEFAULT_IDLE_AUTO_COMPACT_CONTEXT_TOKENS)), ), + /** Fork: voice call with GPT-Live on the Codex ChatGPT login (docs/internals/voice-call.md). */ + enableVoiceCall: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), + voiceCallVoice: VoiceCallVoice.pipe( + Schema.withDecodingDefault(Effect.succeed(DEFAULT_VOICE_CALL_VOICE)), + ), + voiceCallLanguage: VoiceCallLanguage.pipe( + Schema.withDecodingDefault(Effect.succeed<VoiceCallLanguage>("auto")), + ), + voiceCallPersona: VoiceCallPersona.pipe(Schema.withDecodingDefault(Effect.succeed(""))), /** * Kept separate from agent access so enabling the user's Device panel does * not also grant providers control of simulators and emulators. @@ -1456,6 +1471,11 @@ export const ServerSettingsPatch = Schema.Struct({ enableIdleAutoCompact: Schema.optionalKey(Schema.Boolean), idleAutoCompactAfterMinutes: Schema.optionalKey(IdleAutoCompactAfterMinutes), idleAutoCompactMinContextTokens: Schema.optionalKey(IdleAutoCompactMinContextTokens), + // Fork: voice call. + enableVoiceCall: Schema.optionalKey(Schema.Boolean), + voiceCallVoice: Schema.optionalKey(VoiceCallVoice), + voiceCallLanguage: Schema.optionalKey(VoiceCallLanguage), + voiceCallPersona: Schema.optionalKey(VoiceCallPersona), enableDeviceSupport: Schema.optionalKey(Schema.Boolean), deviceOnboardingCompleted: Schema.optionalKey(Schema.Boolean), deviceHosts: Schema.optionalKey(SshDeviceHostConfigs), diff --git a/packages/contracts/src/voiceCall.ts b/packages/contracts/src/voiceCall.ts new file mode 100644 index 0000000000000..a3894cdfede4a --- /dev/null +++ b/packages/contracts/src/voiceCall.ts @@ -0,0 +1,223 @@ +/** + * Fork: voice call. A realtime voice model (GPT-Live on the user's ChatGPT + * login) talks with the user, hands real work to the thread open on screen and + * reads the result back. The client owns the microphone, the speaker and the + * WebRTC peer; the server creates the call, holds the event sideband and is + * the only one that reads or writes realtime events. + * See docs/internals/voice-call.md. + */ +import * as Schema from "effect/Schema"; +import * as Rpc from "effect/rpc/Rpc"; + +import { EnvironmentAuthorizationError } from "./auth.ts"; +import { IsoDateTime, MessageId, RunId, ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts"; + +export const VOICE_CALL_WS_METHODS = { + subscribe: "voiceCall.subscribe", + connect: "voiceCall.connect", + setTarget: "voiceCall.setTarget", + hangUp: "voiceCall.hangUp", +} as const; + +/** + * The voices the realtime v3 session accepts. Any other name ends the session + * with "Voice session access denied.", so settings only offer this list. + */ +export const VOICE_CALL_VOICES = [ + "juniper", + "maple", + "spruce", + "ember", + "vale", + "breeze", + "arbor", + "sol", + "cove", + "aube", + "bubbie", + "haetsal", + "hanul", + "himari", + "leher", + "miko", + "neer", + "porto", + "rindo", + "rio", + "rivage", + "selva", + "sonna", + "tinta", + "viola", +] as const; +export const VoiceCallVoice = Schema.Literals(VOICE_CALL_VOICES); +export type VoiceCallVoice = typeof VoiceCallVoice.Type; +export const DEFAULT_VOICE_CALL_VOICE: VoiceCallVoice = "miko"; + +/** `auto` answers in the user's language; `de` speaks German from the first sentence. */ +export const VoiceCallLanguage = Schema.Literals(["auto", "de"]); +export type VoiceCallLanguage = typeof VoiceCallLanguage.Type; + +export const VOICE_CALL_PERSONA_MAX_LENGTH = 2_000; +export const VoiceCallPersona = Schema.String.check( + Schema.isMaxLength(VOICE_CALL_PERSONA_MAX_LENGTH), +); + +/** Why a call cannot start, so the UI can say so. */ +export const VoiceCallUnavailableReason = Schema.Literals([ + "disabled", + "codex-not-logged-in", + "codex-api-key", + "codex-login-expired", +]); +export type VoiceCallUnavailableReason = typeof VoiceCallUnavailableReason.Type; + +export const VoiceCallAvailability = Schema.Struct({ + available: Schema.Boolean, + reason: Schema.NullOr(VoiceCallUnavailableReason), + message: Schema.NullOr(Schema.String), +}); +export type VoiceCallAvailability = typeof VoiceCallAvailability.Type; + +/** + * `connecting`: a segment is being created. `live`: a realtime session is open. + * `ended`: hung up. `error`: the last segment failed and no retry is running. + */ +export const VoiceCallState = Schema.Literals(["connecting", "live", "ended", "error"]); +export type VoiceCallState = typeof VoiceCallState.Type; + +export const VoiceCallStatus = Schema.Struct({ + callId: TrimmedNonEmptyString, + state: VoiceCallState, + /** The client whose peer carries the audio; any other client shows the call as taken over. */ + ownerClientId: Schema.String, + /** The realtime session now carrying the call, if any. */ + segmentId: Schema.NullOr(Schema.String), + /** The thread delegations go to; null when no thread is on screen. */ + targetThreadId: Schema.NullOr(ThreadId), + startedAt: IsoDateTime, + error: Schema.NullOr(Schema.String), +}); +export type VoiceCallStatus = typeof VoiceCallStatus.Type; + +/** + * `open`: sent to the thread, waiting for its run. `answered`: the result was + * read back. `superseded`: a later request steered the same run and gets the + * answer instead. `failed`: it never reached the thread, or the run failed. + */ +export const VoiceCallDelegationStatus = Schema.Literals([ + "open", + "answered", + "superseded", + "failed", +]); +export type VoiceCallDelegationStatus = typeof VoiceCallDelegationStatus.Type; + +/** One request the voice handed to a thread. */ +export const VoiceCallDelegation = Schema.Struct({ + /** The realtime delegation item id. */ + id: TrimmedNonEmptyString, + segmentId: Schema.String, + text: Schema.String, + threadId: Schema.NullOr(ThreadId), + messageId: Schema.NullOr(MessageId), + runId: Schema.NullOr(RunId), + status: VoiceCallDelegationStatus, + /** What was read back, as sent to the voice. */ + answer: Schema.NullOr(Schema.String), + createdAt: IsoDateTime, +}); +export type VoiceCallDelegation = typeof VoiceCallDelegation.Type; + +export const VoiceCallSpeaker = Schema.Literals(["user", "voice"]); +export type VoiceCallSpeaker = typeof VoiceCallSpeaker.Type; + +/** The first item of a subscription is a snapshot; deltas follow, so a long call never resends its transcript. */ +export const VoiceCallEvent = Schema.Union([ + Schema.Struct({ + type: Schema.Literal("snapshot"), + availability: VoiceCallAvailability, + call: Schema.NullOr(VoiceCallStatus), + delegations: Schema.Array(VoiceCallDelegation), + }), + Schema.Struct({ type: Schema.Literal("availability"), availability: VoiceCallAvailability }), + Schema.Struct({ type: Schema.Literal("call"), call: Schema.NullOr(VoiceCallStatus) }), + /** A transcript delta while a turn runs, then its full text once with `done`. */ + Schema.Struct({ + type: Schema.Literal("transcript"), + speaker: VoiceCallSpeaker, + turnId: Schema.String, + text: Schema.String, + done: Schema.Boolean, + }), + Schema.Struct({ type: Schema.Literal("delegation"), delegation: VoiceCallDelegation }), +]); +export type VoiceCallEvent = typeof VoiceCallEvent.Type; + +export const VoiceCallSubscribeInput = Schema.Struct({ + /** Random per client tab; ties the peer to the subscription that keeps it alive. */ + clientId: TrimmedNonEmptyString, +}); +export type VoiceCallSubscribeInput = typeof VoiceCallSubscribeInput.Type; + +export const VoiceCallConnectInput = Schema.Struct({ + clientId: TrimmedNonEmptyString, + offerSdp: TrimmedNonEmptyString.check(Schema.isMaxLength(64_000)), + target: Schema.NullOr(ThreadId), + /** The segment whose peer never connected; the server closes it and starts a new one. */ + retryOf: Schema.optionalKey(Schema.String), +}); +export type VoiceCallConnectInput = typeof VoiceCallConnectInput.Type; + +export const VoiceCallConnectResult = Schema.Struct({ + callId: TrimmedNonEmptyString, + segmentId: TrimmedNonEmptyString, + answerSdp: Schema.String, +}); +export type VoiceCallConnectResult = typeof VoiceCallConnectResult.Type; + +export const VoiceCallSetTargetInput = Schema.Struct({ + target: Schema.NullOr(ThreadId), +}); +export type VoiceCallSetTargetInput = typeof VoiceCallSetTargetInput.Type; + +export class VoiceCallDisabledError extends Schema.TaggedError<VoiceCallDisabledError>()( + "VoiceCallDisabledError", + { message: Schema.String }, +) {} + +export class VoiceCallError extends Schema.TaggedError<VoiceCallError>()("VoiceCallError", { + message: Schema.String, + reason: Schema.optionalKey(VoiceCallUnavailableReason), +}) {} + +const VoiceCallRpcError = Schema.Union([ + VoiceCallDisabledError, + VoiceCallError, + EnvironmentAuthorizationError, +]); + +export const WsVoiceCallSubscribeRpc = Rpc.make(VOICE_CALL_WS_METHODS.subscribe, { + payload: VoiceCallSubscribeInput, + success: VoiceCallEvent, + error: VoiceCallRpcError, + stream: true, +}); + +export const WsVoiceCallConnectRpc = Rpc.make(VOICE_CALL_WS_METHODS.connect, { + payload: VoiceCallConnectInput, + success: VoiceCallConnectResult, + error: VoiceCallRpcError, +}); + +export const WsVoiceCallSetTargetRpc = Rpc.make(VOICE_CALL_WS_METHODS.setTarget, { + payload: VoiceCallSetTargetInput, + success: Schema.Struct({}), + error: VoiceCallRpcError, +}); + +export const WsVoiceCallHangUpRpc = Rpc.make(VOICE_CALL_WS_METHODS.hangUp, { + payload: Schema.Struct({}), + success: Schema.Struct({}), + error: VoiceCallRpcError, +}); From ef9d9e79b2d5244609a59d242ae1d8686d047008 Mon Sep 17 00:00:00 2001 From: Paul Dittrich <paul.dittrich@esvo.com> Date: Fri, 23 Oct 2026 17:30:05 +0200 Subject: [PATCH 3/3] fix(fork): voice call sends the SDP offer untrimmed Trimming dropped the final line break, and ChatGPT rejected the offer with "failed to unmarshal SDP: EOF". Found in the first real call from headless Chromium. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --- packages/contracts/src/voiceCall.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/contracts/src/voiceCall.ts b/packages/contracts/src/voiceCall.ts index a3894cdfede4a..4b806214b3fed 100644 --- a/packages/contracts/src/voiceCall.ts +++ b/packages/contracts/src/voiceCall.ts @@ -162,7 +162,8 @@ export type VoiceCallSubscribeInput = typeof VoiceCallSubscribeInput.Type; export const VoiceCallConnectInput = Schema.Struct({ clientId: TrimmedNonEmptyString, - offerSdp: TrimmedNonEmptyString.check(Schema.isMaxLength(64_000)), + /** Untrimmed: the SDP parser needs the final line break. */ + offerSdp: Schema.String.check(Schema.isNonEmpty(), Schema.isMaxLength(64_000)), target: Schema.NullOr(ThreadId), /** The segment whose peer never connected; the server closes it and starts a new one. */ retryOf: Schema.optionalKey(Schema.String),