diff --git a/gitnexus-mcp/package.json b/gitnexus-mcp/package.json index 1c9b29d5f..4203ff69f 100644 --- a/gitnexus-mcp/package.json +++ b/gitnexus-mcp/package.json @@ -31,15 +31,17 @@ }, "dependencies": { "@modelcontextprotocol/sdk": "^1.0.0", + "uuid": "^13.0.0", "ws": "^8.16.0" }, "devDependencies": { "@types/node": "^20.0.0", + "@types/uuid": "^10.0.0", "@types/ws": "^8.5.10", - "typescript": "^5.4.0", - "tsx": "^4.0.0" + "tsx": "^4.0.0", + "typescript": "^5.4.0" }, "engines": { "node": ">=18.0.0" } -} \ No newline at end of file +} diff --git a/gitnexus-mcp/src/bridge/protocol.ts b/gitnexus-mcp/src/bridge/protocol.ts index 1578158ac..1d0d63fe4 100644 --- a/gitnexus-mcp/src/bridge/protocol.ts +++ b/gitnexus-mcp/src/bridge/protocol.ts @@ -4,28 +4,28 @@ * JSON-RPC-like protocol for communication between bridge and browser. */ -export interface ToolCallRequest { - id: string; - method: string; - params: Record; -} - -export interface ToolCallResponse { +export interface BridgeMessage { id: string; + type?: 'register_peer' | 'tool_call' | 'tool_result' | 'agent_info' | 'handshake' | 'handshake_ack' | 'context'; + method?: string; + params?: any; result?: any; error?: { code?: number; message: string; }; + agentName?: string; + peerId?: string; } -export type BridgeMessage = ToolCallRequest | ToolCallResponse; +export type ToolCallRequest = BridgeMessage & { method: string }; +export type ToolCallResponse = BridgeMessage & ({ result: any } | { error: any }); /** * Check if message is a request (has method) */ export function isRequest(msg: BridgeMessage): msg is ToolCallRequest { - return 'method' in msg; + return typeof msg.method === 'string'; } /** diff --git a/gitnexus-mcp/src/bridge/websocket-server.ts b/gitnexus-mcp/src/bridge/websocket-server.ts index 927c9dfe1..450414941 100644 --- a/gitnexus-mcp/src/bridge/websocket-server.ts +++ b/gitnexus-mcp/src/bridge/websocket-server.ts @@ -1,22 +1,7 @@ -/** - * WebSocket Bridge - * - * WebSocket server that connects to the GitNexus browser tab. - * Relays tool calls from MCP server to browser and returns results. - */ - import { WebSocketServer, WebSocket } from 'ws'; import { createServer as createNetServer } from 'net'; - -export interface BridgeMessage { - id: string; - method?: string; - params?: any; - result?: any; - error?: { message: string }; - type?: 'context' | string; - agentName?: string; -} +import { BridgeMessage, isRequest, isResponse } from './protocol.js'; +import { v4 as uuidv4 } from 'uuid'; /** * Codebase context sent from the GitNexus browser app @@ -39,13 +24,8 @@ export interface CodebaseContext { folderTree: string; } -type RequestResolver = { - resolve: (result: any) => void; - reject: (error: Error) => void; -}; - /** - * Check if a port is available + * Check if a Port is available */ async function isPortAvailable(port: number): Promise { return new Promise((resolve) => { @@ -60,135 +40,298 @@ async function isPortAvailable(port: number): Promise { } export class WebSocketBridge { - private wss: WebSocketServer | null = null; - private client: WebSocket | null = null; - private pendingRequests: Map = new Map(); + private wss: WebSocketServer | null = null; // Used if we are the Hub + private client: WebSocket | null = null; // Used if we are a Peer (connecting to Hub), OR if we are Hub (clients connecting to us) + + // Hub State + private browserClient: WebSocket | null = null; + private peerClients: Map = new Map(); + + // Common State + private pendingRequests: Map void, reject: (err: any) => void }> = new Map(); private requestId = 0; private started = false; - private _context: CodebaseContext | null = null; - private contextListeners: Set<(context: CodebaseContext | null) => void> = new Set(); + private _context: any | null = null; // CodebaseContext + private contextListeners: Set<(context: any | null) => void> = new Set(); private agentName: string; + private isHub = false; + private port = 54319; - constructor(private port: number = 54319, agentName?: string) { + constructor(port: number = 54319, agentName?: string) { + this.port = port; this.agentName = agentName || process.env.GITNEXUS_AGENT || this.detectAgent(); } private detectAgent(): string { - // Try to detect agent from environment clues if (process.env.CURSOR_SESSION_ID) return 'Cursor'; if (process.env.CLAUDE_CODE) return 'Claude Code'; if (process.env.WINDSURF_SESSION) return 'Windsurf'; - return 'Unknown'; + return 'Unknown Agent'; } - /** - * Start the WebSocket server (handles port-in-use gracefully) - */ - /** - * Start the WebSocket server (handles port-in-use gracefully by scanning range) - */ async start(): Promise { - const MAX_RETRIES = 10; + const available = await isPortAvailable(this.port); - for (let i = 0; i <= MAX_RETRIES; i++) { - const currentPort = this.port + i; - const available = await isPortAvailable(currentPort); + if (available) { + return this.startAsHub(); + } else { + return this.startAsPeer(); + } + } + + // ------------------------------------------------------------------------- + // Hub Implementation (Master) + // ------------------------------------------------------------------------- + + private async startAsHub(): Promise { + console.error(`Starting as MCP Hub on port ${this.port}`); + this.isHub = true; + + return new Promise((resolve) => { + this.wss = new WebSocketServer({ port: this.port }); - if (available) { - return new Promise((resolve) => { - this.wss = new WebSocketServer({ port: currentPort }); - - this.wss.on('connection', (ws) => { - // Only allow one browser connection at a time - if (this.client) { - this.client.close(); + this.wss.on('connection', (ws, req) => { + // Security: Origin check could go here if req.headers.origin available + + ws.on('message', (data) => this.handleHubMessage(ws, data)); + ws.on('close', () => this.handleHubDisconnect(ws)); + ws.on('error', (err) => console.error('Hub client error:', err)); + }); + + this.wss.on('listening', () => { + this.started = true; + resolve(true); + }); + + this.wss.on('error', (err) => { + console.error('Hub server error:', err); + resolve(false); + }); + }); + } + + private handleHubMessage(ws: WebSocket, data: any) { + try { + const msg: BridgeMessage = JSON.parse(data.toString()); + + if (msg.type === 'handshake') { + // Peer verifying we are GitNexus + ws.send(JSON.stringify({ type: 'handshake_ack', id: msg.id })); + return; + } + + if (msg.type === 'register_peer') { + // Peer registering itself + const peerId = uuidv4(); + this.peerClients.set(peerId, ws); + (ws as any).peerId = peerId; + (ws as any).agentName = msg.agentName; + console.error(`Peer connected: ${msg.agentName} (${peerId})`); + + // Forward current context to new peer if available + if (this._context) { + ws.send(JSON.stringify({ type: 'context', params: this._context })); + } + return; + } + + // Handle Context updates (from Browser) + if (msg.type === 'context') { + // Browser identified itself (implicitly) + if (this.browserClient !== ws) { + if (this.browserClient) this.browserClient.close(); + this.browserClient = ws; + console.error('Browser connected to Hub'); + } + + this._context = msg.params; + this.notifyContextListeners(); + + // Broadcast context to all peers + this.broadcastToPeers(msg); + return; + } + + // Handle Tool Calls (Peer/Hub -> Browser) + if (isRequest(msg)) { + // If it came from a ws client (Peer), validation needed? + // We assume it's destined for the Browser + if (this.browserClient && this.browserClient.readyState === WebSocket.OPEN) { + // Attach agent info if missing (for UI) + if (!msg.agentName && (ws as any).agentName) { + msg.agentName = (ws as any).agentName; + } + // Attach peerId so we can route response back + if (!msg.peerId && (ws as any).peerId) { + msg.peerId = (ws as any).peerId; } - this.client = ws; - // Clear context until browser sends an update - this._context = null; - this.notifyContextListeners(); - ws.on('message', (data) => { - try { - const msg: BridgeMessage = JSON.parse(data.toString()); - this.handleMessage(msg); - } catch (error) { - console.error('Failed to parse message:', error); - } - }); - - ws.on('close', () => { - if (this.client === ws) { - this.client = null; - this._context = null; - this.notifyContextListeners(); - } - }); - - ws.on('error', (error) => { - console.error('WebSocket error:', error); - }); - }); - - this.wss.on('listening', () => { - this.started = true; - // Update the port property to reflect the actual bound port - this.port = currentPort; - console.error(`Browser bridge listening on port ${currentPort}`); // Use stderr to not interfere with MCP stdio - resolve(true); - }); - - this.wss.on('error', (error) => { - console.error(`WebSocket server error on port ${currentPort}:`, error); - resolve(false); - }); - }); + this.browserClient.send(JSON.stringify(msg)); + } else { + // Browser not connected, fail + if (msg.id) { + ws.send(JSON.stringify({ + id: msg.id, + error: { message: "Browser not connected. Open GitNexus." } + })); + } + } + return; } + + // Handle Tool Results (Browser -> Peer/Hub) + if (isResponse(msg)) { + // Route to the correct peer + if (msg.peerId && this.peerClients.has(msg.peerId)) { + const peer = this.peerClients.get(msg.peerId); + if (peer?.readyState === WebSocket.OPEN) { + peer.send(JSON.stringify(msg)); + } + } else { + // It might be for Us (the Hub) + this.handleResponseLocal(msg); + } + return; + } + + } catch (e) { + console.error('Hub: Failed to parse message', e); + } + } + + private handleHubDisconnect(ws: WebSocket) { + if (ws === this.browserClient) { + console.error('Browser disconnected from Hub'); + this.browserClient = null; + this._context = null; + this.notifyContextListeners(); + } else { + const peerId = (ws as any).peerId; + if (peerId) { + this.peerClients.delete(peerId); + console.error(`Peer disconnected: ${peerId}`); + } + } + } + + private broadcastToPeers(msg: any) { + for (const client of this.peerClients.values()) { + if (client.readyState === WebSocket.OPEN) { + client.send(JSON.stringify(msg)); + } + } + } + + // ------------------------------------------------------------------------- + // Peer Implementation (Spoke) + // ------------------------------------------------------------------------- + + private async startAsPeer(): Promise { + console.error(`Port ${this.port} busy. Attempting to connect as Peer...`); + + return new Promise((resolve) => { + const ws = new WebSocket(`ws://localhost:${this.port}`); + + const timeout = setTimeout(() => { + console.error('Handshake timeout. Port is busy by unknown app.'); + ws.close(); + resolve(false); + }, 1000); + + ws.on('open', () => { + // Send Handshake + ws.send(JSON.stringify({ type: 'handshake', id: 'init' })); + }); + + ws.on('message', (data) => { + try { + const msg = JSON.parse(data.toString()); + + // Handshake success? + if (msg.type === 'handshake_ack') { + clearTimeout(timeout); + console.error('Handshake successful. Joining as Peer.'); + + // Register ourselves + ws.send(JSON.stringify({ + type: 'register_peer', + agentName: this.agentName + })); + + this.client = ws; + this.started = true; + resolve(true); + return; + } + + // Normal messages from Hub + this.handlePeerMessage(msg); + + } catch (e) { + // ignore garbage + } + }); + + ws.on('error', (err) => { + console.error('Peer connection error:', err); + resolve(false); + }); + + // If connection fails immediately + ws.on('close', () => { + if (!this.started) resolve(false); + else { + this.client = null; + this._context = null; + this.notifyContextListeners(); + } + }); + }); + } + + private handlePeerMessage(msg: BridgeMessage) { + if (msg.type === 'context') { + this._context = msg.params; + this.notifyContextListeners(); + return; } - console.error(`Failed to find available port in range ${this.port}-${this.port + MAX_RETRIES}`); - return false; - } - - private handleMessage(msg: BridgeMessage) { - // Browser can proactively send codebase context - if (msg.type === 'context' && msg.params) { - this._context = msg.params as CodebaseContext; - this.notifyContextListeners(); - return; + if (isResponse(msg)) { + this.handleResponseLocal(msg); } + } - // This is a response to a pending request + // ------------------------------------------------------------------------- + // Shared / Public API + // ------------------------------------------------------------------------- + + private handleResponseLocal(msg: any) { if (msg.id && this.pendingRequests.has(msg.id)) { - const { resolve, reject } = this.pendingRequests.get(msg.id)!; - this.pendingRequests.delete(msg.id); - - if (msg.error) { - reject(new Error(msg.error.message)); - } else { - resolve(msg.result); - } + const { resolve, reject } = this.pendingRequests.get(msg.id)!; + this.pendingRequests.delete(msg.id); + + if (msg.error) { + // We'll reject the promise so caller knows + reject(new Error(msg.error.message)); + } else { + resolve(msg.result); + } } } - /** - * Check if browser is connected - */ get isConnected(): boolean { - return this.client !== null && this.client.readyState === WebSocket.OPEN; + if (this.isHub) { + return this.browserClient !== null && this.browserClient.readyState === WebSocket.OPEN; + } else { + return this.client !== null && this.client.readyState === WebSocket.OPEN; + } } - /** - * Latest context received from browser (if any) - */ - get context(): CodebaseContext | null { + get context(): any { return this._context; } - /** - * Listen for context changes - */ - onContextChange(listener: (context: CodebaseContext | null) => void) { + onContextChange(listener: (context: any) => void) { this.contextListeners.add(listener); return () => this.contextListeners.delete(listener); } @@ -196,20 +339,11 @@ export class WebSocketBridge { private notifyContextListeners() { this.contextListeners.forEach((listener) => listener(this._context)); } - - /** - * Check if server started successfully - */ - get isStarted(): boolean { - return this.started; - } - /** - * Call a tool in the browser - */ async callTool(method: string, params: any): Promise { if (!this.isConnected) { - throw new Error('GitNexus browser not connected. Open GitNexus and enable MCP toggle.'); + if (this.isHub) throw new Error('GitNexus Browser not connected.'); + else throw new Error('GitNexus Hub disonnected.'); } const id = `req_${++this.requestId}`; @@ -217,29 +351,46 @@ export class WebSocketBridge { return new Promise((resolve, reject) => { this.pendingRequests.set(id, { resolve, reject }); - const msg: BridgeMessage = { id, method, params, agentName: this.agentName }; - this.client!.send(JSON.stringify(msg)); + const msg: BridgeMessage = { + id, + method, + params, + agentName: this.agentName, + // type is implicitly request because of method + }; + + if (this.isHub) { + // Send directly to browser + if (this.browserClient && this.browserClient.readyState === WebSocket.OPEN) { + this.browserClient.send(JSON.stringify(msg)); + } else { + this.pendingRequests.delete(id); + reject(new Error('Browser not connected')); + } + } else { + // Send to Hub (who forwards to browser) + if (this.client && this.client.readyState === WebSocket.OPEN) { + this.client.send(JSON.stringify(msg)); + } else { + this.pendingRequests.delete(id); + reject(new Error('Hub disconnected')); + } + } - // Timeout after 30 seconds setTimeout(() => { if (this.pendingRequests.has(id)) { - this.pendingRequests.delete(id); - reject(new Error('Request timeout')); + this.pendingRequests.delete(id); + reject(new Error('Request timeout')); } }, 30000); }); } - /** - * Close the WebSocket server - */ close() { this.wss?.close(); + this.client?.close(); } - /** - * MCP server calls this on shutdown - */ disconnect() { this.close(); } diff --git a/gitnexus/src/core/mcp/mcp-client.ts b/gitnexus/src/core/mcp/mcp-client.ts index d55eb4b3e..a546331f9 100644 --- a/gitnexus/src/core/mcp/mcp-client.ts +++ b/gitnexus/src/core/mcp/mcp-client.ts @@ -85,115 +85,89 @@ type ToolHandler = (params: Record) => Promise; type ActivityListener = (event: ActivityEvent) => void; export class MCPBrowserClient { - private sockets: Map = new Map(); + private ws: WebSocket | null = null; private handlers: Map = new Map(); private connectionListeners: Set<(connected: boolean) => void> = new Set(); private activityListeners: Set = new Set(); private activityLog: ActivityEvent[] = []; private pendingContext: CodebaseContext | null = null; - private _connectedAgents: Map = new Map(); + private _connectedAgent: ConnectedAgent | null = null; - constructor(private startPort = 54319, private endPort = 54329) {} + constructor(private port = 54319) {} /** - * Connect to all available MCP daemons in the port range + * Connect to the MCP daemon */ async connect(): Promise { - const promises: Promise[] = []; - - for (let port = this.startPort; port <= this.endPort; port++) { - promises.push(this.connectToPort(port)); - } - - await Promise.all(promises); - - if (this.sockets.size === 0) { - throw new Error('Failed to connect to any MCP bridge'); - } - - console.log(`[MCP] Connected to ${this.sockets.size} daemon(s)`); - } - - private connectToPort(port: number): Promise { - return new Promise((resolve) => { + return new Promise((resolve, reject) => { try { - const ws = new WebSocket(`ws://localhost:${port}`); + this.ws = new WebSocket(`ws://localhost:${this.port}`); - ws.onopen = () => { - console.log(`[MCP] Connected to daemon on port ${port}`); - this.sockets.set(port, ws); + this.ws.onopen = () => { + console.log('[MCP] Connected to daemon'); this.notifyConnectionListeners(true); // Send pending context if available if (this.pendingContext) { - this.sendContextToSocket(ws, this.pendingContext); + this.sendContext(this.pendingContext); } resolve(); }; - ws.onerror = () => { - // Just resolve on error to continue checking other ports - resolve(); + this.ws.onerror = () => { + this.notifyConnectionListeners(false); + reject(new Error('Failed to connect to MCP bridge')); }; - ws.onmessage = (event) => { + this.ws.onmessage = (event) => { try { const msg: MCPMessage = JSON.parse(event.data); - this.handleMessage(msg, port, ws); + this.handleMessage(msg); } catch (error) { console.error('[MCP] Failed to parse message:', error); } }; - ws.onclose = () => { - this.sockets.delete(port); - this._connectedAgents.delete(port); - if (this.sockets.size === 0) { - this.notifyConnectionListeners(false); - } + this.ws.onclose = () => { + this.ws = null; + this.notifyConnectionListeners(false); }; } catch (error) { - resolve(); + reject(error); } }); } /** - * Send codebase context to all connected daemons + * Send codebase context to daemon * Call this whenever context changes (new repo loaded, etc.) */ sendContext(context: CodebaseContext) { this.pendingContext = context; - for (const ws of this.sockets.values()) { - this.sendContextToSocket(ws, context); - } - console.log(`[MCP] Sent context to ${this.sockets.size} daemon(s):`, context.projectName); - } - - private sendContextToSocket(ws: WebSocket, context: CodebaseContext) { - if (ws.readyState === WebSocket.OPEN) { + if (this.ws?.readyState === WebSocket.OPEN) { const msg = { id: `ctx_${Date.now()}`, type: 'context', params: context, }; - ws.send(JSON.stringify(msg)); + this.ws.send(JSON.stringify(msg)); + console.log('[MCP] Sent context:', context.projectName); } } /** - * Handle incoming messages from a daemon + * Handle incoming messages from daemon */ - private async handleMessage(msg: MCPMessage, port: number, ws: WebSocket) { + private async handleMessage(msg: MCPMessage) { // Handle agent info updates if (msg.type === 'agent_info' && msg.agentName) { - this._connectedAgents.set(port, { + this._connectedAgent = { name: msg.agentName, color: getAgentColor(msg.agentName), - }); - console.log(`[MCP] Agent connected on port ${port}:`, msg.agentName); + }; + console.log('[MCP] Agent connected:', this._connectedAgent); return; } @@ -202,9 +176,8 @@ export class MCPBrowserClient { const handler = this.handlers.get(msg.method); const startTime = Date.now(); - // Get agent info from message or use connected agent for this port - const connectedAgent = this._connectedAgents.get(port); - const agentName = msg.agentName || connectedAgent?.name || 'Unknown'; + // Get agent info from message or use connected agent + const agentName = msg.agentName || this._connectedAgent?.name || 'Unknown'; const agentColor = getAgentColor(agentName); // Create activity event with agent info @@ -222,7 +195,7 @@ export class MCPBrowserClient { if (handler) { try { const result = await handler(msg.params || {}); - this.sendToSocket(ws, { id: msg.id, result }); + this.send({ id: msg.id, result }); // Update activity with success this.updateActivity(msg.id, { @@ -232,7 +205,7 @@ export class MCPBrowserClient { }); } catch (error) { const message = error instanceof Error ? error.message : 'Unknown error'; - this.sendToSocket(ws, { id: msg.id, error: { message } }); + this.send({ id: msg.id, error: { message } }); // Update activity with error this.updateActivity(msg.id, { @@ -242,7 +215,7 @@ export class MCPBrowserClient { }); } } else { - this.sendToSocket(ws, { + this.send({ id: msg.id, error: { message: `Unknown tool: ${msg.method}` } }); @@ -281,11 +254,11 @@ export class MCPBrowserClient { } /** - * Send a message to a specific socket + * Send a message to the daemon */ - private sendToSocket(ws: WebSocket, msg: MCPMessage) { - if (ws.readyState === WebSocket.OPEN) { - ws.send(JSON.stringify(msg)); + private send(msg: MCPMessage) { + if (this.ws?.readyState === WebSocket.OPEN) { + this.ws.send(JSON.stringify(msg)); } } @@ -335,31 +308,25 @@ export class MCPBrowserClient { } /** - * Check if connected to at least one daemon + * Check if connected */ get isConnected(): boolean { - return this.sockets.size > 0; + return this.ws?.readyState === WebSocket.OPEN; } /** - * Get the connected agent info (returns first available) + * Get the connected agent info */ get connectedAgent(): ConnectedAgent | null { - if (this._connectedAgents.size > 0) { - return this._connectedAgents.values().next().value || null; - } - return null; + return this._connectedAgent; } /** - * Disconnect from all daemons + * Disconnect from daemon */ disconnect() { - for (const ws of this.sockets.values()) { - ws.close(); - } - this.sockets.clear(); - this._connectedAgents.clear(); + this.ws?.close(); + this.ws = null; } }