From 0793fa0ac4017245ca65882eb0286b2989ecb5d7 Mon Sep 17 00:00:00 2001 From: snowspeeder Date: Tue, 4 Aug 2026 19:09:33 -0400 Subject: [PATCH] feat: add authenticated voice pipeline API docs --- docs/voice-api.md | 91 ++ package-lock.json | 88 +- package.json | 2 + src/buildContents.test.ts | 4 + src/cli.ts | 48 + src/server/app.ts | 10 + src/server/azureSpeechRoutes.test.ts | 55 +- src/server/azureSpeechRoutes.ts | 9 + .../piSessionService.messages.test.ts | 15 + src/server/sessions/piSessionService.ts | 4 + src/server/voiceApi.test.ts | 478 ++++++++ src/server/voiceApi.ts | 1038 +++++++++++++++++ src/server/voiceApiOpenApi.ts | 39 + src/shared/apiTypes.ts | 2 + 14 files changed, 1874 insertions(+), 9 deletions(-) create mode 100644 docs/voice-api.md create mode 100644 src/server/voiceApi.test.ts create mode 100644 src/server/voiceApi.ts create mode 100644 src/server/voiceApiOpenApi.ts diff --git a/docs/voice-api.md b/docs/voice-api.md new file mode 100644 index 0000000..87f9ddc --- /dev/null +++ b/docs/voice-api.md @@ -0,0 +1,91 @@ +# Voice Conversation API (v1) + +`/api/v1/voice` is a server-side device API for native iOS and Echo-style clients. It is separate from the browser voice feature. Azure credentials remain on the PI WEB host. + +## Security and provisioning + +Create a device token on the PI WEB host: + +```sh +pi-web voice-token create \ + --project project-id --workspace workspace-id \ + --model openai-codex/gpt-5.6-terra --thinking high +``` + +Each restriction flag may be repeated. Omitting a category grants `*` for that category. A restricted model or thinking scope requires the client to explicitly select an allowed value; it cannot inherit a server default. Project and workspace scopes are both enforced. Token records contain a SHA-256 token hash only and are persisted at mode `0600` in `~/.config/pi-web/voice-api.json`; the plaintext (`pwv1_...`) is printed only once. + +```sh +pi-web voice-token list +pi-web voice-token revoke +``` + +Use `Authorization: Bearer pwv1_...` on every HTTP and WebSocket connection. The API requires HTTPS, except loopback for local development/testing. A conversation is owned by the creating token: other valid device tokens receive `404` when they try to inspect, delete, or attach it. A client may select only a registered workspace id, never a filesystem path. + +## HTTP + +`GET /api/v1/voice/targets` returns only the caller's authorized registered workspaces plus effective scopes. + +`POST /api/v1/voice/conversations` + +```json +{ + "workspaceId": "registered-workspace-id", + "model": "openai-codex/gpt-5.6-terra", + "thinking": "high", + "context": "You are helping with the home-automation project." +} +``` + +The response is `201` and includes the opaque conversation id, session id, selected workspace, and `input-ready` status. `GET` and `DELETE /api/v1/voice/conversations/:id` inspect and close the API conversation handle. + +## WebSocket wire protocol + +Connect to `wss://host/api/v1/voice/stream` with the Bearer header. The server first sends: + +```json +{ + "type":"hello", + "protocol":"pi-web.voice.v1", + "pcm":{"input":"s16le/16000/mono","output":"s16le/24000/mono"}, + "binary":{"version":1,"headerBytes":8,"kind":"audio"} +} +``` + +Attach an owned conversation and wait for `input.ready` before sending a turn: + +```json +{"type":"attach","conversationId":"..."} +``` + +### Text fallback + +Send exactly one JSON input while `input-ready`: + +```json +{"type":"input.text","text":"Turn on the kitchen lights."} +``` + +### Audio input + +For speech input, send binary WebSocket frames containing raw signed little-endian 16-bit, 16 kHz, mono PCM (no WAV header). Frames must be nonempty, <=64 KiB, aligned to two bytes, and an utterance is limited to two minutes. Finish with: + +```json +{"type":"input.end"} +``` + +Azure push-stream recognition emits `transcript.partial` and `transcript.final` JSON events. The final transcript is submitted to Pi. This Voice API never returns an Azure token or key to a device. + +### Turn output and lifecycle + +For either input form, the server emits `agent.working`, then `agent.accepted` after Pi accepted the prompt. It subscribes to the session event stream before submission and waits for Pi's native `agent.settled` event (not merely `agent.end`), with a two-minute turn timeout. `assistant.delta` is streamed while Pi answers; `assistant.final` contains only assistant text, never STT transcript. + +Azure synthesis output is signed little-endian 16-bit, 24 kHz, mono PCM. Every binary server frame has this eight-byte header followed by PCM: + +| Bytes | Meaning | +|---|---| +| 0 | protocol version: `1` | +| 1 | kind: `1` (audio) | +| 2–3 | reserved: `0` (big endian) | +| 4–7 | monotonically increasing frame sequence (uint32 big endian) | + +The server then emits `{"type":"audio.end"}` and a fresh `input.ready`. Send `{"type":"close"}` or close the WebSocket to end; disconnecting closes STT/event resources and clears the handle's working state. diff --git a/package-lock.json b/package-lock.json index 47386a1..ccdd20e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -24,6 +24,8 @@ "@codemirror/view": "^6.43.6", "@fastify/compress": "^9.0.0", "@fastify/static": "^9.3.0", + "@fastify/swagger": "^9.8.1", + "@fastify/swagger-ui": "^5.2.6", "@fastify/websocket": "^11.3.0", "@xterm/addon-fit": "^0.11.0", "@xterm/xterm": "^6.0.0", @@ -3907,6 +3909,68 @@ "glob": "^13.0.0" } }, + "node_modules/@fastify/swagger": { + "version": "9.8.1", + "resolved": "https://registry.npmjs.org/@fastify/swagger/-/swagger-9.8.1.tgz", + "integrity": "sha512-VpHMnqZTY8iBZYJE8WWkbKPrXIYWy2rDfIf5qLr6DzZSpQYZ+KxQVcJFiq/AMlvNwI4gCBd66++iUlxXXGT0IQ==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/fastify" + }, + { + "type": "opencollective", + "url": "https://opencollective.com/fastify" + } + ], + "license": "MIT", + "dependencies": { + "fastify-plugin": "^6.0.0", + "json-schema-resolver": "^3.0.0", + "openapi-types": "^12.1.3", + "rfdc": "^1.3.1", + "yaml": "^2.4.2" + } + }, + "node_modules/@fastify/swagger-ui": { + "version": "5.2.6", + "resolved": "https://registry.npmjs.org/@fastify/swagger-ui/-/swagger-ui-5.2.6.tgz", + "integrity": "sha512-OMnms0O5s9wb6wis/K5nlrAMLsgUbr1GA8uphM41IasWe3AFdgxz6r/3bA9HTxlDNUYc2FGGKeqMp3ntxmSiNA==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/fastify" + }, + { + "type": "opencollective", + "url": "https://opencollective.com/fastify" + } + ], + "license": "MIT", + "dependencies": { + "@fastify/static": "^9.1.2", + "fastify-plugin": "^5.0.0", + "openapi-types": "^12.1.3", + "rfdc": "^1.3.1", + "yaml": "^2.4.1" + } + }, + "node_modules/@fastify/swagger-ui/node_modules/fastify-plugin": { + "version": "5.1.0", + "resolved": "https://registry.npmjs.org/fastify-plugin/-/fastify-plugin-5.1.0.tgz", + "integrity": "sha512-FAIDA8eovSt5qcDgcBvDuX/v0Cjz0ohGhENZ/wpc3y+oZCY2afZ9Baqql3g/lC+OHRnciQol4ww7tuthOb9idw==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/fastify" + }, + { + "type": "opencollective", + "url": "https://opencollective.com/fastify" + } + ], + "license": "MIT" + }, "node_modules/@fastify/websocket": { "version": "11.3.0", "resolved": "https://registry.npmjs.org/@fastify/websocket/-/websocket-11.3.0.tgz", @@ -7745,6 +7809,23 @@ "dequal": "^2.0.3" } }, + "node_modules/json-schema-resolver": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/json-schema-resolver/-/json-schema-resolver-3.0.0.tgz", + "integrity": "sha512-HqMnbz0tz2DaEJ3ntsqtx3ezzZyDE7G56A/pPY/NGmrPu76UzsWquOpHFRAf5beTNXoH2LU5cQePVvRli1nchA==", + "license": "MIT", + "dependencies": { + "debug": "^4.1.1", + "fast-uri": "^3.0.5", + "rfdc": "^1.1.4" + }, + "engines": { + "node": ">=20" + }, + "funding": { + "url": "https://github.com/Eomm/json-schema-resolver?sponsor=1" + } + }, "node_modules/json-schema-to-ts": { "version": "3.1.1", "resolved": "https://registry.npmjs.org/json-schema-to-ts/-/json-schema-to-ts-3.1.1.tgz", @@ -8539,6 +8620,12 @@ } } }, + "node_modules/openapi-types": { + "version": "12.1.3", + "resolved": "https://registry.npmjs.org/openapi-types/-/openapi-types-12.1.3.tgz", + "integrity": "sha512-N4YtSYJqghVu4iek2ZUvcN/0aqH1kRDuNqzcycDxhOUpg7GdvLa2F3DgS6yBNhInhv2r/6I0Flkn7CqL8+nIcw==", + "license": "MIT" + }, "node_modules/optionator": { "version": "0.9.4", "resolved": "https://registry.npmjs.org/optionator/-/optionator-0.9.4.tgz", @@ -10139,7 +10226,6 @@ "version": "2.9.0", "resolved": "https://registry.npmjs.org/yaml/-/yaml-2.9.0.tgz", "integrity": "sha512-2AvhNX3mb8zd6Zy7INTtSpl1F15HW6Wnqj0srWlkKLcpYl/gMIMJiyuGq2KeI2YFxUPjdlB+3Lc10seMLtL4cA==", - "dev": true, "license": "ISC", "bin": { "yaml": "bin.mjs" diff --git a/package.json b/package.json index 604a49d..2129e15 100644 --- a/package.json +++ b/package.json @@ -70,6 +70,8 @@ "@codemirror/view": "^6.43.6", "@fastify/compress": "^9.0.0", "@fastify/static": "^9.3.0", + "@fastify/swagger": "^9.8.1", + "@fastify/swagger-ui": "^5.2.6", "@fastify/websocket": "^11.3.0", "@xterm/addon-fit": "^0.11.0", "@xterm/xterm": "^6.0.0", diff --git a/src/buildContents.test.ts b/src/buildContents.test.ts index b59cd6e..093ffc6 100644 --- a/src/buildContents.test.ts +++ b/src/buildContents.test.ts @@ -26,8 +26,12 @@ describe("production build contents", () => { try { const fixtureDist = join(fixtureRoot, "dist", "server"); await mkdir(fixtureDist, { recursive: true }); + // npm pack runs the package's prepare lifecycle even with --ignore-scripts + // on the npm version used in CI, so include its harmless fixture script. + await mkdir(join(fixtureRoot, "scripts"), { recursive: true }); await Promise.all([ copyFile(join(repoRoot, "package.json"), join(fixtureRoot, "package.json")), + copyFile(join(repoRoot, "scripts", "install-git-hooks.mjs"), join(fixtureRoot, "scripts", "install-git-hooks.mjs")), writeFile(join(fixtureDist, "app.js"), "export {};\n", "utf8"), writeFile(join(fixtureDist, "app.testSupport.js"), "export {};\n", "utf8"), writeFile(join(fixtureDist, "app.testSupport.js.map"), "{}\n", "utf8"), diff --git a/src/cli.ts b/src/cli.ts index 16cf6af..155d13e 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -41,6 +41,7 @@ import { nativeServicePrerequisiteShellCheck, } from "./nativeServices/serviceProbe.js"; import { renderLaunchdPlist, renderSystemdUnit } from "./nativeServices/serviceRendering.js"; +import { createVoiceToken, listVoiceTokens, revokeVoiceToken } from "./server/voiceApi.js"; const PI_WEB_PACKAGE_NAME = "@jmfederico/pi-web"; @@ -1060,6 +1061,9 @@ Usage: pi-web start|stop|restart|status|logs pi-web doctor pi-web version + pi-web voice-token create [--model provider/model] [--thinking level] [--project project-id] [--workspace workspace-id] + pi-web voice-token list + pi-web voice-token revoke Recommended install: npm install -g @jmfederico/pi-web --allow-scripts=node-pty @@ -1078,11 +1082,55 @@ async function main(): Promise { else if (command === "logs") logs(); else if (command === "doctor") await doctor(); else if (command === "version") await printPiWebVersionReport(); + else if (command === "voice-token") voiceToken(args); else if (command === "--version" || command === "-v") console.log(packageVersion()); else if (command === "help" || command === "--help" || command === "-h") help(); else throw new Error(`Unknown command: ${command}`); } +function voiceToken(args: string[]): void { + const [action, ...rest] = args; + if (action === "list") { console.log(JSON.stringify(listVoiceTokens(), null, 2)); return; } + if (action === "revoke") { + const id = rest[0]; + if (id === undefined || !revokeVoiceToken(id)) throw new Error("Voice token was not found or is already revoked"); + console.log(`Revoked voice token ${id}`); + return; + } + if (action === "create") { + const models: string[] = []; + const thinking: string[] = []; + const projects: string[] = []; + const workspaces: string[] = []; + for (let index = 0; index < rest.length; index += 1) { + const flag = rest[index]; + const value = rest[index + 1]; + if ( + (flag !== "--model" && + flag !== "--thinking" && + flag !== "--project" && + flag !== "--workspace") || + value === undefined + ) + throw new Error("Usage: pi-web voice-token create [--model provider/model] [--thinking level] [--project project-id] [--workspace workspace-id]"); + if (flag === "--model") models.push(value); + else if (flag === "--thinking") thinking.push(value); + else if (flag === "--project") projects.push(value); + else workspaces.push(value); + index += 1; + } + const created = createVoiceToken({ + ...(models.length === 0 ? {} : { models }), + ...(thinking.length === 0 ? {} : { thinking }), + ...(projects.length === 0 ? {} : { projects }), + ...(workspaces.length === 0 ? {} : { workspaces }), + }); + console.log(`Voice token ${created.id} (shown once): ${created.token}`); + return; + } + throw new Error("Usage: pi-web voice-token create|list|revoke"); +} + export function isCliEntrypoint(entrypoint: string | undefined = process.argv[1], modulePath: string = fileURLToPath(import.meta.url)): boolean { if (entrypoint === undefined) return false; if (entrypoint === modulePath) return true; diff --git a/src/server/app.ts b/src/server/app.ts index 3f91f92..a9f1420 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -6,6 +6,8 @@ import type { ServerOptions as HttpsServerOptions } from "node:https"; import fastifyCompress from "@fastify/compress"; import fastifyStatic from "@fastify/static"; import fastifyWebsocket from "@fastify/websocket"; +import fastifySwagger from "@fastify/swagger"; +import fastifySwaggerUi from "@fastify/swagger-ui"; import { ProjectStore } from "./storage/projectStore.js"; import { ProjectService } from "./projects/projectService.js"; import { WorkspaceService } from "./workspaces/workspaceService.js"; @@ -22,6 +24,8 @@ import { registerTerminalProxyRoutes } from "./terminalProxyRoutes.js"; import { registerWorkspaceDeletionRoutes } from "./workspaces/workspaceDeletionRoutes.js"; import { createFilePiWebConfigService, registerConfigRoutes, registerLocalMachineConfigRoutes, type PiWebConfigService } from "./configRoutes.js"; import { registerAzureSpeechRoutes } from "./azureSpeechRoutes.js"; +import { voiceApiOpenApi } from "./voiceApiOpenApi.js"; +import { registerVoiceApiRoutes } from "./voiceApi.js"; import { PiWebPluginService } from "./piWebPluginService.js"; import { createActiveProfilePiPackageService, type PiPackageService } from "./piPackageService.js"; import { registerPiPackageRoutes } from "./piPackageRoutes.js"; @@ -168,6 +172,11 @@ export async function buildApp(deps: AppDependencies = {}) { threshold: 1024, }); await app.register(fastifyWebsocket); + await app.register(fastifySwagger, { mode: "static", specification: { document: voiceApiOpenApi } }); + await app.register(fastifySwaggerUi, { + routePrefix: "/api/v1/voice/docs", + uiConfig: { docExpansion: "list", persistAuthorization: true }, + }); const projects = deps.projects ?? new ProjectService(new ProjectStore()); const workspaces = deps.workspaces ?? new WorkspaceService(); @@ -223,6 +232,7 @@ export async function buildApp(deps: AppDependencies = {}) { registerLocalMachineConfigRoutes(app, invalidatingConfigService); registerAzureSpeechRoutes(app); registerAzureSpeechRoutes(app, undefined, "/api/machines/local"); + registerVoiceApiRoutes(app, { projects, workspaces, daemon: sessionDaemon }); registerMachineRoutes(app, machines); registerMachinePluginProxyRoutes(app, machines); diff --git a/src/server/azureSpeechRoutes.test.ts b/src/server/azureSpeechRoutes.test.ts index 1a43044..12984b9 100644 --- a/src/server/azureSpeechRoutes.test.ts +++ b/src/server/azureSpeechRoutes.test.ts @@ -1,12 +1,29 @@ import { describe, expect, it, vi } from "vitest"; import Fastify from "fastify"; -import { registerAzureSpeechRoutes, type AzureSpeechService } from "./azureSpeechRoutes.js"; +import { + registerAzureSpeechRoutes, + type AzureSpeechService, +} from "./azureSpeechRoutes.js"; function service(): AzureSpeechService { return { - settings: vi.fn(() => ({ region: "eastus", voice: "en-US-Andrew:DragonHDLatestNeural", hasApiKey: true })), - update: vi.fn(() => ({ region: "eastus", voice: "en-US-Andrew:DragonHDLatestNeural", hasApiKey: true })), - token: vi.fn(async () => ({ token: "short-lived-token", region: "eastus", voice: "en-US-Andrew:DragonHDLatestNeural" })), + settings: vi.fn(() => ({ + region: "eastus", + voice: "en-US-Andrew:DragonHDLatestNeural", + hasApiKey: true, + })), + update: vi.fn(() => ({ + region: "eastus", + voice: "en-US-Andrew:DragonHDLatestNeural", + hasApiKey: true, + })), + token: vi.fn(() => + Promise.resolve({ + token: "short-lived-token", + region: "eastus", + voice: "en-US-Andrew:DragonHDLatestNeural", + }) + ), }; } @@ -18,12 +35,29 @@ describe("Azure Speech routes", () => { const get = await app.inject({ method: "GET", url: "/api/azure-speech" }); expect(get.statusCode).toBe(200); - expect(get.json()).toEqual({ region: "eastus", voice: "en-US-Andrew:DragonHDLatestNeural", hasApiKey: true }); + expect(get.json()).toEqual({ + region: "eastus", + voice: "en-US-Andrew:DragonHDLatestNeural", + hasApiKey: true, + }); expect(get.body).not.toContain("apiKey"); - const put = await app.inject({ method: "PUT", url: "/api/azure-speech", payload: { region: "eastus", voice: "en-US-Andrew:DragonHDLatestNeural", apiKey: "a".repeat(32) } }); + const put = await app.inject({ + method: "PUT", + url: "/api/azure-speech", + payload: { + region: "eastus", + voice: "en-US-Andrew:DragonHDLatestNeural", + apiKey: "a".repeat(32), + }, + }); expect(put.statusCode).toBe(200); - expect(azure.update).toHaveBeenCalledWith({ region: "eastus", voice: "en-US-Andrew:DragonHDLatestNeural", apiKey: "a".repeat(32) }); + // eslint-disable-next-line @typescript-eslint/unbound-method + expect(azure.update).toHaveBeenCalledWith({ + region: "eastus", + voice: "en-US-Andrew:DragonHDLatestNeural", + apiKey: "a".repeat(32), + }); expect(put.body).not.toContain("apiKey"); await app.close(); }); @@ -32,8 +66,13 @@ describe("Azure Speech routes", () => { const app = Fastify(); const azure = service(); registerAzureSpeechRoutes(app, azure); - const response = await app.inject({ method: "PUT", url: "/api/azure-speech", payload: { region: "eastus!", voice: "voice" } }); + const response = await app.inject({ + method: "PUT", + url: "/api/azure-speech", + payload: { region: "eastus!", voice: "voice" }, + }); expect(response.statusCode).toBe(400); + // eslint-disable-next-line @typescript-eslint/unbound-method expect(azure.update).not.toHaveBeenCalled(); await app.close(); }); diff --git a/src/server/azureSpeechRoutes.ts b/src/server/azureSpeechRoutes.ts index bc53c85..f87fd53 100644 --- a/src/server/azureSpeechRoutes.ts +++ b/src/server/azureSpeechRoutes.ts @@ -10,6 +10,15 @@ interface AzureSpeechSecret { apiKey?: string; } +/** Server-only Azure credentials. Never expose this shape through an HTTP route. */ +export interface AzureSpeechServerSettings { region: string; voice: string; apiKey: string } + +export function readAzureSpeechServerSettings(path = join(dirname(piWebConfigPath()), "azure-speech.json")): AzureSpeechServerSettings { + const secret = readSecret(path); + if (secret.apiKey === undefined || secret.region === "") throw new AzureSpeechConfigurationError("Azure Speech is not configured. Add a Speech key and region in Settings → Azure Speech."); + return { region: secret.region, voice: secret.voice, apiKey: secret.apiKey }; +} + export interface AzureSpeechService { settings(): AzureSpeechSettings; update(input: AzureSpeechSettingsUpdate): AzureSpeechSettings; diff --git a/src/server/sessions/piSessionService.messages.test.ts b/src/server/sessions/piSessionService.messages.test.ts index be67a33..277238a 100644 --- a/src/server/sessions/piSessionService.messages.test.ts +++ b/src/server/sessions/piSessionService.messages.test.ts @@ -65,6 +65,21 @@ describe("PiSessionService", () => { await service.dispose(); }); + it("forwards Pi's native agent_settled event without inferring it from agent_end", async () => { + const { fake, service, events } = messagesService([]); + await service.status(sessionRef("session-1")); + + fake.emit({ type: "agent_end" }); + await new Promise((resolve) => setImmediate(resolve)); + expect(events.sessionEvents.map(({ event }) => event).filter((event) => event.type === "agent.settled")).toEqual([]); + + fake.emit({ type: "agent_settled" }); + expect(events.sessionEvents.map(({ event }) => event).filter((event) => event.type === "agent.settled")).toEqual([ + { type: "agent.settled" }, + ]); + await service.dispose(); + }); + it("annotates the join-time stream snapshot partial with the current thinking level", async () => { const streamingMessage = { role: "assistant", diff --git a/src/server/sessions/piSessionService.ts b/src/server/sessions/piSessionService.ts index cc26e1f..355b78d 100644 --- a/src/server/sessions/piSessionService.ts +++ b/src/server/sessions/piSessionService.ts @@ -3224,6 +3224,9 @@ export class PiSessionService implements SessionRouteService { this.publishActivityForEvent(session, event); const eventType = getString(event, "type"); if (eventType === "agent_end") this.abortRunScopedExtensionDialogs(session.sessionId); + // Pi itself emits agent_settled only after automatic retries, compaction, + // and queued continuations are exhausted. Preserve that native lifecycle + // event; never infer it from agent_end or mutable runtime flags. if (eventType === "compaction_end") this.scheduleCompactionQueueDrain(session.sessionId); if (eventType === "agent_start" || eventType === "agent_end") this.scheduleCompactionQueueDrain(session.sessionId); this.publishStatus(session); @@ -4209,6 +4212,7 @@ function toClientEvent(event: unknown, thinkingLevel?: string): SessionUiEvent { } if (eventType === "agent_start") return { type: "agent.start" }; if (eventType === "agent_end") return { type: "agent.end" }; + if (eventType === "agent_settled") return { type: "agent.settled" }; if (eventType === "message_end") { const message = getProperty(event, "message"); if (message === undefined) return { type: "message.end" }; diff --git a/src/server/voiceApi.test.ts b/src/server/voiceApi.test.ts new file mode 100644 index 0000000..e844a38 --- /dev/null +++ b/src/server/voiceApi.test.ts @@ -0,0 +1,478 @@ +import { EventEmitter } from "node:events"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import fastifyWebsocket from "@fastify/websocket"; +import Fastify from "fastify"; +import WebSocket, { type RawData } from "ws"; +import { afterEach, describe, expect, it } from "vitest"; +import { + createVoiceToken, + encodeAudioFrame, + registerVoiceApiRoutes, + revokeVoiceToken, +} from "./voiceApi.js"; + +const temporary: string[] = []; +afterEach(() => { + for (const path of temporary.splice(0)) + rmSync(path, { recursive: true, force: true }); +}); + +function setup() { + const directory = mkdtempSync(join(tmpdir(), "pi-web-voice-api-")); + temporary.push(directory); + const configPath = join(directory, "voice-api.json"); + const daemon = { + request: (method: string, path: string, body?: unknown) => { + if (method === "POST" && path === "/sessions") + return Promise.resolve(response(200, { id: "session-1" })); + if (method === "POST" && path.endsWith("/model")) + return Promise.resolve(response(200, {})); + if (method === "POST" && path.endsWith("/thinking-level")) + return Promise.resolve(response(200, {})); + if (method === "POST" && path.endsWith("/prompt")) + return Promise.resolve(response(200, { accepted: true, body })); + return Promise.resolve(response(404, { error: "not found" })); + }, + connectWebSocket: () => { + throw new Error("not used"); + }, + }; + const app = Fastify(); + registerVoiceApiRoutes(app, { + configPath, + daemon, + projects: { + list: () => + Promise.resolve([ + { + id: "project-1", + name: "Home", + path: "/home/hope/home", + createdAt: new Date(0).toISOString(), + }, + ]), + }, + workspaces: { + list: () => + Promise.resolve([ + { + id: "workspace-1", + projectId: "project-1", + path: "/home/hope/home", + label: "Home", + isMain: true, + isGitRepo: false, + isGitWorktree: false, + }, + ]), + }, + }); + return { app, configPath }; +} + +function response(statusCode: number, body: unknown) { + return { statusCode, headers: {}, body: JSON.stringify(body) }; +} + +function websocketConnectingState(): number { + return WebSocket.CONNECTING; +} + +function rawDataToBuffer(data: RawData): Buffer { + if (Buffer.isBuffer(data)) return data; + if (data instanceof ArrayBuffer) return Buffer.from(data); + return Buffer.concat(data); +} + +function responseId(body: string): string { + const parsed: unknown = JSON.parse(body); + if ( + typeof parsed !== "object" || + parsed === null || + !("id" in parsed) || + typeof parsed.id !== "string" + ) { + throw new Error("conversation response is missing an id"); + } + return parsed.id; +} + +describe("voice API", () => { + it("frames synthesized PCM with a versioned kind and monotonic sequence header", () => { + const frame = encodeAudioFrame(42, Buffer.from([1, 2, 3, 4])); + expect(frame.subarray(0, 8)).toEqual( + Buffer.from([1, 1, 0, 0, 0, 0, 0, 42]) + ); + expect(frame.subarray(8)).toEqual(Buffer.from([1, 2, 3, 4])); + }); + + it("accepts PCM, emits recognition text, waits for agent.settled, then streams framed synthesis", async () => { + const directory = mkdtempSync(join(tmpdir(), "pi-web-voice-api-")); + temporary.push(directory); + const events = Object.assign(new EventEmitter(), { + close: () => undefined, + readyState: websocketConnectingState(), + }); + const daemon = { + request: (method: string, path: string) => { + if (method === "POST" && path === "/sessions") + return Promise.resolve(response(200, { id: "session-1" })); + if (method === "POST" && path.endsWith("/prompt")) { + expect(events.listenerCount("message")).toBeGreaterThan(0); + queueMicrotask(() => { + events.emit( + "message", + Buffer.from( + JSON.stringify({ + type: "assistant.delta", + text: "Settled reply", + }) + ) + ); + events.emit( + "message", + Buffer.from(JSON.stringify({ type: "agent.settled" })) + ); + }); + return Promise.resolve(response(200, { accepted: true })); + } + return Promise.resolve(response(200, {})); + }, + // EventEmitter supplies exactly the message/error/close surface the bridge consumes. + connectWebSocket: () => { + queueMicrotask(() => { + events.readyState = WebSocket.OPEN; + events.emit("open"); + }); + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + return events as unknown as WebSocket; + }, + }; + const speech = { + recognize: (handlers: { + partial(text: string): void; + final(text: string): void; + error(error: Error): void; + }) => ({ + write: () => { + handlers.partial("hello"); + }, + end: () => { + handlers.final("hello"); + return Promise.resolve(); + }, + close: () => undefined, + }), + synthesize: (_text: string, onAudio: (pcm: Buffer) => void) => { + onAudio(Buffer.from([7, 8])); + return Promise.resolve(); + }, + }; + const app = Fastify(); + await app.register(fastifyWebsocket); + const configPath = join(directory, "voice-api.json"); + registerVoiceApiRoutes(app, { + configPath, + daemon, + speech, + projects: { + list: () => + Promise.resolve([ + { id: "p", name: "P", path: "/tmp", createdAt: "" }, + ]), + }, + workspaces: { + list: () => + Promise.resolve([ + { + id: "w", + projectId: "p", + path: "/tmp", + label: "W", + isMain: true, + isGitRepo: false, + isGitWorktree: false, + }, + ]), + }, + }); + const token = createVoiceToken({}, configPath).token; + const created = await app.inject({ + method: "POST", + url: "/api/v1/voice/conversations", + headers: { authorization: `Bearer ${token}` }, + payload: { workspaceId: "w" }, + }); + const address = await app.listen({ port: 0, host: "127.0.0.1" }); + const conversationId = responseId(created.body); + const received: (string | Buffer)[] = []; + await new Promise((resolve, reject) => { + const client = new WebSocket( + address.replace("http", "ws") + "/api/v1/voice/stream", + { headers: { authorization: `Bearer ${token}` } } + ); + client.on("open", () => { + client.send(JSON.stringify({ type: "attach", conversationId })); + client.send(Buffer.from([0, 0])); + client.send(JSON.stringify({ type: "input.end" })); + }); + client.on("message", (data, binary) => { + const message = rawDataToBuffer(data).toString(); + received.push(binary ? rawDataToBuffer(data) : message); + if ( + !binary && + message.includes("input.ready") && + received.some((item) => Buffer.isBuffer(item)) + ) { + client.close(); + resolve(); + } + }); + client.on("error", (error) => { + reject(error); + }); + }); + expect( + received + .filter((item): item is string => typeof item === "string") + .join(" ") + ).toContain("transcript.partial"); + const jsonEvents = received + .filter((item): item is string => typeof item === "string") + .map((item): unknown => JSON.parse(item)); + expect(jsonEvents).toContainEqual({ + type: "assistant.final", + text: "Settled reply", + }); + expect(received.find((item) => Buffer.isBuffer(item))).toEqual( + Buffer.from([1, 1, 0, 0, 0, 0, 0, 0, 7, 8]) + ); + await app.close(); + }); + + it("requires a hashed bearer token and only returns registered workspaces", async () => { + const { app, configPath } = setup(); + const created = createVoiceToken( + { models: ["openai/gpt-5"], thinking: ["high"] }, + configPath + ); + const denied = await app.inject({ + method: "GET", + url: "/api/v1/voice/targets", + }); + expect(denied.statusCode).toBe(401); + const allowed = await app.inject({ + method: "GET", + url: "/api/v1/voice/targets", + headers: { authorization: `Bearer ${created.token}` }, + }); + expect(allowed.json()).toMatchObject({ + workspaces: [{ id: "workspace-1", path: "/home/hope/home" }], + scopes: { models: ["openai/gpt-5"], thinking: ["high"] }, + }); + expect(revokeVoiceToken(created.id, configPath)).toBe(true); + expect( + ( + await app.inject({ + method: "GET", + url: "/api/v1/voice/targets", + headers: { authorization: `Bearer ${created.token}` }, + }) + ).statusCode + ).toBe(401); + await app.close(); + }); + + it("creates a scoped conversation and rejects arbitrary workspace/model choices", async () => { + const { app, configPath } = setup(); + const token = createVoiceToken( + { models: ["openai/gpt-5"], thinking: ["high"] }, + configPath + ).token; + const headers = { authorization: `Bearer ${token}` }; + const missingRestrictedModel = await app.inject({ + method: "POST", + url: "/api/v1/voice/conversations", + headers, + payload: { workspaceId: "workspace-1", thinking: "high" }, + }); + expect(missingRestrictedModel.statusCode).toBe(400); + const badWorkspace = await app.inject({ + method: "POST", + url: "/api/v1/voice/conversations", + headers, + payload: { workspaceId: "/etc", model: "openai/gpt-5", thinking: "high" }, + }); + expect(badWorkspace.statusCode).toBe(400); + const badModel = await app.inject({ + method: "POST", + url: "/api/v1/voice/conversations", + headers, + payload: { + workspaceId: "workspace-1", + model: "other/model", + thinking: "high", + }, + }); + expect(badModel.statusCode).toBe(400); + const created = await app.inject({ + method: "POST", + url: "/api/v1/voice/conversations", + headers, + payload: { + workspaceId: "workspace-1", + model: "openai/gpt-5", + thinking: "high", + context: "Be concise", + }, + }); + expect(created.statusCode).toBe(201); + expect(created.json()).toMatchObject({ + sessionId: "session-1", + workspaceId: "workspace-1", + status: "input-ready", + }); + await app.close(); + }); + + it("filters project/workspace scopes and isolates conversations by token owner", async () => { + const { app, configPath } = setup(); + const owner = createVoiceToken( + { + projects: ["project-1"], + workspaces: ["workspace-1"], + models: ["openai/gpt-5"], + thinking: ["high"], + }, + configPath + ).token; + const other = createVoiceToken({}, configPath).token; + const headers = { authorization: `Bearer ${owner}` }; + const targets = await app.inject({ + method: "GET", + url: "/api/v1/voice/targets", + headers, + }); + expect(targets.json()).toMatchObject({ + scopes: { + projects: ["project-1"], + workspaces: ["workspace-1"], + }, + workspaces: [{ id: "workspace-1" }], + }); + const created = await app.inject({ + method: "POST", + url: "/api/v1/voice/conversations", + headers, + payload: { + workspaceId: "workspace-1", + model: "openai/gpt-5", + thinking: "high", + }, + }); + const id = responseId(created.body); + expect( + ( + await app.inject({ + method: "GET", + url: `/api/v1/voice/conversations/${id}`, + headers: { authorization: `Bearer ${other}` }, + }) + ).statusCode + ).toBe(404); + expect( + ( + await app.inject({ + method: "DELETE", + url: `/api/v1/voice/conversations/${id}`, + headers: { authorization: `Bearer ${other}` }, + }) + ).statusCode + ).toBe(404); + expect( + ( + await app.inject({ + method: "GET", + url: `/api/v1/voice/conversations/${id}`, + headers, + }) + ).statusCode + ).toBe(200); + await app.close(); + }); + + it("rejects WebSocket attachment by a different valid device token", async () => { + const directory = mkdtempSync(join(tmpdir(), "pi-web-voice-api-")); + temporary.push(directory); + const configPath = join(directory, "voice-api.json"); + const app = Fastify(); + await app.register(fastifyWebsocket); + registerVoiceApiRoutes(app, { + configPath, + daemon: { + request: () => Promise.resolve(response(200, { id: "session-1" })), + connectWebSocket: () => { + throw new Error("no turn should start"); + }, + }, + speech: { + recognize: () => ({ + write: () => undefined, + end: () => Promise.resolve(), + close: () => undefined, + }), + synthesize: () => Promise.resolve(), + }, + projects: { + list: () => Promise.resolve([{ id: "p", name: "P", path: "/tmp", createdAt: "" }]), + }, + workspaces: { + list: () => Promise.resolve([{ + id: "w", + projectId: "p", + path: "/tmp", + label: "W", + isMain: true, + isGitRepo: false, + isGitWorktree: false, + }]), + }, + }); + const owner = createVoiceToken({}, configPath).token; + const intruder = createVoiceToken({}, configPath).token; + const created = await app.inject({ + method: "POST", + url: "/api/v1/voice/conversations", + headers: { authorization: `Bearer ${owner}` }, + payload: { workspaceId: "w" }, + }); + const address = await app.listen({ port: 0, host: "127.0.0.1" }); + const result = await new Promise<{ type?: string; error?: string }>((resolve, reject) => { + const client = new WebSocket( + address.replace("http", "ws") + "/api/v1/voice/stream", + { headers: { authorization: `Bearer ${intruder}` } } + ); + client.on("open", () => { + client.send(JSON.stringify({ type: "attach", conversationId: responseId(created.body) })); + }); + client.on("message", (data, binary) => { + if (binary) return; + const payload: unknown = JSON.parse(rawDataToBuffer(data).toString()); + if ( + typeof payload === "object" && + payload !== null && + "error" in payload && + typeof payload.error === "string" + ) { + client.close(); + resolve({ type: "error", error: payload.error }); + } + }); + client.on("error", reject); + }); + expect(result).toEqual({ type: "error", error: "conversation not found" }); + await app.close(); + }); +}); diff --git a/src/server/voiceApi.ts b/src/server/voiceApi.ts new file mode 100644 index 0000000..5142c34 --- /dev/null +++ b/src/server/voiceApi.ts @@ -0,0 +1,1038 @@ +/* eslint-disable @typescript-eslint/consistent-type-assertions, @typescript-eslint/no-base-to-string, @typescript-eslint/prefer-optional-chain, @typescript-eslint/return-await */ +import { createHash, randomBytes, timingSafeEqual } from "node:crypto"; +import { + chmodSync, + existsSync, + mkdirSync, + readFileSync, + writeFileSync, +} from "node:fs"; +import { dirname, join } from "node:path"; +import type { FastifyInstance, FastifyRequest } from "fastify"; +import * as SpeechSDK from "microsoft-cognitiveservices-speech-sdk"; +import type { RawData, WebSocket } from "ws"; +import { piWebConfigPath } from "../config.js"; +import { readAzureSpeechServerSettings } from "./azureSpeechRoutes.js"; +import type { ProjectService } from "./projects/projectService.js"; +import type { SessionProxyDaemon } from "./sessiond/sessionProxyRoutes.js"; +import type { WorkspaceService } from "./workspaces/workspaceService.js"; + +const MAX_CONTEXT = 12_000; +const MAX_TEXT = 16_000; +const PCM_FRAME_BYTES = 64 * 1024; +const MAX_AUDIO_BYTES = 16_000 * 2 * 60; +const MAX_CONVERSATIONS = 256; +const TOKEN_PREFIX = "pwv1_"; +const BINARY_VERSION = 1; +const BINARY_AUDIO = 1; +const OUTPUT_CHUNK = 16 * 1024; +const EVENT_OPEN_TIMEOUT_MS = 10_000; +const TURN_TIMEOUT_MS = 120_000; +interface TokenRecord { + id: string; + hash: string; + models: string[]; + thinking: string[]; + projects: string[]; + workspaces: string[]; + createdAt: string; + revokedAt?: string; +} +interface VoiceConfig { + tokens: TokenRecord[]; +} +export interface VoiceTokenScope { + models?: string[]; + thinking?: string[]; + /** Exact registered project ids this device may use; omit for all projects. */ + projects?: string[]; + /** Exact registered workspace ids this device may use; omit for all workspaces. */ + workspaces?: string[]; +} +export interface VoiceToken { + id: string; + token: string; + models: string[]; + thinking: string[]; + projects: string[]; + workspaces: string[]; +} +export interface VoiceRecognizer { + write(pcm: Buffer): void; + end(): Promise; + close(): void; +} +export interface VoiceSpeechGateway { + recognize(handlers: { + partial(text: string): void; + final(text: string): void; + error(error: Error): void; + }): VoiceRecognizer; + synthesize(text: string, onAudio: (pcm: Buffer) => void): Promise; +} +export interface VoiceApiDependencies { + projects: Pick; + workspaces: Pick; + daemon: SessionProxyDaemon; + configPath?: string; + speech?: VoiceSpeechGateway; +} +interface Conversation { + id: string; + /** The device token that created this handle; never expose it to clients. */ + ownerTokenId: string; + sessionId: string; + cwd: string; + workspaceId: string; + model?: string; + thinking?: string; + context?: string; + status: "input-ready" | "working" | "closed"; + createdAt: number; +} +interface SocketTurn { + recognizer?: VoiceRecognizer; + eventSocket: WebSocket | undefined; + /** VAD/STT result only; never use it as an assistant response. */ + transcript: string; + /** Assistant output collected from Pi's session event stream only. */ + assistant: string; + outputSequence: number; + closed: boolean; +} + +/** Owner-only file store. The plaintext token is deliberately returned once only. */ +export function createVoiceToken( + scope: VoiceTokenScope = {}, + path = voiceConfigPath() +): VoiceToken { + const models = normalizeScope(scope.models); + const thinking = normalizeScope(scope.thinking); + const projects = normalizeScope(scope.projects); + const workspaces = normalizeScope(scope.workspaces); + const token = `${TOKEN_PREFIX}${randomBytes(32).toString("base64url")}`; + const record: TokenRecord = { + id: randomBytes(12).toString("hex"), + hash: hashToken(token), + models, + thinking, + projects, + workspaces, + createdAt: new Date().toISOString(), + }; + const config = readConfig(path); + config.tokens.push(record); + writeConfig(path, config); + return { id: record.id, token, models, thinking, projects, workspaces }; +} +export function listVoiceTokens( + path = voiceConfigPath() +): Omit[] { + return readConfig(path).tokens.map((token) => ({ + id: token.id, + models: token.models, + thinking: token.thinking, + projects: token.projects, + workspaces: token.workspaces, + createdAt: token.createdAt, + ...(token.revokedAt === undefined ? {} : { revokedAt: token.revokedAt }), + })); +} +export function revokeVoiceToken( + id: string, + path = voiceConfigPath() +): boolean { + const config = readConfig(path); + const token = config.tokens.find( + (entry) => entry.id === id && entry.revokedAt === undefined + ); + if (token === undefined) return false; + token.revokedAt = new Date().toISOString(); + writeConfig(path, config); + return true; +} + +export function registerVoiceApiRoutes( + app: FastifyInstance, + deps: VoiceApiDependencies +): void { + const conversations = new Map(); + const configPath = deps.configPath ?? voiceConfigPath(); + let speech = deps.speech; + const gateway = (): VoiceSpeechGateway => { + speech ??= createAzureSpeechGateway(); + return speech; + }; + const requireToken = (request: FastifyRequest): TokenRecord | undefined => { + if (!isSecureVoiceRequest(request)) return undefined; + const value = request.headers.authorization; + if (typeof value !== "string" || !value.startsWith("Bearer ")) + return undefined; + const hash = hashToken(value.slice(7)); + return readConfig(configPath).tokens.find( + (candidate) => + candidate.revokedAt === undefined && equalHash(candidate.hash, hash) + ); + }; + const reject = ( + request: FastifyRequest, + reply: { code(code: number): { send(value: unknown): unknown } } + ): TokenRecord | undefined => { + if (!isSecureVoiceRequest(request)) { + reply.code(426).send({ error: "Voice API requires HTTPS." }); + return undefined; + } + const token = requireToken(request); + if (token === undefined) + reply + .code(401) + .send({ error: "A valid Bearer device token is required." }); + return token; + }; + app.get("/api/v1/voice/targets", async (request, reply) => { + const token = reject(request, reply); + if (token === undefined) return; + const projects = await deps.projects.list(); + const workspaces = ( + await Promise.all( + projects.map(async (project) => deps.workspaces.list(project)) + ) + ).flat().filter((workspace) => workspaceAllowed(token, workspace)); + return { + workspaces: workspaces.map((workspace) => ({ + id: workspace.id, + projectId: workspace.projectId, + label: workspace.label, + path: workspace.path, + })), + scopes: publicScope(token), + }; + }); + app.post<{ Body: unknown }>( + "/api/v1/voice/conversations", + async (request, reply) => { + const token = reject(request, reply); + if (token === undefined) return; + try { + const input = createInput(request.body); + requireScope(token, input.model, input.thinking); + const workspace = await exactWorkspace(deps, input.workspaceId); + if (!workspaceAllowed(token, workspace)) + throw new Error("token is not allowed to use this workspace"); + const created = await daemonJson(deps.daemon, "POST", "/sessions", { + cwd: workspace.path, + }); + const sessionId = requiredString(created, "id"); + if (input.model !== undefined) { + const [provider, modelId] = splitModel(input.model); + await daemonJson( + deps.daemon, + "POST", + `/sessions/${encodeURIComponent(sessionId)}/model`, + { cwd: workspace.path, provider, modelId } + ); + } + if (input.thinking !== undefined) + await daemonJson( + deps.daemon, + "POST", + `/sessions/${encodeURIComponent(sessionId)}/thinking-level`, + { cwd: workspace.path, level: input.thinking } + ); + pruneConversations(conversations); + const conversation: Conversation = { + id: randomBytes(16).toString("hex"), + ownerTokenId: token.id, + sessionId, + cwd: workspace.path, + workspaceId: workspace.id, + status: "input-ready", + createdAt: Date.now(), + ...(input.model === undefined ? {} : { model: input.model }), + ...(input.thinking === undefined ? {} : { thinking: input.thinking }), + ...(input.context === undefined ? {} : { context: input.context }), + }; + conversations.set(conversation.id, conversation); + return reply.code(201).send(publicConversation(conversation)); + } catch (error) { + return reply.code(400).send({ error: message(error) }); + } + } + ); + app.get<{ Params: { id: string } }>( + "/api/v1/voice/conversations/:id", + async (request, reply) => { + const token = reject(request, reply); + if (token === undefined) return; + const conversation = conversations.get(request.params.id); + return conversation === undefined || conversation.ownerTokenId !== token.id + ? reply.code(404).send({ error: "Conversation not found" }) + : publicConversation(conversation); + } + ); + app.delete<{ Params: { id: string } }>( + "/api/v1/voice/conversations/:id", + async (request, reply) => { + const token = reject(request, reply); + if (token === undefined) return; + const conversation = conversations.get(request.params.id); + if (conversation === undefined || conversation.ownerTokenId !== token.id) + return reply.code(404).send({ error: "Conversation not found" }); + conversation.status = "closed"; + conversations.delete(conversation.id); + return reply.code(204).send(); + } + ); + app.get("/api/v1/voice/stream", { websocket: true }, (socket, request) => { + const token = requireToken(request); + if (token === undefined) { + socket.close(1008, "Unauthorized or insecure voice connection"); + return; + } + wireVoiceSocket(socket, conversations, deps, gateway(), token); + }); +} + +function wireVoiceSocket( + socket: WebSocket, + conversations: Map, + deps: VoiceApiDependencies, + speech: VoiceSpeechGateway, + token: TokenRecord +): void { + let conversation: Conversation | undefined; + let inputBytes = 0; + const turn: SocketTurn = { + transcript: "", + assistant: "", + outputSequence: 0, + closed: false, + eventSocket: undefined, + }; + sendJson(socket, { + type: "hello", + protocol: "pi-web.voice.v1", + pcm: { input: "s16le/16000/mono", output: "s16le/24000/mono" }, + binary: { version: BINARY_VERSION, headerBytes: 8, kind: "audio" }, + }); + const cleanup = () => { + if (turn.closed) return; + turn.closed = true; + turn.recognizer?.close(); + delete turn.recognizer; + turn.eventSocket?.close(); + turn.eventSocket = undefined; + if (conversation !== undefined && conversation.status === "working") + conversation.status = "input-ready"; + }; + socket.on("close", cleanup); + socket.on("error", cleanup); + socket.on("message", (data, isBinary) => { + if (isBinary) { + if ( + conversation === undefined || + conversation.status !== "input-ready" || + turn.recognizer === undefined + ) { + protocolError(socket, "audio is only accepted while input-ready"); + return; + } + const pcm = rawDataToBuffer(data); + if ( + pcm.length === 0 || + pcm.length > PCM_FRAME_BYTES || + pcm.length % 2 !== 0 || + (inputBytes += pcm.length) > MAX_AUDIO_BYTES + ) { + protocolError(socket, "invalid or oversized PCM frame"); + return; + } + turn.recognizer.write(pcm); + return; + } + let control: unknown; + try { + control = JSON.parse(data.toString()); + } catch { + protocolError(socket, "invalid JSON control frame"); + return; + } + if (!isRecord(control) || typeof control["type"] !== "string") { + protocolError(socket, "invalid control frame"); + return; + } + if (control["type"] === "attach") { + const found = + typeof control["conversationId"] === "string" + ? conversations.get(control["conversationId"]) + : undefined; + if ( + found === undefined || + found.status === "closed" || + found.ownerTokenId !== token.id + ) { + protocolError(socket, "conversation not found"); + return; + } + if (conversation?.status === "working") { + protocolError(socket, "a turn is already working"); + return; + } + conversation = found; + inputBytes = 0; + turn.transcript = ""; + turn.assistant = ""; + turn.recognizer?.close(); + turn.recognizer = speech.recognize({ + partial: (text) => { + sendJson(socket, { type: "transcript.partial", text }); + }, + final: (text) => { + turn.transcript = `${turn.transcript} ${text}`.trim(); + sendJson(socket, { type: "transcript.final", text }); + }, + error: (error) => { + sendJson(socket, { type: "error", error: error.message }); + }, + }); + sendJson(socket, { + type: "input.ready", + conversation: publicConversation(found), + }); + return; + } + if (control["type"] === "input.text") { + if (conversation === undefined || conversation.status !== "input-ready") { + protocolError(socket, "conversation is not input-ready"); + return; + } + const text = + typeof control["text"] === "string" ? control["text"].trim() : ""; + if (text === "" || text.length > MAX_TEXT) { + protocolError(socket, "text must be 1..16000 characters"); + return; + } + // Typed fallback does not use the active Azure recognizer; release it + // before the Pi turn so disconnects and long answers retain no STT work. + turn.recognizer?.close(); + delete turn.recognizer; + void submitText(socket, turn, conversation, deps, speech, text); + return; + } + if (control["type"] === "input.end") { + if ( + conversation === undefined || + conversation.status !== "input-ready" || + turn.recognizer === undefined + ) { + protocolError(socket, "audio input is not active"); + return; + } + const recognizer = turn.recognizer; + const activeConversation = conversation; + delete turn.recognizer; + void recognizer + .end() + .then(() => { + recognizer.close(); + const text = turn.transcript.trim(); + if (text === "") { + protocolError(socket, "no speech was recognized"); + return; + } + return submitText( + socket, + turn, + activeConversation, + deps, + speech, + text + ); + }) + .catch((error: unknown) => { + recognizer.close(); + sendJson(socket, { type: "error", error: message(error) }); + }); + return; + } + if (control["type"] === "close") { + socket.close(1000); + return; + } + protocolError(socket, `unsupported control type: ${control["type"]}`); + }); +} + +async function submitText( + socket: WebSocket, + turn: SocketTurn, + conversation: Conversation, + deps: VoiceApiDependencies, + speech: VoiceSpeechGateway, + text: string +): Promise { + if (turn.closed) return; + conversation.status = "working"; + turn.assistant = ""; + sendJson(socket, { type: "agent.working" }); + try { + await withTimeout( + runTurn(socket, turn, conversation, deps, speech, text), + TURN_TIMEOUT_MS, + "voice turn timed out waiting for Pi to settle", + () => { + closeEventSocket(turn); + } + ); + } catch (error) { + sendTurnError(socket, turn, error); + } finally { + closeEventSocket(turn); + resetConversationAfterTurn(socket, turn, conversation); + } +} + +async function runTurn( + socket: WebSocket, + turn: SocketTurn, + conversation: Conversation, + deps: VoiceApiDependencies, + speech: VoiceSpeechGateway, + text: string +): Promise { + const prompt = conversation.context === undefined ? text : `${conversation.context}\n\n${text}`; + // The event subscription is fully open before the prompt is submitted, so + // a fast Pi run cannot emit agent_settled before this device starts waiting. + const { settled } = await startSettledWaiter( + deps.daemon, + conversation.sessionId, + turn, + socket + ); + try { + await daemonJson( + deps.daemon, + "POST", + `/sessions/${encodeURIComponent(conversation.sessionId)}/prompt`, + { cwd: conversation.cwd, text: prompt } + ); + } catch (error) { + closeEventSocket(turn); + try { + await settled; + } catch { + // The prompt request error is the actionable failure. + } + throw error; + } + sendJson(socket, { type: "agent.accepted" }); + await settled; + const finalText = turn.assistant.trim(); + sendJson(socket, { type: "assistant.final", text: finalText }); + if (finalText !== "") { + await speech.synthesize(finalText, (pcm) => { + if (!turn.closed) sendAudio(socket, turn, pcm); + }); + } + sendJson(socket, { type: "audio.end" }); +} + +async function startSettledWaiter( + daemon: SessionProxyDaemon, + sessionId: string, + turn: SocketTurn, + socket: WebSocket +): Promise<{ settled: Promise }> { + const events = daemon.connectWebSocket( + `/sessions/${encodeURIComponent(sessionId)}/events` + ); + turn.eventSocket = events; + await waitForWebSocketOpen(events); + const settled = new Promise((resolve, reject) => { + let done = false; + const finish = (error?: Error) => { + if (done) return; + done = true; + if (error === undefined) resolve(); + else reject(error); + }; + events.on("error", (error) => { + finish(error instanceof Error ? error : new Error(String(error))); + }); + events.on("message", (data) => { + try { + const event: unknown = JSON.parse(data.toString()); + if (!isRecord(event)) return; + if ( + event["type"] === "assistant.delta" && + typeof event["text"] === "string" + ) { + turn.assistant += event["text"]; + sendJson(socket, { type: "assistant.delta", text: event["text"] }); + } + if (event["type"] === "message.end") { + const text = assistantText(event["message"]); + if (text !== "") turn.assistant = text; + } + // This is Pi's native session-level event, forwarded by + // PiSessionService/sessiond without deriving it from agent_end. + if (event["type"] === "agent.settled") finish(); + } catch (error) { + finish(error instanceof Error ? error : new Error(String(error))); + } + }); + events.on("close", () => { + finish(new Error("session event bridge closed before the agent settled")); + }); + }); + return { settled }; +} + +function waitForWebSocketOpen(socket: WebSocket): Promise { + if (socket.readyState === 1) return Promise.resolve(); + return withTimeout( + new Promise((resolve, reject) => { + socket.once("open", resolve); + socket.once("error", (error) => { + reject(error instanceof Error ? error : new Error(String(error))); + }); + socket.once("close", () => { + reject(new Error("session event bridge closed before opening")); + }); + }), + EVENT_OPEN_TIMEOUT_MS, + "timed out opening the session event bridge" + ); +} + +function closeEventSocket(turn: SocketTurn): void { + const events = turn.eventSocket; + turn.eventSocket = undefined; + events?.close(); +} + +function sendTurnError(socket: WebSocket, turn: SocketTurn, error: unknown): void { + if (!turn.closed) sendJson(socket, { type: "error", error: message(error) }); +} + +function resetConversationAfterTurn( + socket: WebSocket, + turn: SocketTurn, + conversation: Conversation +): void { + if (conversation.status !== "closed") conversation.status = "input-ready"; + if (!turn.closed) { + sendJson(socket, { + type: "input.ready", + conversation: publicConversation(conversation), + }); + } +} + +function withTimeout( + promise: Promise, + timeoutMs: number, + error: string, + onTimeout?: () => void +): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + onTimeout?.(); + reject(new Error(error)); + }, timeoutMs); + promise.then( + (value) => { + clearTimeout(timer); + resolve(value); + }, + (reason: unknown) => { + clearTimeout(timer); + reject(reason instanceof Error ? reason : new Error(String(reason))); + } + ); + }); +} + +/** Azure server gateway; credentials stay in the owner-only azure-speech.json file. */ +export function createAzureSpeechGateway(): VoiceSpeechGateway { + const settings = readAzureSpeechServerSettings(); + const config = SpeechSDK.SpeechConfig.fromSubscription( + settings.apiKey, + settings.region + ); + config.speechRecognitionLanguage = "en-US"; + config.setProperty( + SpeechSDK.PropertyId.Speech_SegmentationSilenceTimeoutMs, + "1200" + ); + return { + recognize(handlers) { + const input = SpeechSDK.AudioInputStream.createPushStream( + SpeechSDK.AudioStreamFormat.getWaveFormatPCM(16_000, 16, 1) + ); + const recognizer = new SpeechSDK.SpeechRecognizer( + config, + SpeechSDK.AudioConfig.fromStreamInput(input) + ); + recognizer.recognizing = (_sender, event) => { + if (event.result.text.trim() !== "") + handlers.partial(event.result.text); + }; + recognizer.recognized = (_sender, event) => { + if ( + event.result.reason === SpeechSDK.ResultReason.RecognizedSpeech && + event.result.text.trim() !== "" + ) + handlers.final(event.result.text); + }; + recognizer.canceled = (_sender, event) => { + handlers.error( + new Error(event.errorDetails || "Azure Speech recognition cancelled") + ); + }; + recognizer.startContinuousRecognitionAsync(undefined, (error) => { + handlers.error(new Error(error)); + }); + return { + write: (pcm) => { + input.write(Uint8Array.from(pcm).buffer); + }, + end: () => + new Promise((resolve, reject) => { + input.close(); + recognizer.stopContinuousRecognitionAsync(resolve, (error) => { + reject(new Error(error)); + }); + }), + close: () => { + input.close(); + recognizer.close(); + }, + }; + }, + synthesize(text, onAudio) { + return new Promise((resolve, reject) => { + const synthesis = SpeechSDK.SpeechConfig.fromSubscription( + settings.apiKey, + settings.region + ); + if (settings.voice !== "") + synthesis.speechSynthesisVoiceName = settings.voice; + synthesis.speechSynthesisOutputFormat = + SpeechSDK.SpeechSynthesisOutputFormat.Raw24Khz16BitMonoPcm; + const synthesizer = new SpeechSDK.SpeechSynthesizer(synthesis, null); + synthesizer.synthesizing = (_sender, event) => { + if (event.result.audioData.byteLength > 0) + onAudio(Buffer.from(event.result.audioData)); + }; + synthesizer.speakTextAsync( + text, + (result) => { + synthesizer.close(); + if ( + result.reason === + SpeechSDK.ResultReason.SynthesizingAudioCompleted + ) { + resolve(); + } else { + reject( + new Error( + result.errorDetails || "Azure Speech synthesis failed" + ) + ); + } + }, + (error) => { + synthesizer.close(); + reject(new Error(error)); + } + ); + }); + }, + }; +} + +export function encodeAudioFrame(sequence: number, pcm: Buffer): Buffer { + const header = Buffer.allocUnsafe(8); + header.writeUInt8(BINARY_VERSION, 0); + header.writeUInt8(BINARY_AUDIO, 1); + header.writeUInt16BE(0, 2); + header.writeUInt32BE(sequence, 4); + return Buffer.concat([header, pcm]); +} +function sendAudio(socket: WebSocket, turn: SocketTurn, pcm: Buffer): void { + for (let offset = 0; offset < pcm.length; offset += OUTPUT_CHUNK) { + if ( + socket.readyState !== 1 || + ((socket as unknown as { bufferedAmount?: number }).bufferedAmount !== + undefined && + (socket as unknown as { bufferedAmount: number }).bufferedAmount > + 1_000_000) + ) + throw new Error("voice output backpressure exceeded"); + socket.send( + encodeAudioFrame( + turn.outputSequence++, + pcm.subarray(offset, offset + OUTPUT_CHUNK) + ), + { binary: true } + ); + } +} +function rawDataToBuffer(data: RawData): Buffer { + if (Buffer.isBuffer(data)) return data; + if (data instanceof ArrayBuffer) return Buffer.from(data); + return Buffer.concat(data); +} +function assistantText(value: unknown): string { + if ( + !isRecord(value) || + value["role"] !== "assistant" || + !Array.isArray(value["content"]) + ) + return ""; + return value["content"] + .filter( + (part) => + isRecord(part) && + part["type"] === "text" && + typeof part["text"] === "string" + ) + .map((part) => (part as { text: string }).text) + .join("\n") + .trim(); +} +function protocolError(socket: WebSocket, error: string): void { + sendJson(socket, { type: "error", error }); +} +function sendJson(socket: WebSocket, value: unknown): void { + if (socket.readyState === 1) socket.send(JSON.stringify(value)); +} +function pruneConversations(conversations: Map): void { + while (conversations.size >= MAX_CONVERSATIONS) { + const oldest = [...conversations.values()].sort( + (a, b) => a.createdAt - b.createdAt + )[0]; + if (oldest === undefined) return; + conversations.delete(oldest.id); + } +} +function voiceConfigPath(): string { + return join(dirname(piWebConfigPath()), "voice-api.json"); +} +function hashToken(token: string): string { + return createHash("sha256").update(token).digest("hex"); +} +function equalHash(a: string, b: string): boolean { + const left = Buffer.from(a, "hex"); + const right = Buffer.from(b, "hex"); + return left.length === right.length && timingSafeEqual(left, right); +} +function readConfig(path: string): VoiceConfig { + if (!existsSync(path)) return { tokens: [] }; + + let value: unknown; + try { + value = JSON.parse(readFileSync(path, "utf8")); + } catch { + throw new Error("Voice API token configuration is invalid"); + } + + if (!isRecord(value) || !Array.isArray(value["tokens"])) + throw new Error("Voice API token configuration is invalid"); + return { tokens: value["tokens"].map(tokenRecord) }; +} +function tokenRecord(value: unknown): TokenRecord { + if ( + !isRecord(value) || + typeof value["id"] !== "string" || + typeof value["hash"] !== "string" || + !Array.isArray(value["models"]) || + !Array.isArray(value["thinking"]) || + typeof value["createdAt"] !== "string" + ) + throw new Error("Voice API token record is invalid"); + return { + id: value["id"], + hash: value["hash"], + models: stringArray(value["models"]), + thinking: stringArray(value["thinking"]), + // Older unrestricted tokens predate target scopes; defaulting them to + // wildcard preserves their explicitly broad authorization. + projects: value["projects"] === undefined ? ["*"] : stringArrayValue(value["projects"], "projects"), + workspaces: value["workspaces"] === undefined ? ["*"] : stringArrayValue(value["workspaces"], "workspaces"), + createdAt: value["createdAt"], + ...(typeof value["revokedAt"] === "string" + ? { revokedAt: value["revokedAt"] } + : {}), + }; +} +function writeConfig(path: string, value: VoiceConfig): void { + mkdirSync(dirname(path), { recursive: true, mode: 0o700 }); + writeFileSync(path, `${JSON.stringify(value, null, 2)}\n`, { mode: 0o600 }); + chmodSync(path, 0o600); +} +function normalizeScope(value: string[] | undefined): string[] { + return value === undefined || value.length === 0 + ? ["*"] + : [ + ...new Set( + value + .filter((item) => typeof item === "string" && item.trim() !== "") + .map((item) => item.trim()) + ), + ]; +} +function stringArray(value: unknown[]): string[] { + if (!value.every((item) => typeof item === "string")) + throw new Error("Voice API scope is invalid"); + return value; +} +function stringArrayValue(value: unknown, name: string): string[] { + if (!Array.isArray(value)) + throw new Error(`Voice API ${name} scope is invalid`); + return stringArray(value); +} +function isSecureVoiceRequest(request: FastifyRequest): boolean { + return ( + request.protocol === "https" || + request.ip === "127.0.0.1" || + request.ip === "::1" + ); +} +function publicScope(token: TokenRecord) { + return { + models: token.models, + thinking: token.thinking, + projects: token.projects, + workspaces: token.workspaces, + }; +} +function publicConversation(value: Conversation) { + return { + id: value.id, + sessionId: value.sessionId, + workspaceId: value.workspaceId, + cwd: value.cwd, + status: value.status, + ...(value.model === undefined ? {} : { model: value.model }), + ...(value.thinking === undefined ? {} : { thinking: value.thinking }), + }; +} +function createInput(value: unknown): { + workspaceId: string; + model?: string; + thinking?: string; + context?: string; +} { + if ( + !isRecord(value) || + typeof value["workspaceId"] !== "string" || + value["workspaceId"].trim() === "" + ) + throw new Error("workspaceId is required"); + const model = optionalBounded(value["model"], "model", 512); + const thinking = optionalBounded(value["thinking"], "thinking", 64); + const context = optionalBounded(value["context"], "context", MAX_CONTEXT); + return { + workspaceId: value["workspaceId"], + ...(model === undefined ? {} : { model }), + ...(thinking === undefined ? {} : { thinking }), + ...(context === undefined ? {} : { context }), + }; +} +function optionalBounded( + value: unknown, + name: string, + max: number +): string | undefined { + if (value === undefined) return undefined; + if (typeof value !== "string" || value.trim() === "" || value.length > max) + throw new Error( + `${name} must be a non-empty string up to ${String(max)} characters` + ); + return value.trim(); +} +async function exactWorkspace(deps: VoiceApiDependencies, workspaceId: string) { + const projects = await deps.projects.list(); + const workspaces = ( + await Promise.all( + projects.map(async (project) => deps.workspaces.list(project)) + ) + ).flat(); + const workspace = workspaces.find((entry) => entry.id === workspaceId); + if (workspace === undefined) + throw new Error("workspaceId must identify a registered workspace"); + return workspace; +} +function requireScope( + token: TokenRecord, + model: string | undefined, + thinking: string | undefined +): void { + requireScopedValue(token.models, model, "model"); + requireScopedValue(token.thinking, thinking, "thinking level"); +} +function requireScopedValue( + permitted: readonly string[], + selected: string | undefined, + label: string +): void { + if (permitted.includes("*")) return; + // A restricted device cannot inherit a potentially different server/session + // default. It must select one of its granted values for every conversation. + if (selected === undefined) + throw new Error(`restricted token must explicitly select a ${label}`); + if (!permitted.includes(selected)) + throw new Error(`token is not allowed to use this ${label}`); +} +function workspaceAllowed( + token: TokenRecord, + workspace: { id: string; projectId: string } +): boolean { + return ( + (token.projects.includes("*") || token.projects.includes(workspace.projectId)) && + (token.workspaces.includes("*") || token.workspaces.includes(workspace.id)) + ); +} +async function daemonJson( + daemon: SessionProxyDaemon, + method: string, + path: string, + body: unknown +): Promise> { + const response = await daemon.request(method, path, body); + let parsed: unknown; + try { + parsed = response.body === "" ? {} : JSON.parse(response.body); + } catch { + throw new Error("session daemon returned invalid JSON"); + } + if (response.statusCode < 200 || response.statusCode >= 300) + throw new Error( + isRecord(parsed) && typeof parsed["error"] === "string" + ? parsed["error"] + : `session daemon returned HTTP ${String(response.statusCode)}` + ); + if (!isRecord(parsed)) + throw new Error("session daemon returned invalid response"); + return parsed; +} +function requiredString(value: Record, field: string): string { + const result = value[field]; + if (typeof result !== "string" || result === "") + throw new Error(`session daemon response is missing ${field}`); + return result; +} +function splitModel(model: string): [string, string] { + const slash = model.indexOf("/"); + if (slash <= 0 || slash === model.length - 1) + throw new Error("model must use provider/model-id format"); + return [model.slice(0, slash), model.slice(slash + 1)]; +} +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} +function message(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} diff --git a/src/server/voiceApiOpenApi.ts b/src/server/voiceApiOpenApi.ts new file mode 100644 index 0000000..d12c2bb --- /dev/null +++ b/src/server/voiceApiOpenApi.ts @@ -0,0 +1,39 @@ +import type { OpenAPIV3 } from "openapi-types"; + +export const voiceApiOpenApi: OpenAPIV3.Document = { + openapi: "3.0.3", + info: { + title: "PI WEB Voice Pipeline API", + version: "1.0.0", + description: "Authenticated, server-side Azure Speech voice conversations for native and embedded clients. Audio is sent over the documented WebSocket protocol; Azure credentials never leave PI WEB.", + }, + servers: [{ url: "/", description: "Current PI WEB HTTPS origin" }], + components: { + securitySchemes: { + bearerAuth: { type: "http", scheme: "bearer", bearerFormat: "pwv1 device token" }, + }, + schemas: { + Error: { type: "object", required: ["error"], properties: { error: { type: "string" } } }, + VoiceTarget: { type: "object", required: ["id", "projectId", "label", "path"], properties: { id: { type: "string" }, projectId: { type: "string" }, label: { type: "string" }, path: { type: "string", description: "Registered workspace directory" } } }, + CreateConversation: { type: "object", required: ["workspaceId"], properties: { workspaceId: { type: "string" }, model: { type: "string", example: "openai-codex/gpt-5.6-terra" }, thinking: { type: "string", example: "high" }, context: { type: "string", maxLength: 12000, description: "Additional instructions appended to the turn" } } }, + Conversation: { type: "object", required: ["id", "sessionId", "workspaceId", "cwd", "status"], properties: { id: { type: "string" }, sessionId: { type: "string" }, workspaceId: { type: "string" }, cwd: { type: "string" }, status: { type: "string", enum: ["input-ready", "working", "closed"] }, model: { type: "string" }, thinking: { type: "string" } } }, + }, + }, + security: [{ bearerAuth: [] }], + paths: { + "/api/v1/voice/targets": { + get: { summary: "List authorized voice targets", responses: { "200": { description: "Registered workspaces permitted by this device token", content: { "application/json": { schema: { type: "object", required: ["workspaces", "scopes"], properties: { workspaces: { type: "array", items: { $ref: "#/components/schemas/VoiceTarget" } }, scopes: { type: "object" } } } } } }, "401": { description: "Missing or invalid token", content: { "application/json": { schema: { $ref: "#/components/schemas/Error" } } } } } }, + }, + "/api/v1/voice/conversations": { + post: { summary: "Create a voice conversation", requestBody: { required: true, content: { "application/json": { schema: { $ref: "#/components/schemas/CreateConversation" } } } }, responses: { "201": { description: "Conversation created", content: { "application/json": { schema: { $ref: "#/components/schemas/Conversation" } } } }, "400": { description: "Invalid or unauthorized requested scope", content: { "application/json": { schema: { $ref: "#/components/schemas/Error" } } } } } }, + }, + "/api/v1/voice/conversations/{id}": { + parameters: [{ name: "id", in: "path", required: true, schema: { type: "string" } }], + get: { summary: "Read a conversation", responses: { "200": { description: "Conversation", content: { "application/json": { schema: { $ref: "#/components/schemas/Conversation" } } } }, "404": { description: "Not found or not owned by this token" } } }, + delete: { summary: "Close a conversation", responses: { "204": { description: "Closed" }, "404": { description: "Not found or not owned by this token" } } }, + }, + "/api/v1/voice/stream": { + get: { summary: "Voice audio WebSocket", description: "Upgrade to WebSocket using the bearer token. Send `attach`, then binary signed 16-bit little-endian PCM at 16 kHz mono, followed by `input.end`. Server emits transcript JSON, `assistant.final`, and raw 24 kHz mono PCM binary frames. See docs/voice-api.md for the binary frame layout.", responses: { "101": { description: "WebSocket upgraded" }, "401": { description: "Invalid token" } } }, + }, + }, +}; diff --git a/src/shared/apiTypes.ts b/src/shared/apiTypes.ts index ccfd1e8..3a41793 100644 --- a/src/shared/apiTypes.ts +++ b/src/shared/apiTypes.ts @@ -1283,6 +1283,8 @@ type SessionUiEventBody = | { type: "shell.end"; output?: string; exitCode?: number | null; cancelled?: boolean; truncated?: boolean; fullOutputPath?: string; isError?: boolean } | { type: "agent.start" } | { type: "agent.end" } + /** Pi-native completion: no retry, compaction retry, or queued continuation remains. */ + | { type: "agent.settled" } | { type: "message.end"; message?: unknown } | { type: "status.update"; status: SessionStatus } | { type: "activity.update"; activity: SessionActivity }