diff --git a/src/main/ipc/ssh.test.ts b/src/main/ipc/ssh.test.ts index cde245ee4..0c5a51d9a 100644 --- a/src/main/ipc/ssh.test.ts +++ b/src/main/ipc/ssh.test.ts @@ -50,7 +50,8 @@ const { onRequest: vi.fn().mockReturnValue(() => {}), onDispose: vi.fn().mockReturnValue(() => {}), request: vi.fn().mockResolvedValue({}), - notify: vi.fn() + notify: vi.fn(), + probeLiveness: vi.fn().mockResolvedValue(false) }, mockPtyProvider: { onData: vi.fn(), @@ -316,6 +317,7 @@ describe('SSH IPC handlers', () => { mockMux.isDisposed.mockReset().mockReturnValue(false) mockMux.onNotification.mockReset() mockMux.onDispose.mockReset().mockReturnValue(() => {}) + mockMux.probeLiveness.mockReset().mockResolvedValue(false) mockPtyProvider.onData.mockReset() mockPtyProvider.onExit.mockReset() mockPtyProvider.onReplay.mockReset() @@ -1401,7 +1403,7 @@ describe('SSH IPC handlers', () => { expect(mockConnectionManager.disconnect).toHaveBeenCalledWith('ssh-1') }) - it('forces active SSH sessions to reconnect when the system resumes from sleep', async () => { + it('reconnects on system resume when the relay liveness probe fails', async () => { const target: SshTarget = { id: 'ssh-1', label: 'Server', @@ -1419,6 +1421,7 @@ describe('SSH IPC handlers', () => { error: null, reconnectAttempt: 0 }) + mockMux.probeLiveness.mockResolvedValue(false) await handlers.get('ssh:connect')!(null, { targetId: 'ssh-1' }) @@ -1427,7 +1430,77 @@ describe('SSH IPC handlers', () => { resumeListener() - expect(mockConnectionManager.reconnect).toHaveBeenCalledWith('ssh-1') + await vi.waitFor(() => expect(mockConnectionManager.reconnect).toHaveBeenCalledWith('ssh-1')) + // Why: a failed first probe gets one retry before teardown (slow post-wake network). + expect(mockMux.probeLiveness).toHaveBeenCalledTimes(2) + }) + + it('skips reconnect on system resume when the relay link is still alive', async () => { + const target: SshTarget = { + id: 'ssh-1', + label: 'Server', + host: 'example.com', + port: 22, + username: 'deploy' + } + const conn = {} + mockSshStore.getTarget.mockReturnValue(target) + mockConnectionManager.connect.mockResolvedValue(conn) + mockConnectionManager.getConnection.mockReturnValue(conn) + mockConnectionManager.getState.mockReturnValue({ + targetId: 'ssh-1', + status: 'connected', + error: null, + reconnectAttempt: 0 + }) + mockMux.probeLiveness.mockResolvedValue(true) + + await handlers.get('ssh:connect')!(null, { targetId: 'ssh-1' }) + + const resumeListener = powerMonitorOnMock.mock.calls.find(([event]) => event === 'resume')?.[1] + expect(resumeListener).toBeTypeOf('function') + + resumeListener() + + await vi.waitFor(() => expect(mockMux.probeLiveness).toHaveBeenCalledTimes(1)) + // Let the async resume handler settle before asserting no teardown happened. + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(mockConnectionManager.reconnect).not.toHaveBeenCalled() + }) + + it('does not reconnect after resume when the target was disconnected during the probe', async () => { + const target: SshTarget = { + id: 'ssh-1', + label: 'Server', + host: 'example.com', + port: 22, + username: 'deploy' + } + const conn = {} + mockSshStore.getTarget.mockReturnValue(target) + mockConnectionManager.connect.mockResolvedValue(conn) + mockConnectionManager.getConnection.mockReturnValue(conn) + mockConnectionManager.getState.mockReturnValue({ + targetId: 'ssh-1', + status: 'connected', + error: null, + reconnectAttempt: 0 + }) + mockMux.probeLiveness.mockResolvedValue(false) + + await handlers.get('ssh:connect')!(null, { targetId: 'ssh-1' }) + + const resumeListener = powerMonitorOnMock.mock.calls.find(([event]) => event === 'resume')?.[1] + expect(resumeListener).toBeTypeOf('function') + + resumeListener() + // Why: the probe window is seconds long; a user disconnect during it must + // win — reconnecting afterwards would resurrect the torn-down target. + mockConnectionManager.getConnection.mockReturnValue(undefined) + + await vi.waitFor(() => expect(mockMux.probeLiveness).toHaveBeenCalledTimes(2)) + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(mockConnectionManager.reconnect).not.toHaveBeenCalled() }) it('extends active relay grace while the system is suspending', async () => { diff --git a/src/main/ipc/ssh.ts b/src/main/ipc/ssh.ts index 58391c4c0..8b5756bff 100644 --- a/src/main/ipc/ssh.ts +++ b/src/main/ipc/ssh.ts @@ -442,6 +442,24 @@ function registerAdvertisedUrlRefresh(getMainWindow: () => BrowserWindow | null) }) } +// Why: macOS can resume the process before the network stack is back up, so +// a failed first probe gets one retry before the link is declared dead (#7773). +const RESUME_PROBE_TIMEOUT_MS = 5_000 +const RESUME_PROBE_ATTEMPTS = 2 + +async function isRelayLinkAliveAfterResume(session: SshRelaySession): Promise { + const mux = session.getMux() + if (!mux || mux.isDisposed()) { + return false + } + for (let attempt = 0; attempt < RESUME_PROBE_ATTEMPTS; attempt++) { + if (await mux.probeLiveness(RESUME_PROBE_TIMEOUT_MS)) { + return true + } + } + return false +} + function registerPowerMonitorReconnect(): void { powerMonitorUnsubscribe?.() const onSuspend = (): void => { @@ -450,19 +468,35 @@ function registerPowerMonitorReconnect(): void { } } const onResume = (): void => { - for (const targetId of activeSessions.keys()) { - const conn = connectionManager?.getConnection(targetId) + for (const [targetId, session] of activeSessions) { + const manager = connectionManager + const conn = manager?.getConnection(targetId) if (!conn) { continue } - const reconnect = connectionManager?.reconnect(targetId) - void reconnect?.catch((err) => { - console.warn( - `[ssh] Failed to reconnect ${targetId} after system resume: ${ - err instanceof Error ? err.message : String(err) - }` - ) - }) + void (async () => { + // Why: unconditional reconnect on every wake tore down live sessions + // and flashed the reconnect overlay (#7773). Only reconnect targets + // whose relay link actually died during sleep. + if (await isRelayLinkAliveAfterResume(session)) { + return + } + // Why: the probe can take ~10s. If the user disconnected or the + // session/connection was replaced meanwhile, reconnecting would + // resurrect a connection that was intentionally torn down. + if (activeSessions.get(targetId) !== session || manager?.getConnection(targetId) !== conn) { + return + } + try { + await manager?.reconnect(targetId) + } catch (err) { + console.warn( + `[ssh] Failed to reconnect ${targetId} after system resume: ${ + err instanceof Error ? err.message : String(err) + }` + ) + } + })() } } powerMonitor.on('suspend', onSuspend) diff --git a/src/main/ssh/ssh-channel-multiplexer.test.ts b/src/main/ssh/ssh-channel-multiplexer.test.ts index 8c56ae145..10123d56a 100644 --- a/src/main/ssh/ssh-channel-multiplexer.test.ts +++ b/src/main/ssh/ssh-channel-multiplexer.test.ts @@ -307,6 +307,81 @@ describe('SshChannelMultiplexer', () => { }) }) + describe('wake guard (timer pause across system sleep, #7773)', () => { + it('does not kill a healthy link on the first tick after a long timer pause', () => { + // Reach steady state with pending unacked keepalives (<5s old at pause). + vi.advanceTimersByTime(5_000) + expect(mux.isDisposed()).toBe(false) + + // Simulate sleep/App Nap: wall clock jumps far ahead with no ticks. + // Without the guard, the first post-wake tick sees lastReceivedAt and + // the pre-pause keepalive both >20s stale and disposes the mux. + vi.setSystemTime(Date.now() + 60 * 60 * 1000) + const writesBefore = transport.written.length + vi.advanceTimersByTime(5_000) + + expect(mux.isDisposed()).toBe(false) + // The guard probes immediately with a fresh keepalive. + expect(transport.written.length).toBeGreaterThan(writesBefore) + expect(transport.written.at(-1)![0]).toBe(MessageType.KeepAlive) + }) + + it('keeps the link alive after wake when frames resume', () => { + vi.advanceTimersByTime(5_000) + vi.setSystemTime(Date.now() + 60 * 60 * 1000) + vi.advanceTimersByTime(5_000) // guard tick + + // The relay answers the post-wake probe; the link must stay up. + let seq = 1 + for (let i = 0; i < 8; i++) { + vi.advanceTimersByTime(5_000) + transport.dataCallbacks[0](encodeKeepAliveFrame(seq++, 0)) + } + expect(mux.isDisposed()).toBe(false) + }) + + it('still detects a genuinely dead link within the next window after wake', () => { + vi.advanceTimersByTime(5_000) + vi.setSystemTime(Date.now() + 60 * 60 * 1000) + vi.advanceTimersByTime(5_000) // guard tick: reset + probe, no kill + + expect(mux.isDisposed()).toBe(false) + // No frames arrive after the guard reset; the honest window expires. + vi.advanceTimersByTime(25_000) + expect(mux.isDisposed()).toBe(true) + }) + }) + + describe('probeLiveness', () => { + it('sends a keepalive and resolves true when any frame arrives', async () => { + const writesBefore = transport.written.length + const probe = mux.probeLiveness(5_000) + + expect(transport.written.length).toBe(writesBefore + 1) + expect(transport.written.at(-1)![0]).toBe(MessageType.KeepAlive) + + transport.dataCallbacks[0](encodeKeepAliveFrame(1, 0)) + await expect(probe).resolves.toBe(true) + }) + + it('resolves false when no frame arrives before the timeout', async () => { + const probe = mux.probeLiveness(5_000) + vi.advanceTimersByTime(5_000) + await expect(probe).resolves.toBe(false) + }) + + it('resolves false when the mux is disposed while probing', async () => { + const probe = mux.probeLiveness(5_000) + mux.dispose() + await expect(probe).resolves.toBe(false) + }) + + it('resolves false immediately on a disposed mux', async () => { + mux.dispose() + await expect(mux.probeLiveness(5_000)).resolves.toBe(false) + }) + }) + describe('dispose', () => { it('rejects all pending requests on dispose', async () => { const promise = mux.request('pty.spawn') diff --git a/src/main/ssh/ssh-channel-multiplexer.ts b/src/main/ssh/ssh-channel-multiplexer.ts index 09642387e..8cb0a24f3 100644 --- a/src/main/ssh/ssh-channel-multiplexer.ts +++ b/src/main/ssh/ssh-channel-multiplexer.ts @@ -34,6 +34,9 @@ export type MethodNotificationHandler = (params: Record) => voi export type RequestHandler = (params: Record) => Promise | unknown const REQUEST_TIMEOUT_MS = 30_000 +// Why: a tick gap far beyond the interval means the process was paused +// (system sleep, App Nap timer throttling) — not that the link is dead (#7773). +const WAKE_GAP_MS = KEEPALIVE_SEND_MS * 3 export class SshChannelMultiplexer { private decoder: FrameDecoder @@ -58,6 +61,10 @@ export class SshChannelMultiplexer { // Track the oldest unacked outgoing message timestamp private unackedTimestamps = new Map() + // Why: liveness probes (#7773) resolve on the first frame of any kind — + // a keepalive ack proves the relay round-trip without a full RPC. + private livenessProbeWaiters: { succeed: () => void; fail: () => void }[] = [] + constructor(transport: MultiplexerTransport) { this.transport = transport @@ -232,6 +239,31 @@ export class SshChannelMultiplexer { this.sendMessage(msg) } + /** + * Send a fresh keepalive and resolve true when any frame arrives before the + * timeout. Used on system resume to distinguish a link that survived sleep + * from a dead one before tearing the session down (#7773). + */ + probeLiveness(timeoutMs: number): Promise { + if (this.disposed) { + return Promise.resolve(false) + } + return new Promise((resolve) => { + const settle = (alive: boolean): void => { + clearTimeout(timer) + const idx = this.livenessProbeWaiters.indexOf(waiter) + if (idx !== -1) { + this.livenessProbeWaiters.splice(idx, 1) + } + resolve(alive) + } + const waiter = { succeed: () => settle(true), fail: () => settle(false) } + const timer = setTimeout(() => settle(false), timeoutMs) + this.livenessProbeWaiters.push(waiter) + this.sendKeepAlive() + }) + } + dispose(reason: 'shutdown' | 'connection_lost' = 'shutdown'): void { if (this.disposed) { return @@ -259,6 +291,10 @@ export class SshChannelMultiplexer { reason === 'connection_lost' ? 'SSH connection lost, reconnecting...' : 'Multiplexer disposed' const errorCode = reason === 'connection_lost' ? 'CONNECTION_LOST' : 'DISPOSED' + for (const waiter of this.livenessProbeWaiters.splice(0)) { + waiter.fail() + } + for (const [id, pending] of this.pendingRequests) { pending.cleanup() const err = new Error(errorMessage) as Error & { code: string } @@ -322,6 +358,12 @@ export class SshChannelMultiplexer { } private handleFrame(frame: DecodedFrame): void { + // Why: any decoded frame proves the relay round-trip is alive; resolve + // pending resume probes before ordinary dispatch (#7773). + for (const waiter of this.livenessProbeWaiters.splice(0)) { + waiter.succeed() + } + // Update ack tracking if (frame.id > this.highestReceivedSeq) { this.highestReceivedSeq = frame.id @@ -453,12 +495,28 @@ export class SshChannelMultiplexer { } private startTimeoutCheck(): void { + let lastTickAt = Date.now() this.timeoutTimer = setInterval(() => { if (this.disposed) { return } const now = Date.now() + const sinceLastTick = now - lastTickAt + lastTickAt = now + // Why: after sleep/App Nap the pre-pause keepalive looks stale on the + // first post-wake tick, killing a healthy link (#7773). Reset staleness + // tracking, probe with a fresh keepalive, and let the NEXT full window + // make an honest liveness determination. + if (sinceLastTick > WAKE_GAP_MS) { + this.lastReceivedAt = now + for (const seq of this.unackedTimestamps.keys()) { + this.unackedTimestamps.set(seq, now) + } + this.sendKeepAlive() + return + } + const noDataReceived = now - this.lastReceivedAt > TIMEOUT_MS // Check oldest unacked message diff --git a/src/relay/relay-diagnostic-log.test.ts b/src/relay/relay-diagnostic-log.test.ts new file mode 100644 index 000000000..38adbcb76 --- /dev/null +++ b/src/relay/relay-diagnostic-log.test.ts @@ -0,0 +1,21 @@ +import { describe, expect, it, vi } from 'vitest' +import { relayLogLine } from './relay-diagnostic-log' + +describe('relayLogLine', () => { + it('prefixes each log line with an ISO timestamp', () => { + const writeSpy = vi.spyOn(process.stderr, 'write').mockImplementation(() => true) + try { + relayLogLine('[relay] Grace started (stdin ended)') + + expect(writeSpy).toHaveBeenCalledTimes(1) + const line = writeSpy.mock.calls[0][0] as string + // Why: relay.log correlation depends on a grep-stable + // " " shape (#7773). + expect(line).toMatch( + /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z \[relay\] Grace started \(stdin ended\)\n$/ + ) + } finally { + writeSpy.mockRestore() + } + }) +}) diff --git a/src/relay/relay-diagnostic-log.ts b/src/relay/relay-diagnostic-log.ts new file mode 100644 index 000000000..a42106101 --- /dev/null +++ b/src/relay/relay-diagnostic-log.ts @@ -0,0 +1,7 @@ +// Why: the relay daemon's stderr is redirected to relay.log on the remote +// host. Without timestamps, reconnect flaps in that log cannot be correlated +// with user activity or sleep/wake windows during diagnosis (#7773). Keep the +// format grep-stable: ISO timestamp, single space, then the original line. +export function relayLogLine(message: string): void { + process.stderr.write(`${new Date().toISOString()} ${message}\n`) +} diff --git a/src/relay/relay-handshake.ts b/src/relay/relay-handshake.ts index 6db263bcc..fb64c57ec 100644 --- a/src/relay/relay-handshake.ts +++ b/src/relay/relay-handshake.ts @@ -17,6 +17,7 @@ import { parseHandshakeMessage, type DecodedFrame } from './protocol' +import { relayLogLine } from './relay-diagnostic-log' // Why: a unique exit code reserved for the wire-level version-mismatch terminal // condition. The client (waitForSentinel + ssh.ts) maps this exit code to a @@ -132,22 +133,18 @@ function handleDaemonHandshakeFrame( try { msg = parseHandshakeMessage(frame.payload) } catch (err) { - process.stderr.write( - `[relay] Could not parse handshake: ${(err as Error).message}; closing socket\n` - ) + relayLogLine(`[relay] Could not parse handshake: ${(err as Error).message}; closing socket`) sock.destroy() return false } if (msg.type !== 'orca-relay-handshake') { - process.stderr.write( - `[relay] Unexpected handshake type from client: ${msg.type}; closing socket\n` - ) + relayLogLine(`[relay] Unexpected handshake type from client: ${msg.type}; closing socket`) sock.destroy() return false } if (msg.version !== launchVersion) { - process.stderr.write( - `[relay] Handshake mismatch: own=${launchVersion}, client=${msg.version}; closing socket\n` + relayLogLine( + `[relay] Handshake mismatch: own=${launchVersion}, client=${msg.version}; closing socket` ) try { sock.write( diff --git a/src/relay/relay.ts b/src/relay/relay.ts index 992252279..65f90e886 100644 --- a/src/relay/relay.ts +++ b/src/relay/relay.ts @@ -59,6 +59,7 @@ import { resolveOpenCodeSourceConfigDir, resolvePiSourceAgentDir } from './plugi import { detectPiAgentKindFromCommand } from '../shared/pi-agent-kind' import { resolveSetupAgentSequenceLaunchCommand } from '../shared/setup-agent-sequencing' import { pickRemoteCliEnv } from './remote-cli-env' +import { relayLogLine } from './relay-diagnostic-log' import { remoteCliRequestTimeoutMs } from './remote-cli-timeout' import { shouldReadRemoteCliStdin } from './remote-cli-stdin' @@ -358,13 +359,13 @@ async function main(): Promise { // would risk silent data corruption or zombie PTYs. We log for diagnostics // and then exit so the client can detect the disconnect and reconnect cleanly. process.on('uncaughtException', (err) => { - process.stderr.write(`[relay] Uncaught exception: ${err.message}\n${err.stack}\n`) + relayLogLine(`[relay] Uncaught exception: ${err.message}\n${err.stack}`) cleanupOwnedSocket() process.exit(1) }) process.on('unhandledRejection', (reason) => { - process.stderr.write(`[relay] Unhandled rejection: ${reason}\n`) + relayLogLine(`[relay] Unhandled rejection: ${reason}`) }) // Why: stdoutAlive tracks whether process.stdout is still writable. @@ -529,8 +530,8 @@ async function main(): Promise { try { await hookServer.start({ publishEndpoint: false }) } catch (err) { - process.stderr.write( - `[relay] agent-hook server failed to start: ${err instanceof Error ? err.message : String(err)}\n` + relayLogLine( + `[relay] agent-hook server failed to start: ${err instanceof Error ? err.message : String(err)}` ) } @@ -688,7 +689,7 @@ async function main(): Promise { function cancelGrace(reason: string): void { if (ptyHandler.graceTimerActive) { - process.stderr.write(`[relay] Grace canceled: ${reason}\n`) + relayLogLine(`[relay] Grace canceled: ${reason}`) } graceDeadlineAt = null graceReason = null @@ -704,8 +705,8 @@ async function main(): Promise { hasAcceptedSocketClient = true acceptedSocketConnections++ - process.stderr.write( - `[relay] Socket client accepted (clients=${socketClients.size + 1}, accepted=${acceptedSocketConnections})\n` + relayLogLine( + `[relay] Socket client accepted (clients=${socketClients.size + 1}, accepted=${acceptedSocketConnections})` ) cancelGrace('socket client accepted') @@ -779,7 +780,7 @@ async function main(): Promise { if (clientId !== undefined) { dispatcher.detachClient(clientId) } - process.stderr.write(`[relay] Socket client closed (clients=${socketClients.size})\n`) + relayLogLine(`[relay] Socket client closed (clients=${socketClients.size})`) if (!stdoutAlive && socketClients.size === 0) { startGrace('socket client closed') } @@ -821,9 +822,9 @@ async function main(): Promise { ownsSocketPath = true ownedSocketIdentity = readSocketIdentity(sockPath) server.on('error', (err) => { - process.stderr.write(`[relay] Socket server error: ${err.message}\n`) + relayLogLine(`[relay] Socket server error: ${err.message}`) }) - process.stderr.write(`[relay] Socket server listening: ${sockPath}\n`) + relayLogLine(`[relay] Socket server listening: ${sockPath}`) resolve() } @@ -831,11 +832,11 @@ async function main(): Promise { removeStartupListeners() restoreUmask() if (err.code === 'EADDRINUSE') { - process.stderr.write( - `[relay] Socket path already in use: ${sockPath}; another relay is likely active. Use --connect instead of starting a new daemon.\n` + relayLogLine( + `[relay] Socket path already in use: ${sockPath}; another relay is likely active. Use --connect instead of starting a new daemon.` ) } else { - process.stderr.write(`[relay] Socket server error before listen: ${err.message}\n`) + relayLogLine(`[relay] Socket server error before listen: ${err.message}`) } reject(err) } @@ -903,9 +904,7 @@ async function main(): Promise { failInitial(err) return } - process.stderr.write( - `[relay] Removed stale socket at ${sockPath} and retrying listen\n` - ) + relayLogLine(`[relay] Removed stale socket at ${sockPath} and retrying listen`) removeStartupListeners() listenForStartupError(failInitial) }) @@ -958,11 +957,11 @@ async function main(): Promise { : graceTimeMs graceDeadlineAt = timeoutMs === 0 ? null : Date.now() + timeoutMs graceReason = reason - process.stderr.write( - `[relay] Grace started (${reason}): timeoutMs=${timeoutMs}, startupEmptyDetached=${startupEmptyDetached}, ptys=${ptyHandler.activePtyCount}, clients=${socketClients.size}\n` + relayLogLine( + `[relay] Grace started (${reason}): timeoutMs=${timeoutMs}, startupEmptyDetached=${startupEmptyDetached}, ptys=${ptyHandler.activePtyCount}, clients=${socketClients.size}` ) ptyHandler.startGraceTimer(() => { - process.stderr.write(`[relay] Grace expired (${reason}); shutting down\n`) + relayLogLine(`[relay] Grace expired (${reason}); shutting down`) shutdown() }, timeoutMs) } @@ -1006,8 +1005,8 @@ async function main(): Promise { } function shutdown(): void { - process.stderr.write( - `[relay] Shutdown: ptys=${ptyHandler.activePtyCount}, clients=${socketClients.size}, ownsSocket=${ownsSocketPath}\n` + relayLogLine( + `[relay] Shutdown: ptys=${ptyHandler.activePtyCount}, clients=${socketClients.size}, ownsSocket=${ownsSocketPath}` ) graceDeadlineAt = null graceReason = null @@ -1034,10 +1033,10 @@ async function main(): Promise { // window — a reconnecting client can then bridge to the live relay via // --connect and reattach to the still-running PTY sessions. process.on('SIGHUP', () => { - process.stderr.write('[relay] Received SIGHUP (SSH session dropped), ignoring\n') + relayLogLine('[relay] Received SIGHUP (SSH session dropped), ignoring') }) process.on('exit', (code) => { - process.stderr.write(`[relay] Process exiting with code ${code}\n`) + relayLogLine(`[relay] Process exiting with code ${code}`) }) // Signal readiness to the client — the client watches for this exact @@ -1059,8 +1058,8 @@ function cleanupSocket(sockPath: string): void { } void main().catch((err) => { - process.stderr.write( - `[relay] Fatal startup error: ${err instanceof Error ? (err.stack ?? err.message) : String(err)}\n` + relayLogLine( + `[relay] Fatal startup error: ${err instanceof Error ? (err.stack ?? err.message) : String(err)}` ) process.exit(1) })