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

feat: implement daemon connectivity updates and relay protocol versioning

Browse Source
This commit is contained in:
Mohamed Boudra
2026-02-20 19:27:56 +07:00
parent 6dad3212da
commit 0350f00840
49 changed files with 4146 additions and 1351 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/client/daemon-client.test.ts
+207 -33
View File
@@ -83,7 +83,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -166,12 +166,177 @@ describe('DaemonClient', () => {
})
})
test('does not reconnect after close when ensureConnected is called', async () => {
const logger = createMockLogger()
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
})
clients.push(client)
const connectPromise = client.connect()
mock.triggerOpen()
await connectPromise
expect(client.getConnectionState().status).toBe('connected')
await client.close()
expect(client.getConnectionState().status).toBe('disposed')
client.ensureConnected()
expect(client.getConnectionState().status).toBe('disposed')
})
test('transitions out of connecting when connect timeout elapses', async () => {
vi.useFakeTimers()
try {
const logger = createMockLogger()
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
connectTimeoutMs: 100,
transportFactory: () => mock.transport,
})
clients.push(client)
const pendingConnect = client.connect().then(
() => ({ ok: true as const }),
(error) => ({ ok: false as const, error })
)
expect(client.getConnectionState().status).toBe('connecting')
await vi.advanceTimersByTimeAsync(120)
const result = await pendingConnect
expect(result.ok).toBe(false)
if (!result.ok) {
expect(result.error).toBeInstanceOf(Error)
expect((result.error as Error).message).toContain('Connection timed out')
}
expect(client.getConnectionState().status).toBe('disconnected')
} finally {
vi.useRealTimers()
}
})
test('reconnects after relay close with replaced-by-new-connection reason', async () => {
vi.useFakeTimers()
try {
const logger = createMockLogger()
const first = createMockTransport()
const second = createMockTransport()
const transports = [first, second]
let transportIndex = 0
const client = new DaemonClient({
url: 'ws://relay.test/ws?role=client&serverId=srv_test&clientId=clsk_test&v=2',
logger,
reconnect: {
enabled: true,
baseDelayMs: 5,
maxDelayMs: 5,
},
transportFactory: () => {
const next = transports[Math.min(transportIndex, transports.length - 1)]
transportIndex += 1
return next.transport
},
})
clients.push(client)
const connectPromise = client.connect()
first.triggerOpen()
await connectPromise
expect(client.getConnectionState().status).toBe('connected')
first.triggerClose({ code: 1008, reason: 'Replaced by new connection' })
expect(client.getConnectionState().status).toBe('disconnected')
await vi.advanceTimersByTimeAsync(10)
expect(client.getConnectionState().status).toBe('connecting')
second.triggerOpen()
expect(client.getConnectionState().status).toBe('connected')
} finally {
vi.useRealTimers()
}
})
test('requires explicit relay session identity when URL clientId is missing', () => {
expect(() => {
new DaemonClient({
url: 'ws://relay.test/ws?role=client&serverId=srv_test&v=2',
reconnect: { enabled: false },
})
}).toThrow('Relay client requires clientSessionKey or URL clientId')
})
test('rejects relay identity mismatch between URL clientId and clientSessionKey', () => {
expect(() => {
new DaemonClient({
url: 'ws://relay.test/ws?role=client&serverId=srv_test&clientId=clsk_a&v=2',
clientSessionKey: 'clsk_b',
reconnect: { enabled: false },
})
}).toThrow('Relay clientId and clientSessionKey must match when both are provided')
})
test('requires explicit direct session identity when URL clientSessionKey is missing', () => {
expect(() => {
new DaemonClient({
url: 'ws://127.0.0.1:6767/ws',
reconnect: { enabled: false },
})
}).toThrow('Direct client requires clientSessionKey or URL clientSessionKey')
})
test('rejects direct identity mismatch between URL and config clientSessionKey', () => {
expect(() => {
new DaemonClient({
url: 'ws://127.0.0.1:6767/ws?clientSessionKey=clsk_a',
clientSessionKey: 'clsk_b',
reconnect: { enabled: false },
})
}).toThrow('Direct URL clientSessionKey and config clientSessionKey must match when both are provided')
})
test('logs configured runtime generation in connection transition events', async () => {
const logger = createMockLogger()
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
runtimeGeneration: 7,
transportFactory: () => mock.transport,
})
clients.push(client)
const connectPromise = client.connect()
mock.triggerOpen()
await connectPromise
const transitionPayloads = logger.info.mock.calls
.filter(([, message]) => message === 'DaemonClientTransition')
.map(([payload]) => payload as { generation?: number | null })
expect(transitionPayloads.length).toBeGreaterThan(0)
for (const payload of transitionPayloads) {
expect(payload.generation).toBe(7)
}
})
test('subscribes to checkout diff updates via RPC handshake', async () => {
const logger = createMockLogger()
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -234,7 +399,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -302,7 +467,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -361,7 +526,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -432,7 +597,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -494,7 +659,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -540,7 +705,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -621,7 +786,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -664,7 +829,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -716,7 +881,7 @@ describe('DaemonClient', () => {
})
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory,
@@ -743,7 +908,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -797,7 +962,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -872,7 +1037,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -928,7 +1093,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -974,7 +1139,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -1016,7 +1181,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -1037,7 +1202,12 @@ describe('DaemonClient', () => {
})
)
expect(mock.sent).toHaveLength(0)
expect(mock.sent).toHaveLength(1)
const bufferedAck = decodeBinaryMuxFrame(asUint8Array(mock.sent[0])!)
expect(bufferedAck?.messageType).toBe(TerminalBinaryMessageType.Ack)
expect(bufferedAck?.streamId).toBe(19)
expect(bufferedAck?.offset).toBe(8)
mock.sent.length = 0
const seen: string[] = []
client.onTerminalStreamData(19, (chunk) => {
@@ -1045,11 +1215,7 @@ describe('DaemonClient', () => {
})
expect(seen.join('')).toContain('buffered')
expect(mock.sent).toHaveLength(1)
const ackFrame = decodeBinaryMuxFrame(asUint8Array(mock.sent[0])!)
expect(ackFrame?.messageType).toBe(TerminalBinaryMessageType.Ack)
expect(ackFrame?.streamId).toBe(19)
expect(ackFrame?.offset).toBe(8)
expect(mock.sent).toHaveLength(0)
})
test('stops delivering and acking after terminal_stream_exit', async () => {
@@ -1057,7 +1223,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -1106,7 +1272,11 @@ describe('DaemonClient', () => {
)
expect(seen).toEqual(['before-exit'])
expect(mock.sent).toHaveLength(0)
expect(mock.sent).toHaveLength(1)
const postExitAck = decodeBinaryMuxFrame(asUint8Array(mock.sent[0])!)
expect(postExitAck?.messageType).toBe(TerminalBinaryMessageType.Ack)
expect(postExitAck?.streamId).toBe(23)
expect(postExitAck?.offset).toBe(22)
unsubscribe()
})
@@ -1115,7 +1285,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -1165,7 +1335,11 @@ describe('DaemonClient', () => {
)
expect(seen).toEqual([])
expect(mock.sent).toHaveLength(0)
expect(mock.sent).toHaveLength(1)
const detachedAck = decodeBinaryMuxFrame(asUint8Array(mock.sent[0])!)
expect(detachedAck?.messageType).toBe(TerminalBinaryMessageType.Ack)
expect(detachedAck?.streamId).toBe(31)
expect(detachedAck?.offset).toBe(12)
})
test('parses canonical agent_stream tool_call payloads without crashing', async () => {
@@ -1173,7 +1347,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -1244,7 +1418,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -1296,7 +1470,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -1388,7 +1562,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -1458,7 +1632,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
@@ -1494,7 +1668,7 @@ describe('DaemonClient', () => {
const mock = createMockTransport()
const client = new DaemonClient({
url: 'ws://test',
url: 'ws://test?clientSessionKey=clsk_unit_test',
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
packages/server/src/client/daemon-client.ts
+222 -69
View File
@@ -77,7 +77,6 @@ import {
describeTransportClose,
describeTransportError,
encodeUtf8String,
safeRandomId,
type DaemonTransport,
type DaemonTransportFactory,
type WebSocketFactory,
@@ -110,6 +109,7 @@ export type ConnectionState =
| { status: 'connecting'; attempt: number }
| { status: 'connected' }
| { status: 'disconnected'; reason?: string }
| { status: 'disposed' }
export type DaemonEvent =
| {
@@ -144,11 +144,14 @@ export type DaemonEventHandler = (event: DaemonEvent) => void
export type DaemonClientConfig = {
url: string
clientSessionKey?: string
runtimeGeneration?: number | null
authHeader?: string
suppressSendErrors?: boolean
transportFactory?: DaemonTransportFactory
webSocketFactory?: WebSocketFactory
logger?: Logger
connectTimeoutMs?: number
e2ee?: {
enabled?: boolean
daemonPublicKeyB64?: string
@@ -297,6 +300,7 @@ class DaemonRpcError extends Error {
const DEFAULT_RECONNECT_BASE_DELAY_MS = 1500
const DEFAULT_RECONNECT_MAX_DELAY_MS = 30000
const DEFAULT_CONNECT_TIMEOUT_MS = 15000
/** Default timeout for waiting for connection before sending queued messages */
const DEFAULT_SEND_QUEUE_TIMEOUT_MS = 10000
@@ -308,6 +312,45 @@ function isWaiterTimeoutError(error: unknown): boolean {
return error instanceof Error && error.message.startsWith('Timeout waiting for message')
}
function normalizeClientSessionKey(value: unknown): string | null {
if (typeof value !== 'string') {
return null
}
const trimmed = value.trim()
return trimmed.length > 0 ? trimmed : null
}
function hashForLog(value: string): string {
let hash = 0
for (let index = 0; index < value.length; index += 1) {
hash = (hash * 31 + value.charCodeAt(index)) | 0
}
return `h_${Math.abs(hash).toString(16)}`
}
function toReasonCode(reason: string | null | undefined): string | null {
if (!reason) {
return null
}
const normalized = reason.toLowerCase()
if (normalized.includes('timed out')) {
return 'connect_timeout'
}
if (normalized.includes('disposed')) {
return 'disposed'
}
if (normalized.includes('client closed')) {
return 'client_closed'
}
if (normalized.includes('transport')) {
return 'transport_error'
}
if (normalized.includes('failed to connect')) {
return 'connect_failed'
}
return 'unknown'
}
interface PendingSend {
message: SessionInboundMessage
resolve: () => void
@@ -328,6 +371,7 @@ export class DaemonClient {
private checkoutStatusInFlight: Map<string, Promise<CheckoutStatusPayload>> = new Map()
private connectionListeners: Set<(status: ConnectionState) => void> = new Set()
private reconnectTimeout: ReturnType<typeof setTimeout> | null = null
private connectTimeout: ReturnType<typeof setTimeout> | null = null
private pendingGenericTransportErrorTimeout: ReturnType<typeof setTimeout> | null = null
private reconnectAttempt = 0
private shouldReconnect = true
@@ -343,11 +387,30 @@ export class DaemonClient {
private terminalDirectorySubscriptions = new Set<string>()
private logger: Logger
private pendingSendQueue: PendingSend[] = []
private relayClientId: string | null = null
private terminalStreams: TerminalStreamManager
private readonly logConnectionPath: 'direct' | 'relay'
private readonly logServerId: string | null
private readonly logClientSessionKeyHash: string
private readonly logGeneration: number | null
constructor(private config: DaemonClientConfig) {
this.logger = config.logger ?? consoleLogger
this.logConnectionPath = isRelayClientWebSocketUrl(this.config.url) ? 'relay' : 'direct'
let parsedUrlForLog: URL | null = null
try {
parsedUrlForLog = new URL(this.config.url)
} catch {
parsedUrlForLog = null
}
const parsedServerIdForLog = normalizeClientSessionKey(
parsedUrlForLog?.searchParams.get('serverId')
)
this.logServerId = parsedServerIdForLog ?? parsedUrlForLog?.host ?? null
this.logClientSessionKeyHash = 'h_unresolved'
this.logGeneration =
typeof this.config.runtimeGeneration === 'number' && Number.isFinite(this.config.runtimeGeneration)
? this.config.runtimeGeneration
: null
this.terminalStreams = new TerminalStreamManager({
sendAck: (ack) => {
this.sendBinaryFrame({
@@ -359,20 +422,47 @@ export class DaemonClient {
})
},
})
const configClientSessionKey = normalizeClientSessionKey(this.config.clientSessionKey)
let parsed: URL | null = null
try {
parsed = new URL(this.config.url)
} catch {
// ignore - invalid URL will be handled on connect
}
// Relay requires a clientId so the daemon can create an independent
// socket + E2EE channel per connected client. Generate one per DaemonClient
// instance (stable across reconnects in this tab/app session).
// socket + E2EE channel per connected client.
if (isRelayClientWebSocketUrl(this.config.url)) {
try {
const parsed = new URL(this.config.url)
if (!parsed.searchParams.get('clientId')) {
this.relayClientId = `clt_${safeRandomId()}`
parsed.searchParams.set('clientId', this.relayClientId)
this.config.url = parsed.toString()
}
} catch {
// ignore - invalid URL will be handled on connect
const urlClientId = normalizeClientSessionKey(parsed?.searchParams.get('clientId'))
if (urlClientId && configClientSessionKey && configClientSessionKey !== urlClientId) {
throw new Error('Relay clientId and clientSessionKey must match when both are provided')
}
const resolvedClientSessionKey = configClientSessionKey ?? urlClientId
if (!resolvedClientSessionKey) {
throw new Error('Relay client requires clientSessionKey or URL clientId')
}
this.config.clientSessionKey = resolvedClientSessionKey
this.logClientSessionKeyHash = hashForLog(resolvedClientSessionKey)
if (parsed && !urlClientId) {
parsed.searchParams.set('clientId', resolvedClientSessionKey)
this.config.url = parsed.toString()
}
return
}
const urlClientSessionKey = normalizeClientSessionKey(parsed?.searchParams.get('clientSessionKey'))
if (urlClientSessionKey && configClientSessionKey && configClientSessionKey !== urlClientSessionKey) {
throw new Error('Direct URL clientSessionKey and config clientSessionKey must match when both are provided')
}
const resolvedClientSessionKey = configClientSessionKey ?? urlClientSessionKey
if (!resolvedClientSessionKey) {
throw new Error('Direct client requires clientSessionKey or URL clientSessionKey')
}
this.config.clientSessionKey = resolvedClientSessionKey
this.logClientSessionKeyHash = hashForLog(resolvedClientSessionKey)
if (parsed && !urlClientSessionKey) {
parsed.searchParams.set('clientSessionKey', resolvedClientSessionKey)
this.config.url = parsed.toString()
}
}
@@ -381,6 +471,9 @@ export class DaemonClient {
// ============================================================================
async connect(): Promise<void> {
if (this.connectionState.status === 'disposed') {
throw new Error('Daemon client is disposed')
}
if (this.connectionState.status === 'connected') {
return
}
@@ -399,6 +492,10 @@ export class DaemonClient {
}
private attemptConnect(): void {
if (this.connectionState.status === 'disposed') {
this.rejectConnect(new Error('Daemon client is disposed'))
return
}
if (!this.shouldReconnect) {
this.rejectConnect(new Error('Daemon client is closed'))
return
@@ -414,10 +511,8 @@ export class DaemonClient {
}
try {
// If we reconnect while the previous socket is still open (common in browsers
// where `onerror` may fire before `onclose`), we can end up with multiple
// concurrent relay sockets. Cloudflare then closes the old one with
// "Replaced by new connection", causing a disconnect loop.
// Reconnect can overlap with browser close/error delivery ordering.
// Always dispose previous transport before constructing the next one.
this.disposeTransport()
const baseTransportFactory =
this.config.transportFactory ??
@@ -443,62 +538,55 @@ export class DaemonClient {
this.updateConnectionState({
status: 'connecting',
attempt: this.reconnectAttempt,
})
}, { event: 'CONNECT_REQUEST' })
this.resetConnectTimeout()
const timeoutMs = Math.max(1, this.config.connectTimeoutMs ?? DEFAULT_CONNECT_TIMEOUT_MS)
this.connectTimeout = setTimeout(() => {
if (this.connectionState.status !== 'connecting') {
return
}
this.lastErrorValue = 'Connection timed out'
this.disposeTransport(1001, 'Connection timed out')
this.scheduleReconnect({
reason: 'Connection timed out',
event: 'CONNECT_TIMEOUT',
reasonCode: 'connect_timeout',
})
}, timeoutMs)
this.transportCleanup = [
transport.onOpen(() => {
this.resetConnectTimeout()
if (this.pendingGenericTransportErrorTimeout) {
clearTimeout(this.pendingGenericTransportErrorTimeout)
this.pendingGenericTransportErrorTimeout = null
}
this.lastErrorValue = null
this.reconnectAttempt = 0
this.updateConnectionState({ status: 'connected' })
this.updateConnectionState({ status: 'connected' }, { event: 'TRANSPORT_OPEN' })
this.resubscribeCheckoutDiffSubscriptions()
this.resubscribeTerminalDirectorySubscriptions()
this.flushPendingSendQueue()
this.resolveConnect()
}),
transport.onClose((event) => {
this.resetConnectTimeout()
if (this.pendingGenericTransportErrorTimeout) {
clearTimeout(this.pendingGenericTransportErrorTimeout)
this.pendingGenericTransportErrorTimeout = null
}
const closeRecord = event as { code?: unknown; reason?: unknown } | null
const closeCode =
closeRecord && typeof closeRecord === 'object' && typeof closeRecord.code === 'number'
? closeRecord.code
: null
const closeReason =
closeRecord && typeof closeRecord === 'object' && typeof closeRecord.reason === 'string'
? closeRecord.reason
: null
const reason = describeTransportClose(event)
if (reason) {
this.lastErrorValue = reason
}
this.updateConnectionState({
status: 'disconnected',
...(reason ? { reason } : {}),
this.scheduleReconnect({
reason,
event: 'TRANSPORT_CLOSE',
reasonCode: 'transport_closed',
})
// When connecting over the relay, only one client connection is allowed at a time.
// If another device/tab takes over, we should not auto-reconnect and "fight" the new
// connection (which causes flapping where both sides repeatedly replace each other).
if (
isRelayClientWebSocketUrl(this.config.url) &&
closeCode === 1008 &&
(closeReason ?? reason) === 'Replaced by new connection'
) {
this.shouldReconnect = false
this.clearWaiters(new Error(reason ?? 'Replaced by new connection'))
this.rejectPendingSendQueue(new Error(reason ?? 'Replaced by new connection'))
this.rejectConnect(new Error(reason ?? 'Replaced by new connection'))
return
}
this.scheduleReconnect(reason)
}),
transport.onError((event) => {
this.resetConnectTimeout()
const reason = describeTransportError(event)
const isGeneric = reason === 'Transport error'
// Browser WebSocket.onerror often provides no useful details and is followed
@@ -514,8 +602,11 @@ export class DaemonClient {
this.connectionState.status === 'connecting'
) {
this.lastErrorValue = reason
this.updateConnectionState({ status: 'disconnected', reason })
this.scheduleReconnect(reason)
this.scheduleReconnect({
reason,
event: 'TRANSPORT_ERROR',
reasonCode: 'transport_error',
})
}
}, 250)
}
@@ -527,15 +618,23 @@ export class DaemonClient {
this.pendingGenericTransportErrorTimeout = null
}
this.lastErrorValue = reason
this.updateConnectionState({ status: 'disconnected', reason })
this.scheduleReconnect(reason)
this.scheduleReconnect({
reason,
event: 'TRANSPORT_ERROR',
reasonCode: 'transport_error',
})
}),
transport.onMessage((data) => this.handleTransportMessage(data)),
]
} catch (error) {
this.resetConnectTimeout()
const message = error instanceof Error ? error.message : 'Failed to connect'
this.lastErrorValue = message
this.scheduleReconnect(message)
this.scheduleReconnect({
reason: message,
event: 'CONNECT_FAILED',
reasonCode: 'connect_failed',
})
this.rejectConnect(error instanceof Error ? error : new Error(message))
}
}
@@ -559,6 +658,9 @@ export class DaemonClient {
}
async close(): Promise<void> {
if (this.connectionState.status === 'disposed') {
return
}
this.shouldReconnect = false
this.connectPromise = null
this.connectResolve = null
@@ -567,16 +669,21 @@ export class DaemonClient {
clearTimeout(this.reconnectTimeout)
this.reconnectTimeout = null
}
this.resetConnectTimeout()
this.disposeTransport(1000, 'Client closed')
this.clearWaiters(new Error('Daemon client closed'))
this.rejectPendingSendQueue(new Error('Daemon client closed'))
this.terminalStreams.clearAll()
this.updateConnectionState({
status: 'disconnected',
reason: 'client_closed',
})
this.updateConnectionState(
{ status: 'disposed' },
{ event: 'DISPOSE', reason: 'Client closed', reasonCode: 'disposed' }
)
}
ensureConnected(): void {
if (this.connectionState.status === 'disposed') {
return
}
if (!this.shouldReconnect) {
this.shouldReconnect = true
}
@@ -2505,6 +2612,7 @@ export class DaemonClient {
}
private cleanupTransport(): void {
this.resetConnectTimeout()
if (this.pendingGenericTransportErrorTimeout) {
clearTimeout(this.pendingGenericTransportErrorTimeout)
this.pendingGenericTransportErrorTimeout = null
@@ -2519,6 +2627,14 @@ export class DaemonClient {
this.transportCleanup = []
}
private resetConnectTimeout(): void {
if (!this.connectTimeout) {
return
}
clearTimeout(this.connectTimeout)
this.connectTimeout = null
}
private handleTransportMessage(data: unknown): void {
const rawData =
data && typeof data === 'object' && 'data' in data ? (data as { data: unknown }).data : data
@@ -2573,8 +2689,32 @@ export class DaemonClient {
}
}
private updateConnectionState(next: ConnectionState): void {
private updateConnectionState(
next: ConnectionState,
metadata?: { event: string; reason?: string; reasonCode?: string }
): void {
const previous = this.connectionState
this.connectionState = next
const reasonFromNext =
next.status === 'disconnected' && typeof next.reason === 'string'
? next.reason
: null
const reason = metadata?.reason ?? reasonFromNext
const reasonCode = metadata?.reasonCode ?? toReasonCode(reason)
this.logger.info(
{
serverId: this.logServerId,
clientSessionKeyHash: this.logClientSessionKeyHash,
from: previous.status,
to: next.status,
event: metadata?.event ?? 'STATE_UPDATE',
connectionPath: this.logConnectionPath,
generation: this.logGeneration,
reasonCode,
reason,
},
'DaemonClientTransition'
)
for (const listener of this.connectionListeners) {
try {
listener(next)
@@ -2584,22 +2724,17 @@ export class DaemonClient {
}
}
private scheduleReconnect(reason?: string): void {
private scheduleReconnect(input?: {
reason?: string
event?: string
reasonCode?: string
}): void {
if (this.reconnectTimeout) {
clearTimeout(this.reconnectTimeout)
this.reconnectTimeout = null
}
if (!this.shouldReconnect || this.config.reconnect?.enabled === false) {
this.rejectConnect(new Error(reason ?? 'Transport disconnected before connect'))
return
}
const attempt = this.reconnectAttempt
const baseDelay = this.config.reconnect?.baseDelayMs ?? DEFAULT_RECONNECT_BASE_DELAY_MS
const maxDelay = this.config.reconnect?.maxDelayMs ?? DEFAULT_RECONNECT_MAX_DELAY_MS
const delay = Math.min(baseDelay * 2 ** attempt, maxDelay)
this.reconnectAttempt = attempt + 1
const wasDisposed = this.connectionState.status === 'disposed'
const reason = input?.reason
if (typeof reason === 'string' && reason.trim().length > 0) {
this.lastErrorValue = reason.trim()
@@ -2611,10 +2746,28 @@ export class DaemonClient {
this.rejectPendingSendQueue(new Error(reason ?? 'Connection lost'))
this.terminalStreams.clearAll()
if (wasDisposed) {
this.rejectConnect(new Error(reason ?? 'Daemon client is disposed'))
return
}
this.updateConnectionState({
status: 'disconnected',
...(reason ? { reason } : {}),
}, {
event: input?.event ?? 'TRANSPORT_CLOSE',
...(reason ? { reason } : {}),
...(input?.reasonCode ? { reasonCode: input.reasonCode } : {}),
})
if (!this.shouldReconnect || this.config.reconnect?.enabled === false) {
this.rejectConnect(new Error(reason ?? 'Transport disconnected before connect'))
return
}
const attempt = this.reconnectAttempt
const baseDelay = this.config.reconnect?.baseDelayMs ?? DEFAULT_RECONNECT_BASE_DELAY_MS
const maxDelay = this.config.reconnect?.maxDelayMs ?? DEFAULT_RECONNECT_MAX_DELAY_MS
const delay = Math.min(baseDelay * 2 ** attempt, maxDelay)
this.reconnectAttempt = attempt + 1
this.reconnectTimeout = setTimeout(() => {
this.reconnectTimeout = null
if (!this.shouldReconnect) {
packages/server/src/server/agent/providers/claude-agent.interrupt-restart-regression.test.ts
+242
View File
@@ -0,0 +1,242 @@
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { createTestLogger } from "../../../test-utils/test-logger.js";
import { ClaudeAgentClient } from "./claude-agent.js";
import type { AgentStreamEvent } from "../agent-sdk-types.js";
type Deferred<T> = {
promise: Promise<T>;
resolve: (value: T) => void;
reject: (error: unknown) => void;
};
const sdkMocks = vi.hoisted(() => ({
query: vi.fn(),
firstQuery: null as QueryMock | null,
secondQuery: null as QueryMock | null,
releaseOldAssistant: null as (() => void) | null,
}));
vi.mock("@anthropic-ai/claude-agent-sdk", () => ({
query: sdkMocks.query,
}));
type QueryMock = {
next: ReturnType<typeof vi.fn>;
interrupt: ReturnType<typeof vi.fn>;
return: ReturnType<typeof vi.fn>;
setPermissionMode: ReturnType<typeof vi.fn>;
setModel: ReturnType<typeof vi.fn>;
supportedModels: ReturnType<typeof vi.fn>;
supportedCommands: ReturnType<typeof vi.fn>;
rewindFiles: ReturnType<typeof vi.fn>;
};
function deferred<T>(): Deferred<T> {
let resolve!: (value: T) => void;
let reject!: (error: unknown) => void;
const promise = new Promise<T>((res, rej) => {
resolve = res;
reject = rej;
});
return { promise, resolve, reject };
}
function buildUsage() {
return {
input_tokens: 1,
cache_read_input_tokens: 0,
output_tokens: 1,
};
}
function buildFirstQueryMock(
allowOldAssistant: Promise<void>
): QueryMock {
let step = 0;
return {
next: vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "system",
subtype: "init",
session_id: "interrupt-regression-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
await allowOldAssistant;
step += 1;
return {
done: false,
value: {
type: "assistant",
message: {
content: "OLD_TURN_RESPONSE",
},
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
return { done: true, value: undefined };
}),
interrupt: vi.fn(async () => {
throw new Error("simulated interrupt failure");
}),
return: vi.fn(async () => undefined),
setPermissionMode: vi.fn(async () => undefined),
setModel: vi.fn(async () => undefined),
supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]),
supportedCommands: vi.fn(async () => []),
rewindFiles: vi.fn(async () => ({ canRewind: true })),
};
}
function buildSecondQueryMock(): QueryMock {
let step = 0;
return {
next: vi.fn(async () => {
if (step === 0) {
step += 1;
return {
done: false,
value: {
type: "system",
subtype: "init",
session_id: "interrupt-regression-session",
permissionMode: "default",
model: "opus",
},
};
}
if (step === 1) {
step += 1;
return {
done: false,
value: {
type: "assistant",
message: {
content: "NEW_TURN_RESPONSE",
},
},
};
}
if (step === 2) {
step += 1;
return {
done: false,
value: {
type: "result",
subtype: "success",
usage: buildUsage(),
total_cost_usd: 0,
},
};
}
return { done: true, value: undefined };
}),
interrupt: vi.fn(async () => undefined),
return: vi.fn(async () => undefined),
setPermissionMode: vi.fn(async () => undefined),
setModel: vi.fn(async () => undefined),
supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]),
supportedCommands: vi.fn(async () => []),
rewindFiles: vi.fn(async () => ({ canRewind: true })),
};
}
async function collectUntilTerminal(
stream: AsyncGenerator<AgentStreamEvent>
): Promise<AgentStreamEvent[]> {
const events: AgentStreamEvent[] = [];
for await (const event of stream) {
events.push(event);
if (
event.type === "turn_completed" ||
event.type === "turn_failed" ||
event.type === "turn_canceled"
) {
break;
}
}
return events;
}
function collectAssistantText(events: AgentStreamEvent[]): string {
return events
.filter(
(event): event is Extract<AgentStreamEvent, { type: "timeline" }> =>
event.type === "timeline" && event.item.type === "assistant_message"
)
.map((event) => event.item.text)
.join("");
}
describe("ClaudeAgentSession interrupt restart regression", () => {
beforeEach(() => {
const allowOldAssistant = deferred<void>();
let queryCreateCount = 0;
sdkMocks.query.mockImplementation(() => {
queryCreateCount += 1;
if (queryCreateCount === 1) {
const mock = buildFirstQueryMock(allowOldAssistant.promise);
sdkMocks.firstQuery = mock;
return mock;
}
const mock = buildSecondQueryMock();
sdkMocks.secondQuery = mock;
return mock;
});
sdkMocks.releaseOldAssistant = () => allowOldAssistant.resolve();
});
afterEach(() => {
sdkMocks.query.mockReset();
sdkMocks.firstQuery = null;
sdkMocks.secondQuery = null;
sdkMocks.releaseOldAssistant = null;
});
test("starts a fresh query after interrupt failure to avoid stale old-turn response", async () => {
const logger = createTestLogger();
const client = new ClaudeAgentClient({ logger });
const session = await client.createSession({
provider: "claude",
cwd: process.cwd(),
});
const firstTurn = session.stream("first prompt");
await firstTurn.next();
const secondTurnPromise = collectUntilTerminal(session.stream("second prompt"));
await Promise.resolve();
sdkMocks.releaseOldAssistant?.();
const secondTurnEvents = await secondTurnPromise;
const secondAssistantText = collectAssistantText(secondTurnEvents);
expect(sdkMocks.query).toHaveBeenCalledTimes(2);
expect(secondAssistantText).toContain("NEW_TURN_RESPONSE");
expect(secondAssistantText).not.toContain("OLD_TURN_RESPONSE");
await firstTurn.return?.();
await session.close();
});
});
packages/server/src/server/agent/providers/claude-agent.ts
+3
View File
@@ -1607,6 +1607,9 @@ class ClaudeAgentSession implements AgentSession {
this.queryRestartNeeded = false;
} catch (error) {
this.logger.warn({ err: error }, "Failed to interrupt active turn");
// If interrupt fails, the SDK iterator may remain in an indeterminate state.
// Force a teardown/recreate path so the next turn cannot reuse stale query state.
this.queryRestartNeeded = true;
}
}
packages/server/src/server/daemon-e2e/checkout-debug.ts
+3 -1
View File
@@ -32,13 +32,15 @@ class LoggingWebSocket extends OriginalWebSocket {
const PASEO_HOME = process.env.PASEO_HOME ?? `${os.homedir()}/.paseo`;
const PASEO_LISTEN = process.env.PASEO_LISTEN ?? "127.0.0.1:6767";
const DAEMON_URL = `ws://${PASEO_LISTEN}/ws`;
const CLIENT_SESSION_KEY = "clsk_checkout_debug";
async function testMultiAgentSequence() {
console.log("\n=== Testing multi-agent checkout sequence ===");
console.log(`Daemon URL: ${DAEMON_URL}`);
const client = new DaemonClient({
url: DAEMON_URL,
url: `${DAEMON_URL}?clientSessionKey=${CLIENT_SESSION_KEY}`,
clientSessionKey: CLIENT_SESSION_KEY,
webSocketFactory: (url) => new LoggingWebSocket(url) as any,
reconnect: { enabled: false },
});
packages/server/src/server/relay-transport.test.ts
+1 -1
View File
@@ -216,7 +216,7 @@ describe("relay-transport control lifecycle", () => {
dataSocket,
{
transport: "relay",
externalSessionKey: "relay:clt_test",
externalSessionKey: "session:clt_test",
}
);
});
packages/server/src/server/relay-transport.ts
+1 -1
View File
@@ -324,7 +324,7 @@ export function startRelayTransport({
attached = true;
const externalMetadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: `relay:${clientId}`,
externalSessionKey: `session:${clientId}`,
};
if (daemonKeyPair) {
void attachEncryptedSocket(
packages/server/src/server/test-utils/daemon-client.ts
+9
View File
@@ -16,10 +16,19 @@ export type DaemonClientConfig = Omit<
export type CreateAgentOptions = CreateAgentRequestOptions;
export { type SendMessageOptions, type DaemonEvent, type DaemonEventHandler };
let testClientCounter = 0;
function nextTestClientSessionKey(): string {
testClientCounter += 1;
return `clsk_test_client_${testClientCounter}`;
}
export class DaemonClient extends SharedDaemonClient {
constructor(config: DaemonClientConfig) {
const clientSessionKey = config.clientSessionKey ?? nextTestClientSessionKey();
super({
...config,
clientSessionKey,
webSocketFactory: (url, options) =>
new WebSocket(url, { headers: options?.headers }) as unknown as WebSocketLike,
});
packages/server/src/server/websocket-server.relay-reconnect.test.ts
+154 -7
View File
@@ -256,7 +256,7 @@ describe("relay external socket reconnect behavior", () => {
const server = createServer();
const metadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: "relay:client-1",
externalSessionKey: "session:client-1",
};
const socket1 = new MockSocket();
@@ -278,11 +278,158 @@ describe("relay external socket reconnect behavior", () => {
await server.close();
});
test("rejects direct socket attach when clientSessionKey is missing", async () => {
const server = createServer();
const request = {
headers: {
host: "localhost:6767",
origin: "http://localhost:6767",
"user-agent": "vitest",
},
socket: {
remoteAddress: "127.0.0.1",
},
url: "/ws",
};
const socket = new MockSocket();
let closeCode: number | null = null;
let closeReason = "";
socket.on("close", (code: unknown, reason: unknown) => {
closeCode = typeof code === "number" ? code : null;
closeReason = typeof reason === "string" ? reason : String(reason ?? "");
});
await (server as any).attachSocket(socket, request);
expect(closeCode).toBe(1008);
expect(closeReason).toBe("Missing clientSessionKey");
expect(sessionMock.instances).toHaveLength(0);
await server.close();
});
test("attaches multiple sockets to the same relay external session key", async () => {
const server = createServer();
const metadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: "session:client-multi",
};
const socket1 = new MockSocket();
await server.attachExternalSocket(socket1, metadata);
expect(sessionMock.instances).toHaveLength(1);
const session = sessionMock.instances[0]!;
const socket2 = new MockSocket();
await server.attachExternalSocket(socket2, metadata);
expect(sessionMock.instances).toHaveLength(1);
const onMessage = session.args.onMessage as
| ((msg: { type: "status"; payload: { status: string } }) => void)
| undefined;
expect(onMessage).toBeTypeOf("function");
onMessage?.({
type: "status",
payload: { status: "ok" },
});
expect(socket1.sent.length).toBeGreaterThan(1);
expect(socket2.sent.length).toBeGreaterThan(1);
await server.close();
});
test("reuses direct session when clientSessionKey reconnects within grace window", async () => {
const server = createServer();
const request = {
headers: {
host: "localhost:6767",
origin: "http://localhost:6767",
"user-agent": "vitest",
},
socket: {
remoteAddress: "127.0.0.1",
},
url: "/ws?clientSessionKey=clsk_direct_reconnect",
};
const socket1 = new MockSocket();
await (server as any).attachSocket(socket1, request);
expect(sessionMock.instances).toHaveLength(1);
const session = sessionMock.instances[0]!;
socket1.emit("close", 1006, "");
await vi.advanceTimersByTimeAsync(1_000);
expect(session.cleanup).not.toHaveBeenCalled();
const socket2 = new MockSocket();
await (server as any).attachSocket(socket2, request);
expect(sessionMock.instances).toHaveLength(1);
await vi.advanceTimersByTimeAsync(20_000);
expect(session.cleanup).not.toHaveBeenCalled();
await server.close();
});
test("reuses one session when switching from direct to relay with the same session key", async () => {
const server = createServer();
const clientSessionKey = "clsk_switch_path";
const directRequest = {
headers: {
host: "localhost:6767",
origin: "http://localhost:6767",
"user-agent": "vitest",
},
socket: {
remoteAddress: "127.0.0.1",
},
url: `/ws?clientSessionKey=${clientSessionKey}`,
};
const relayMetadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: `session:${clientSessionKey}`,
};
const directSocket = new MockSocket();
await (server as any).attachSocket(directSocket, directRequest);
expect(sessionMock.instances).toHaveLength(1);
const session = sessionMock.instances[0]!;
const relaySocket = new MockSocket();
await server.attachExternalSocket(relaySocket, relayMetadata);
expect(sessionMock.instances).toHaveLength(1);
const onMessage = session.args.onMessage as
| ((msg: { type: "status"; payload: { status: string } }) => void)
| undefined;
expect(onMessage).toBeTypeOf("function");
onMessage?.({
type: "status",
payload: { status: "ok" },
});
expect(directSocket.sent.length).toBeGreaterThan(1);
expect(relaySocket.sent.length).toBeGreaterThan(1);
directSocket.emit("close", 1006, "");
await vi.advanceTimersByTimeAsync(1_000);
expect(session.cleanup).not.toHaveBeenCalled();
relaySocket.emit("close", 1006, "");
await vi.advanceTimersByTimeAsync(90_000);
expect(session.cleanup).toHaveBeenCalledTimes(1);
await server.close();
});
test("cleans up relay session when reconnect grace expires", async () => {
const server = createServer();
const metadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: "relay:client-2",
externalSessionKey: "session:client-2",
};
const socket1 = new MockSocket();
@@ -302,7 +449,7 @@ describe("relay external socket reconnect behavior", () => {
const server = createServer({ speechReadiness });
const metadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: "relay:client-server-info-capabilities",
externalSessionKey: "session:client-server-info-capabilities",
};
const socket = new MockSocket();
@@ -339,7 +486,7 @@ describe("relay external socket reconnect behavior", () => {
const server = createServer();
const metadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: "relay:client-server-info-broadcast",
externalSessionKey: "session:client-server-info-broadcast",
};
const socket = new MockSocket();
@@ -368,7 +515,7 @@ describe("relay external socket reconnect behavior", () => {
const server = createServer();
const metadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: "relay:client-server-info-download-guidance",
externalSessionKey: "session:client-server-info-download-guidance",
};
const socket = new MockSocket();
await server.attachExternalSocket(socket, metadata);
@@ -397,7 +544,7 @@ describe("relay external socket reconnect behavior", () => {
const server = createServer();
const metadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: "relay:client-binary-inbound",
externalSessionKey: "session:client-binary-inbound",
};
const socket = new MockSocket();
@@ -438,7 +585,7 @@ describe("relay external socket reconnect behavior", () => {
const server = createServer();
const metadata: ExternalSocketMetadata = {
transport: "relay",
externalSessionKey: "relay:client-binary-outbound",
externalSessionKey: "session:client-binary-outbound",
};
const socket = new MockSocket();
packages/server/src/server/websocket-server.ts
+106 -33
View File
@@ -151,7 +151,7 @@ type SessionConnection = {
session: Session;
clientId: string;
connectionLogger: pino.Logger;
socketRef: { current: WebSocketLike };
sockets: Set<WebSocketLike>;
externalSessionKey: string | null;
externalDisconnectCleanupTimeout: ReturnType<typeof setTimeout> | null;
};
@@ -357,19 +357,20 @@ export class VoiceAssistantWebSocketServer {
connection.externalDisconnectCleanupTimeout = null;
}
const ws = connection.socketRef.current;
cleanupPromises.push(connection.session.cleanup());
cleanupPromises.push(
new Promise<void>((resolve) => {
// WebSocket.CLOSED = 3
if (ws.readyState === 3) {
resolve();
return;
}
ws.once("close", () => resolve());
ws.close();
})
);
for (const ws of connection.sockets) {
cleanupPromises.push(
new Promise<void>((resolve) => {
// WebSocket.CLOSED = 3
if (ws.readyState === 3) {
resolve();
return;
}
ws.once("close", () => resolve());
ws.close();
})
);
}
}
await Promise.all(cleanupPromises);
this.sessions.clear();
@@ -394,15 +395,54 @@ export class VoiceAssistantWebSocketServer {
ws.send(encodeBinaryMuxFrame(frame));
}
private sendToConnection(connection: SessionConnection, message: WSOutboundMessage): void {
for (const ws of connection.sockets) {
this.sendToClient(ws, message);
}
}
private sendBinaryToConnection(
connection: SessionConnection,
frame: Parameters<typeof encodeBinaryMuxFrame>[0]
): void {
for (const ws of connection.sockets) {
this.sendBinaryToClient(ws, frame);
}
}
private async attachSocket(
ws: WebSocketLike,
request?: unknown,
metadata?: ExternalSocketMetadata
): Promise<void> {
const externalSessionKey =
const requestMetadata = extractSocketRequestMetadata(request);
const relayExternalSessionKey =
metadata?.transport === "relay" && metadata.externalSessionKey.trim().length > 0
? metadata.externalSessionKey
: null;
const directExternalSessionKey =
typeof requestMetadata.clientSessionKey === "string" &&
requestMetadata.clientSessionKey.trim().length > 0
? `session:${requestMetadata.clientSessionKey.trim()}`
: null;
const externalSessionKey = relayExternalSessionKey ?? directExternalSessionKey;
if (metadata?.transport !== "relay" && !directExternalSessionKey) {
this.logger.warn(
{
host: requestMetadata.host,
origin: requestMetadata.origin,
remoteAddress: requestMetadata.remoteAddress,
},
"Rejected direct connection without clientSessionKey"
);
try {
ws.close(1008, "Missing clientSessionKey");
} catch {
// ignore close errors
}
return;
}
if (externalSessionKey) {
const existing = this.externalSessionsByKey.get(externalSessionKey);
@@ -412,12 +452,7 @@ export class VoiceAssistantWebSocketServer {
existing.externalDisconnectCleanupTimeout = null;
}
const previousSocket = existing.socketRef.current;
if (previousSocket !== ws) {
this.sessions.delete(previousSocket);
existing.socketRef.current = ws;
}
existing.sockets.add(ws);
this.sessions.set(ws, existing);
this.sendServerInfo(ws);
existing.connectionLogger.trace(
@@ -434,10 +469,9 @@ export class VoiceAssistantWebSocketServer {
}
const clientId = `client-${++this.clientIdCounter}`;
const requestMetadata = extractSocketRequestMetadata(request);
const connectionLoggerFields: Record<string, string> = {
clientId,
transport: externalSessionKey ? "relay" : "direct",
transport: metadata?.transport === "relay" ? "relay" : "direct",
};
if (requestMetadata.host) {
connectionLoggerFields.host = requestMetadata.host;
@@ -452,15 +486,21 @@ export class VoiceAssistantWebSocketServer {
connectionLoggerFields.remoteAddress = requestMetadata.remoteAddress;
}
const connectionLogger = this.logger.child(connectionLoggerFields);
const socketRef = { current: ws };
let connection: SessionConnection | null = null;
const session = new Session({
clientId,
onMessage: (msg) => {
this.sendToClient(socketRef.current, wrapSessionMessage(msg));
if (!connection) {
return;
}
this.sendToConnection(connection, wrapSessionMessage(msg));
},
onBinaryMessage: (frame) => {
this.sendBinaryToClient(socketRef.current, frame);
if (!connection) {
return;
}
this.sendBinaryToConnection(connection, frame);
},
logger: connectionLogger.child({ module: "session" }),
downloadTokenStore: this.downloadTokenStore,
@@ -493,11 +533,11 @@ export class VoiceAssistantWebSocketServer {
agentProviderRuntimeSettings: this.agentProviderRuntimeSettings,
});
const connection: SessionConnection = {
connection = {
session,
clientId,
connectionLogger,
socketRef,
sockets: new Set([ws]),
externalSessionKey,
externalDisconnectCleanupTimeout: null,
};
@@ -593,11 +633,9 @@ export class VoiceAssistantWebSocketServer {
const activeConnection = this.sessions.get(ws);
if (activeConnection !== connection) return;
this.sessions.delete(ws);
connection.sockets.delete(ws);
if (
connection.externalSessionKey &&
connection.socketRef.current === ws
) {
if (connection.externalSessionKey && connection.sockets.size === 0) {
if (connection.externalDisconnectCleanupTimeout) {
clearTimeout(connection.externalDisconnectCleanupTimeout);
}
@@ -623,6 +661,19 @@ export class VoiceAssistantWebSocketServer {
return;
}
if (connection.sockets.size > 0) {
connection.connectionLogger.trace(
{
clientId: connection.clientId,
remainingSockets: connection.sockets.size,
code: details.code,
reason: stringifyCloseReason(details.reason),
},
"Client socket disconnected; session remains attached"
);
return;
}
await this.cleanupConnection(connection, "Client disconnected");
}
@@ -635,8 +686,10 @@ export class VoiceAssistantWebSocketServer {
connection.externalDisconnectCleanupTimeout = null;
}
const currentSocket = connection.socketRef.current;
this.sessions.delete(currentSocket);
for (const socket of connection.sockets) {
this.sessions.delete(socket);
}
connection.sockets.clear();
if (connection.externalSessionKey) {
const existing = this.externalSessionsByKey.get(connection.externalSessionKey);
if (existing === connection) {
@@ -899,6 +952,7 @@ type SocketRequestMetadata = {
origin?: string;
userAgent?: string;
remoteAddress?: string;
clientSessionKey?: string;
};
function extractSocketRequestMetadata(request: unknown): SocketRequestMetadata {
@@ -912,6 +966,7 @@ function extractSocketRequestMetadata(request: unknown): SocketRequestMetadata {
origin?: unknown;
"user-agent"?: unknown;
};
url?: unknown;
socket?: {
remoteAddress?: unknown;
};
@@ -928,12 +983,30 @@ function extractSocketRequestMetadata(request: unknown): SocketRequestMetadata {
typeof record.socket?.remoteAddress === "string"
? record.socket.remoteAddress
: undefined;
const rawUrl = typeof record.url === "string" ? record.url : null;
const clientSessionKey = (() => {
if (!rawUrl) {
return undefined;
}
try {
const parsed = new URL(rawUrl, "http://localhost");
const value = parsed.searchParams.get("clientSessionKey");
if (!value) {
return undefined;
}
const trimmed = value.trim();
return trimmed.length > 0 ? trimmed : undefined;
} catch {
return undefined;
}
})();
return {
...(host ? { host } : {}),
...(origin ? { origin } : {}),
...(userAgent ? { userAgent } : {}),
...(remoteAddress ? { remoteAddress } : {}),
...(clientSessionKey ? { clientSessionKey } : {}),
};
}
packages/server/src/shared/daemon-endpoints.test.ts
+47
View File
@@ -0,0 +1,47 @@
import { describe, expect, test } from "vitest";
import {
buildRelayWebSocketUrl,
CURRENT_RELAY_PROTOCOL_VERSION,
normalizeRelayProtocolVersion,
} from "./daemon-endpoints.js";
describe("relay websocket URL versioning", () => {
test("defaults relay URLs to v2", () => {
const url = new URL(
buildRelayWebSocketUrl({
endpoint: "relay.paseo.sh:443",
serverId: "srv_test",
role: "client",
clientSessionKey: "clsk_test",
})
);
expect(url.searchParams.get("v")).toBe(CURRENT_RELAY_PROTOCOL_VERSION);
expect(url.searchParams.get("clientId")).toBe("clsk_test");
});
test("allows explicitly requesting v1 relay URLs", () => {
const url = new URL(
buildRelayWebSocketUrl({
endpoint: "relay.paseo.sh:443",
serverId: "srv_test",
role: "server",
version: "1",
})
);
expect(url.searchParams.get("v")).toBe("1");
});
test("normalizes numeric relay versions", () => {
expect(normalizeRelayProtocolVersion(2)).toBe("2");
expect(normalizeRelayProtocolVersion(1)).toBe("1");
});
test("rejects unsupported relay versions", () => {
expect(() => normalizeRelayProtocolVersion("3")).toThrow(
'Relay version must be "1" or "2"'
);
});
});
packages/server/src/shared/daemon-endpoints.ts
+47 -4
View File
@@ -5,6 +5,28 @@ export type HostPortParts = {
};
export type RelayRole = "server" | "client";
export type RelayProtocolVersion = "1" | "2";
export const CURRENT_RELAY_PROTOCOL_VERSION: RelayProtocolVersion = "2";
export function normalizeRelayProtocolVersion(
value: unknown,
fallback: RelayProtocolVersion = CURRENT_RELAY_PROTOCOL_VERSION
): RelayProtocolVersion {
if (value == null) {
return fallback;
}
const normalized =
typeof value === "string" ? value.trim() : typeof value === "number" ? String(value) : "";
if (!normalized) {
return fallback;
}
if (normalized === "1" || normalized === "2") {
return normalized;
}
throw new Error('Relay version must be "1" or "2"');
}
function parsePort(portStr: string, context: string): number {
const port = Number(portStr);
@@ -60,11 +82,21 @@ function shouldUseSecureWebSocket(port: number): boolean {
return port === 443;
}
export function buildDaemonWebSocketUrl(endpoint: string): string {
export function buildDaemonWebSocketUrl(
endpoint: string,
params?: { clientSessionKey?: string }
): string {
const { host, port, isIpv6 } = parseHostPort(endpoint);
const protocol = shouldUseSecureWebSocket(port) ? "wss" : "ws";
const hostPart = isIpv6 ? `[${host}]` : host;
return `${protocol}://${hostPart}:${port}/ws`;
const url = new URL(`${protocol}://${hostPart}:${port}/ws`);
if (
typeof params?.clientSessionKey === "string" &&
params.clientSessionKey.trim().length > 0
) {
url.searchParams.set("clientSessionKey", params.clientSessionKey.trim());
}
return url.toString();
}
export function buildRelayWebSocketUrl(params: {
@@ -72,6 +104,8 @@ export function buildRelayWebSocketUrl(params: {
serverId: string;
role: RelayRole;
clientId?: string;
clientSessionKey?: string;
version?: RelayProtocolVersion | 1 | 2;
}): string {
const { host, port, isIpv6 } = parseHostPort(params.endpoint);
const protocol = shouldUseSecureWebSocket(port) ? "wss" : "ws";
@@ -79,8 +113,17 @@ export function buildRelayWebSocketUrl(params: {
const url = new URL(`${protocol}://${hostPart}:${port}/ws`);
url.searchParams.set("serverId", params.serverId);
url.searchParams.set("role", params.role);
if (params.clientId) {
url.searchParams.set("clientId", params.clientId);
url.searchParams.set("v", normalizeRelayProtocolVersion(params.version));
if (
params.clientId &&
params.clientSessionKey &&
params.clientId !== params.clientSessionKey
) {
throw new Error("clientId and clientSessionKey must match when both are provided");
}
const resolvedClientId = params.clientId ?? params.clientSessionKey;
if (resolvedClientId) {
url.searchParams.set("clientId", resolvedClientId);
}
return url.toString();
}