diff --git a/README.md b/README.md index dcf8665b..47173e4f 100644 --- a/README.md +++ b/README.md @@ -216,6 +216,7 @@ pairux login # Optional: label your agents as yours (browser pairux join ABC123 --name Claude # Join with the session's join code pairux listen --json # Follow chat and who is here; keeps the agent present pairux say "Tests pass" # Post in the chat +pairux voice join --json # Hear call audio; stdin speaks through C0mpute TTS pairux who # Who is in the session pairux leave ``` diff --git a/apps/desktop/electron.vite.config.ts b/apps/desktop/electron.vite.config.ts index 8e09fba9..25f94a8a 100644 --- a/apps/desktop/electron.vite.config.ts +++ b/apps/desktop/electron.vite.config.ts @@ -54,6 +54,7 @@ export default defineConfig({ rollupOptions: { input: { index: resolve(__dirname, 'src/renderer/index.html'), + 'agent-voice': resolve(__dirname, 'src/renderer/agent-voice.html'), }, }, }, diff --git a/apps/desktop/public/agent-voice-worklet.js b/apps/desktop/public/agent-voice-worklet.js new file mode 100644 index 00000000..5526c2aa --- /dev/null +++ b/apps/desktop/public/agent-voice-worklet.js @@ -0,0 +1,15 @@ +class PairuxVoiceCaptureProcessor extends AudioWorkletProcessor { + process(inputs, outputs) { + const output = outputs[0]?.[0]; + output?.fill(0); + + const input = inputs[0]?.[0]; + if (input?.length) { + const samples = input.slice(); + this.port.postMessage(samples, [samples.buffer]); + } + return true; + } +} + +registerProcessor('pairux-voice-capture', PairuxVoiceCaptureProcessor); diff --git a/apps/desktop/src/main/agent-cli/api.ts b/apps/desktop/src/main/agent-cli/api.ts index 3c0e5596..c32034af 100644 --- a/apps/desktop/src/main/agent-cli/api.ts +++ b/apps/desktop/src/main/agent-cli/api.ts @@ -42,6 +42,17 @@ export interface JoinResult { owned: boolean; } +export interface AgentVoiceToken { + token: string; + url: string; + roomName: string; + iceServers?: { + urls: string | string[]; + username?: string; + credential?: string; + }[]; +} + export class ApiError extends Error { constructor( readonly status: number, @@ -133,6 +144,13 @@ export class PairuxClient { }); } + voiceToken(participantId: string) { + return this.request( + `/api/v1/agents/${encodeURIComponent(participantId)}/voice-token`, + { method: 'POST', body: '{}' } + ); + } + /** Look up a live session by its 6-character join code. */ lookupJoinCode(joinCode: string) { return this.request<{ id: string; status?: string; scheduled?: boolean }>( diff --git a/apps/desktop/src/main/agent-cli/commands.ts b/apps/desktop/src/main/agent-cli/commands.ts index b427a4da..9a249c32 100644 --- a/apps/desktop/src/main/agent-cli/commands.ts +++ b/apps/desktop/src/main/agent-cli/commands.ts @@ -8,6 +8,7 @@ * pairux join ABC123 --name "Claude" * pairux listen follow chat + roster (stays present) * pairux say "Tests pass on my side" + * pairux voice join --json hear call audio and speak through C0mpute * pairux who who is in the session * pairux leave * pairux room link to a fresh anonymous E2EE room @@ -30,6 +31,7 @@ import { revoke, } from './auth'; import { ApiError, PairuxClient, type ChatMessage, type RosterEntry } from './api'; +import type { AgentVoiceRunOptions } from './voice-protocol'; import { newRoomLink } from './room'; import { COPILOT_HELP, runCopilot } from './copilot'; import { UsageError } from './errors'; @@ -49,6 +51,8 @@ export interface Deps { maxPolls?: number; /** Tests stop `copilot watch` after N stream connections; production reconnects. */ maxStreams?: number; + /** Starts the hidden WebRTC renderer used by `pairux voice join`. */ + startVoice?: (options: AgentVoiceRunOptions) => Promise; login?: typeof listenForCallback; /** The app version, for the User-Agent and `pairux version`. */ version?: string; @@ -72,6 +76,8 @@ Usage: pairux who [--json] List who is in the session pairux status [--json] Show the agent this CLI is running as pairux leave Leave the session + pairux voice join [--json] Join call audio; stdout is transcripts, stdin speaks + Each input line is spoken via C0mpute TTS. Send /quit to leave. pairux room Print a link to a fresh anonymous room: end-to-end encrypted voice/video, no account needed to join pairux redeem Redeem a credit code (adds days of a paid plan) @@ -86,9 +92,16 @@ Environment: PAIRUX_AGENT_NAME default display name for join PAIRUX_AGENT_CLIENT default --client for join PAIRUX_CONFIG_DIR where config.json lives (default ~/.config/pairux) + C0MPUTE_SPEECH_URL speech gateway; default http://127.0.0.1:8787 + C0MPUTE_SPEECH_TOKEN token used by \`c0mpute speech serve\` (required for voice) + C0MPUTE_TTS_ENGINE default kokoro + C0MPUTE_TTS_VOICE default af_sarah + C0MPUTE_TTS_LOCALE default en-US + C0MPUTE_STT_LANGUAGE optional language hint, e.g. en -Agents watch and chat. They cannot see or control anyone's screen, and the -host can remove them at any time. Docs: https://pairux.com/docs/agents`; +Agents can follow chat and call audio, and speak into the call. They cannot see +or control anyone's screen, and the host can remove them at any time. +Docs: https://pairux.com/docs/agents`; function defaultName(env: NodeJS.ProcessEnv): string { if (env.PAIRUX_AGENT_NAME) return env.PAIRUX_AGENT_NAME; @@ -315,7 +328,7 @@ export async function run(argv: string[], deps: Deps): Promise { deps.out( `Joined ${joinCode} as ${joined.displayName}${joined.owned ? ' (your agent)' : ' (anonymous agent)'}.` ); - deps.out('Run `pairux listen` to follow the chat and stay present.'); + deps.out('Run `pairux listen` for chat or `pairux voice join` to join call audio.'); return 0; } @@ -335,6 +348,60 @@ export async function run(argv: string[], deps: Deps): Promise { return 0; } + case 'voice': { + const [action, ...voiceArgs] = rest; + if (!action || ['help', '--help', '-h'].includes(action)) { + deps.out( + 'Usage: pairux voice join [--json]\nJoin the active call as this agent. Each stdin line is spoken with C0mpute TTS; transcript events go to stdout. Send /quit to leave.' + ); + return 0; + } + if (action !== 'join') throw new UsageError(`Unknown voice command "${action}".`); + const { values, positionals } = parseArgs({ + args: voiceArgs, + allowPositionals: true, + options: { json: { type: 'boolean', default: false } }, + }); + if (positionals.length > 0) throw new UsageError('Usage: pairux voice join [--json]'); + if (!deps.startVoice) { + throw new UsageError( + 'The PairUX desktop CLI voice runtime is unavailable in this build.' + ); + } + + const agent = requireAgent(); + const speechToken = deps.env.C0MPUTE_SPEECH_TOKEN; + if (!speechToken) { + throw new UsageError( + 'Set C0MPUTE_SPEECH_TOKEN to the token used by `c0mpute speech serve` before joining voice.' + ); + } + const livekit = await client.voiceToken(agent.participantId); + return await deps.startVoice({ + participantId: agent.participantId, + displayName: agent.displayName, + token: livekit.token, + url: livekit.url, + ...(livekit.iceServers ? { iceServers: livekit.iceServers } : {}), + speechUrl: deps.env.C0MPUTE_SPEECH_URL ?? 'http://127.0.0.1:8787', + speechToken, + ttsEngine: deps.env.C0MPUTE_TTS_ENGINE ?? 'kokoro', + ttsVoice: deps.env.C0MPUTE_TTS_VOICE ?? 'af_sarah', + ttsLocale: deps.env.C0MPUTE_TTS_LOCALE ?? 'en-US', + sttModel: 'base', + ...(deps.env.C0MPUTE_STT_LANGUAGE ? { language: deps.env.C0MPUTE_STT_LANGUAGE } : {}), + json: values.json, + out: deps.out, + err: deps.err, + heartbeat: async () => { + const result = await client.poll(agent.participantId, new Date().toISOString()); + if (result.session.status === 'ended') { + throw new ApiError(410, 'The session has ended'); + } + }, + }); + } + case 'who': case 'status': { const { values } = parseArgs({ diff --git a/apps/desktop/src/main/agent-cli/index.ts b/apps/desktop/src/main/agent-cli/index.ts index 8f83af7a..58eebec8 100644 --- a/apps/desktop/src/main/agent-cli/index.ts +++ b/apps/desktop/src/main/agent-cli/index.ts @@ -1,5 +1,5 @@ /** - * `pairux login|logout|whoami|join|listen|say|who|status|leave|room|redeem|invites|copilot` + * `pairux login|logout|whoami|join|listen|say|who|status|leave|voice|room|redeem|invites|copilot` * * The installed `pairux` launcher passes any argument it does not handle * itself (update, uninstall, --help, --version) straight to the app, the same @@ -21,6 +21,7 @@ export const AGENT_COMMANDS = new Set([ 'who', 'status', 'leave', + 'voice', 'room', 'redeem', 'invites', @@ -56,6 +57,10 @@ export function runAgentCommand(args: string[], version: string): Promise new Promise((resolve) => setTimeout(resolve, ms)), readStdin: readAllStdin, + startVoice: async (options) => { + const { runAgentVoice } = await import('./voice-runtime'); + return runAgentVoice(options); + }, version, }); } diff --git a/apps/desktop/src/main/agent-cli/voice-protocol.ts b/apps/desktop/src/main/agent-cli/voice-protocol.ts new file mode 100644 index 00000000..ebb18eeb --- /dev/null +++ b/apps/desktop/src/main/agent-cli/voice-protocol.ts @@ -0,0 +1,24 @@ +export interface VoiceIceServer { + urls: string | string[]; + username?: string; + credential?: string; +} + +export interface AgentVoiceRunOptions { + participantId: string; + displayName: string; + token: string; + url: string; + iceServers?: VoiceIceServer[]; + speechUrl: string; + speechToken: string; + ttsEngine: string; + ttsVoice: string; + ttsLocale: string; + sttModel: string; + language?: string; + json: boolean; + out: (line: string) => void; + err: (line: string) => void; + heartbeat: () => Promise; +} diff --git a/apps/desktop/src/main/agent-cli/voice-runtime.ts b/apps/desktop/src/main/agent-cli/voice-runtime.ts new file mode 100644 index 00000000..01b0a2bf --- /dev/null +++ b/apps/desktop/src/main/agent-cli/voice-runtime.ts @@ -0,0 +1,272 @@ +import { app, BrowserWindow, ipcMain } from 'electron'; +import { createInterface } from 'node:readline'; +import { join } from 'node:path'; +import type { AgentVoiceRunOptions } from './voice-protocol'; + +interface RendererEvent { + type?: string; + text?: string; + error?: string; + audio?: ArrayBuffer | Uint8Array; +} + +interface VoiceConnectPayload { + token: string; + url: string; + iceServers?: AgentVoiceRunOptions['iceServers']; +} + +function gatewayUrl(base: string, path: string): string { + return `${base.replace(/\/+$/, '')}${path}`; +} + +function describeError(error: unknown): string { + if (error instanceof Error) return error.message; + if (typeof error === 'string') return error; + return JSON.stringify(error); +} + +function responseError(body: unknown, fallback: string): string { + if ( + typeof body === 'object' && + body !== null && + 'error' in body && + typeof body.error === 'string' + ) { + return body.error; + } + return fallback; +} + +function stringField(event: Record, field: string): string { + const value = event[field]; + return typeof value === 'string' ? value : ''; +} + +function emit(options: AgentVoiceRunOptions, event: Record): void { + if (options.json) options.out(JSON.stringify(event)); + else if (event.type === 'transcript') options.out(stringField(event, 'text')); + else if (event.type === 'ended') options.out(`Voice call ended: ${stringField(event, 'reason')}`); + else if (event.type === 'connected') options.out(`Voice connected as ${options.displayName}.`); + else if (event.type === 'warning') options.err(`Voice: ${stringField(event, 'message')}`); +} + +async function synthesize(options: AgentVoiceRunOptions, text: string): Promise { + const response = await fetch(gatewayUrl(options.speechUrl, '/v1/tts'), { + method: 'POST', + headers: { + Authorization: `Bearer ${options.speechToken}`, + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ + text, + engine: options.ttsEngine, + voice: options.ttsVoice, + locale: options.ttsLocale, + speed: 1, + persona: '', + }), + signal: AbortSignal.timeout(5 * 60_000), + }); + if (!response.ok) { + const body: unknown = await response.json().catch(() => ({})); + throw new Error( + `C0mpute TTS failed (${String(response.status)}): ${responseError(body, response.statusText)}` + ); + } + return response.arrayBuffer(); +} + +async function transcribe(options: AgentVoiceRunOptions, audio: ArrayBuffer | Uint8Array) { + const query = new URLSearchParams({ model: options.sttModel }); + if (options.language) query.set('language', options.language); + const response = await fetch(gatewayUrl(options.speechUrl, `/v1/stt?${query.toString()}`), { + method: 'POST', + headers: { + Authorization: `Bearer ${options.speechToken}`, + 'Content-Type': 'audio/wav', + }, + body: Buffer.from(audio instanceof ArrayBuffer ? new Uint8Array(audio) : audio), + signal: AbortSignal.timeout(5 * 60_000), + }); + if (!response.ok) { + const body: unknown = await response.json().catch(() => ({})); + throw new Error( + `C0mpute STT failed (${String(response.status)}): ${responseError(body, response.statusText)}` + ); + } + return (await response.json()) as { text?: string; language?: string }; +} + +function parseSpeechInput(line: string): string | null { + const value = line.trim(); + if (!value) return null; + try { + const parsed = JSON.parse(value) as { type?: string; say?: string; text?: string }; + if (parsed.type && parsed.type !== 'speak') return null; + return parsed.say ?? parsed.text ?? null; + } catch { + return value; + } +} + +/** Run a hidden Chromium renderer that provides WebRTC and WebAudio to the CLI. */ +export async function runAgentVoice(options: AgentVoiceRunOptions): Promise { + await app.whenReady(); + const win = new BrowserWindow({ + show: false, + width: 1, + height: 1, + webPreferences: { + preload: join(__dirname, '../preload/index.js'), + contextIsolation: true, + nodeIntegration: false, + sandbox: true, + backgroundThrottling: false, + }, + }); + + return await new Promise((resolve) => { + let finished = false; + let connected = false; + let transcribing = false; + const audioQueue: (ArrayBuffer | Uint8Array)[] = []; + let ttsQueue = Promise.resolve(); + const input = createInterface({ + input: process.stdin, + crlfDelay: Infinity, + terminal: process.stdin.isTTY, + }); + + const cleanup = () => { + clearInterval(heartbeatTimer); + input.close(); + process.removeListener('SIGINT', stop); + process.removeListener('SIGTERM', stop); + ipcMain.removeListener('agent-voice:event', onRendererEvent); + win.removeAllListeners('closed'); + if (!win.isDestroyed()) { + win.webContents.send('agent-voice:stop'); + win.destroy(); + } + }; + const finish = (code: number) => { + if (finished) return; + finished = true; + cleanup(); + resolve(code); + }; + const stop = () => { + finish(0); + }; + const drainAudioQueue = async () => { + if (transcribing) return; + transcribing = true; + try { + while (!finished && audioQueue.length > 0) { + const audio = audioQueue.shift(); + if (!audio) continue; + try { + const result = await transcribe(options, audio); + if (result.text?.trim()) { + emit(options, { type: 'transcript', text: result.text.trim() }); + } + } catch (error) { + emit(options, { type: 'warning', message: String(error) }); + } + } + } finally { + transcribing = false; + if (!finished && audioQueue.length > 0) void drainAudioQueue(); + } + }; + const sendConnect = (connection: VoiceConnectPayload) => { + if (!win.isDestroyed()) win.webContents.send('agent-voice:connect', connection); + }; + const onRendererEvent = (_event: Electron.IpcMainEvent, raw: unknown) => { + if (finished || _event.sender !== win.webContents) return; + const event = raw as RendererEvent; + if (event.type === 'ready') { + sendConnect({ token: options.token, url: options.url, iceServers: options.iceServers }); + } else if (event.type === 'connected') { + connected = true; + emit(options, { type: 'connected' }); + } else if (event.type === 'transcript' && event.text?.trim()) { + emit(options, { type: 'transcript', text: event.text.trim() }); + } else if (event.type === 'audio' && event.audio) { + if (audioQueue.length >= 3) { + emit(options, { + type: 'warning', + message: 'C0mpute STT is behind; dropping an audio chunk.', + }); + } else { + audioQueue.push(event.audio); + void drainAudioQueue(); + } + } else if (event.type === 'error') { + emit(options, { type: 'warning', message: event.error ?? 'Audio connection error' }); + if (!connected) finish(1); + } else if (event.type === 'disconnected' && connected) { + connected = false; + emit(options, { type: 'warning', message: event.error ?? 'LiveKit disconnected' }); + finish(1); + } + }; + + ipcMain.on('agent-voice:event', onRendererEvent); + process.once('SIGINT', stop); + process.once('SIGTERM', stop); + win.on('closed', () => { + finish(0); + }); + + input.on('line', (line) => { + const trimmed = line.trim(); + if (trimmed === '/quit') { + finish(0); + return; + } + const text = parseSpeechInput(line); + if (!text) return; + ttsQueue = ttsQueue.then(async () => { + try { + const audio = await synthesize(options, text); + if (!finished && !win.isDestroyed()) win.webContents.send('agent-voice:speak', audio); + } catch (error) { + emit(options, { type: 'warning', message: String(error) }); + } + }); + }); + + const heartbeatTimer = setInterval(() => { + void options.heartbeat().catch((error: unknown) => { + if ( + typeof error === 'object' && + error !== null && + 'status' in error && + error.status === 410 + ) { + emit(options, { type: 'ended', reason: describeError(error) }); + finish(0); + } else { + emit(options, { + type: 'warning', + message: `Could not refresh agent status: ${describeError(error)}`, + }); + } + }); + }, 30_000); + const rendererUrl = process.env.ELECTRON_RENDERER_URL; + const loadRenderer = + !app.isPackaged && rendererUrl + ? win.loadURL(new URL('agent-voice.html', `${rendererUrl.replace(/\/+$/, '')}/`).toString()) + : win.loadFile(join(__dirname, '../renderer/agent-voice.html')); + void loadRenderer.catch((error: unknown) => { + emit(options, { + type: 'warning', + message: `Could not load PairUX voice runtime: ${describeError(error)}`, + }); + finish(1); + }); + }); +} diff --git a/apps/desktop/src/main/index.ts b/apps/desktop/src/main/index.ts index 38cfc95f..25d9a381 100644 --- a/apps/desktop/src/main/index.ts +++ b/apps/desktop/src/main/index.ts @@ -87,6 +87,10 @@ const isDaemonMode = process.argv.includes('daemon') || process.argv.includes('- // by it. See ./agent-cli. const agentArgs = parseAgentArgv(process.argv, app.isPackaged); if (agentArgs) { + if (agentArgs[0] === 'voice') { + // The hidden voice renderer has no user gesture to unlock WebAudio. + app.commandLine.appendSwitch('autoplay-policy', 'no-user-gesture-required'); + } if (process.platform === 'darwin') app.dock?.hide(); void runAgentCommand(agentArgs, app.getVersion()).then( (code) => { diff --git a/apps/desktop/src/preload/index.ts b/apps/desktop/src/preload/index.ts index 0bda73c0..e6812ee3 100644 --- a/apps/desktop/src/preload/index.ts +++ b/apps/desktop/src/preload/index.ts @@ -35,5 +35,33 @@ const electronAPI = { // Expose API to renderer contextBridge.exposeInMainWorld('electronAPI', electronAPI); +// The agent voice page is a tiny, hidden renderer launched by `pairux voice` +// so the CLI can use Chromium's WebRTC stack without opening the desktop UI. +// Keep its bridge separate from the broad app IPC surface. +contextBridge.exposeInMainWorld('agentVoice', { + onCommand: (callback: (command: { type: string; payload?: unknown }) => void) => { + const connect = (_event: IpcRendererEvent, payload: unknown) => { + callback({ type: 'connect', payload }); + }; + const speak = (_event: IpcRendererEvent, payload: unknown) => { + callback({ type: 'speak', payload }); + }; + const stop = () => { + callback({ type: 'stop' }); + }; + ipcRenderer.on('agent-voice:connect', connect); + ipcRenderer.on('agent-voice:speak', speak); + ipcRenderer.on('agent-voice:stop', stop); + return () => { + ipcRenderer.removeListener('agent-voice:connect', connect); + ipcRenderer.removeListener('agent-voice:speak', speak); + ipcRenderer.removeListener('agent-voice:stop', stop); + }; + }, + send: (event: unknown) => { + ipcRenderer.send('agent-voice:event', event); + }, +}); + // Type declaration for use in renderer export type ElectronAPI = typeof electronAPI; diff --git a/apps/desktop/src/renderer/agent-voice.html b/apps/desktop/src/renderer/agent-voice.html new file mode 100644 index 00000000..9a6fc6b8 --- /dev/null +++ b/apps/desktop/src/renderer/agent-voice.html @@ -0,0 +1,15 @@ + + + + + + + PairUX voice runtime + + + + + diff --git a/apps/desktop/src/renderer/agent-voice.ts b/apps/desktop/src/renderer/agent-voice.ts new file mode 100644 index 00000000..8d1a7fd4 --- /dev/null +++ b/apps/desktop/src/renderer/agent-voice.ts @@ -0,0 +1,215 @@ +import { Room, RoomEvent, Track } from 'livekit-client'; + +interface VoiceConnectConfig { + token: string; + url: string; + iceServers?: RTCIceServer[]; +} + +interface VoiceCommand { + type: 'connect' | 'speak' | 'stop'; + payload?: VoiceConnectConfig | ArrayBuffer; +} + +interface AgentVoiceBridge { + onCommand(callback: (command: VoiceCommand) => void): () => void; + send(event: { type: string; text?: string; error?: string; audio?: Uint8Array }): void; +} + +declare global { + interface Window { + agentVoice: AgentVoiceBridge; + } +} + +const bridge = window.agentVoice; +let room: Room | null = null; +let audioContext: AudioContext | null = null; +let mixer: GainNode | null = null; +let processor: AudioWorkletNode | null = null; +let mutedOutput: GainNode | null = null; +let ttsDestination: MediaStreamAudioDestinationNode | null = null; +let capturing = false; +let intentionalDisconnect = false; +let pendingSamples: number[] = []; +const tracks = new Map(); + +const CHUNK_SECONDS = 5; +const STT_RATE = 16_000; + +function sendError(error: unknown) { + bridge.send({ type: 'error', error: error instanceof Error ? error.message : String(error) }); +} + +function encodeWav(input: Float32Array, sourceRate: number): Uint8Array { + const ratio = sourceRate / STT_RATE; + const outputLength = Math.floor(input.length / ratio); + const bytes = new Uint8Array(44 + outputLength * 2); + const view = new DataView(bytes.buffer); + const write = (offset: number, value: string) => { + for (let index = 0; index < value.length; index++) + view.setUint8(offset + index, value.charCodeAt(index)); + }; + write(0, 'RIFF'); + view.setUint32(4, 36 + outputLength * 2, true); + write(8, 'WAVE'); + write(12, 'fmt '); + view.setUint32(16, 16, true); + view.setUint16(20, 1, true); + view.setUint16(22, 1, true); + view.setUint32(24, STT_RATE, true); + view.setUint32(28, STT_RATE * 2, true); + view.setUint16(32, 2, true); + view.setUint16(34, 16, true); + write(36, 'data'); + view.setUint32(40, outputLength * 2, true); + for (let index = 0; index < outputLength; index++) { + const sample = Math.max(-1, Math.min(1, input[Math.floor(index * ratio)] ?? 0)); + view.setInt16(44 + index * 2, sample < 0 ? sample * 0x8000 : sample * 0x7fff, true); + } + return bytes; +} + +function handlePcm(channel: Float32Array) { + if (!capturing) return; + for (const sample of channel) { + pendingSamples.push(sample); + } + const wanted = Math.floor((audioContext?.sampleRate ?? 48_000) * CHUNK_SECONDS); + if (pendingSamples.length < wanted) return; + + const audio = new Float32Array(pendingSamples.splice(0, wanted)); + let squareSum = 0; + for (const sample of audio) squareSum += sample * sample; + const rms = Math.sqrt(squareSum / audio.length); + if (rms < 0.003) return; + bridge.send({ type: 'audio', audio: encodeWav(audio, audioContext?.sampleRate ?? 48_000) }); +} + +function addRemoteAudio(track: MediaStreamTrack, sid: string) { + if (!audioContext || !mixer || tracks.has(sid)) return; + const source = audioContext.createMediaStreamSource(new MediaStream([track])); + source.connect(mixer); + tracks.set(sid, { source, track }); +} + +function removeRemoteAudio(sid: string) { + const entry = tracks.get(sid); + if (!entry) return; + entry.source.disconnect(); + tracks.delete(sid); +} + +async function connect(config: VoiceConnectConfig) { + if (room) { + intentionalDisconnect = true; + await room.disconnect(); + room = null; + tracks.clear(); + } + if (!audioContext) { + audioContext = new AudioContext(); + await audioContext.resume(); + await audioContext.audioWorklet.addModule( + new URL('./agent-voice-worklet.js', window.location.href) + ); + mixer = audioContext.createGain(); + mixer.channelCount = 1; + mixer.channelCountMode = 'explicit'; + processor = new AudioWorkletNode(audioContext, 'pairux-voice-capture', { + numberOfInputs: 1, + numberOfOutputs: 1, + channelCount: 1, + channelCountMode: 'explicit', + outputChannelCount: [1], + }); + processor.port.onmessage = (event: MessageEvent) => { + handlePcm(event.data); + }; + mutedOutput = audioContext.createGain(); + mutedOutput.gain.value = 0; + ttsDestination = audioContext.createMediaStreamDestination(); + mixer.connect(processor); + processor.connect(mutedOutput); + mutedOutput.connect(audioContext.destination); + } + + const nextRoom = new Room({ adaptiveStream: false, dynacast: false }); + room = nextRoom; + nextRoom.on(RoomEvent.TrackSubscribed, (track, publication) => { + if (track.kind === Track.Kind.Audio) { + addRemoteAudio(track.mediaStreamTrack, publication.trackSid); + } + }); + nextRoom.on(RoomEvent.TrackUnsubscribed, (_track, publication) => { + removeRemoteAudio(publication.trackSid); + }); + nextRoom.on(RoomEvent.Disconnected, (reason) => { + if (!intentionalDisconnect) { + bridge.send({ type: 'disconnected', error: reason ? String(reason) : undefined }); + } + intentionalDisconnect = false; + }); + + for (let attempt = 0; attempt < 3; attempt++) { + try { + await nextRoom.connect(config.url, config.token, { + rtcConfig: config.iceServers ? { iceServers: config.iceServers } : undefined, + }); + break; + } catch (error) { + await nextRoom.disconnect().catch(() => undefined); + if (attempt === 2) { + throw error instanceof Error ? error : new Error(String(error)); + } + await new Promise((resolve) => setTimeout(resolve, 800 * (attempt + 1))); + } + } + + if (!ttsDestination) throw new Error('Audio output is unavailable'); + const outgoing = ttsDestination.stream.getAudioTracks()[0]; + await nextRoom.localParticipant.publishTrack(outgoing, { source: Track.Source.Microphone }); + capturing = true; + bridge.send({ type: 'connected' }); +} + +async function speak(audio: ArrayBuffer) { + if (!audioContext || !ttsDestination) throw new Error('Voice call has not connected yet'); + await audioContext.resume(); + const copy = audio.slice(0); + const buffer = await audioContext.decodeAudioData(copy); + const source = audioContext.createBufferSource(); + source.buffer = buffer; + source.connect(ttsDestination); + source.onended = () => { + source.disconnect(); + }; + source.start(); +} + +async function stop() { + capturing = false; + pendingSamples = []; + for (const sid of tracks.keys()) removeRemoteAudio(sid); + if (room) await room.disconnect().catch(() => undefined); + room = null; + if (processor) processor.disconnect(); + if (mixer) mixer.disconnect(); + if (mutedOutput) mutedOutput.disconnect(); + await audioContext?.close().catch(() => undefined); + audioContext = null; +} + +bridge.onCommand((command) => { + if (command.type === 'connect') { + void connect(command.payload as VoiceConnectConfig).catch(sendError); + } else if (command.type === 'speak') { + void speak(command.payload as ArrayBuffer).catch((error: unknown) => { + bridge.send({ type: 'error', error: error instanceof Error ? error.message : String(error) }); + }); + } else { + void stop(); + } +}); + +bridge.send({ type: 'ready' }); diff --git a/apps/desktop/src/renderer/components/ParticipantList.tsx b/apps/desktop/src/renderer/components/ParticipantList.tsx index 1d1ef9ee..caf8ff5e 100644 --- a/apps/desktop/src/renderer/components/ParticipantList.tsx +++ b/apps/desktop/src/renderer/components/ParticipantList.tsx @@ -179,7 +179,7 @@ export function ParticipantList({ {activeParticipants.map((participant) => { const isCurrentUser = participant.user_id === currentUserId; const isParticipantHost = participant.role === 'host'; - // Agents watch and chat only: no media to mute, no control to grant. + // Agents can publish voice but never receive remote control. const isAgent = participant.kind === 'agent'; const hasAnyActions = Boolean(onGrantControl) || diff --git a/apps/installer/scripts/install.ps1 b/apps/installer/scripts/install.ps1 index 81e55d91..0435a000 100644 --- a/apps/installer/scripts/install.ps1 +++ b/apps/installer/scripts/install.ps1 @@ -164,7 +164,7 @@ if /I "%~1"=="update" goto :update if /I "%~1"=="upgrade" goto :update if /I "%~1"=="uninstall" goto :uninstall if /I "%~1"=="remove" goto :uninstall -for %%C in (login logout whoami join listen say who status leave room redeem invites) do if /I "%~1"=="%%C" goto :agent +for %%C in (login logout whoami join listen say who status leave voice room redeem invites) do if /I "%~1"=="%%C" goto :agent goto :run :help @@ -183,6 +183,7 @@ echo login^|logout^|whoami Sign in so your agents are labelled as yours echo join CODE Join a session as an agent (--name, --client) echo listen [--json] Follow chat and who is here; keeps the agent present echo say MESSAGE Post in the session chat +echo voice join [--json] Hear and speak in the call via C0mpute echo who ^| status ^| leave Roster, this agent, leave the session echo redeem CODE Redeem a credit code (adds days of a paid plan) echo invites [create] Your invites and referral link diff --git a/apps/installer/scripts/install.sh b/apps/installer/scripts/install.sh index dd19458b..a6c0d1a5 100644 --- a/apps/installer/scripts/install.sh +++ b/apps/installer/scripts/install.sh @@ -777,6 +777,7 @@ case "\${1-}" in echo " join Join a session as an agent (--name, --client)" echo " listen [--json] Follow chat and who is here; keeps the agent present" echo " say Post in the session chat" + echo " voice join [--json] Hear and speak in the call via C0mpute" echo " who | status | leave Roster, this agent, leave the session" echo " redeem Redeem a credit code (adds days of a paid plan)" echo " invites [create] Your invites and referral link" @@ -990,6 +991,7 @@ case "\${1-}" in echo " join Join a session as an agent (--name, --client)" echo " listen [--json] Follow chat and who is here; keeps the agent present" echo " say Post in the session chat" + echo " voice join [--json] Hear and speak in the call via C0mpute" echo " who | status | leave Roster, this agent, leave the session" echo " redeem Redeem a credit code (adds days of a paid plan)" echo " invites [create] Your invites and referral link" diff --git a/apps/web/src/app/api/v1/agents/[participantId]/voice-token/route.ts b/apps/web/src/app/api/v1/agents/[participantId]/voice-token/route.ts new file mode 100644 index 00000000..8c052e61 --- /dev/null +++ b/apps/web/src/app/api/v1/agents/[participantId]/voice-token/route.ts @@ -0,0 +1,104 @@ +import { z } from 'zod'; +import { AccessToken, TrackSource } from 'livekit-server-sdk'; +import type { VideoGrant } from 'livekit-server-sdk'; +import { serviceClient } from '@/lib/supabase/service'; +import { successResponse, errorResponse, handleApiError } from '@/lib/api'; +import { getIceServers } from '@/lib/ice-servers'; +import { FixedWindowRateLimiter, getClientIp } from '@/lib/rate-limit'; + +interface RouteParams { + params: Promise<{ participantId: string }>; +} + +const idSchema = z.guid('Invalid participant ID'); +const tokensByIp = new FixedWindowRateLimiter(30, 60_000); +const tokensByParticipant = new FixedWindowRateLimiter(12, 60_000); + +/** + * POST /api/v1/agents/[participantId]/voice-token + * + * The random participant id is the agent's existing CLI capability. A live + * agent can mint a session-scoped LiveKit token for microphone audio only. + */ +export async function POST(request: Request, { params }: RouteParams) { + try { + const ipLimit = tokensByIp.check(getClientIp(request)); + if (!ipLimit.success) { + return errorResponse( + `Too many voice token requests. Try again in ${String(ipLimit.retryAfterSeconds)} seconds.`, + 429 + ); + } + + const participantId = idSchema.parse((await params).participantId); + const participantLimit = tokensByParticipant.check(participantId); + if (!participantLimit.success) { + return errorResponse( + `Too many voice token requests. Try again in ${String(participantLimit.retryAfterSeconds)} seconds.`, + 429 + ); + } + + const apiKey = process.env.LIVEKIT_API_KEY; + const apiSecret = process.env.LIVEKIT_API_SECRET; + const url = process.env.NEXT_PUBLIC_LIVEKIT_URL; + if (!apiKey || !apiSecret || !url) return errorResponse('LiveKit not configured', 503); + + const db = serviceClient(); + const { data: participant } = (await db + .from('session_participants') + .select('id, session_id, display_name, kind, left_at, agent_owner_id') + .eq('id', participantId) + .maybeSingle()) as { + data: { + id: string; + session_id: string; + display_name: string; + kind: string; + left_at: string | null; + agent_owner_id: string | null; + } | null; + }; + + if (participant?.kind !== 'agent') return errorResponse('Agent not found', 404); + if (participant.left_at) return errorResponse('This agent has left or was removed', 410); + + const { data: session } = (await db + .from('sessions') + .select('id, mode, status') + .eq('id', participant.session_id) + .maybeSingle()) as { + data: { id: string; mode: string; status: string } | null; + }; + if (!session) return errorResponse('Session not found', 404); + if (session.status === 'ended') return errorResponse('The session has ended', 410); + if (session.mode !== 'sfu') { + return errorResponse('Voice CLI requires a session running in SFU mode', 409); + } + + const roomName = `session-${session.id}`; + const token = new AccessToken(apiKey, apiSecret, { + identity: participant.id, + name: participant.display_name, + ttl: '24h', + metadata: JSON.stringify({ role: 'agent', userId: participant.agent_owner_id }), + }); + const grant: VideoGrant = { + room: roomName, + roomJoin: true, + canPublishSources: [TrackSource.MICROPHONE], + canSubscribe: true, + canPublishData: false, + }; + token.addGrant(grant); + + return successResponse({ + token: await token.toJwt(), + url, + roomName, + iceServers: await getIceServers(), + }); + } catch (error) { + return handleApiError(error); + } +} diff --git a/apps/web/src/app/docs/agents/page.tsx b/apps/web/src/app/docs/agents/page.tsx index 177503e9..2447ff5d 100644 --- a/apps/web/src/app/docs/agents/page.tsx +++ b/apps/web/src/app/docs/agents/page.tsx @@ -31,6 +31,10 @@ const commands: { cmd: string; what: string }[] = [ cmd: 'pairux listen', what: 'Print new chat messages and people joining or leaving. Keeps the agent present.', }, + { + cmd: 'pairux voice join --json', + what: 'Join call audio as this agent. Emits transcript JSONL; JSONL stdin speaks through C0mpute TTS.', + }, { cmd: 'pairux say "Tests pass on my side"', what: 'Post in the session chat.' }, { cmd: 'pairux who', what: 'List who is in the session.' }, { cmd: 'pairux leave', what: 'Leave the session.' }, @@ -49,6 +53,10 @@ const api: { route: string; what: string }[] = [ route: 'POST /api/v1/agents/:participantId/messages', what: '{ content } up to 500 characters, 10 per minute.', }, + { + route: 'POST /api/v1/agents/:participantId/voice-token', + what: 'Mint a 24-hour LiveKit token scoped to this session and microphone audio. The session must use SFU mode.', + }, { route: 'DELETE /api/v1/agents/:participantId', what: 'Leave the session.' }, ]; @@ -73,11 +81,11 @@ export default function AgentDocsPage() {

)}

