refactor: replace welcome message with server_info status payload
Unify the initial connection handshake and capability update paths. The server now sends a server_info status message on connect and whenever capabilities change, enabling the onboarding flow where voice becomes available after models are configured.
This commit is contained in:
@@ -29,7 +29,7 @@ function createMockTransport() {
|
||||
let onOpen: () => void = () => {}
|
||||
let onClose: (_event?: unknown) => void = () => {}
|
||||
let onError: (_event?: unknown) => void = () => {}
|
||||
let welcomeOrdinal = 1
|
||||
let serverInfoOrdinal = 1
|
||||
|
||||
const transport: DaemonTransport = {
|
||||
send: (data) => sent.push(data),
|
||||
@@ -61,11 +61,16 @@ function createMockTransport() {
|
||||
sent.length = 0
|
||||
onMessage(
|
||||
JSON.stringify({
|
||||
type: 'welcome',
|
||||
serverId: `srv_test_${welcomeOrdinal++}`,
|
||||
hostname: null,
|
||||
version: null,
|
||||
resumed: false,
|
||||
type: 'session',
|
||||
message: {
|
||||
type: 'status',
|
||||
payload: {
|
||||
status: 'server_info',
|
||||
serverId: `srv_test_${serverInfoOrdinal++}`,
|
||||
hostname: null,
|
||||
version: null,
|
||||
},
|
||||
},
|
||||
})
|
||||
)
|
||||
},
|
||||
|
||||
@@ -4,10 +4,11 @@ import {
|
||||
AgentCreatedStatusPayloadSchema,
|
||||
AgentRefreshedStatusPayloadSchema,
|
||||
AgentResumedStatusPayloadSchema,
|
||||
parseServerInfoStatusPayload,
|
||||
RestartRequestedStatusPayloadSchema,
|
||||
ShutdownRequestedStatusPayloadSchema,
|
||||
SessionInboundMessageSchema,
|
||||
type WSWelcomeMessage,
|
||||
type ServerInfoStatusPayload,
|
||||
WSOutboundMessageSchema,
|
||||
} from '../shared/messages.js'
|
||||
import type {
|
||||
@@ -450,7 +451,7 @@ export class DaemonClient {
|
||||
private readonly logServerId: string | null
|
||||
private readonly logClientIdHash: string
|
||||
private readonly logGeneration: number | null
|
||||
private lastWelcomeMessage: WSWelcomeMessage | null = null
|
||||
private lastServerInfoMessage: ServerInfoStatusPayload | null = null
|
||||
|
||||
constructor(private config: DaemonClientConfig) {
|
||||
this.logger = config.logger ?? consoleLogger
|
||||
@@ -565,7 +566,7 @@ export class DaemonClient {
|
||||
const transportUrl = this.resolveTransportUrlForAttempt()
|
||||
const transport = transportFactory({ url: transportUrl, headers })
|
||||
this.transport = transport
|
||||
this.lastWelcomeMessage = null
|
||||
this.lastServerInfoMessage = null
|
||||
|
||||
this.updateConnectionState({
|
||||
status: 'connecting',
|
||||
@@ -700,7 +701,7 @@ export class DaemonClient {
|
||||
this.clearWaiters(new Error('Daemon client closed'))
|
||||
this.rejectPendingSendQueue(new Error('Daemon client closed'))
|
||||
this.terminalStreams.clearAll()
|
||||
this.lastWelcomeMessage = null
|
||||
this.lastServerInfoMessage = null
|
||||
this.updateConnectionState(
|
||||
{ status: 'disposed' },
|
||||
{ event: 'DISPOSE', reason: 'Client closed', reasonCode: 'disposed' }
|
||||
@@ -2796,8 +2797,8 @@ export class DaemonClient {
|
||||
return requestId ?? crypto.randomUUID()
|
||||
}
|
||||
|
||||
getLastWelcomeMessage(): WSWelcomeMessage | null {
|
||||
return this.lastWelcomeMessage
|
||||
getLastServerInfoMessage(): ServerInfoStatusPayload | null {
|
||||
return this.lastServerInfoMessage
|
||||
}
|
||||
|
||||
private resolveTransportUrlForAttempt(): string {
|
||||
@@ -2962,18 +2963,6 @@ export class DaemonClient {
|
||||
return
|
||||
}
|
||||
|
||||
if (parsed.data.type === 'welcome') {
|
||||
this.lastWelcomeMessage = parsed.data
|
||||
this.resetConnectTimeout()
|
||||
this.reconnectAttempt = 0
|
||||
this.updateConnectionState({ status: 'connected' }, { event: 'HELLO_WELCOME' })
|
||||
this.resubscribeCheckoutDiffSubscriptions()
|
||||
this.resubscribeTerminalDirectorySubscriptions()
|
||||
this.flushPendingSendQueue()
|
||||
this.resolveConnect()
|
||||
return
|
||||
}
|
||||
|
||||
this.handleSessionMessage(parsed.data.message)
|
||||
}
|
||||
|
||||
@@ -3050,7 +3039,7 @@ export class DaemonClient {
|
||||
this.clearWaiters(new Error(reason ?? 'Connection lost'))
|
||||
this.rejectPendingSendQueue(new Error(reason ?? 'Connection lost'))
|
||||
this.terminalStreams.clearAll()
|
||||
this.lastWelcomeMessage = null
|
||||
this.lastServerInfoMessage = null
|
||||
|
||||
if (wasDisposed) {
|
||||
this.rejectConnect(new Error(reason ?? 'Daemon client is disposed'))
|
||||
@@ -3084,6 +3073,22 @@ export class DaemonClient {
|
||||
}
|
||||
|
||||
private handleSessionMessage(msg: SessionOutboundMessage): void {
|
||||
if (msg.type === 'status') {
|
||||
const serverInfo = parseServerInfoStatusPayload(msg.payload)
|
||||
if (serverInfo) {
|
||||
this.lastServerInfoMessage = serverInfo
|
||||
if (this.connectionState.status === 'connecting') {
|
||||
this.resetConnectTimeout()
|
||||
this.reconnectAttempt = 0
|
||||
this.updateConnectionState({ status: 'connected' }, { event: 'HELLO_SERVER_INFO' })
|
||||
this.resubscribeCheckoutDiffSubscriptions()
|
||||
this.resubscribeTerminalDirectorySubscriptions()
|
||||
this.flushPendingSendQueue()
|
||||
this.resolveConnect()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (msg.type === 'terminal_stream_exit') {
|
||||
this.terminalStreams.clearStream({ streamId: msg.payload.streamId })
|
||||
}
|
||||
|
||||
@@ -335,17 +335,16 @@ describe("daemon client E2E", () => {
|
||||
}
|
||||
}, 30000);
|
||||
|
||||
test("receives welcome on websocket connect", async () => {
|
||||
test("receives server_info on websocket connect", async () => {
|
||||
const client = new DaemonClient({
|
||||
url: `ws://127.0.0.1:${ctx.daemon.port}/ws`,
|
||||
clientId: `cid-e2e-${randomUUID()}`,
|
||||
clientType: "cli",
|
||||
});
|
||||
await client.connect();
|
||||
const welcome = client.getLastWelcomeMessage();
|
||||
expect(welcome).not.toBeNull();
|
||||
expect(welcome?.serverId.length).toBeGreaterThan(0);
|
||||
expect(typeof welcome?.resumed).toBe("boolean");
|
||||
const serverInfo = client.getLastServerInfoMessage();
|
||||
expect(serverInfo).not.toBeNull();
|
||||
expect(serverInfo?.serverId.length).toBeGreaterThan(0);
|
||||
|
||||
await client.close();
|
||||
}, 15000);
|
||||
@@ -369,8 +368,8 @@ describe("daemon client E2E", () => {
|
||||
|
||||
try {
|
||||
await client.connect();
|
||||
const welcome = client.getLastWelcomeMessage();
|
||||
const voice = welcome?.capabilities?.voice;
|
||||
const serverInfo = client.getLastServerInfoMessage();
|
||||
const voice = serverInfo?.capabilities?.voice;
|
||||
expect(voice).toBeTruthy();
|
||||
|
||||
expect(voice?.dictation.enabled).toBe(false);
|
||||
|
||||
@@ -243,7 +243,13 @@ async function waitForRelayWebSocketReady(port: number, timeout = 60000): Promis
|
||||
try {
|
||||
const payload =
|
||||
typeof data === "string" ? JSON.parse(data) : data;
|
||||
if (payload && typeof payload === "object" && (payload as any).type === "welcome") {
|
||||
if (
|
||||
payload &&
|
||||
typeof payload === "object" &&
|
||||
(payload as any).type === "session" &&
|
||||
(payload as any).message?.type === "status" &&
|
||||
(payload as any).message?.payload?.status === "server_info"
|
||||
) {
|
||||
if (!pingSent && channelRef) {
|
||||
pingSent = true;
|
||||
void channelRef.send(JSON.stringify({ type: "ping" }));
|
||||
@@ -365,7 +371,13 @@ async function waitForRelayWebSocketReady(port: number, timeout = 60000): Promis
|
||||
const channel = await createClientChannel(transport, daemonPublicKeyB64, {
|
||||
onmessage: (data) => {
|
||||
const payload = typeof data === "string" ? JSON.parse(data) : data;
|
||||
if (payload && typeof payload === "object" && (payload as any).type === "welcome") {
|
||||
if (
|
||||
payload &&
|
||||
typeof payload === "object" &&
|
||||
(payload as any).type === "session" &&
|
||||
(payload as any).message?.type === "status" &&
|
||||
(payload as any).message?.payload?.status === "server_info"
|
||||
) {
|
||||
if (!pingSent && channelRef) {
|
||||
pingSent = true;
|
||||
void channelRef.send(JSON.stringify({ type: "ping" }));
|
||||
|
||||
@@ -459,7 +459,7 @@ const shouldRun = !process.env.CI;
|
||||
const helloReady = new Promise<void>((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
cleanup();
|
||||
reject(new Error("Timed out waiting for websocket welcome"));
|
||||
reject(new Error("Timed out waiting for initial websocket server_info"));
|
||||
}, 10000);
|
||||
|
||||
const onError = (error: Error) => {
|
||||
@@ -478,8 +478,15 @@ const shouldRun = !process.env.CI;
|
||||
}
|
||||
const text = typeof raw === "string" ? raw : raw.toString("utf8");
|
||||
try {
|
||||
const parsed = JSON.parse(text) as { type?: string };
|
||||
if (parsed.type === "welcome") {
|
||||
const parsed = JSON.parse(text) as {
|
||||
type?: string;
|
||||
message?: { type?: string; payload?: { status?: string } };
|
||||
};
|
||||
if (
|
||||
parsed.type === "session" &&
|
||||
parsed.message?.type === "status" &&
|
||||
parsed.message.payload?.status === "server_info"
|
||||
) {
|
||||
cleanup();
|
||||
resolve();
|
||||
}
|
||||
|
||||
@@ -282,13 +282,15 @@ async function attachRelayAndHello(params: {
|
||||
params.socket.emit("message", JSON.stringify(createHelloMessage(params.clientId)));
|
||||
await Promise.resolve();
|
||||
expect(params.socket.sent.length).toBeGreaterThan(0);
|
||||
const welcome = JSON.parse(params.socket.sent[0] as string) as {
|
||||
const envelope = JSON.parse(params.socket.sent[0] as string) as {
|
||||
type?: unknown;
|
||||
resumed?: unknown;
|
||||
serverId?: unknown;
|
||||
message?: { type?: unknown; payload?: unknown };
|
||||
};
|
||||
expect(welcome.type).toBe("welcome");
|
||||
return welcome;
|
||||
expect(envelope.type).toBe("session");
|
||||
const serverInfo = parseServerInfoStatusPayload(envelope.message?.payload);
|
||||
expect(envelope.message?.type).toBe("status");
|
||||
expect(serverInfo).not.toBeNull();
|
||||
return serverInfo!;
|
||||
}
|
||||
|
||||
async function attachDirectAndHello(params: {
|
||||
@@ -300,13 +302,15 @@ async function attachDirectAndHello(params: {
|
||||
params.socket.emit("message", JSON.stringify(createHelloMessage(params.clientId)));
|
||||
await Promise.resolve();
|
||||
expect(params.socket.sent.length).toBeGreaterThan(0);
|
||||
const welcome = JSON.parse(params.socket.sent[0] as string) as {
|
||||
const envelope = JSON.parse(params.socket.sent[0] as string) as {
|
||||
type?: unknown;
|
||||
resumed?: unknown;
|
||||
serverId?: unknown;
|
||||
message?: { type?: unknown; payload?: unknown };
|
||||
};
|
||||
expect(welcome.type).toBe("welcome");
|
||||
return welcome;
|
||||
expect(envelope.type).toBe("session");
|
||||
const serverInfo = parseServerInfoStatusPayload(envelope.message?.payload);
|
||||
expect(envelope.message?.type).toBe("status");
|
||||
expect(serverInfo).not.toBeNull();
|
||||
return serverInfo!;
|
||||
}
|
||||
|
||||
describe("relay external socket reconnect behavior", () => {
|
||||
@@ -324,12 +328,11 @@ describe("relay external socket reconnect behavior", () => {
|
||||
const clientId = "cid-relay-reconnect";
|
||||
|
||||
const socket1 = new MockSocket();
|
||||
const firstWelcome = await attachRelayAndHello({
|
||||
await attachRelayAndHello({
|
||||
server,
|
||||
socket: socket1,
|
||||
clientId,
|
||||
});
|
||||
expect(firstWelcome.resumed).toBe(false);
|
||||
expect(sessionMock.instances).toHaveLength(1);
|
||||
const session = sessionMock.instances[0]!;
|
||||
|
||||
@@ -338,12 +341,11 @@ describe("relay external socket reconnect behavior", () => {
|
||||
expect(session.cleanup).not.toHaveBeenCalled();
|
||||
|
||||
const socket2 = new MockSocket();
|
||||
const secondWelcome = await attachRelayAndHello({
|
||||
await attachRelayAndHello({
|
||||
server,
|
||||
socket: socket2,
|
||||
clientId,
|
||||
});
|
||||
expect(secondWelcome.resumed).toBe(true);
|
||||
expect(sessionMock.instances).toHaveLength(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
@@ -373,50 +375,46 @@ describe("relay external socket reconnect behavior", () => {
|
||||
await server.close();
|
||||
});
|
||||
|
||||
test("marks hello as resumed when clientId already has a session", async () => {
|
||||
test("returns server_info when clientId reconnects with existing session", async () => {
|
||||
const server = createServer();
|
||||
const clientId = "cid-resume-flag";
|
||||
|
||||
const firstSocket = new MockSocket();
|
||||
const firstWelcome = await attachRelayAndHello({
|
||||
await attachRelayAndHello({
|
||||
server,
|
||||
socket: firstSocket,
|
||||
clientId,
|
||||
});
|
||||
expect(firstWelcome.resumed).toBe(false);
|
||||
|
||||
firstSocket.emit("close", 1006, "");
|
||||
await vi.advanceTimersByTimeAsync(1_000);
|
||||
|
||||
const secondSocket = new MockSocket();
|
||||
const secondWelcome = await attachRelayAndHello({
|
||||
await attachRelayAndHello({
|
||||
server,
|
||||
socket: secondSocket,
|
||||
clientId,
|
||||
});
|
||||
expect(secondWelcome.resumed).toBe(true);
|
||||
|
||||
await server.close();
|
||||
});
|
||||
|
||||
test("marks hello as not resumed for new clientIds", async () => {
|
||||
test("returns server_info for distinct clientIds", async () => {
|
||||
const server = createServer();
|
||||
|
||||
const firstSocket = new MockSocket();
|
||||
const firstWelcome = await attachRelayAndHello({
|
||||
await attachRelayAndHello({
|
||||
server,
|
||||
socket: firstSocket,
|
||||
clientId: "cid-new-1",
|
||||
});
|
||||
expect(firstWelcome.resumed).toBe(false);
|
||||
|
||||
const secondSocket = new MockSocket();
|
||||
const secondWelcome = await attachRelayAndHello({
|
||||
await attachRelayAndHello({
|
||||
server,
|
||||
socket: secondSocket,
|
||||
clientId: "cid-new-2",
|
||||
});
|
||||
expect(secondWelcome.resumed).toBe(false);
|
||||
expect(sessionMock.instances).toHaveLength(2);
|
||||
|
||||
await server.close();
|
||||
@@ -456,12 +454,11 @@ describe("relay external socket reconnect behavior", () => {
|
||||
const clientId = "cid-direct-reconnect";
|
||||
|
||||
const socket1 = new MockSocket();
|
||||
const firstWelcome = await attachDirectAndHello({
|
||||
await attachDirectAndHello({
|
||||
server,
|
||||
socket: socket1,
|
||||
clientId,
|
||||
});
|
||||
expect(firstWelcome.resumed).toBe(false);
|
||||
expect(sessionMock.instances).toHaveLength(1);
|
||||
const session = sessionMock.instances[0]!;
|
||||
|
||||
@@ -470,12 +467,11 @@ describe("relay external socket reconnect behavior", () => {
|
||||
expect(session.cleanup).not.toHaveBeenCalled();
|
||||
|
||||
const socket2 = new MockSocket();
|
||||
const secondWelcome = await attachDirectAndHello({
|
||||
await attachDirectAndHello({
|
||||
server,
|
||||
socket: socket2,
|
||||
clientId,
|
||||
});
|
||||
expect(secondWelcome.resumed).toBe(true);
|
||||
expect(sessionMock.instances).toHaveLength(1);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
@@ -548,12 +544,12 @@ describe("relay external socket reconnect behavior", () => {
|
||||
await server.close();
|
||||
});
|
||||
|
||||
test("includes voice capabilities in welcome when speech readiness exists", async () => {
|
||||
test("includes voice capabilities in initial server_info when speech readiness exists", async () => {
|
||||
const speechReadiness = createReadySpeechReadinessSnapshot();
|
||||
const server = createServer({ speechReadiness });
|
||||
|
||||
const socket = new MockSocket();
|
||||
const welcome = await attachRelayAndHello({
|
||||
const serverInfo = await attachRelayAndHello({
|
||||
server,
|
||||
socket,
|
||||
clientId: "cid-server-info-capabilities",
|
||||
@@ -566,15 +562,15 @@ describe("relay external socket reconnect behavior", () => {
|
||||
};
|
||||
};
|
||||
};
|
||||
expect(welcome.version).toBe(TEST_DAEMON_VERSION);
|
||||
expect(welcome.capabilities?.voice?.dictation?.enabled).toBe(
|
||||
expect(serverInfo.version).toBe(TEST_DAEMON_VERSION);
|
||||
expect(serverInfo.capabilities?.voice?.dictation?.enabled).toBe(
|
||||
speechReadiness.dictation.enabled
|
||||
);
|
||||
expect(welcome.capabilities?.voice?.dictation?.reason).toBe("");
|
||||
expect(welcome.capabilities?.voice?.voice?.enabled).toBe(
|
||||
expect(serverInfo.capabilities?.voice?.dictation?.reason).toBe("");
|
||||
expect(serverInfo.capabilities?.voice?.voice?.enabled).toBe(
|
||||
speechReadiness.realtimeVoice.enabled
|
||||
);
|
||||
expect(welcome.capabilities?.voice?.voice?.reason).toBe("");
|
||||
expect(serverInfo.capabilities?.voice?.voice?.reason).toBe("");
|
||||
|
||||
await server.close();
|
||||
});
|
||||
|
||||
@@ -12,7 +12,6 @@ import type { ProjectRegistry, WorkspaceRegistry } from "./workspace-registry.js
|
||||
import {
|
||||
type ServerInfoStatusPayload,
|
||||
type WSHelloMessage,
|
||||
type WSWelcomeMessage,
|
||||
WSInboundMessageSchema,
|
||||
type ServerCapabilityState,
|
||||
type ServerCapabilities,
|
||||
@@ -678,17 +677,6 @@ export class VoiceAssistantWebSocketServer {
|
||||
return pending;
|
||||
}
|
||||
|
||||
private buildWelcomeMessage(params: { resumed: boolean }): WSWelcomeMessage {
|
||||
return {
|
||||
type: "welcome",
|
||||
serverId: this.serverId,
|
||||
hostname: getHostname(),
|
||||
version: this.daemonVersion,
|
||||
resumed: params.resumed,
|
||||
...(this.serverCapabilities ? { capabilities: this.serverCapabilities } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
private handleHello(params: {
|
||||
ws: WebSocketLike;
|
||||
message: WSHelloMessage;
|
||||
@@ -735,7 +723,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
}
|
||||
existing.sockets.add(ws);
|
||||
this.sessions.set(ws, existing);
|
||||
this.sendToClient(ws, this.buildWelcomeMessage({ resumed: true }));
|
||||
this.sendToClient(ws, this.createServerInfoMessage());
|
||||
existing.connectionLogger.trace(
|
||||
{
|
||||
clientId,
|
||||
@@ -756,7 +744,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
});
|
||||
this.sessions.set(ws, connection);
|
||||
this.externalSessionsByKey.set(clientId, connection);
|
||||
this.sendToClient(ws, this.buildWelcomeMessage({ resumed: false }));
|
||||
this.sendToClient(ws, this.createServerInfoMessage());
|
||||
connection.connectionLogger.trace(
|
||||
{
|
||||
clientId,
|
||||
@@ -777,13 +765,18 @@ export class VoiceAssistantWebSocketServer {
|
||||
};
|
||||
}
|
||||
|
||||
private broadcastCapabilitiesUpdate(): void {
|
||||
this.broadcast(
|
||||
wrapSessionMessage({
|
||||
private createServerInfoMessage(): WSOutboundMessage {
|
||||
return {
|
||||
type: "session",
|
||||
message: {
|
||||
type: "status",
|
||||
payload: this.buildServerInfoStatusPayload(),
|
||||
})
|
||||
);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
private broadcastCapabilitiesUpdate(): void {
|
||||
this.broadcast(this.createServerInfoMessage());
|
||||
}
|
||||
|
||||
private bindSocketHandlers(ws: WebSocketLike): void {
|
||||
|
||||
@@ -2388,15 +2388,6 @@ export const WSHelloMessageSchema = z.object({
|
||||
.optional(),
|
||||
})
|
||||
|
||||
export const WSWelcomeMessageSchema = z.object({
|
||||
type: z.literal('welcome'),
|
||||
serverId: z.string(),
|
||||
hostname: z.string().nullable(),
|
||||
version: z.string().nullable(),
|
||||
resumed: z.boolean(),
|
||||
capabilities: ServerCapabilitiesSchema.optional(),
|
||||
})
|
||||
|
||||
export const WSRecordingStateMessageSchema = z.object({
|
||||
type: z.literal('recording_state'),
|
||||
isRecording: z.boolean(),
|
||||
@@ -2423,14 +2414,12 @@ export const WSInboundMessageSchema = z.discriminatedUnion('type', [
|
||||
|
||||
export const WSOutboundMessageSchema = z.discriminatedUnion('type', [
|
||||
WSPongMessageSchema,
|
||||
WSWelcomeMessageSchema,
|
||||
WSSessionOutboundSchema,
|
||||
])
|
||||
|
||||
export type WSInboundMessage = z.infer<typeof WSInboundMessageSchema>
|
||||
export type WSOutboundMessage = z.infer<typeof WSOutboundMessageSchema>
|
||||
export type WSHelloMessage = z.infer<typeof WSHelloMessageSchema>
|
||||
export type WSWelcomeMessage = z.infer<typeof WSWelcomeMessageSchema>
|
||||
|
||||
// ============================================================================
|
||||
// Helper functions for message conversion
|
||||
|
||||
Reference in New Issue
Block a user