Major refactor: Multi-user Chrome MCP extension with remote server architecture
This commit is contained in:
nasir@endelospay.com
317
app/remote-server/src/server/livekit-agent-manager.ts
Normal file
317
app/remote-server/src/server/livekit-agent-manager.ts
Normal file
@@ -0,0 +1,317 @@
|
||||
import { Logger } from 'pino';
|
||||
import { spawn, ChildProcess } from 'child_process';
|
||||
import { SessionManager, ExtensionConnection } from './session-manager.js';
|
||||
import path from 'path';
|
||||
|
||||
export interface LiveKitAgentInstance {
|
||||
userId: string;
|
||||
sessionId: string;
|
||||
process: ChildProcess;
|
||||
roomName: string;
|
||||
startedAt: number;
|
||||
status: 'starting' | 'running' | 'stopping' | 'stopped' | 'error';
|
||||
pid?: number;
|
||||
}
|
||||
|
||||
export class LiveKitAgentManager {
|
||||
private logger: Logger;
|
||||
private sessionManager: SessionManager;
|
||||
private agentInstances: Map<string, LiveKitAgentInstance> = new Map(); // sessionId -> agent
|
||||
private userToAgent: Map<string, string> = new Map(); // userId -> sessionId
|
||||
private agentPath: string;
|
||||
private liveKitConfig: any;
|
||||
|
||||
constructor(logger: Logger, sessionManager: SessionManager, agentPath?: string) {
|
||||
this.logger = logger;
|
||||
this.sessionManager = sessionManager;
|
||||
this.agentPath = agentPath || path.join(process.cwd(), '../../agent-livekit');
|
||||
this.liveKitConfig = this.loadLiveKitConfig();
|
||||
}
|
||||
|
||||
private loadLiveKitConfig(): any {
|
||||
// Default LiveKit configuration
|
||||
return {
|
||||
livekit_url: process.env.LIVEKIT_URL || 'ws://localhost:7880',
|
||||
api_key: process.env.LIVEKIT_API_KEY || 'devkey',
|
||||
api_secret: process.env.LIVEKIT_API_SECRET || 'secret',
|
||||
room_prefix: 'mcp-chrome-user-',
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Start a LiveKit agent for a Chrome extension connection
|
||||
*/
|
||||
async startAgentForConnection(connection: ExtensionConnection): Promise<LiveKitAgentInstance> {
|
||||
const { userId, sessionId } = connection;
|
||||
|
||||
// Check if agent already exists for this user
|
||||
const existingSessionId = this.userToAgent.get(userId);
|
||||
if (existingSessionId && this.agentInstances.has(existingSessionId)) {
|
||||
const existingAgent = this.agentInstances.get(existingSessionId)!;
|
||||
if (existingAgent.status === 'running' || existingAgent.status === 'starting') {
|
||||
this.logger.info(`Agent already running for user ${userId}, reusing existing agent`);
|
||||
return existingAgent;
|
||||
}
|
||||
}
|
||||
|
||||
// Create room name based on user ID
|
||||
const roomName = `${this.liveKitConfig.room_prefix}${userId}`;
|
||||
|
||||
this.logger.info(
|
||||
`Starting LiveKit agent for user ${userId}, session ${sessionId}, room ${roomName}`,
|
||||
);
|
||||
|
||||
// Create agent instance record
|
||||
const agentInstance: LiveKitAgentInstance = {
|
||||
userId,
|
||||
sessionId,
|
||||
process: null as any, // Will be set below
|
||||
roomName,
|
||||
startedAt: Date.now(),
|
||||
status: 'starting',
|
||||
};
|
||||
|
||||
try {
|
||||
// Spawn the full LiveKit agent process directly
|
||||
const agentProcess = spawn(
|
||||
'python',
|
||||
[
|
||||
'livekit_agent.py',
|
||||
'start',
|
||||
'--url',
|
||||
this.liveKitConfig.livekit_url,
|
||||
'--api-key',
|
||||
this.liveKitConfig.api_key,
|
||||
'--api-secret',
|
||||
this.liveKitConfig.api_secret,
|
||||
],
|
||||
{
|
||||
cwd: this.agentPath,
|
||||
env: {
|
||||
...process.env,
|
||||
LIVEKIT_URL: this.liveKitConfig.livekit_url,
|
||||
LIVEKIT_API_KEY: this.liveKitConfig.api_key,
|
||||
LIVEKIT_API_SECRET: this.liveKitConfig.api_secret,
|
||||
MCP_SERVER_URL: 'http://localhost:3001/mcp',
|
||||
CHROME_USER_ID: userId, // Pass the user ID as environment variable
|
||||
// Voice processing optimization
|
||||
LIVEKIT_ROOM_NAME: roomName,
|
||||
OPENAI_API_KEY: process.env.OPENAI_API_KEY || '',
|
||||
DEEPGRAM_API_KEY: process.env.DEEPGRAM_API_KEY || '',
|
||||
},
|
||||
stdio: ['pipe', 'pipe', 'pipe'],
|
||||
},
|
||||
);
|
||||
|
||||
agentInstance.process = agentProcess;
|
||||
agentInstance.pid = agentProcess.pid;
|
||||
|
||||
// Set up process event handlers
|
||||
agentProcess.stdout?.on('data', (data) => {
|
||||
const output = data.toString();
|
||||
this.logger.info(`[Agent ${userId}] ${output.trim()}`);
|
||||
|
||||
// Check for successful startup
|
||||
if (
|
||||
output.includes('Agent initialized successfully') ||
|
||||
output.includes('LiveKit agent started')
|
||||
) {
|
||||
agentInstance.status = 'running';
|
||||
this.logger.info(`LiveKit agent for user ${userId} is now running`);
|
||||
}
|
||||
});
|
||||
|
||||
agentProcess.stderr?.on('data', (data) => {
|
||||
const error = data.toString();
|
||||
this.logger.error(`[Agent ${userId}] ERROR: ${error.trim()}`);
|
||||
});
|
||||
|
||||
agentProcess.on('close', (code) => {
|
||||
this.logger.info(`LiveKit agent for user ${userId} exited with code ${code}`);
|
||||
agentInstance.status = code === 0 ? 'stopped' : 'error';
|
||||
|
||||
// Clean up mappings
|
||||
this.agentInstances.delete(sessionId);
|
||||
this.userToAgent.delete(userId);
|
||||
});
|
||||
|
||||
agentProcess.on('error', (error) => {
|
||||
this.logger.error(`Failed to start LiveKit agent for user ${userId}:`, error);
|
||||
agentInstance.status = 'error';
|
||||
});
|
||||
|
||||
// Store the agent instance
|
||||
this.agentInstances.set(sessionId, agentInstance);
|
||||
this.userToAgent.set(userId, sessionId);
|
||||
|
||||
this.logger.info(
|
||||
`LiveKit agent process started for user ${userId} with PID ${agentProcess.pid}`,
|
||||
);
|
||||
|
||||
return agentInstance;
|
||||
} catch (error) {
|
||||
this.logger.error(`Error starting LiveKit agent for user ${userId}:`, error);
|
||||
agentInstance.status = 'error';
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop a LiveKit agent for a user
|
||||
*/
|
||||
async stopAgentForUser(userId: string): Promise<boolean> {
|
||||
const sessionId = this.userToAgent.get(userId);
|
||||
if (!sessionId) {
|
||||
this.logger.warn(`No agent found for user ${userId}`);
|
||||
return false;
|
||||
}
|
||||
|
||||
return this.stopAgentForSession(sessionId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop a LiveKit agent for a session
|
||||
*/
|
||||
async stopAgentForSession(sessionId: string): Promise<boolean> {
|
||||
const agentInstance = this.agentInstances.get(sessionId);
|
||||
if (!agentInstance) {
|
||||
this.logger.warn(`No agent found for session ${sessionId}`);
|
||||
return false;
|
||||
}
|
||||
|
||||
this.logger.info(
|
||||
`Stopping LiveKit agent for user ${agentInstance.userId}, session ${sessionId}`,
|
||||
);
|
||||
|
||||
agentInstance.status = 'stopping';
|
||||
|
||||
try {
|
||||
if (agentInstance.process && !agentInstance.process.killed) {
|
||||
// Try graceful shutdown first
|
||||
agentInstance.process.kill('SIGTERM');
|
||||
|
||||
// Force kill after 5 seconds if still running
|
||||
setTimeout(() => {
|
||||
if (agentInstance.process && !agentInstance.process.killed) {
|
||||
this.logger.warn(`Force killing LiveKit agent for user ${agentInstance.userId}`);
|
||||
agentInstance.process.kill('SIGKILL');
|
||||
}
|
||||
}, 5000);
|
||||
}
|
||||
|
||||
return true;
|
||||
} catch (error) {
|
||||
this.logger.error(`Error stopping LiveKit agent for user ${agentInstance.userId}:`, error);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle Chrome extension connection
|
||||
*/
|
||||
async onChromeExtensionConnected(connection: ExtensionConnection): Promise<void> {
|
||||
this.logger.info(
|
||||
`Chrome extension connected, starting LiveKit agent for user ${connection.userId}`,
|
||||
);
|
||||
|
||||
try {
|
||||
await this.startAgentForConnection(connection);
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to start LiveKit agent for Chrome connection:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle Chrome extension disconnection
|
||||
*/
|
||||
async onChromeExtensionDisconnected(connection: ExtensionConnection): Promise<void> {
|
||||
this.logger.info(
|
||||
`Chrome extension disconnected, stopping LiveKit agent for user ${connection.userId}`,
|
||||
);
|
||||
|
||||
try {
|
||||
await this.stopAgentForUser(connection.userId);
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to stop LiveKit agent for Chrome disconnection:`, error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get agent instance for a user
|
||||
*/
|
||||
getAgentForUser(userId: string): LiveKitAgentInstance | null {
|
||||
const sessionId = this.userToAgent.get(userId);
|
||||
return sessionId ? this.agentInstances.get(sessionId) || null : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get agent instance for a session
|
||||
*/
|
||||
getAgentForSession(sessionId: string): LiveKitAgentInstance | null {
|
||||
return this.agentInstances.get(sessionId) || null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all active agents
|
||||
*/
|
||||
getAllActiveAgents(): LiveKitAgentInstance[] {
|
||||
return Array.from(this.agentInstances.values()).filter(
|
||||
(agent) => agent.status === 'running' || agent.status === 'starting',
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get agent statistics
|
||||
*/
|
||||
getAgentStats(): any {
|
||||
const agents = Array.from(this.agentInstances.values());
|
||||
return {
|
||||
totalAgents: agents.length,
|
||||
runningAgents: agents.filter((a) => a.status === 'running').length,
|
||||
startingAgents: agents.filter((a) => a.status === 'starting').length,
|
||||
stoppedAgents: agents.filter((a) => a.status === 'stopped').length,
|
||||
errorAgents: agents.filter((a) => a.status === 'error').length,
|
||||
agentsByUser: Object.fromEntries(this.userToAgent.entries()),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Cleanup stopped agents
|
||||
*/
|
||||
cleanupStoppedAgents(): void {
|
||||
const stoppedAgents: string[] = [];
|
||||
|
||||
for (const [sessionId, agent] of this.agentInstances.entries()) {
|
||||
if (agent.status === 'stopped' || agent.status === 'error') {
|
||||
stoppedAgents.push(sessionId);
|
||||
}
|
||||
}
|
||||
|
||||
for (const sessionId of stoppedAgents) {
|
||||
const agent = this.agentInstances.get(sessionId);
|
||||
if (agent) {
|
||||
this.agentInstances.delete(sessionId);
|
||||
this.userToAgent.delete(agent.userId);
|
||||
this.logger.info(`Cleaned up stopped agent for user ${agent.userId}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Shutdown all agents
|
||||
*/
|
||||
async shutdownAllAgents(): Promise<void> {
|
||||
this.logger.info('Shutting down all LiveKit agents...');
|
||||
|
||||
const shutdownPromises = Array.from(this.agentInstances.keys()).map((sessionId) =>
|
||||
this.stopAgentForSession(sessionId),
|
||||
);
|
||||
|
||||
await Promise.all(shutdownPromises);
|
||||
|
||||
this.agentInstances.clear();
|
||||
this.userToAgent.clear();
|
||||
|
||||
this.logger.info('All LiveKit agents shut down');
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user