mirror of
https://github.com/RooVetGit/Roo-Code.git
synced 2026-09-07 08:26:51 +00:00
fix: improve IPC stability in headless Docker environments with xvfb
- Add robust error handling and recovery mechanisms to IPC server/client - Implement graceful shutdown handling for IPC connections - Add connection timeouts and retry logic for headless environments - Enhance logging for debugging in virtual display scenarios - Handle SIGTERM, SIGINT, and SIGHUP signals properly - Add socket cleanup and directory creation for Docker containers - Implement reconnection logic with exponential backoff This fix addresses the issue where VSCode gets killed (signal 9) when running RooCode in headless Docker environments with xvfb during IPC operations. The improvements make IPC communication more resilient to the challenges of virtual display environments. Fixes #7814
This commit is contained in:
Roo Code
parent
195f4eb245
commit
ba2077c40a
4 changed files with 642 additions and 103 deletions
|
|
@ -12,12 +12,27 @@ import {
|
|||
ipcMessageSchema,
|
||||
} from "@roo-code/types"
|
||||
|
||||
// Configuration for headless environments
|
||||
const HEADLESS_CONFIG = {
|
||||
// Increase retry attempts for headless environments
|
||||
maxRetries: 10,
|
||||
// Increase retry delay for slower environments
|
||||
retryDelay: 1000,
|
||||
// Connection timeout for headless environments (ms)
|
||||
connectionTimeout: 30000,
|
||||
// Enable verbose logging in headless mode
|
||||
verboseLogging: process.env.DISPLAY === ":99" || process.env.XVFB_DISPLAY !== undefined,
|
||||
}
|
||||
|
||||
export class IpcClient extends EventEmitter<IpcClientEvents> {
|
||||
private readonly _socketPath: string
|
||||
private readonly _id: string
|
||||
private readonly _log: (...args: unknown[]) => void
|
||||
private _isConnected = false
|
||||
private _clientId?: string
|
||||
private _connectionTimeout?: NodeJS.Timeout
|
||||
private _shutdownInProgress = false
|
||||
private _reconnectAttempts = 0
|
||||
|
||||
constructor(socketPath: string, log = console.log) {
|
||||
super()
|
||||
|
|
@ -26,72 +41,213 @@ export class IpcClient extends EventEmitter<IpcClientEvents> {
|
|||
this._id = `roo-code-evals-${crypto.randomBytes(6).toString("hex")}`
|
||||
this._log = log
|
||||
|
||||
// Configure IPC for headless environments
|
||||
ipc.config.silent = true
|
||||
ipc.config.retry = HEADLESS_CONFIG.retryDelay
|
||||
ipc.config.maxRetries = HEADLESS_CONFIG.maxRetries
|
||||
ipc.config.stopRetrying = false
|
||||
|
||||
ipc.connectTo(this._id, this.socketPath, () => {
|
||||
ipc.of[this._id]?.on("connect", () => this.onConnect())
|
||||
ipc.of[this._id]?.on("disconnect", () => this.onDisconnect())
|
||||
ipc.of[this._id]?.on("message", (data) => this.onMessage(data))
|
||||
})
|
||||
this.setupConnection()
|
||||
this.setupShutdownHandlers()
|
||||
}
|
||||
|
||||
private setupConnection() {
|
||||
try {
|
||||
ipc.connectTo(this._id, this.socketPath, () => {
|
||||
ipc.of[this._id]?.on("connect", () => this.onConnect())
|
||||
ipc.of[this._id]?.on("disconnect", () => this.onDisconnect())
|
||||
ipc.of[this._id]?.on("message", (data) => this.onMessage(data))
|
||||
ipc.of[this._id]?.on("error", (error) => this.onError(error))
|
||||
})
|
||||
|
||||
// Set connection timeout for headless environments
|
||||
if (HEADLESS_CONFIG.verboseLogging) {
|
||||
this._connectionTimeout = setTimeout(() => {
|
||||
if (!this._isConnected && !this._shutdownInProgress) {
|
||||
this.log(
|
||||
`[client#setupConnection] Connection timeout after ${HEADLESS_CONFIG.connectionTimeout}ms`,
|
||||
)
|
||||
this.handleConnectionFailure()
|
||||
}
|
||||
}, HEADLESS_CONFIG.connectionTimeout)
|
||||
}
|
||||
} catch (error) {
|
||||
this.log(`[client#setupConnection] Error setting up connection: ${error}`)
|
||||
this.handleConnectionFailure()
|
||||
}
|
||||
}
|
||||
|
||||
private setupShutdownHandlers() {
|
||||
const gracefulShutdown = async (signal: string) => {
|
||||
if (this._shutdownInProgress) {
|
||||
return
|
||||
}
|
||||
|
||||
this._shutdownInProgress = true
|
||||
this.log(`[IpcClient] Received ${signal}, initiating graceful shutdown...`)
|
||||
|
||||
try {
|
||||
await this.shutdown()
|
||||
} catch (error) {
|
||||
this.log(`[IpcClient] Error during shutdown: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
// Handle various termination signals
|
||||
process.once("SIGTERM", () => gracefulShutdown("SIGTERM"))
|
||||
process.once("SIGINT", () => gracefulShutdown("SIGINT"))
|
||||
process.once("SIGHUP", () => gracefulShutdown("SIGHUP"))
|
||||
}
|
||||
|
||||
private handleConnectionFailure() {
|
||||
if (this._shutdownInProgress) {
|
||||
return
|
||||
}
|
||||
|
||||
this._reconnectAttempts++
|
||||
|
||||
if (this._reconnectAttempts >= HEADLESS_CONFIG.maxRetries) {
|
||||
this.log(
|
||||
`[client#handleConnectionFailure] Max reconnection attempts (${HEADLESS_CONFIG.maxRetries}) reached`,
|
||||
)
|
||||
this.emit(IpcMessageType.Disconnect)
|
||||
return
|
||||
}
|
||||
|
||||
this.log(
|
||||
`[client#handleConnectionFailure] Attempting reconnection ${this._reconnectAttempts}/${HEADLESS_CONFIG.maxRetries}`,
|
||||
)
|
||||
|
||||
// Clear existing connection
|
||||
if (ipc.of[this._id]) {
|
||||
ipc.disconnect(this._id)
|
||||
}
|
||||
|
||||
// Wait before reconnecting
|
||||
setTimeout(() => {
|
||||
if (!this._shutdownInProgress) {
|
||||
this.setupConnection()
|
||||
}
|
||||
}, HEADLESS_CONFIG.retryDelay * this._reconnectAttempts)
|
||||
}
|
||||
|
||||
private onError(error: unknown) {
|
||||
this.log(`[client#onError] IPC client error: ${error}`)
|
||||
|
||||
// In headless environments, try to recover from errors
|
||||
if (HEADLESS_CONFIG.verboseLogging && !this._shutdownInProgress) {
|
||||
this.log("[client#onError] Attempting to recover from error in headless environment...")
|
||||
this.handleConnectionFailure()
|
||||
}
|
||||
}
|
||||
|
||||
private onConnect() {
|
||||
if (this._isConnected) {
|
||||
if (this._isConnected || this._shutdownInProgress) {
|
||||
return
|
||||
}
|
||||
|
||||
// Clear connection timeout
|
||||
if (this._connectionTimeout) {
|
||||
clearTimeout(this._connectionTimeout)
|
||||
this._connectionTimeout = undefined
|
||||
}
|
||||
|
||||
this.log("[client#onConnect]")
|
||||
this._isConnected = true
|
||||
this._reconnectAttempts = 0 // Reset reconnection attempts on successful connection
|
||||
this.emit(IpcMessageType.Connect)
|
||||
}
|
||||
|
||||
private onDisconnect() {
|
||||
if (!this._isConnected) {
|
||||
if (!this._isConnected || this._shutdownInProgress) {
|
||||
return
|
||||
}
|
||||
|
||||
this.log("[client#onDisconnect]")
|
||||
this._isConnected = false
|
||||
this._clientId = undefined
|
||||
|
||||
// Clear connection timeout
|
||||
if (this._connectionTimeout) {
|
||||
clearTimeout(this._connectionTimeout)
|
||||
this._connectionTimeout = undefined
|
||||
}
|
||||
|
||||
this.emit(IpcMessageType.Disconnect)
|
||||
|
||||
// Attempt reconnection in headless environments
|
||||
if (HEADLESS_CONFIG.verboseLogging && !this._shutdownInProgress) {
|
||||
this.log("[client#onDisconnect] Attempting reconnection in headless environment...")
|
||||
this.handleConnectionFailure()
|
||||
}
|
||||
}
|
||||
|
||||
private onMessage(data: unknown) {
|
||||
if (typeof data !== "object") {
|
||||
this._log("[client#onMessage] invalid data", data)
|
||||
if (this._shutdownInProgress) {
|
||||
this.log("[client#onMessage] Ignoring message - shutdown in progress")
|
||||
return
|
||||
}
|
||||
|
||||
const result = ipcMessageSchema.safeParse(data)
|
||||
try {
|
||||
if (typeof data !== "object") {
|
||||
this._log("[client#onMessage] invalid data", data)
|
||||
return
|
||||
}
|
||||
|
||||
if (!result.success) {
|
||||
this.log("[client#onMessage] invalid payload", result.error, data)
|
||||
return
|
||||
}
|
||||
const result = ipcMessageSchema.safeParse(data)
|
||||
|
||||
const payload = result.data
|
||||
if (!result.success) {
|
||||
this.log("[client#onMessage] invalid payload", result.error, data)
|
||||
return
|
||||
}
|
||||
|
||||
if (payload.origin === IpcOrigin.Server) {
|
||||
switch (payload.type) {
|
||||
case IpcMessageType.Ack:
|
||||
this._clientId = payload.data.clientId
|
||||
this.emit(IpcMessageType.Ack, payload.data)
|
||||
break
|
||||
case IpcMessageType.TaskEvent:
|
||||
this.emit(IpcMessageType.TaskEvent, payload.data)
|
||||
break
|
||||
const payload = result.data
|
||||
|
||||
if (payload.origin === IpcOrigin.Server) {
|
||||
switch (payload.type) {
|
||||
case IpcMessageType.Ack:
|
||||
this._clientId = payload.data.clientId
|
||||
this.emit(IpcMessageType.Ack, payload.data)
|
||||
break
|
||||
case IpcMessageType.TaskEvent:
|
||||
this.emit(IpcMessageType.TaskEvent, payload.data)
|
||||
break
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
this.log(`[client#onMessage] Error processing message: ${error}`)
|
||||
if (HEADLESS_CONFIG.verboseLogging) {
|
||||
this.log(`[client#onMessage] Message data: ${JSON.stringify(data)}`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private log(...args: unknown[]) {
|
||||
this._log(...args)
|
||||
// Add timestamp and process info in headless mode
|
||||
if (HEADLESS_CONFIG.verboseLogging) {
|
||||
const timestamp = new Date().toISOString()
|
||||
const processInfo = `[PID:${process.pid}]`
|
||||
this._log(timestamp, processInfo, ...args)
|
||||
} else {
|
||||
this._log(...args)
|
||||
}
|
||||
}
|
||||
|
||||
public sendCommand(command: TaskCommand) {
|
||||
if (this._shutdownInProgress) {
|
||||
this.log("[client#sendCommand] Cannot send command - shutdown in progress")
|
||||
return
|
||||
}
|
||||
|
||||
if (!this._clientId) {
|
||||
this.log("[client#sendCommand] Cannot send command - no client ID")
|
||||
return
|
||||
}
|
||||
|
||||
const message: IpcMessage = {
|
||||
type: IpcMessageType.TaskCommand,
|
||||
origin: IpcOrigin.Client,
|
||||
clientId: this._clientId!,
|
||||
clientId: this._clientId,
|
||||
data: command,
|
||||
}
|
||||
|
||||
|
|
@ -99,18 +255,63 @@ export class IpcClient extends EventEmitter<IpcClientEvents> {
|
|||
}
|
||||
|
||||
public sendMessage(message: IpcMessage) {
|
||||
ipc.of[this._id]?.emit("message", message)
|
||||
if (this._shutdownInProgress) {
|
||||
this.log("[client#sendMessage] Cannot send message - shutdown in progress")
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
const connection = ipc.of[this._id]
|
||||
if (connection) {
|
||||
connection.emit("message", message)
|
||||
} else {
|
||||
this.log("[client#sendMessage] IPC connection not available")
|
||||
}
|
||||
} catch (error) {
|
||||
this.log(`[client#sendMessage] Error sending message: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
public disconnect() {
|
||||
try {
|
||||
ipc.disconnect(this._id)
|
||||
// @TODO: Should we set _disconnect here?
|
||||
this._isConnected = false
|
||||
this._clientId = undefined
|
||||
|
||||
if (this._connectionTimeout) {
|
||||
clearTimeout(this._connectionTimeout)
|
||||
this._connectionTimeout = undefined
|
||||
}
|
||||
|
||||
if (ipc.of[this._id]) {
|
||||
ipc.disconnect(this._id)
|
||||
}
|
||||
} catch (error) {
|
||||
this.log("[client#disconnect] error disconnecting", error)
|
||||
}
|
||||
}
|
||||
|
||||
public async shutdown(): Promise<void> {
|
||||
this.log("[IpcClient] Starting graceful shutdown...")
|
||||
|
||||
try {
|
||||
this._shutdownInProgress = true
|
||||
|
||||
// Clear connection timeout
|
||||
if (this._connectionTimeout) {
|
||||
clearTimeout(this._connectionTimeout)
|
||||
this._connectionTimeout = undefined
|
||||
}
|
||||
|
||||
// Disconnect from server
|
||||
this.disconnect()
|
||||
|
||||
this.log("[IpcClient] Graceful shutdown completed")
|
||||
} catch (error) {
|
||||
this.log(`[IpcClient] Error during shutdown: ${error}`)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
public get socketPath() {
|
||||
return this._socketPath
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
import EventEmitter from "node:events"
|
||||
import { Socket } from "node:net"
|
||||
import * as crypto from "node:crypto"
|
||||
import * as fs from "node:fs"
|
||||
import * as path from "node:path"
|
||||
|
||||
import ipc from "node-ipc"
|
||||
|
||||
|
|
@ -13,10 +15,24 @@ import {
|
|||
ipcMessageSchema,
|
||||
} from "@roo-code/types"
|
||||
|
||||
// Configuration for headless environments
|
||||
const HEADLESS_CONFIG = {
|
||||
// Increase retry attempts for headless environments
|
||||
maxRetries: 10,
|
||||
// Increase retry delay for slower environments
|
||||
retryDelay: 1000,
|
||||
// Socket timeout for headless environments (ms)
|
||||
socketTimeout: 30000,
|
||||
// Enable verbose logging in headless mode
|
||||
verboseLogging: process.env.DISPLAY === ":99" || process.env.XVFB_DISPLAY !== undefined,
|
||||
}
|
||||
|
||||
export class IpcServer extends EventEmitter<IpcServerEvents> implements RooCodeIpcServer {
|
||||
private readonly _socketPath: string
|
||||
private readonly _log: (...args: unknown[]) => void
|
||||
private readonly _clients: Map<string, Socket>
|
||||
private _shutdownInProgress = false
|
||||
private _connectionTimeouts: Map<string, NodeJS.Timeout> = new Map()
|
||||
|
||||
private _isListening = false
|
||||
|
||||
|
|
@ -26,34 +42,197 @@ export class IpcServer extends EventEmitter<IpcServerEvents> implements RooCodeI
|
|||
this._socketPath = socketPath
|
||||
this._log = log
|
||||
this._clients = new Map()
|
||||
|
||||
// Setup graceful shutdown handlers
|
||||
this.setupShutdownHandlers()
|
||||
}
|
||||
|
||||
private setupShutdownHandlers() {
|
||||
const gracefulShutdown = async (signal: string) => {
|
||||
if (this._shutdownInProgress) {
|
||||
return
|
||||
}
|
||||
|
||||
this._shutdownInProgress = true
|
||||
this.log(`[IpcServer] Received ${signal}, initiating graceful shutdown...`)
|
||||
|
||||
try {
|
||||
await this.shutdown()
|
||||
} catch (error) {
|
||||
this.log(`[IpcServer] Error during shutdown: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
// Handle various termination signals
|
||||
process.once("SIGTERM", () => gracefulShutdown("SIGTERM"))
|
||||
process.once("SIGINT", () => gracefulShutdown("SIGINT"))
|
||||
process.once("SIGHUP", () => gracefulShutdown("SIGHUP"))
|
||||
|
||||
// Handle uncaught exceptions in headless environments
|
||||
if (HEADLESS_CONFIG.verboseLogging) {
|
||||
process.on("uncaughtException", (error) => {
|
||||
this.log(`[IpcServer] Uncaught exception in headless environment: ${error}`)
|
||||
this.log(`[IpcServer] Stack trace: ${error.stack}`)
|
||||
})
|
||||
|
||||
process.on("unhandledRejection", (reason, promise) => {
|
||||
this.log(`[IpcServer] Unhandled rejection at: ${promise}, reason: ${reason}`)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
public listen() {
|
||||
if (this._shutdownInProgress) {
|
||||
this.log("[IpcServer] Cannot start listening - shutdown in progress")
|
||||
return
|
||||
}
|
||||
|
||||
this._isListening = true
|
||||
|
||||
// Configure IPC for headless environments
|
||||
ipc.config.silent = true
|
||||
ipc.config.retry = HEADLESS_CONFIG.retryDelay
|
||||
ipc.config.maxRetries = HEADLESS_CONFIG.maxRetries
|
||||
|
||||
ipc.serve(this.socketPath, () => {
|
||||
ipc.server.on("connect", (socket) => this.onConnect(socket))
|
||||
ipc.server.on("socket.disconnected", (socket) => this.onDisconnect(socket))
|
||||
ipc.server.on("message", (data) => this.onMessage(data))
|
||||
})
|
||||
// Ensure socket directory exists (important for Docker containers)
|
||||
const socketDir = path.dirname(this.socketPath)
|
||||
if (!fs.existsSync(socketDir)) {
|
||||
try {
|
||||
fs.mkdirSync(socketDir, { recursive: true })
|
||||
this.log(`[IpcServer] Created socket directory: ${socketDir}`)
|
||||
} catch (error) {
|
||||
this.log(`[IpcServer] Failed to create socket directory: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
ipc.server.start()
|
||||
// Clean up any existing socket file
|
||||
if (fs.existsSync(this.socketPath)) {
|
||||
try {
|
||||
fs.unlinkSync(this.socketPath)
|
||||
this.log(`[IpcServer] Removed existing socket file: ${this.socketPath}`)
|
||||
} catch (error) {
|
||||
this.log(`[IpcServer] Failed to remove existing socket: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
ipc.serve(this.socketPath, () => {
|
||||
ipc.server.on("connect", (socket) => this.onConnect(socket))
|
||||
ipc.server.on("socket.disconnected", (socket) => this.onDisconnect(socket))
|
||||
ipc.server.on("message", (data) => this.onMessage(data))
|
||||
ipc.server.on("error", (error) => this.onError(error))
|
||||
})
|
||||
|
||||
ipc.server.start()
|
||||
|
||||
if (HEADLESS_CONFIG.verboseLogging) {
|
||||
this.log(`[IpcServer] Started listening on ${this.socketPath} in headless mode`)
|
||||
}
|
||||
} catch (error) {
|
||||
this.log(`[IpcServer] Failed to start IPC server: ${error}`)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
private onError(error: unknown) {
|
||||
this.log(`[IpcServer] IPC server error: ${error}`)
|
||||
|
||||
// In headless environments, try to recover from errors
|
||||
if (HEADLESS_CONFIG.verboseLogging && !this._shutdownInProgress) {
|
||||
this.log("[IpcServer] Attempting to recover from error in headless environment...")
|
||||
|
||||
// Clear all client connections
|
||||
this._clients.clear()
|
||||
this._connectionTimeouts.forEach((timeout) => clearTimeout(timeout))
|
||||
this._connectionTimeouts.clear()
|
||||
|
||||
// Emit disconnect events for cleanup
|
||||
this.emit(IpcMessageType.Disconnect, "error-recovery")
|
||||
}
|
||||
}
|
||||
|
||||
private onConnect(socket: Socket) {
|
||||
if (this._shutdownInProgress) {
|
||||
this.log("[server#onConnect] Rejecting connection - shutdown in progress")
|
||||
socket.destroy()
|
||||
return
|
||||
}
|
||||
|
||||
const clientId = crypto.randomBytes(6).toString("hex")
|
||||
this._clients.set(clientId, socket)
|
||||
this.log(`[server#onConnect] clientId = ${clientId}, # clients = ${this._clients.size}`)
|
||||
|
||||
this.send(socket, {
|
||||
type: IpcMessageType.Ack,
|
||||
origin: IpcOrigin.Server,
|
||||
data: { clientId, pid: process.pid, ppid: process.ppid },
|
||||
})
|
||||
try {
|
||||
// Configure socket for headless environments
|
||||
socket.setKeepAlive(true, 5000) // Keep-alive every 5 seconds
|
||||
socket.setTimeout(HEADLESS_CONFIG.socketTimeout)
|
||||
|
||||
this.emit(IpcMessageType.Connect, clientId)
|
||||
// Handle socket timeout
|
||||
socket.on("timeout", () => {
|
||||
this.log(`[server#onConnect] Socket timeout for client ${clientId}`)
|
||||
this.handleClientDisconnect(clientId, socket)
|
||||
})
|
||||
|
||||
// Handle socket errors
|
||||
socket.on("error", (error) => {
|
||||
this.log(`[server#onConnect] Socket error for client ${clientId}: ${error}`)
|
||||
this.handleClientDisconnect(clientId, socket)
|
||||
})
|
||||
|
||||
this._clients.set(clientId, socket)
|
||||
|
||||
// Set up connection timeout for headless environments
|
||||
if (HEADLESS_CONFIG.verboseLogging) {
|
||||
const timeout = setTimeout(() => {
|
||||
if (this._clients.has(clientId)) {
|
||||
this.log(`[server#onConnect] Client ${clientId} connection timeout in headless mode`)
|
||||
this.handleClientDisconnect(clientId, socket)
|
||||
}
|
||||
}, HEADLESS_CONFIG.socketTimeout)
|
||||
|
||||
this._connectionTimeouts.set(clientId, timeout)
|
||||
}
|
||||
|
||||
this.log(`[server#onConnect] clientId = ${clientId}, # clients = ${this._clients.size}`)
|
||||
|
||||
this.send(socket, {
|
||||
type: IpcMessageType.Ack,
|
||||
origin: IpcOrigin.Server,
|
||||
data: { clientId, pid: process.pid, ppid: process.ppid },
|
||||
})
|
||||
|
||||
this.emit(IpcMessageType.Connect, clientId)
|
||||
} catch (error) {
|
||||
this.log(`[server#onConnect] Error setting up client ${clientId}: ${error}`)
|
||||
this.handleClientDisconnect(clientId, socket)
|
||||
}
|
||||
}
|
||||
|
||||
private handleClientDisconnect(clientId: string, socket: Socket) {
|
||||
try {
|
||||
// Clear connection timeout if exists
|
||||
const timeout = this._connectionTimeouts.get(clientId)
|
||||
if (timeout) {
|
||||
clearTimeout(timeout)
|
||||
this._connectionTimeouts.delete(clientId)
|
||||
}
|
||||
|
||||
// Remove client from map
|
||||
if (this._clients.has(clientId)) {
|
||||
this._clients.delete(clientId)
|
||||
this.log(
|
||||
`[server#handleClientDisconnect] Removed client ${clientId}, # clients = ${this._clients.size}`,
|
||||
)
|
||||
}
|
||||
|
||||
// Safely destroy socket
|
||||
if (socket && !socket.destroyed) {
|
||||
socket.destroy()
|
||||
}
|
||||
|
||||
// Emit disconnect event
|
||||
this.emit(IpcMessageType.Disconnect, clientId)
|
||||
} catch (error) {
|
||||
this.log(`[server#handleClientDisconnect] Error disconnecting client ${clientId}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
private onDisconnect(destroyedSocket: Socket) {
|
||||
|
|
@ -62,65 +241,155 @@ export class IpcServer extends EventEmitter<IpcServerEvents> implements RooCodeI
|
|||
for (const [clientId, socket] of this._clients.entries()) {
|
||||
if (socket === destroyedSocket) {
|
||||
disconnectedClientId = clientId
|
||||
this._clients.delete(clientId)
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
this.log(`[server#socket.disconnected] clientId = ${disconnectedClientId}, # clients = ${this._clients.size}`)
|
||||
|
||||
if (disconnectedClientId) {
|
||||
this.emit(IpcMessageType.Disconnect, disconnectedClientId)
|
||||
this.handleClientDisconnect(disconnectedClientId, destroyedSocket)
|
||||
} else {
|
||||
this.log(`[server#socket.disconnected] Unknown socket disconnected`)
|
||||
}
|
||||
}
|
||||
|
||||
private onMessage(data: unknown) {
|
||||
if (typeof data !== "object") {
|
||||
this.log("[server#onMessage] invalid data", data)
|
||||
if (this._shutdownInProgress) {
|
||||
this.log("[server#onMessage] Ignoring message - shutdown in progress")
|
||||
return
|
||||
}
|
||||
|
||||
const result = ipcMessageSchema.safeParse(data)
|
||||
try {
|
||||
if (typeof data !== "object") {
|
||||
this.log("[server#onMessage] invalid data", data)
|
||||
return
|
||||
}
|
||||
|
||||
if (!result.success) {
|
||||
this.log("[server#onMessage] invalid payload", result.error.format(), data)
|
||||
return
|
||||
}
|
||||
const result = ipcMessageSchema.safeParse(data)
|
||||
|
||||
const payload = result.data
|
||||
if (!result.success) {
|
||||
this.log("[server#onMessage] invalid payload", result.error.format(), data)
|
||||
return
|
||||
}
|
||||
|
||||
if (payload.origin === IpcOrigin.Client) {
|
||||
switch (payload.type) {
|
||||
case IpcMessageType.TaskCommand:
|
||||
this.emit(IpcMessageType.TaskCommand, payload.clientId, payload.data)
|
||||
break
|
||||
default:
|
||||
this.log(`[server#onMessage] unhandled payload: ${JSON.stringify(payload)}`)
|
||||
break
|
||||
const payload = result.data
|
||||
|
||||
// Clear connection timeout on successful message from client
|
||||
if (payload.origin === IpcOrigin.Client && "clientId" in payload) {
|
||||
const clientId = payload.clientId
|
||||
if (clientId && this._connectionTimeouts.has(clientId)) {
|
||||
clearTimeout(this._connectionTimeouts.get(clientId)!)
|
||||
this._connectionTimeouts.delete(clientId)
|
||||
}
|
||||
|
||||
switch (payload.type) {
|
||||
case IpcMessageType.TaskCommand:
|
||||
this.emit(IpcMessageType.TaskCommand, payload.clientId, payload.data)
|
||||
break
|
||||
default:
|
||||
this.log(`[server#onMessage] unhandled payload: ${JSON.stringify(payload)}`)
|
||||
break
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
this.log(`[server#onMessage] Error processing message: ${error}`)
|
||||
if (HEADLESS_CONFIG.verboseLogging) {
|
||||
this.log(`[server#onMessage] Message data: ${JSON.stringify(data)}`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private log(...args: unknown[]) {
|
||||
this._log(...args)
|
||||
// Add timestamp and process info in headless mode
|
||||
if (HEADLESS_CONFIG.verboseLogging) {
|
||||
const timestamp = new Date().toISOString()
|
||||
const processInfo = `[PID:${process.pid}]`
|
||||
this._log(timestamp, processInfo, ...args)
|
||||
} else {
|
||||
this._log(...args)
|
||||
}
|
||||
}
|
||||
|
||||
public broadcast(message: IpcMessage) {
|
||||
// this.log("[server#broadcast] message =", message)
|
||||
ipc.server.broadcast("message", message)
|
||||
if (this._shutdownInProgress) {
|
||||
this.log("[server#broadcast] Cannot broadcast - shutdown in progress")
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
// this.log("[server#broadcast] message =", message)
|
||||
ipc.server.broadcast("message", message)
|
||||
} catch (error) {
|
||||
this.log(`[server#broadcast] Error broadcasting message: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
public send(client: string | Socket, message: IpcMessage) {
|
||||
// this.log("[server#send] message =", message)
|
||||
if (this._shutdownInProgress) {
|
||||
this.log("[server#send] Cannot send - shutdown in progress")
|
||||
return
|
||||
}
|
||||
|
||||
if (typeof client === "string") {
|
||||
const socket = this._clients.get(client)
|
||||
try {
|
||||
// this.log("[server#send] message =", message)
|
||||
|
||||
if (socket) {
|
||||
ipc.server.emit(socket, "message", message)
|
||||
if (typeof client === "string") {
|
||||
const socket = this._clients.get(client)
|
||||
|
||||
if (socket && !socket.destroyed) {
|
||||
ipc.server.emit(socket, "message", message)
|
||||
} else {
|
||||
this.log(`[server#send] Client ${client} not found or socket destroyed`)
|
||||
}
|
||||
} else {
|
||||
if (!client.destroyed) {
|
||||
ipc.server.emit(client, "message", message)
|
||||
} else {
|
||||
this.log("[server#send] Socket is destroyed")
|
||||
}
|
||||
}
|
||||
} else {
|
||||
ipc.server.emit(client, "message", message)
|
||||
} catch (error) {
|
||||
this.log(`[server#send] Error sending message: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
public async shutdown(): Promise<void> {
|
||||
this.log("[IpcServer] Starting graceful shutdown...")
|
||||
|
||||
try {
|
||||
// Clear all timeouts
|
||||
this._connectionTimeouts.forEach((timeout) => clearTimeout(timeout))
|
||||
this._connectionTimeouts.clear()
|
||||
|
||||
// Disconnect all clients gracefully
|
||||
for (const [clientId, socket] of this._clients.entries()) {
|
||||
try {
|
||||
this.log(`[IpcServer] Disconnecting client ${clientId}`)
|
||||
this.handleClientDisconnect(clientId, socket)
|
||||
} catch (error) {
|
||||
this.log(`[IpcServer] Error disconnecting client ${clientId}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
// Stop the IPC server
|
||||
if (ipc.server) {
|
||||
ipc.server.stop()
|
||||
}
|
||||
|
||||
// Clean up socket file
|
||||
if (fs.existsSync(this.socketPath)) {
|
||||
try {
|
||||
fs.unlinkSync(this.socketPath)
|
||||
this.log(`[IpcServer] Removed socket file: ${this.socketPath}`)
|
||||
} catch (error) {
|
||||
this.log(`[IpcServer] Failed to remove socket file: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
this._isListening = false
|
||||
this.log("[IpcServer] Graceful shutdown completed")
|
||||
} catch (error) {
|
||||
this.log(`[IpcServer] Error during shutdown: ${error}`)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -358,6 +358,28 @@ export async function activate(context: vscode.ExtensionContext) {
|
|||
export async function deactivate() {
|
||||
outputChannel.appendLine(`${Package.name} extension deactivated`)
|
||||
|
||||
// Store the API instance if it was returned from activate
|
||||
let apiInstance: API | undefined
|
||||
|
||||
// Get the API instance from the extension exports
|
||||
const extension = vscode.extensions.getExtension(Package.name)
|
||||
if (extension && extension.isActive) {
|
||||
apiInstance = extension.exports as API
|
||||
}
|
||||
|
||||
// Clean up API resources (including IPC server)
|
||||
if (apiInstance && typeof apiInstance.cleanup === "function") {
|
||||
try {
|
||||
outputChannel.appendLine("Cleaning up API resources...")
|
||||
await apiInstance.cleanup()
|
||||
outputChannel.appendLine("API cleanup completed")
|
||||
} catch (error) {
|
||||
outputChannel.appendLine(
|
||||
`Failed to clean up API resources: ${error instanceof Error ? error.message : String(error)}`,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (cloudService && CloudService.hasInstance()) {
|
||||
try {
|
||||
if (authStateChangedHandler) {
|
||||
|
|
|
|||
|
|
@ -60,39 +60,67 @@ export class API extends EventEmitter<RooCodeEvents> implements RooCodeAPI {
|
|||
this.registerListeners(this.sidebarProvider)
|
||||
|
||||
if (socketPath) {
|
||||
const ipc = (this.ipc = new IpcServer(socketPath, this.log))
|
||||
try {
|
||||
const ipc = (this.ipc = new IpcServer(socketPath, this.log))
|
||||
|
||||
ipc.listen()
|
||||
this.log(`[API] ipc server started: socketPath=${socketPath}, pid=${process.pid}, ppid=${process.ppid}`)
|
||||
ipc.listen()
|
||||
this.log(`[API] ipc server started: socketPath=${socketPath}, pid=${process.pid}, ppid=${process.ppid}`)
|
||||
|
||||
ipc.on(IpcMessageType.TaskCommand, async (_clientId, { commandName, data }) => {
|
||||
switch (commandName) {
|
||||
case TaskCommandName.StartNewTask:
|
||||
this.log(`[API] StartNewTask -> ${data.text}, ${JSON.stringify(data.configuration)}`)
|
||||
await this.startNewTask(data)
|
||||
break
|
||||
case TaskCommandName.CancelTask:
|
||||
this.log(`[API] CancelTask -> ${data}`)
|
||||
await this.cancelTask(data)
|
||||
break
|
||||
case TaskCommandName.CloseTask:
|
||||
this.log(`[API] CloseTask -> ${data}`)
|
||||
await vscode.commands.executeCommand("workbench.action.files.saveFiles")
|
||||
await vscode.commands.executeCommand("workbench.action.closeWindow")
|
||||
break
|
||||
case TaskCommandName.ResumeTask:
|
||||
this.log(`[API] ResumeTask -> ${data}`)
|
||||
try {
|
||||
await this.resumeTask(data)
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
this.log(`[API] ResumeTask failed for taskId ${data}: ${errorMessage}`)
|
||||
// Don't rethrow - we want to prevent IPC server crashes
|
||||
// The error is logged for debugging purposes
|
||||
}
|
||||
break
|
||||
// Log environment info for debugging headless issues
|
||||
if (process.env.DISPLAY || process.env.XVFB_DISPLAY) {
|
||||
this.log(
|
||||
`[API] Running in headless environment - DISPLAY=${process.env.DISPLAY}, XVFB_DISPLAY=${process.env.XVFB_DISPLAY}`,
|
||||
)
|
||||
}
|
||||
})
|
||||
|
||||
ipc.on(IpcMessageType.TaskCommand, async (_clientId, { commandName, data }) => {
|
||||
try {
|
||||
switch (commandName) {
|
||||
case TaskCommandName.StartNewTask:
|
||||
this.log(`[API] StartNewTask -> ${data.text}, ${JSON.stringify(data.configuration)}`)
|
||||
await this.startNewTask(data)
|
||||
break
|
||||
case TaskCommandName.CancelTask:
|
||||
this.log(`[API] CancelTask -> ${data}`)
|
||||
await this.cancelTask(data)
|
||||
break
|
||||
case TaskCommandName.CloseTask:
|
||||
this.log(`[API] CloseTask -> ${data}`)
|
||||
await vscode.commands.executeCommand("workbench.action.files.saveFiles")
|
||||
await vscode.commands.executeCommand("workbench.action.closeWindow")
|
||||
break
|
||||
case TaskCommandName.ResumeTask:
|
||||
this.log(`[API] ResumeTask -> ${data}`)
|
||||
try {
|
||||
await this.resumeTask(data)
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
this.log(`[API] ResumeTask failed for taskId ${data}: ${errorMessage}`)
|
||||
// Don't rethrow - we want to prevent IPC server crashes
|
||||
// The error is logged for debugging purposes
|
||||
}
|
||||
break
|
||||
}
|
||||
} catch (error) {
|
||||
// Catch any unhandled errors to prevent IPC server crashes
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
this.log(`[API] Error handling IPC command ${commandName}: ${errorMessage}`)
|
||||
}
|
||||
})
|
||||
|
||||
// Handle IPC server errors
|
||||
ipc.on(IpcMessageType.Disconnect, (clientId) => {
|
||||
this.log(`[API] IPC client disconnected: ${clientId}`)
|
||||
})
|
||||
|
||||
ipc.on(IpcMessageType.Connect, (clientId) => {
|
||||
this.log(`[API] IPC client connected: ${clientId}`)
|
||||
})
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
this.log(`[API] Failed to initialize IPC server: ${errorMessage}`)
|
||||
// Continue without IPC server - extension should still work
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -443,4 +471,23 @@ export class API extends EventEmitter<RooCodeEvents> implements RooCodeAPI {
|
|||
await this.sidebarProvider.activateProviderProfile({ name })
|
||||
return this.getActiveProfile()
|
||||
}
|
||||
|
||||
// Cleanup method for graceful shutdown
|
||||
public async cleanup(): Promise<void> {
|
||||
try {
|
||||
// Shutdown IPC server if it exists
|
||||
if (this.ipc && "shutdown" in this.ipc && typeof this.ipc.shutdown === "function") {
|
||||
this.log("[API] Shutting down IPC server...")
|
||||
await (this.ipc as any).shutdown()
|
||||
}
|
||||
|
||||
// Clear task map
|
||||
this.taskMap.clear()
|
||||
|
||||
this.log("[API] Cleanup completed")
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
this.log(`[API] Error during cleanup: ${errorMessage}`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue