voice

/** * Realtime voice observer bridge. * * Local HTTP server that: * - Serves a browser speech-to-speech client (xAI Realtime via ephemeral token) * - SSE `/api/events` for live status + harness→observer injects + harness status text * - POST `/api/to-harness` when the voice agent calls send_message_to_coding_harness * - In-process sendToObserver() for the pi tool send_message_to_observer * - In-process setHarnessStatus() for the pi tool set_harness_status (SSE → frontend HUD) */ import { createServer, type IncomingMessage, type Server, type ServerResponse, } from "node:http"; import { exec, spawn } from "node:child_process"; import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { Type, type Static } from "typebox"; const API_BASE = "https://api.x.ai/v1"; const DEFAULT_MODEL = "grok-voice-latest"; const DEFAULT_PORT = 3847; const sendToObserverParams = Type.Object({ message: Type.String({ description: "Concise update or answer for the voice observer to convey to the user", }), }); type SendToObserverParams = Static; const setHarnessStatusParams = Type.Object({ status: Type.String({ description: 'Short status text for the coding harness HUD (what you are doing, or the latest completion). Prefer a lasting completion line like "Done: fixed auth bug" over clearing. Do not pass empty string to clear unless explicitly asked.', }), }); type SetHarnessStatusParams = Static; /** Appended to the coding agent system prompt while /realtime-voice is running. */ const CODING_AGENT_OBSERVER_PROMPT = `REALTIME VOICE OBSERVER A voice observer co-pilot is watching this coding harness (started with /realtime-voice). The observer speaks with the user in a browser and cannot read your normal terminal output, tool results, or assistant messages. You MUST use these tools to keep the observer and user in the loop: - send_message_to_observer — send a concise update or answer the observer can speak. Call this when work finishes, when answering a question that came from the observer, or when something important happens the user should hear. Bias toward spoken completion messages (what finished and the outcome), not just mid-progress chatter. - set_harness_status — keep a short live status line on the observer UI up to date. Set it when work starts, update it as you progress, and when finished leave a clear completion status (e.g. "Done: added login tests" or "Failed: type error in auth.ts"). Do NOT clear the status line — leave the latest completion/failure text visible so the observer UI still shows what happened. Only clear if the user explicitly asks you to. Do not assume the observer saw anything you only printed in the terminal. Prefer short spoken-ready messages.`; interface RealtimeVoiceOptions { /** Fresh xAI bearer (OAuth or API key). */ getToken: () => Promise; /** Prefer config voice (leo, eve, ara, …). Default leo. */ voice?: string; model?: string; port?: number; /** Inject into the pi coding session. */ onHarnessMessage: (message: string) => void | Promise; /** Called when the server stops itself (browser closed, no SSE clients). */ onSelfStop?: () => void; /** Optional instructions override for the voice agent. */ instructions?: string; } interface RealtimeVoiceServer { port: number; url: string; sendToObserver: (message: string) => boolean; /** Push coding-harness status text to the browser HUD over SSE. */ setHarnessStatus: (status: string) => boolean; getHarnessStatus: () => string; stop: () => Promise; clientCount: () => number; } type SseClient = { res: ServerResponse; id: number; }; function json(res: ServerResponse, status: number, body: unknown): void { const data = JSON.stringify(body); res.writeHead(status, { "Content-Type": "application/json; charset=utf-8", "Content-Length": Buffer.byteLength(data), "Cache-Control": "no-store", }); res.end(data); } function readBody(req: IncomingMessage): Promise { return new Promise((resolve, reject) => { const chunks: Buffer[] = []; req.on("data", (c) => chunks.push(Buffer.isBuffer(c) ? c : Buffer.from(c))); req.on("end", () => resolve(Buffer.concat(chunks).toString("utf8"))); req.on("error", reject); }); } function openBrowser(url: string): void { // Detached spawn so the pi process is not coupled to the browser lifetime. const trySpawn = (command: string, args: string[]): boolean => { try { const child = spawn(command, args, { detached: true, stdio: "ignore", }); child.unref(); return true; } catch { return false; } }; if (process.platform === "darwin") { if (trySpawn("open", [url])) return; } else if (process.platform === "win32") { // `start` is a cmd built-in if (trySpawn("cmd", ["/c", "start", "", url])) return; } else { // Linux / BSD: prefer xdg-open, then common browsers. for (const cmd of [ "xdg-open", "gio", "gnome-open", "kde-open", "firefox", "chromium", "google-chrome", "chrome", ]) { if (cmd === "gio") { if (trySpawn("gio", ["open", url])) return; continue; } if (trySpawn(cmd, [url])) return; } } // Last resort: shell (may still work when PATH differs for non-login shells). let shellCmd = `xdg-open ${JSON.stringify(url)}`; if (process.platform === "darwin") shellCmd = `open ${JSON.stringify(url)}`; else if (process.platform === "win32") shellCmd = `start "" ${JSON.stringify(url)}`; exec(shellCmd, () => {}); } const DEFAULT_VOICE = "leo"; const DEFAULT_INSTRUCTIONS = `You are the voice co-pilot for a coding session powered by pi. You speak with the user in real time. Core rule: silently delegate real work to the coding agent via send_message_to_coding_harness — and never mention that agent, the harness, the handoff, or that you are waiting on anything. To the user, you are simply doing the work. How to sound: - Short natural acks while work is out: "On it!", "One moment.", "Working on that.", "Got it." - Never say: coding harness/agent, "I'll send that", "waiting on", "when it's done", "I told the terminal", tool relays, or any back-channel. - When [coding_harness] updates arrive, treat them as your own results and speak the substance in first person ("Done — fixed the auth bug."). Do not say you received a message from anywhere. What to do: - Answer quick questions yourself when you can. - For code changes, investigation, runs, deep explanations, or media: call send_message_to_coding_harness with a clear self-contained instruction, then ack like you're on it. - ALWAYS end every send_message_to_coding_harness message with an explicit instruction that the coding agent must call send_message_to_observer when done (success, failure, or answer), with a concise spoken-ready summary of the outcome. Example closing line: "When finished, call send_message_to_observer with a short summary of what you did and the result." Never omit this — the observer cannot see terminal output. - For opening a page/docs/PR/URL: call open_browser_tab with a full http(s) URL. Media you can take on (silent delegate — never name the handoff): - Images: Grok Imagine generate/edit (text-to-image, multi-image edit, aspect ratio, 1K/2K). - Video: image-to-video, reference-to-video, edit, short extensions. - Speech files: text-to-speech to a path; speech-to-text from files/URLs. Spell out paths and preferences in the delegated task, then say something like "On it — generating that now." Keep spoken replies concise. Prefer short turns. - Do not speak IP addresses or file paths unless the user specifically asks for them. Summarize locations in plain language instead (e.g. "the config file", "the local server") when enough context is clear.`; function clientHtml(opts: { model: string; voice: string }): string { const model = JSON.stringify(opts.model); const voice = JSON.stringify(opts.voice); const instructions = JSON.stringify(DEFAULT_INSTRUCTIONS); return ` voice

listening

`; } async function startRealtimeVoiceServer( options: RealtimeVoiceOptions, ): Promise { const port = options.port ?? DEFAULT_PORT; const model = options.model ?? DEFAULT_MODEL; const voice = options.voice ?? DEFAULT_VOICE; const instructions = options.instructions ?? DEFAULT_INSTRUCTIONS; let sseId = 0; const sseClients = new Set(); let closed = false; /** Latest coding-harness status line shown in the browser HUD. */ let harnessStatus = ""; // Auto-stop when the browser goes away: once a client has connected, // if all SSE clients disappear for longer than the grace window, self-stop // (equivalent of /realtime-voice-stop). Grace covers page refreshes. const EMPTY_GRACE_MS = 6000; let everHadClient = false; let emptyTimer: ReturnType | null = null; function clearEmptyTimer(): void { if (emptyTimer) { clearTimeout(emptyTimer); emptyTimer = null; } } function armEmptyStop(): void { clearEmptyTimer(); if (!everHadClient || closed || sseClients.size > 0) return; emptyTimer = setTimeout(() => { emptyTimer = null; if (!closed && sseClients.size === 0) void selfStop(); }, EMPTY_GRACE_MS); } async function selfStop(): Promise { if (closed) return; closed = true; clearEmptyTimer(); for (const c of sseClients) { try { c.res.end(); } catch { /* ignore */ } } sseClients.clear(); await new Promise((resolve) => { server.close(() => resolve()); setTimeout(() => resolve(), 500).unref?.(); }); options.onSelfStop?.(); } function broadcast(payload: unknown): void { const data = `data: ${JSON.stringify(payload)}\n\n`; for (const c of [...sseClients]) { try { c.res.write(data); } catch { sseClients.delete(c); } } } function sendToObserver(message: string): boolean { const text = message.trim(); if (!text) return false; broadcast({ type: "to_observer", message: text, ts: Date.now() }); return sseClients.size > 0; } function setHarnessStatus(status: string): boolean { harnessStatus = String(status ?? "").trim(); broadcast({ type: "harness_status", text: harnessStatus, ts: Date.now(), }); return sseClients.size > 0; } async function handleClientSecret(res: ServerResponse): Promise { const token = await options.getToken(); const r = await fetch(`${API_BASE}/realtime/client_secrets`, { method: "POST", headers: { Authorization: `Bearer ${token}`, "Content-Type": "application/json", }, body: JSON.stringify({ expires_after: { seconds: 300 } }), }); const text = await r.text(); if (!r.ok) { json(res, r.status, { error: text || r.statusText }); return; } try { json(res, 200, JSON.parse(text)); } catch { json(res, 200, { value: text.trim() }); } } const server: Server = createServer(async (req, res) => { try { const url = new URL(req.url || "/", `http://127.0.0.1:${port}`); const path = url.pathname; const method = req.method || "GET"; // Same-origin browser UI only (127.0.0.1). No CORS wildcard. if (method === "OPTIONS") { res.writeHead(204); res.end(); return; } if (method === "GET" && (path === "/" || path === "/index.html")) { const html = clientHtml({ model, voice }); res.writeHead(200, { "Content-Type": "text/html; charset=utf-8", "Cache-Control": "no-store", }); res.end(html); return; } if (method === "GET" && path === "/api/voices") { try { const token = await options.getToken(); const r = await fetch(`${API_BASE}/tts/voices`, { headers: { Authorization: `Bearer ${token}` }, }); const text = await r.text(); if (!r.ok) { json(res, r.status, { error: text || r.statusText }); return; } json(res, 200, JSON.parse(text)); } catch (e) { json(res, 500, { error: e instanceof Error ? e.message : String(e) }); } return; } if (method === "GET" && path === "/api/health") { json(res, 200, { ok: true, model, voice, clients: sseClients.size, harnessStatus, instructionsPreview: instructions.slice(0, 120), }); return; } if (method === "GET" && path === "/api/events") { res.writeHead(200, { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache, no-transform", Connection: "keep-alive", }); res.write( `data: ${JSON.stringify({ type: "status", text: "sse connected" })}\n\n`, ); // Replay current harness status so late joiners / refreshes stay in sync. if (harnessStatus) { res.write( `data: ${JSON.stringify({ type: "harness_status", text: harnessStatus, ts: Date.now() })}\n\n`, ); } const client: SseClient = { res, id: ++sseId }; sseClients.add(client); everHadClient = true; clearEmptyTimer(); const ping = setInterval(() => { try { res.write( `data: ${JSON.stringify({ type: "ping", ts: Date.now() })}\n\n`, ); } catch { /* ignore */ } }, 15000); req.on("close", () => { clearInterval(ping); sseClients.delete(client); armEmptyStop(); }); return; } if (method === "POST" && path === "/api/client-secret") { await handleClientSecret(res); return; } if (method === "POST" && path === "/api/to-harness") { const raw = await readBody(req); let message = ""; try { const body = JSON.parse(raw || "{}") as { message?: string }; message = String(body.message || "").trim(); } catch { json(res, 400, { error: "invalid JSON" }); return; } if (!message) { json(res, 400, { error: "message required" }); return; } broadcast({ type: "status", text: `to-harness: ${message.slice(0, 200)}`, }); try { await options.onHarnessMessage(message); json(res, 200, { ok: true }); } catch (e) { json(res, 500, { error: e instanceof Error ? e.message : String(e) }); } return; } if (method === "POST" && path === "/api/to-observer") { const raw = await readBody(req); let message = ""; try { const body = JSON.parse(raw || "{}") as { message?: string }; message = String(body.message || "").trim(); } catch { json(res, 400, { error: "invalid JSON" }); return; } if (!message) { json(res, 400, { error: "message required" }); return; } const delivered = sendToObserver(message); json(res, 200, { ok: true, delivered, clients: sseClients.size }); return; } if (method === "POST" && path === "/api/harness-status") { const raw = await readBody(req); let status = ""; try { const body = JSON.parse(raw || "{}") as { status?: string; text?: string; }; status = String(body.status ?? body.text ?? ""); } catch { json(res, 400, { error: "invalid JSON" }); return; } const delivered = setHarnessStatus(status); json(res, 200, { ok: true, delivered, status: harnessStatus, clients: sseClients.size, }); return; } if (method === "POST" && path === "/api/open-tab") { const raw = await readBody(req); let target = ""; try { const body = JSON.parse(raw || "{}") as { url?: string }; target = String(body.url || "").trim(); } catch { json(res, 400, { error: "invalid JSON" }); return; } let parsed: URL; try { parsed = new URL(target); } catch { json(res, 400, { error: "invalid url" }); return; } if (parsed.protocol !== "http:" && parsed.protocol !== "https:") { json(res, 400, { error: "only http(s) urls are allowed" }); return; } const href = parsed.toString(); openBrowser(href); broadcast({ type: "status", text: `open-tab: ${href}` }); json(res, 200, { ok: true, url: href }); return; } json(res, 404, { error: "not found" }); } catch (e) { json(res, 500, { error: e instanceof Error ? e.message : String(e) }); } }); await new Promise((resolve, reject) => { server.once("error", reject); server.listen(port, "127.0.0.1", () => resolve()); }); const url = `http://127.0.0.1:${port}/`; return { port, url, sendToObserver, setHarnessStatus, getHarnessStatus: () => harnessStatus, clientCount: () => sseClients.size, stop: () => new Promise((resolve) => { if (closed) return resolve(); closed = true; clearEmptyTimer(); broadcast({ type: "status", text: "server stopping" }); for (const c of sseClients) { try { c.res.end(); } catch { /* ignore */ } } sseClients.clear(); server.close(() => resolve()); // force-close hangers setTimeout(() => resolve(), 500).unref?.(); }), }; } function openRealtimeVoiceBrowser(url: string): void { openBrowser(url); } const OBSERVER_TOOL_NAMES = [ "send_message_to_observer", "set_harness_status", ] as const; /** Wire slash commands + harness tools onto an ExtensionAPI. */ export function registerRealtimeVoice( pi: ExtensionAPI, deps: { getToken: (ctx: { modelRegistry: { getApiKeyForProvider(provider: string): Promise; }; }) => Promise; readVoice: () => Promise; }, ): void { let server: RealtimeVoiceServer | null = null; let observerToolsActive = false; let statusCtx: { ui?: { setStatus?(k: string, v: string | undefined): void; notify?(m: string, l?: string): void; }; hasUI?: boolean; } | null = null; const setFooter = (label: string | undefined) => { if (statusCtx?.hasUI && statusCtx.ui?.setStatus) { statusCtx.ui.setStatus("spacexai-realtime", label); } }; /** Register (or re-register) observer tools and add them to the active tool set. */ function enableObserverTools(): void { pi.registerTool({ name: "send_message_to_observer", label: "Send Message to Voice Observer", description: "Send information to the realtime speech-to-speech observer session. The voice agent will hear/see this update and can speak it to the user. The observer cannot read normal terminal output.", promptSnippet: "Send a message to the realtime voice observer", promptGuidelines: [ "When you complete work the user asked about via voice, or when answering an observer question, call send_message_to_observer with a concise status or answer.", "Do not call this tool if the realtime voice server is not running.", ], parameters: sendToObserverParams, async execute(_id, params: SendToObserverParams, _signal, _update, _ctx) { if (!server) { throw new Error( "Realtime voice is not running. Start it with /realtime-voice", ); } const message = String(params.message || "").trim(); if (!message) throw new Error("message is required"); const delivered = server.sendToObserver(message); return { content: [ { type: "text" as const, text: delivered ? `Delivered to observer (${server.clientCount()} SSE client(s)).` : `Queued/broadcast to observer but no browser SSE client is connected yet. Open ${server.url}`, }, ], details: { delivered, clients: server.clientCount(), message }, }; }, }); pi.registerTool({ name: "set_harness_status", label: "Set Coding Harness Status", description: 'Update the live coding-harness status text shown in the realtime voice observer UI (browser HUD over SSE). Use short phrases for in-progress work, and prefer a lasting completion/failure line when done (e.g. "Done: fixed flaky test"). Do not clear the status unless the user asks.', promptSnippet: "Update live harness status on the voice observer UI", promptGuidelines: [ "Keep set_harness_status up to date as work starts and progresses. When work finishes, set a completion or failure status and leave it — do not clear the status line.", "Bias toward completion messages on the status line so the observer UI still shows the latest outcome after you stop working.", "Status text is visual-only for the observer UI — it is not spoken. Use send_message_to_observer for spoken updates (also prefer completion answers there).", ], parameters: setHarnessStatusParams, async execute( _id, params: SetHarnessStatusParams, _signal, _update, _ctx, ) { if (!server) { throw new Error( "Realtime voice is not running. Start it with /realtime-voice", ); } const status = String(params.status ?? ""); const delivered = server.setHarnessStatus(status); const current = server.getHarnessStatus(); const suffix = delivered ? "" : " (no SSE client yet)"; const text = current ? `Harness status set${suffix}: ${current}` : `Harness status cleared${suffix}.`; return { content: [{ type: "text" as const, text }], details: { delivered, clients: server.clientCount(), status: current, }, }; }, }); // First registration auto-activates; re-start after stop must re-add explicitly. const active = new Set(pi.getActiveTools()); for (const name of OBSERVER_TOOL_NAMES) active.add(name); pi.setActiveTools([...active]); observerToolsActive = true; } /** Drop observer tools from the active set so the LLM can no longer call them. */ function disableObserverTools(): void { if (!observerToolsActive) return; const drop = new Set(OBSERVER_TOOL_NAMES); pi.setActiveTools(pi.getActiveTools().filter((n) => !drop.has(n))); observerToolsActive = false; } function teardownServer(): void { // Clear HUD status before dropping the server reference. try { server?.setHarnessStatus(""); } catch { /* ignore */ } server = null; disableObserverTools(); setFooter(undefined); } pi.registerCommand("realtime-voice", { description: "Start Grok speech-to-speech observer (browser) bridged to this coding session", handler: async (args, ctx) => { statusCtx = ctx; if (server) { ctx.ui.notify( `Realtime voice already running at ${server.url}`, "warning", ); openRealtimeVoiceBrowser(server.url); return; } const parts = args.trim().split(/\s+/).filter(Boolean); let port = DEFAULT_PORT; for (const p of parts) { const n = Number(p); if (Number.isFinite(n) && n > 0 && n < 65536) port = Math.floor(n); } try { const voice = (await deps.readVoice()) || DEFAULT_VOICE; server = await startRealtimeVoiceServer({ port, voice, getToken: () => deps.getToken(ctx), onSelfStop: () => { // Browser closed and no SSE clients returned — same as /realtime-voice-stop teardownServer(); if (ctx.hasUI) { ctx.ui.notify("Realtime voice stopped (browser closed)", "info"); } }, onHarnessMessage: async (message) => { const payload = { source: "observer" as const, message }; // Custom message participates in LLM context; trigger a turn when idle. pi.sendMessage( { customType: "spacexai-observer", content: JSON.stringify(payload), display: true, details: payload, }, { triggerTurn: true, deliverAs: "followUp" }, ); if (ctx.hasUI) { ctx.ui.notify( `Observer → harness: ${message.slice(0, 120)}`, "info", ); } }, }); enableObserverTools(); setFooter(`voice:${server.port}`); ctx.ui.notify( `Realtime voice on ${server.url} (voice=${voice}). Opened browser.`, "info", ); openRealtimeVoiceBrowser(server.url); } catch (e) { teardownServer(); ctx.ui.notify( `Failed to start realtime voice: ${e instanceof Error ? e.message : String(e)}`, "error", ); } }, }); pi.registerCommand("realtime-voice-stop", { description: "Stop the realtime voice observer server", handler: async (_args, ctx) => { if (!server) { ctx.ui.notify("Realtime voice is not running", "warning"); return; } await server.stop(); teardownServer(); ctx.ui.notify("Realtime voice stopped", "info"); }, }); // While the observer is live, remind the coding agent how to communicate with it. pi.on("before_agent_start", async (event) => { if (!server) return; return { systemPrompt: `${event.systemPrompt}\n\n${CODING_AGENT_OBSERVER_PROMPT}`, }; }); pi.on("session_shutdown", async () => { if (server) { await server.stop(); } teardownServer(); }); }