jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add voice capability status reporting to client

Browse Source
This commit is contained in:
Mohamed Boudra
2026-02-14 13:14:57 +07:00
parent c4bebb97ab
commit a644adbf9b
16 changed files with 585 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/bootstrap.ts
+7
View File
@@ -299,6 +299,7 @@ export async function createPaseoDaemon(
);
let wsServer: VoiceAssistantWebSocketServer | null = null;
let voiceMcpBridgeManager: VoiceMcpSocketBridgeManager | null = null;
let unsubscribeSpeechReadiness: (() => void) | null = null;
// Create in-memory transport for Session's Agent MCP client (voice assistant tools)
const createInMemoryAgentMcpTransport = async (): Promise<InMemoryTransport> => {
@@ -461,6 +462,7 @@ export async function createPaseoDaemon(
resolveVoiceTts,
resolveDictationStt,
getSpeechReadiness,
subscribeSpeechReadiness,
cleanup: cleanupSpeechRuntime,
localModelConfig,
} = await initializeSpeechRuntime({
@@ -503,6 +505,9 @@ export async function createPaseoDaemon(
},
config.agentProviderSettings
);
unsubscribeSpeechReadiness = subscribeSpeechReadiness((snapshot) => {
wsServer?.publishSpeechReadiness(snapshot);
});
const start = async () => {
// Start main HTTP server
@@ -582,6 +587,8 @@ export async function createPaseoDaemon(
runtimeSettings: config.agentProviderSettings,
});
terminalManager.killAll();
unsubscribeSpeechReadiness?.();
unsubscribeSpeechReadiness = null;
cleanupSpeechRuntime();
await relayTransport?.stop().catch(() => undefined);
if (wsServer) {
packages/server/src/server/daemon-client.e2e.test.ts
+4 -4
View File
@@ -11,6 +11,7 @@ import {
DaemonClient,
} from "./test-utils/index.js";
import { getFullAccessConfig, getAskModeConfig } from "./daemon-e2e/agent-configs.js";
import { parseServerInfoStatusPayload } from "./messages.js";
import {
chunkPcm16,
parsePcm16MonoWav,
@@ -175,10 +176,9 @@ describe("daemon client E2E", () => {
const infoPromise = waitForSignal<{ serverId: string }>(5000, (resolve) => {
const unsubscribe = client.on("status", (message) => {
if (message.type !== "status") return;
const payload = message.payload as { status?: unknown; serverId?: unknown };
if (payload.status !== "server_info") return;
if (typeof payload.serverId !== "string" || payload.serverId.trim().length === 0) return;
resolve({ serverId: payload.serverId.trim() });
const payload = parseServerInfoStatusPayload(message.payload);
if (!payload) return;
resolve({ serverId: payload.serverId });
});
return unsubscribe;
});
packages/server/src/server/speech/speech-runtime.ts
+63 -1
View File
@@ -308,6 +308,9 @@ export type InitializedSpeechRuntime = {
resolveVoiceTts: () => TextToSpeechProvider | null;
resolveDictationStt: () => SpeechToTextProvider | null;
getSpeechReadiness: () => SpeechReadinessSnapshot;
subscribeSpeechReadiness: (
listener: (snapshot: SpeechReadinessSnapshot) => void
) => () => void;
cleanup: () => void;
localModelConfig: {
modelsDir: string;
@@ -359,6 +362,9 @@ export async function initializeSpeechRuntime(params: {
let stopped = false;
let monitorTimeout: ReturnType<typeof setTimeout> | null = null;
let reconcileInFlight: Promise<void> | null = null;
const readinessListeners = new Set<(snapshot: SpeechReadinessSnapshot) => void>();
let lastReadinessFingerprint: string | null = null;
let lastPublishedReadinessSnapshot: SpeechReadinessSnapshot | null = null;
const computeReadinessSnapshot = (): SpeechReadinessSnapshot => {
const realtimeVoice = buildRealtimeVoiceReadiness({
@@ -398,6 +404,54 @@ export async function initializeSpeechRuntime(params: {
};
};
const readinessFingerprint = (snapshot: SpeechReadinessSnapshot): string =>
JSON.stringify({
...snapshot,
generatedAt: "",
});
const publishReadinessIfChanged = (): void => {
const snapshot = computeReadinessSnapshot();
const fingerprint = readinessFingerprint(snapshot);
if (fingerprint === lastReadinessFingerprint) {
return;
}
lastReadinessFingerprint = fingerprint;
lastPublishedReadinessSnapshot = snapshot;
for (const listener of readinessListeners) {
try {
listener(snapshot);
} catch (error) {
logger.warn(
{ err: error },
"Speech readiness listener threw an error"
);
}
}
};
const subscribeSpeechReadiness = (
listener: (snapshot: SpeechReadinessSnapshot) => void
): (() => void) => {
readinessListeners.add(listener);
const snapshot = lastPublishedReadinessSnapshot ?? computeReadinessSnapshot();
if (!lastPublishedReadinessSnapshot) {
lastPublishedReadinessSnapshot = snapshot;
lastReadinessFingerprint = readinessFingerprint(snapshot);
}
try {
listener(snapshot);
} catch (error) {
logger.warn(
{ err: error },
"Speech readiness listener threw an error during subscribe"
);
}
return () => {
readinessListeners.delete(listener);
};
};
const refreshMissingLocalModels = async (): Promise<void> => {
missingLocalModelIds = await findMissingRequiredLocalModels({
modelsDir: localModelConfig?.modelsDir ?? null,
@@ -470,12 +524,14 @@ export async function initializeSpeechRuntime(params: {
const runReconcile = async (params: { modelEnsureAutoDownload: boolean }): Promise<void> => {
if (reconcileInFlight) {
await reconcileInFlight;
publishReadinessIfChanged();
return;
}
reconcileInFlight = reconcileServices(params.modelEnsureAutoDownload).finally(() => {
reconcileInFlight = null;
});
await reconcileInFlight;
publishReadinessIfChanged();
};
const scheduleMonitor = (): void => {
@@ -500,6 +556,7 @@ export async function initializeSpeechRuntime(params: {
backgroundDownloadInProgress = true;
backgroundDownloadError = null;
publishReadinessIfChanged();
logger.info(
{
@@ -521,6 +578,7 @@ export async function initializeSpeechRuntime(params: {
backgroundDownloadError = null;
} catch (error) {
backgroundDownloadError = error instanceof Error ? error.message : String(error);
publishReadinessIfChanged();
logger.error(
{
err: error,
@@ -533,6 +591,7 @@ export async function initializeSpeechRuntime(params: {
await refreshMissingLocalModels().catch((error) => {
logger.warn({ err: error }, "Failed to refresh local speech model status after download");
});
publishReadinessIfChanged();
scheduleMonitor();
}
})();
@@ -565,6 +624,7 @@ export async function initializeSpeechRuntime(params: {
} catch (error) {
logger.warn({ err: error }, "Speech runtime monitor tick failed");
} finally {
publishReadinessIfChanged();
scheduleMonitor();
}
};
@@ -591,7 +651,9 @@ export async function initializeSpeechRuntime(params: {
resolveVoiceStt: () => sttService,
resolveVoiceTts: () => ttsService,
resolveDictationStt: () => dictationSttService,
getSpeechReadiness: () => computeReadinessSnapshot(),
getSpeechReadiness: () =>
lastPublishedReadinessSnapshot ?? computeReadinessSnapshot(),
subscribeSpeechReadiness,
cleanup,
localModelConfig,
};
packages/server/src/server/websocket-server.relay-reconnect.test.ts
+115 -2
View File
@@ -76,6 +76,8 @@ import {
VoiceAssistantWebSocketServer,
type ExternalSocketMetadata,
} from "./websocket-server";
import { parseServerInfoStatusPayload } from "./messages.js";
import type { SpeechReadinessSnapshot } from "./speech/speech-runtime.js";
class MockSocket {
readyState = 1;
@@ -133,7 +135,8 @@ function createLogger() {
return logger;
}
function createServer() {
function createServer(options?: { speechReadiness?: SpeechReadinessSnapshot | null }) {
const speechReadiness = options?.speechReadiness ?? null;
return new VoiceAssistantWebSocketServer(
{} as any,
createLogger() as any,
@@ -146,10 +149,54 @@ function createServer() {
{} as any,
"/tmp/paseo-test",
async () => ({} as any),
{ allowedOrigins: new Set() }
{ allowedOrigins: new Set() },
undefined,
undefined,
undefined,
speechReadiness
? {
getSpeechReadiness: () => speechReadiness,
}
: undefined
);
}
function createReadySpeechReadinessSnapshot(): SpeechReadinessSnapshot {
return {
generatedAt: "2026-02-14T00:00:00.000Z",
requiredLocalModelIds: [],
missingLocalModelIds: [],
download: {
inProgress: false,
error: null,
},
dictation: {
enabled: true,
available: true,
reasonCode: "ready",
message: "Dictation is ready.",
retryable: false,
missingModelIds: [],
},
realtimeVoice: {
enabled: true,
available: true,
reasonCode: "ready",
message: "Realtime voice is ready.",
retryable: false,
missingModelIds: [],
},
voiceFeature: {
enabled: true,
available: true,
reasonCode: "ready",
message: "Voice features are ready.",
retryable: false,
missingModelIds: [],
},
};
}
describe("relay external socket reconnect behavior", () => {
beforeEach(() => {
sessionMock.instances.length = 0;
@@ -205,6 +252,72 @@ describe("relay external socket reconnect behavior", () => {
await server.close();
});
test("includes voice capabilities in server_info when speech readiness exists", async () => {
const speechReadiness = createReadySpeechReadinessSnapshot();
const server = createServer({ speechReadiness });
const metadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: "relay:client-server-info-capabilities",
};
const socket = new MockSocket();
await server.attachExternalSocket(socket, metadata);
expect(socket.sent).toHaveLength(1);
expect(typeof socket.sent[0]).toBe("string");
const envelope = JSON.parse(socket.sent[0] as string) as {
type?: unknown;
message?: {
type?: unknown;
payload?: unknown;
};
};
expect(envelope.type).toBe("session");
expect(envelope.message?.type).toBe("status");
const payload = parseServerInfoStatusPayload(envelope.message?.payload);
expect(payload?.status).toBe("server_info");
expect(payload?.capabilities?.voice?.dictation.enabled).toBe(
speechReadiness.dictation.enabled
);
expect(payload?.capabilities?.voice?.dictation.reason).toBe("");
expect(payload?.capabilities?.voice?.voice.enabled).toBe(
speechReadiness.realtimeVoice.enabled
);
expect(payload?.capabilities?.voice?.voice.reason).toBe("");
await server.close();
});
test("broadcasts updated server_info when capabilities change", async () => {
const server = createServer();
const metadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: "relay:client-server-info-broadcast",
};
const socket = new MockSocket();
await server.attachExternalSocket(socket, metadata);
expect(socket.sent).toHaveLength(1);
const speechReadiness = createReadySpeechReadinessSnapshot();
server.publishSpeechReadiness(speechReadiness);
expect(socket.sent).toHaveLength(2);
const secondEnvelope = JSON.parse(socket.sent[1] as string) as {
message?: { payload?: unknown };
};
const secondPayload = parseServerInfoStatusPayload(secondEnvelope.message?.payload);
expect(secondPayload?.capabilities?.voice?.dictation.enabled).toBe(true);
expect(secondPayload?.capabilities?.voice?.voice.enabled).toBe(true);
// Same readiness should not produce another server_info broadcast.
server.publishSpeechReadiness(speechReadiness);
expect(socket.sent).toHaveLength(2);
await server.close();
});
test("routes inbound binary mux frames to session.handleBinaryFrame", async () => {
const server = createServer();
const metadata: ExternalSocketMetadata = {
packages/server/src/server/websocket-server.ts
+72 -5
View File
@@ -9,7 +9,10 @@ import type { DownloadTokenStore } from "./file-download/token-store.js";
import type { TerminalManager } from "../terminal/terminal-manager.js";
import type pino from "pino";
import {
type ServerInfoStatusPayload,
WSInboundMessageSchema,
type ServerCapabilityState,
type ServerCapabilities,
type WSOutboundMessage,
wrapSessionMessage,
} from "./messages.js";
@@ -51,6 +54,37 @@ type WebSocketServerConfig = {
allowedHosts?: AllowedHostsConfig;
};
function toServerCapabilityState(
state: SpeechReadinessSnapshot["dictation"]
): ServerCapabilityState {
return {
enabled: state.enabled,
reason: state.available ? "" : state.message,
};
}
function buildServerCapabilities(params: {
readiness: SpeechReadinessSnapshot | null;
}): ServerCapabilities | undefined {
const readiness = params.readiness;
if (!readiness) {
return undefined;
}
return {
voice: {
dictation: toServerCapabilityState(readiness.dictation),
voice: toServerCapabilityState(readiness.realtimeVoice),
},
};
}
function areServerCapabilitiesEqual(
current: ServerCapabilities | undefined,
next: ServerCapabilities | undefined
): boolean {
return JSON.stringify(current ?? null) === JSON.stringify(next ?? null);
}
function bufferFromWsData(data: Buffer | ArrayBuffer | Buffer[] | string): Buffer {
if (typeof data === "string") return Buffer.from(data, "utf8");
if (Array.isArray(data)) {
@@ -123,6 +157,7 @@ export class VoiceAssistantWebSocketServer {
>();
private readonly voiceCallerContexts = new Map<string, VoiceCallerContext>();
private readonly agentProviderRuntimeSettings: AgentProviderRuntimeSettingsMap | undefined;
private serverCapabilities: ServerCapabilities | undefined;
constructor(
server: HTTPServer,
@@ -168,6 +203,9 @@ export class VoiceAssistantWebSocketServer {
this.voice = voice ?? null;
this.dictation = dictation ?? null;
this.agentProviderRuntimeSettings = agentProviderRuntimeSettings;
this.serverCapabilities = buildServerCapabilities({
readiness: this.dictation?.getSpeechReadiness?.() ?? null,
});
const pushLogger = this.logger.child({ module: "push" });
this.pushTokenStore = new PushTokenStore(
@@ -230,6 +268,21 @@ export class VoiceAssistantWebSocketServer {
}
}
public publishSpeechReadiness(readiness: SpeechReadinessSnapshot | null): void {
this.updateServerCapabilities(buildServerCapabilities({ readiness }));
}
public updateServerCapabilities(
capabilities: ServerCapabilities | null | undefined
): void {
const next = capabilities ?? undefined;
if (areServerCapabilitiesEqual(this.serverCapabilities, next)) {
return;
}
this.serverCapabilities = next;
this.broadcastServerInfo();
}
public async attachExternalSocket(
ws: WebSocketLike,
metadata?: ExternalSocketMetadata
@@ -410,17 +463,31 @@ export class VoiceAssistantWebSocketServer {
this.bindSocketHandlers(ws, connection);
}
private buildServerInfoStatusPayload(): ServerInfoStatusPayload {
return {
status: "server_info",
serverId: this.serverId,
hostname: getHostname(),
...(this.serverCapabilities ? { capabilities: this.serverCapabilities } : {}),
};
}
private broadcastServerInfo(): void {
this.broadcast(
wrapSessionMessage({
type: "status",
payload: this.buildServerInfoStatusPayload(),
})
);
}
private sendServerInfo(ws: WebSocketLike): void {
// Advertise stable server identity immediately on connect (used for URL/shareable IDs).
this.sendToClient(
ws,
wrapSessionMessage({
type: "status",
payload: {
status: "server_info",
serverId: this.serverId,
hostname: getHostname(),
},
payload: this.buildServerInfoStatusPayload(),
})
);
}
packages/server/src/shared/messages.ts
+67
View File
@@ -1168,6 +1168,59 @@ export const DictationStreamErrorMessageSchema = z.object({
}),
});
export const ServerCapabilityStateSchema = z.object({
enabled: z.boolean(),
reason: z.string(),
});
export const ServerVoiceCapabilitiesSchema = z.object({
dictation: ServerCapabilityStateSchema,
voice: ServerCapabilityStateSchema,
});
export const ServerCapabilitiesSchema = z
.object({
voice: ServerVoiceCapabilitiesSchema.optional(),
})
.passthrough();
const ServerInfoHostnameSchema = z
.unknown()
.transform((value): string | null => {
if (typeof value !== "string") {
return null;
}
const trimmed = value.trim();
return trimmed.length > 0 ? trimmed : null;
});
const ServerCapabilitiesFromUnknownSchema = z
.unknown()
.optional()
.transform((value): z.infer<typeof ServerCapabilitiesSchema> | undefined => {
if (value === undefined) {
return undefined;
}
const parsed = ServerCapabilitiesSchema.safeParse(value);
if (!parsed.success) {
return undefined;
}
return parsed.data;
});
export const ServerInfoStatusPayloadSchema = z
.object({
status: z.literal("server_info"),
serverId: z.string().trim().min(1),
hostname: ServerInfoHostnameSchema.optional(),
capabilities: ServerCapabilitiesFromUnknownSchema,
})
.passthrough()
.transform((payload) => ({
...payload,
hostname: payload.hostname ?? null,
}));
export const StatusMessageSchema = z.object({
type: z.literal("status"),
payload: z
@@ -1975,6 +2028,10 @@ export type AssistantChunkMessage = z.infer<typeof AssistantChunkMessageSchema>;
export type AudioOutputMessage = z.infer<typeof AudioOutputMessageSchema>;
export type TranscriptionResultMessage = z.infer<typeof TranscriptionResultMessageSchema>;
export type StatusMessage = z.infer<typeof StatusMessageSchema>;
export type ServerCapabilityState = z.infer<typeof ServerCapabilityStateSchema>;
export type ServerVoiceCapabilities = z.infer<typeof ServerVoiceCapabilitiesSchema>;
export type ServerCapabilities = z.infer<typeof ServerCapabilitiesSchema>;
export type ServerInfoStatusPayload = z.infer<typeof ServerInfoStatusPayloadSchema>;
export type RpcErrorMessage = z.infer<typeof RpcErrorMessageSchema>;
export type ArtifactMessage = z.infer<typeof ArtifactMessageSchema>;
export type AgentUpdateMessage = z.infer<typeof AgentUpdateMessageSchema>;
@@ -2189,3 +2246,13 @@ export function wrapSessionMessage(
message: sessionMsg,
};
}
export function parseServerInfoStatusPayload(
payload: unknown
): ServerInfoStatusPayload | null {
const parsed = ServerInfoStatusPayloadSchema.safeParse(payload);
if (!parsed.success) {
return null;
}
return parsed.data;
}