- An AI agent can join a PairUX session next to the people in it. It uses the same join - code as everyone else, reads the chat, answers in it, and appears in the participant - list with an Agent badge in the browser, the installed web app, the - desktop app and the mobile app. Use it to let Claude Code, moshcode or your own script - follow a pairing session and report back while you work. + An AI agent can join a PairUX session next to the people in it. Each agent uses its own + CLI process and participant identity. Agents read chat, hear call audio, speak through + C0mpute TTS, and appear in the participant list with an Agent badge. + Use this to let Claude Code, moshcode or your own script participate directly in a + pairing call.

@@ -101,6 +109,33 @@ export default function AgentDocsPage() { names the agent "Claude Code" and tags it claude-code unless you pass --name or --client.

+

Voice setup

+

+ Voice requires an SFU session and a reachable C0mpute speech gateway. Set up its + speech workers and run the gateway with a token shared with the CLI. If you are + providing the compute, start a worker in a separate process: +

+ 
+              {`c0mpute tts setup --engine all
+c0mpute stt setup
+c0mpute worker start --roles tts,stt &
+export C0MPUTE_SPEECH_TOKEN="$(openssl rand -hex 32)"
+c0mpute speech serve &
+pairux join ABC123 --name "Claude Code"
+pairux voice join --json`}
+
+

+ The gateway defaults to http://127.0.0.1:8787. Set{' '} + C0MPUTE_SPEECH_URL if it runs elsewhere. Each agent needs its own CLI + configuration and must run pairux join; the voice process stays attached + to that agent until /quit or removal by the host. Write plain text or{' '} + {'{"say":"Hello"}'} to stdin to speak. With --json, each + transcript is one JSON line on stdout. Configure C0MPUTE_TTS_ENGINE,{' '} + C0MPUTE_TTS_VOICE and C0MPUTE_TTS_LOCALE for the voice.{' '} + C0mpute STT works in audio chunks, so transcripts can arrive several seconds after a + person speaks. The speech gateway's default maximum job price is zero; PairUX + does not fall back to another provider. +

@@ -110,16 +145,20 @@ export default function AgentDocsPage() {
  • It can read the public chat, post in it, and see who is in the session.
  • - It cannot see or hear the shared screen, take remote control, or send direct - messages. + In SFU sessions, it can hear call audio and publish speech audio generated by + C0mpute. +
    • +
  • + It cannot see the shared screen, take remote control, or send direct messages.
  • The host can remove it at any time from the participant list. The agent notices on its next poll and stops.
  • - A session holds at most 5 agents. An agent that stops polling for two minutes is - treated as gone. + Hosts can disable agent joins. Agent processes that stop heartbeating for two + minutes are treated as gone. A session allows up to 20 agents in addition to its + human participant allowance.
  • Signing in with pairux login uses OAuth 2.1 with PKCE: you approve the @@ -134,9 +173,9 @@ export default function AgentDocsPage() {