diff --git a/docs/voice-api.md b/docs/voice-api.md index 87f9ddc..3d3ba15 100644 --- a/docs/voice-api.md +++ b/docs/voice-api.md @@ -23,7 +23,9 @@ Use `Authorization: Bearer pwv1_...` on every HTTP and WebSocket connection. The ## HTTP -`GET /api/v1/voice/targets` returns only the caller's authorized registered workspaces plus effective scopes. +`GET /api/v1/voice/targets` returns only the caller's authorized registered workspaces plus effective scopes. Use its `workspaces[].id` to choose a workspace; paths are returned for display but are never accepted as an API input. + +`GET /api/v1/voice/conversations?workspaceId=` lists prior Pi sessions rooted at that authorized workspace. This lets a device show a conversation picker without learning about sessions outside its workspace scope. `POST /api/v1/voice/conversations` @@ -36,7 +38,21 @@ Use `Authorization: Bearer pwv1_...` on every HTTP and WebSocket connection. The } ``` -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. +For an existing directory beneath the device-safe root `/home/hope/workspaces` that is not yet registered as a PI WEB workspace, use `POST /api/v1/voice/conversations/path` with `path` instead of `workspaceId`. The path must already exist, resolve beneath that root, and is validated before a session starts. + +The response is `201` and includes the opaque conversation id, session id, selected workspace, and `input-ready` status. `GET /api/v1/voice/conversations/:id/models` lists the models currently available to that conversation's Pi session, filtered by the device token's model scope. `GET` and `DELETE /api/v1/voice/conversations/:id` inspect and close the API conversation handle. + +To resume a listed session, create a new token-owned voice handle: + +```json +POST /api/v1/voice/conversations/resume +{ + "workspaceId": "registered-workspace-id", + "sessionId": "previous-pi-session-id" +} +``` + +The session must be in the selected workspace and within the caller's token scope. ## WebSocket wire protocol @@ -77,7 +93,7 @@ Azure push-stream recognition emits `transcript.partial` and `transcript.final` ### 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. +For either input form, the server emits `agent.working`, then `agent.accepted` after Pi accepted the prompt. While it waits for Pi's native `agent.settled` event (not merely `agent.end`), it repeats the existing `agent.working` frame every 15 seconds; those progress frames stop before `assistant.final` and never overlap synthesized audio. The wait has 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: diff --git a/package-lock.json b/package-lock.json index ccdd20e..67be55b 100644 --- a/package-lock.json +++ b/package-lock.json @@ -55,6 +55,7 @@ "globals": "^17.7.0", "happy-dom": "^20.11.1", "knip": "^6.25.0", + "openapi-types": "^12.1.3", "tsx": "^4.23.0", "typescript": "^6.0.3", "typescript-eslint": "^8.63.0", diff --git a/package.json b/package.json index 2129e15..f3ca325 100644 --- a/package.json +++ b/package.json @@ -96,6 +96,7 @@ "globals": "^17.7.0", "happy-dom": "^20.11.1", "knip": "^6.25.0", + "openapi-types": "^12.1.3", "tsx": "^4.23.0", "typescript": "^6.0.3", "typescript-eslint": "^8.63.0", diff --git a/src/server/voiceApi.test.ts b/src/server/voiceApi.test.ts index e844a38..3d34fb3 100644 --- a/src/server/voiceApi.test.ts +++ b/src/server/voiceApi.test.ts @@ -5,16 +5,19 @@ 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 { afterEach, describe, expect, it, vi } from "vitest"; import { createVoiceToken, encodeAudioFrame, registerVoiceApiRoutes, revokeVoiceToken, + clearWorkingProgress, + startWorkingProgress, } from "./voiceApi.js"; const temporary: string[] = []; afterEach(() => { + vi.useRealTimers(); for (const path of temporary.splice(0)) rmSync(path, { recursive: true, force: true }); }); @@ -99,6 +102,14 @@ function responseId(body: string): string { return parsed.id; } +function jsonEventType(data: RawData): string | undefined { + const value: unknown = JSON.parse(rawDataToBuffer(data).toString()); + if (typeof value !== "object" || value === null || !("type" in value)) + return undefined; + const type = value.type; + return typeof type === "string" ? type : undefined; +} + 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])); @@ -108,6 +119,118 @@ describe("voice API", () => { expect(frame.subarray(8)).toEqual(Buffer.from([1, 2, 3, 4])); }); + it("repeats working progress every 15 seconds and stops it before final/audio", () => { + vi.useFakeTimers(); + try { + const sent: unknown[] = []; + const socket: Pick = { + readyState: WebSocket.OPEN, + send: (value) => { + if (typeof value !== "string") throw new Error("expected JSON progress"); + sent.push(JSON.parse(value)); + }, + }; + const turn: { closed: boolean; workingTimer: ReturnType | undefined } = { + closed: false, + workingTimer: undefined, + }; + + startWorkingProgress(socket, turn); + vi.advanceTimersByTime(45_000); + clearWorkingProgress(turn); + vi.advanceTimersByTime(30_000); + + expect(sent).toEqual([ + { type: "agent.working" }, + { type: "agent.working" }, + { type: "agent.working" }, + ]); + expect(turn.workingTimer).toBeUndefined(); + } finally { + vi.useRealTimers(); + } + }); + + it("begins 15-second working repeats before a delayed event bridge opens and stops on settlement", async () => { + const directory = mkdtempSync(join(tmpdir(), "pi-web-voice-api-")); + temporary.push(directory); + const events = Object.assign(new EventEmitter(), { + close: () => undefined, + readyState: websocketConnectingState(), + }); + const app = Fastify(); + await app.register(fastifyWebsocket); + const configPath = join(directory, "voice-api.json"); + registerVoiceApiRoutes(app, { + configPath, + 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")) + return Promise.resolve(response(200, { accepted: true })); + return Promise.resolve(response(404, {})); + }, + connectWebSocket: () => { + setTimeout(() => { + events.readyState = WebSocket.OPEN; + events.emit("open"); + }, 9_500); + // EventEmitter supplies exactly the event bridge surface the route consumes. + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + return events as unknown as WebSocket; + }, + }, + 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 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 received: { type: string }[] = []; + let client: WebSocket | undefined; + vi.useFakeTimers({ + toFake: ["setTimeout", "clearTimeout", "setInterval", "clearInterval", "Date"], + }); + try { + await new Promise((resolve, reject) => { + 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: responseId(created.body) })); + client?.send(JSON.stringify({ type: "input.text", text: "hello" })); + }); + client.on("message", (data, binary) => { + if (binary) return; + const type = jsonEventType(data); + if (type === undefined) return; + received.push({ type }); + if (type === "agent.working") resolve(); + }); + client.on("error", reject); + }); + await vi.advanceTimersByTimeAsync(15_000); + expect(received.filter((event) => event.type === "agent.working")).toHaveLength(2); + + const completed = new Promise((resolve) => { + client?.on("message", (data, binary) => { + if (!binary && jsonEventType(data) === "audio.end") resolve(); + }); + }); + events.emit("message", Buffer.from(JSON.stringify({ type: "agent.settled" }))); + await completed; + await vi.advanceTimersByTimeAsync(30_000); + expect(received.filter((event) => event.type === "agent.working")).toHaveLength(2); + } finally { + client?.close(); + await app.close(); + vi.useRealTimers(); + } + }); + 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); diff --git a/src/server/voiceApi.ts b/src/server/voiceApi.ts index 5142c34..5e58497 100644 --- a/src/server/voiceApi.ts +++ b/src/server/voiceApi.ts @@ -5,9 +5,10 @@ import { existsSync, mkdirSync, readFileSync, + statSync, writeFileSync, } from "node:fs"; -import { dirname, join } from "node:path"; +import { dirname, join, relative, resolve } from "node:path"; import type { FastifyInstance, FastifyRequest } from "fastify"; import * as SpeechSDK from "microsoft-cognitiveservices-speech-sdk"; import type { RawData, WebSocket } from "ws"; @@ -28,6 +29,8 @@ const BINARY_AUDIO = 1; const OUTPUT_CHUNK = 16 * 1024; const EVENT_OPEN_TIMEOUT_MS = 10_000; const TURN_TIMEOUT_MS = 120_000; +const WORKING_PROGRESS_INTERVAL_MS = 15_000; +const DEVICE_WORKSPACE_ROOT = "/home/hope/workspaces"; interface TokenRecord { id: string; hash: string; @@ -93,6 +96,7 @@ interface Conversation { interface SocketTurn { recognizer?: VoiceRecognizer; eventSocket: WebSocket | undefined; + workingTimer: ReturnType | undefined; /** VAD/STT result only; never use it as an assistant response. */ transcript: string; /** Assistant output collected from Pi's session event stream only. */ @@ -208,6 +212,48 @@ export function registerVoiceApiRoutes( scopes: publicScope(token), }; }); + app.post<{ Body: unknown }>( + "/api/v1/voice/conversations/path", + async (request, reply) => { + const token = reject(request, reply); + if (token === undefined) return; + try { + const input = createPathInput(request.body); + requireScope(token, input.model, input.thinking); + const cwd = allowedDeviceWorkspacePath(input.path); + const created = await daemonJson(deps.daemon, "POST", "/sessions", { cwd }); + 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, provider, modelId }); + } + if (input.thinking !== undefined) await daemonJson(deps.daemon, "POST", `/sessions/${encodeURIComponent(sessionId)}/thinking-level`, { cwd, level: input.thinking }); + pruneConversations(conversations); + const conversation: Conversation = { id: randomBytes(16).toString("hex"), ownerTokenId: token.id, sessionId, cwd, workspaceId: `path:${createHash("sha256").update(cwd).digest("hex").slice(0, 16)}`, 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<{ Querystring: { workspaceId?: string } }>( + "/api/v1/voice/conversations", + async (request, reply) => { + const token = reject(request, reply); + if (token === undefined) return; + try { + const workspaceId = request.query.workspaceId; + if (workspaceId === undefined || workspaceId.trim() === "") throw new Error("workspaceId query parameter is required"); + const workspace = await exactWorkspace(deps, workspaceId); + if (!workspaceAllowed(token, workspace)) throw new Error("token is not allowed to use this workspace"); + const sessions = await workspaceSessions(deps.daemon, workspace.path); + return { workspaceId: workspace.id, cwd: workspace.path, sessions }; + } catch (error) { + return reply.code(400).send({ error: message(error) }); + } + } + ); + app.post<{ Body: unknown }>( "/api/v1/voice/conversations", async (request, reply) => { @@ -259,6 +305,36 @@ export function registerVoiceApiRoutes( } } ); + app.post<{ Body: unknown }>( + "/api/v1/voice/conversations/resume", + async (request, reply) => { + const token = reject(request, reply); + if (token === undefined) return; + try { + const input = resumeInput(request.body); + const workspace = await exactWorkspace(deps, input.workspaceId); + if (!workspaceAllowed(token, workspace)) throw new Error("token is not allowed to use this workspace"); + const sessions = await workspaceSessions(deps.daemon, workspace.path); + const session = sessions.find((candidate) => candidate.id === input.sessionId); + if (session === undefined) throw new Error("sessionId does not belong to this workspace"); + pruneConversations(conversations); + const conversation: Conversation = { + id: randomBytes(16).toString("hex"), + ownerTokenId: token.id, + sessionId: input.sessionId, + cwd: workspace.path, + workspaceId: workspace.id, + status: "input-ready", + createdAt: Date.now(), + }; + 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) => { @@ -270,6 +346,35 @@ export function registerVoiceApiRoutes( : publicConversation(conversation); } ); + app.get<{ Params: { id: string } }>( + "/api/v1/voice/conversations/:id/models", + 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" }); + try { + const response = await daemonJson( + deps.daemon, + "GET", + `/sessions/${encodeURIComponent(conversation.sessionId)}/models?cwd=${encodeURIComponent(conversation.cwd)}`, + undefined + ); + const models = Array.isArray(response["models"]) ? response["models"].filter(isRecord) : []; + return { + models: models.filter((model) => { + const provider = model["provider"]; + const id = model["id"]; + return typeof provider === "string" && typeof id === "string" && scopeAllows(token.models, `${provider}/${id}`); + }), + }; + } catch (error) { + return reply.code(502).send({ error: message(error) }); + } + } + ); + app.delete<{ Params: { id: string } }>( "/api/v1/voice/conversations/:id", async (request, reply) => { @@ -308,6 +413,7 @@ function wireVoiceSocket( outputSequence: 0, closed: false, eventSocket: undefined, + workingTimer: undefined, }; sendJson(socket, { type: "hello", @@ -320,6 +426,7 @@ function wireVoiceSocket( turn.closed = true; turn.recognizer?.close(); delete turn.recognizer; + clearWorkingProgress(turn); turn.eventSocket?.close(); turn.eventSocket = undefined; if (conversation !== undefined && conversation.status === "working") @@ -475,6 +582,7 @@ async function submitText( conversation.status = "working"; turn.assistant = ""; sendJson(socket, { type: "agent.working" }); + startWorkingProgress(socket, turn); try { await withTimeout( runTurn(socket, turn, conversation, deps, speech, text), @@ -553,6 +661,7 @@ async function startSettledWaiter( const finish = (error?: Error) => { if (done) return; done = true; + clearWorkingProgress(turn); if (error === undefined) resolve(); else reject(error); }; @@ -605,7 +714,28 @@ function waitForWebSocketOpen(socket: WebSocket): Promise { ); } +export function startWorkingProgress( + socket: Pick, + turn: Pick +): void { + clearWorkingProgress(turn); + const timer = setInterval(() => { + if (!turn.closed && turn.workingTimer === timer && socket.readyState === 1) + socket.send(JSON.stringify({ type: "agent.working" })); + }, WORKING_PROGRESS_INTERVAL_MS); + timer.unref(); + turn.workingTimer = timer; +} + +export function clearWorkingProgress( + turn: Pick +): void { + if (turn.workingTimer !== undefined) clearInterval(turn.workingTimer); + turn.workingTimer = undefined; +} + function closeEventSocket(turn: SocketTurn): void { + clearWorkingProgress(turn); const events = turn.eventSocket; turn.eventSocket = undefined; events?.close(); @@ -919,6 +1049,30 @@ function publicConversation(value: Conversation) { ...(value.thinking === undefined ? {} : { thinking: value.thinking }), }; } +function createPathInput(value: unknown): { path: string; model?: string; thinking?: string; context?: string } { + if (!isRecord(value) || typeof value["path"] !== "string" || value["path"].trim() === "") throw new Error("path 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 { path: value["path"].trim(), ...(model === undefined ? {} : { model }), ...(thinking === undefined ? {} : { thinking }), ...(context === undefined ? {} : { context }) }; +} + +function allowedDeviceWorkspacePath(value: string): string { + const root = resolve(DEVICE_WORKSPACE_ROOT); + const path = resolve(value); + const child = relative(root, path); + if (child === "" || child.startsWith("..") || child.includes("../")) throw new Error(`path must be a directory beneath ${DEVICE_WORKSPACE_ROOT}`); + if (!statSync(path).isDirectory()) throw new Error("path must be an existing directory"); + return path; +} + +function resumeInput(value: unknown): { workspaceId: string; sessionId: string } { + if (!isRecord(value) || typeof value["workspaceId"] !== "string" || value["workspaceId"].trim() === "" || typeof value["sessionId"] !== "string" || value["sessionId"].trim() === "") { + throw new Error("workspaceId and sessionId are required"); + } + return { workspaceId: value["workspaceId"].trim(), sessionId: value["sessionId"].trim() }; +} + function createInput(value: unknown): { workspaceId: string; model?: string; @@ -973,6 +1127,9 @@ function requireScope( requireScopedValue(token.models, model, "model"); requireScopedValue(token.thinking, thinking, "thinking level"); } +function scopeAllows(permitted: readonly string[], value: string): boolean { + return permitted.includes("*") || permitted.includes(value); +} function requireScopedValue( permitted: readonly string[], selected: string | undefined, @@ -995,6 +1152,22 @@ function workspaceAllowed( (token.workspaces.includes("*") || token.workspaces.includes(workspace.id)) ); } +async function workspaceSessions(daemon: SessionProxyDaemon, cwd: string): Promise<{ id: string; name?: string; updatedAt?: string }[]> { + const response = await daemon.request("GET", `/sessions?cwd=${encodeURIComponent(cwd)}`); + if (response.statusCode < 200 || response.statusCode >= 300) throw new Error(`session daemon returned HTTP ${String(response.statusCode)}`); + let parsed: unknown; + try { parsed = JSON.parse(response.body); } catch { throw new Error("session daemon returned invalid JSON"); } + if (!Array.isArray(parsed)) throw new Error("session daemon returned invalid session list"); + return parsed.flatMap((entry) => { + if (!isRecord(entry) || typeof entry["id"] !== "string" || entry["id"] === "") return []; + return [{ + id: entry["id"], + ...(typeof entry["name"] === "string" ? { name: entry["name"] } : {}), + ...(typeof entry["updatedAt"] === "string" ? { updatedAt: entry["updatedAt"] } : {}), + }]; + }); +} + async function daemonJson( daemon: SessionProxyDaemon, method: string, diff --git a/src/server/voiceApiOpenApi.ts b/src/server/voiceApiOpenApi.ts index d12c2bb..5e9eb79 100644 --- a/src/server/voiceApiOpenApi.ts +++ b/src/server/voiceApiOpenApi.ts @@ -25,8 +25,18 @@ export const voiceApiOpenApi: OpenAPIV3.Document = { 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": { + get: { summary: "List prior sessions in an authorized workspace", parameters: [{ name: "workspaceId", in: "query", required: true, schema: { type: "string" } }], responses: { "200": { description: "Previous PI sessions rooted in the workspace" }, "400": { description: "Invalid or unauthorized workspace" } } }, 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/path": { + post: { summary: "Create a conversation in an existing directory beneath the device-safe root", requestBody: { required: true, content: { "application/json": { schema: { type: "object", required: ["path"], properties: { path: { type: "string", example: "/home/hope/workspaces/new-project" }, model: { type: "string" }, thinking: { type: "string" }, context: { type: "string" } } } } } }, responses: { "201": { description: "Conversation created" }, "400": { description: "Path is missing, outside the safe root, or not a directory" } } }, + }, + "/api/v1/voice/conversations/resume": { + post: { summary: "Resume a prior workspace session", requestBody: { required: true, content: { "application/json": { schema: { type: "object", required: ["workspaceId", "sessionId"], properties: { workspaceId: { type: "string" }, sessionId: { type: "string" } } } } } }, responses: { "201": { description: "Conversation handle attached to prior session", content: { "application/json": { schema: { $ref: "#/components/schemas/Conversation" } } } }, "400": { description: "Session is not in the authorized workspace" } } }, + }, + "/api/v1/voice/conversations/{id}/models": { + get: { summary: "List available models for a conversation", parameters: [{ name: "id", in: "path", required: true, schema: { type: "string" } }], responses: { "200": { description: "Models available to the session and permitted by this device token" }, "404": { description: "Conversation not found or not owned by this token" } } }, + }, "/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" } } },