From 2323a576a68547ee2c0f7e5840ccdc5474b767c8 Mon Sep 17 00:00:00 2001 From: Haileyesus <118998054+blackmammoth@users.noreply.github.com> Date: Sat, 25 Apr 2026 15:39:22 +0300 Subject: [PATCH] refactor(websocket): move websocket logic to its own module --- package-lock.json | 11 + package.json | 1 + server/index.js | 727 ++---------------- .../projects/services/projects.service.ts | 7 +- .../services/sessions-watcher.service.ts | 3 +- server/modules/websocket/README.md | 267 +++++++ server/modules/websocket/index.ts | 2 + .../services/chat-websocket.service.ts | 271 +++++++ .../plugin-websocket-proxy.service.ts | 65 ++ .../services/shell-websocket.service.ts | 453 +++++++++++ .../services/websocket-auth.service.ts | 54 ++ .../services/websocket-server.service.ts | 58 ++ .../services/websocket-state.service.ts | 16 + .../services/websocket-writer.service.ts | 38 + server/shared/types.ts | 39 + server/shared/utils.ts | 56 ++ 16 files changed, 1407 insertions(+), 661 deletions(-) create mode 100644 server/modules/websocket/README.md create mode 100644 server/modules/websocket/index.ts create mode 100644 server/modules/websocket/services/chat-websocket.service.ts create mode 100644 server/modules/websocket/services/plugin-websocket-proxy.service.ts create mode 100644 server/modules/websocket/services/shell-websocket.service.ts create mode 100644 server/modules/websocket/services/websocket-auth.service.ts create mode 100644 server/modules/websocket/services/websocket-server.service.ts create mode 100644 server/modules/websocket/services/websocket-state.service.ts create mode 100644 server/modules/websocket/services/websocket-writer.service.ts diff --git a/package-lock.json b/package-lock.json index 2bdf35ab..b5a0ba87 100644 --- a/package-lock.json +++ b/package-lock.json @@ -80,6 +80,7 @@ "@types/node": "^22.19.7", "@types/react": "^18.2.43", "@types/react-dom": "^18.2.17", + "@types/ws": "^8.18.1", "@vitejs/plugin-react": "^4.6.0", "auto-changelog": "^2.5.0", "autoprefixer": "^10.4.16", @@ -4142,6 +4143,16 @@ "integrity": "sha512-ko/gIFJRv177XgZsZcBwnqJN5x/Gien8qNOn0D5bQU/zAzVf9Zt3BlcUiLqhV9y4ARk0GbT3tnUiPNgnTXzc/Q==", "license": "MIT" }, + "node_modules/@types/ws": { + "version": "8.18.1", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz", + "integrity": "sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@typescript-eslint/eslint-plugin": { "version": "8.56.1", "resolved": "https://registry.npmjs.org/@typescript-eslint/eslint-plugin/-/eslint-plugin-8.56.1.tgz", diff --git a/package.json b/package.json index 3388f7a4..eeb9d6b6 100644 --- a/package.json +++ b/package.json @@ -132,6 +132,7 @@ "@types/node": "^22.19.7", "@types/react": "^18.2.43", "@types/react-dom": "^18.2.17", + "@types/ws": "^8.18.1", "@vitejs/plugin-react": "^4.6.0", "auto-changelog": "^2.5.0", "autoprefixer": "^10.4.16", diff --git a/server/index.js b/server/index.js index 9fa6abf6..9061c2ba 100755 --- a/server/index.js +++ b/server/index.js @@ -5,7 +5,7 @@ import fs from 'fs'; import path from 'path'; import { findAppRoot, getModuleDir } from './utils/runtime-paths.js'; -import { AppError, createNormalizedMessage } from '@/shared/utils.js'; +import { AppError } from '@/shared/utils.js'; const __dirname = getModuleDir(import.meta.url); @@ -19,15 +19,15 @@ import { c } from './utils/colors.js'; console.log('SERVER_PORT from env:', process.env.SERVER_PORT); import express from 'express'; -import { WebSocketServer, WebSocket } from 'ws'; import os from 'os'; import http from 'http'; import cors from 'cors'; import { promises as fsPromises } from 'fs'; import { spawn } from 'child_process'; -import pty from 'node-pty'; import mime from 'mime-types'; +import { closeSessionsWatcher, initializeSessionsWatcher } from '@/modules/providers/index.js'; import { getProjectsWithSessions } from '@/modules/projects/index.js'; +import { createWebSocketServer } from '@/modules/websocket/index.js'; import { getSessionsById, @@ -38,11 +38,40 @@ import { getProjectPathById, searchConversations, } from './projects.js'; -import { queryClaudeSDK, abortClaudeSDKSession, isClaudeSDKSessionActive, getActiveClaudeSDKSessions, resolveToolApproval, getPendingApprovalsForSession, reconnectSessionWriter } from './claude-sdk.js'; -import { spawnCursor, abortCursorSession, isCursorSessionActive, getActiveCursorSessions } from './cursor-cli.js'; -import { queryCodex, abortCodexSession, isCodexSessionActive, getActiveCodexSessions } from './openai-codex.js'; -import { spawnGemini, abortGeminiSession, isGeminiSessionActive, getActiveGeminiSessions } from './gemini-cli.js'; +import { + queryClaudeSDK, + abortClaudeSDKSession, + isClaudeSDKSessionActive, + getActiveClaudeSDKSessions, + resolveToolApproval, + getPendingApprovalsForSession, + reconnectSessionWriter, +} from './claude-sdk.js'; +import { + spawnCursor, + abortCursorSession, + isCursorSessionActive, + getActiveCursorSessions, +} from './cursor-cli.js'; +import { + queryCodex, + abortCodexSession, + isCodexSessionActive, + getActiveCodexSessions, +} from './openai-codex.js'; +import { + spawnGemini, + abortGeminiSession, + isGeminiSessionActive, + getActiveGeminiSessions, +} from './gemini-cli.js'; import sessionManager from './sessionManager.js'; +import { + stripAnsiSequences, + normalizeDetectedUrl, + extractUrlsFromText, + shouldAutoOpenUrlFromOutput, +} from './utils/url-detection.js'; import gitRoutes from './routes/git.js'; import authRoutes from './routes/auth.js'; import cursorRoutes from './routes/cursor.js'; @@ -67,53 +96,44 @@ import { getConnectableHost } from '../shared/networkHosts.js'; const VALID_PROVIDERS = ['claude', 'codex', 'cursor', 'gemini']; -export const connectedClients = new Set(); - const app = express(); const server = http.createServer(app); -const ptySessionsMap = new Map(); -const PTY_SESSION_TIMEOUT = 30 * 60 * 1000; -const SHELL_URL_PARSE_BUFFER_LIMIT = 32768; -import { stripAnsiSequences, normalizeDetectedUrl, extractUrlsFromText, shouldAutoOpenUrlFromOutput } from './utils/url-detection.js'; -import { closeSessionsWatcher, initializeSessionsWatcher } from '@/modules/providers/index.js'; - -// Single WebSocket server that handles both paths -const wss = new WebSocketServer({ - server, - verifyClient: (info) => { - console.log('WebSocket connection attempt to:', info.req.url); - - // Platform mode: always allow connection - if (IS_PLATFORM) { - const user = authenticateWebSocket(null); // Will return first user - if (!user) { - console.log('[WARN] Platform mode: No user found in database'); - return false; - } - info.req.user = user; - console.log('[OK] Platform mode WebSocket authenticated for user:', user.username); - return true; - } - - // Normal mode: verify token - // Extract token from query parameters or headers - const url = new URL(info.req.url, 'http://localhost'); - const token = url.searchParams.get('token') || - info.req.headers.authorization?.split(' ')[1]; - - // Verify token - const user = authenticateWebSocket(token); - if (!user) { - console.log('[WARN] WebSocket authentication failed'); - return false; - } - - // Store user info in the request for later use - info.req.user = user; - console.log('[OK] WebSocket authenticated for user:', user.username); - return true; - } +// Single WebSocket server that handles chat, shell, and plugin proxy paths. +const wss = createWebSocketServer(server, { + verifyClient: { + isPlatform: IS_PLATFORM, + authenticateWebSocket, + }, + chat: { + queryClaudeSDK, + spawnCursor, + queryCodex, + spawnGemini, + abortClaudeSDKSession, + abortCursorSession, + abortCodexSession, + abortGeminiSession, + resolveToolApproval, + isClaudeSDKSessionActive, + isCursorSessionActive, + isCodexSessionActive, + isGeminiSessionActive, + reconnectSessionWriter, + getPendingApprovalsForSession, + getActiveClaudeSDKSessions, + getActiveCursorSessions, + getActiveCodexSessions, + getActiveGeminiSessions, + }, + shell: { + getSessionById: (sessionId) => sessionManager.getSession(sessionId), + stripAnsiSequences, + normalizeDetectedUrl, + extractUrlsFromText, + shouldAutoOpenUrlFromOutput, + }, + getPluginPort, }); // Make WebSocket server available to routes @@ -1175,611 +1195,6 @@ const uploadFilesHandler = async (req, res) => { app.post('/api/projects/:projectId/files/upload', authenticateToken, uploadFilesHandler); -/** - * Proxy an authenticated client WebSocket to a plugin's internal WS server. - * Auth is enforced by verifyClient before this function is reached. - */ -function handlePluginWsProxy(clientWs, pathname) { - const pluginName = pathname.replace('/plugin-ws/', ''); - if (!pluginName || /[^a-zA-Z0-9_-]/.test(pluginName)) { - clientWs.close(4400, 'Invalid plugin name'); - return; - } - - const port = getPluginPort(pluginName); - if (!port) { - clientWs.close(4404, 'Plugin not running'); - return; - } - - const upstream = new WebSocket(`ws://127.0.0.1:${port}/ws`); - - upstream.on('open', () => { - console.log(`[Plugins] WS proxy connected to "${pluginName}" on port ${port}`); - }); - - // Relay messages bidirectionally - upstream.on('message', (data) => { - if (clientWs.readyState === WebSocket.OPEN) clientWs.send(data); - }); - clientWs.on('message', (data) => { - if (upstream.readyState === WebSocket.OPEN) upstream.send(data); - }); - - // Propagate close in both directions - upstream.on('close', () => { if (clientWs.readyState === WebSocket.OPEN) clientWs.close(); }); - clientWs.on('close', () => { if (upstream.readyState === WebSocket.OPEN) upstream.close(); }); - - upstream.on('error', (err) => { - console.error(`[Plugins] WS proxy error for "${pluginName}":`, err.message); - if (clientWs.readyState === WebSocket.OPEN) clientWs.close(4502, 'Upstream error'); - }); - clientWs.on('error', () => { - if (upstream.readyState === WebSocket.OPEN) upstream.close(); - }); -} - -// WebSocket connection handler that routes based on URL path -wss.on('connection', (ws, request) => { - const url = request.url; - console.log('[INFO] Client connected to:', url); - - // Parse URL to get pathname without query parameters - const urlObj = new URL(url, 'http://localhost'); - const pathname = urlObj.pathname; - - if (pathname === '/shell') { - handleShellConnection(ws); - } else if (pathname === '/ws') { - handleChatConnection(ws, request); - } else if (pathname.startsWith('/plugin-ws/')) { - handlePluginWsProxy(ws, pathname); - } else { - console.log('[WARN] Unknown WebSocket path:', pathname); - ws.close(); - } -}); - -/** - * WebSocket Writer - Wrapper for WebSocket to match SSEStreamWriter interface - * - * Provider files use `createNormalizedMessage()` from `shared/utils.js` and - * adapter `normalizeMessage()` to produce unified NormalizedMessage events. - * The writer simply serialises and sends. - */ -class WebSocketWriter { - constructor(ws, userId = null) { - this.ws = ws; - this.sessionId = null; - this.userId = userId; - this.isWebSocketWriter = true; // Marker for transport detection - } - - send(data) { - if (this.ws.readyState === 1) { // WebSocket.OPEN - this.ws.send(JSON.stringify(data)); - } - } - - updateWebSocket(newRawWs) { - this.ws = newRawWs; - } - - setSessionId(sessionId) { - this.sessionId = sessionId; - } - - getSessionId() { - return this.sessionId; - } -} - -// Handle chat WebSocket connections -function handleChatConnection(ws, request) { - console.log('[INFO] Chat WebSocket connected'); - - // Add to connected clients for project updates - connectedClients.add(ws); - - // Wrap WebSocket with writer for consistent interface with SSEStreamWriter - const writer = new WebSocketWriter(ws, request?.user?.id ?? request?.user?.userId ?? null); - - ws.on('message', async (message) => { - try { - const data = JSON.parse(message); - - if (data.type === 'claude-command') { - console.log('[DEBUG] User message:', data.command || '[Continue/Resume]'); - console.log('๐Ÿ“ Project:', data.options?.projectPath || 'Unknown'); - console.log('๐Ÿ”„ Session:', data.options?.sessionId ? 'Resume' : 'New'); - - // Use Claude Agents SDK - await queryClaudeSDK(data.command, data.options, writer); - } else if (data.type === 'cursor-command') { - console.log('[DEBUG] Cursor message:', data.command || '[Continue/Resume]'); - console.log('๐Ÿ“ Project:', data.options?.cwd || 'Unknown'); - console.log('๐Ÿ”„ Session:', data.options?.sessionId ? 'Resume' : 'New'); - console.log('๐Ÿค– Model:', data.options?.model || 'default'); - await spawnCursor(data.command, data.options, writer); - } else if (data.type === 'codex-command') { - console.log('[DEBUG] Codex message:', data.command || '[Continue/Resume]'); - console.log('๐Ÿ“ Project:', data.options?.projectPath || data.options?.cwd || 'Unknown'); - console.log('๐Ÿ”„ Session:', data.options?.sessionId ? 'Resume' : 'New'); - console.log('๐Ÿค– Model:', data.options?.model || 'default'); - await queryCodex(data.command, data.options, writer); - } else if (data.type === 'gemini-command') { - console.log('[DEBUG] Gemini message:', data.command || '[Continue/Resume]'); - console.log('๐Ÿ“ Project:', data.options?.projectPath || data.options?.cwd || 'Unknown'); - console.log('๐Ÿ”„ Session:', data.options?.sessionId ? 'Resume' : 'New'); - console.log('๐Ÿค– Model:', data.options?.model || 'default'); - await spawnGemini(data.command, data.options, writer); - } else if (data.type === 'cursor-resume') { - // Backward compatibility: treat as cursor-command with resume and no prompt - console.log('[DEBUG] Cursor resume session (compat):', data.sessionId); - await spawnCursor('', { - sessionId: data.sessionId, - resume: true, - cwd: data.options?.cwd - }, writer); - } else if (data.type === 'abort-session') { - console.log('[DEBUG] Abort session request:', data.sessionId); - const provider = data.provider || 'claude'; - let success; - - if (provider === 'cursor') { - success = abortCursorSession(data.sessionId); - } else if (provider === 'codex') { - success = abortCodexSession(data.sessionId); - } else if (provider === 'gemini') { - success = abortGeminiSession(data.sessionId); - } else { - // Use Claude Agents SDK - success = await abortClaudeSDKSession(data.sessionId); - } - - writer.send(createNormalizedMessage({ kind: 'complete', exitCode: success ? 0 : 1, aborted: true, success, sessionId: data.sessionId, provider })); - } else if (data.type === 'claude-permission-response') { - // Relay UI approval decisions back into the SDK control flow. - // This does not persist permissions; it only resolves the in-flight request, - // introduced so the SDK can resume once the user clicks Allow/Deny. - if (data.requestId) { - resolveToolApproval(data.requestId, { - allow: Boolean(data.allow), - updatedInput: data.updatedInput, - message: data.message, - rememberEntry: data.rememberEntry - }); - } - } else if (data.type === 'cursor-abort') { - console.log('[DEBUG] Abort Cursor session:', data.sessionId); - const success = abortCursorSession(data.sessionId); - writer.send(createNormalizedMessage({ kind: 'complete', exitCode: success ? 0 : 1, aborted: true, success, sessionId: data.sessionId, provider: 'cursor' })); - } else if (data.type === 'check-session-status') { - // Check if a specific session is currently processing - const provider = data.provider || 'claude'; - const sessionId = data.sessionId; - let isActive; - - if (provider === 'cursor') { - isActive = isCursorSessionActive(sessionId); - } else if (provider === 'codex') { - isActive = isCodexSessionActive(sessionId); - } else if (provider === 'gemini') { - isActive = isGeminiSessionActive(sessionId); - } else { - // Use Claude Agents SDK - isActive = isClaudeSDKSessionActive(sessionId); - if (isActive) { - // Reconnect the session's writer to the new WebSocket so - // subsequent SDK output flows to the refreshed client. - reconnectSessionWriter(sessionId, ws); - } - } - - writer.send({ - type: 'session-status', - sessionId, - provider, - isProcessing: isActive - }); - } else if (data.type === 'get-pending-permissions') { - // Return pending permission requests for a session - const sessionId = data.sessionId; - if (sessionId && isClaudeSDKSessionActive(sessionId)) { - const pending = getPendingApprovalsForSession(sessionId); - writer.send({ - type: 'pending-permissions-response', - sessionId, - data: pending - }); - } - } else if (data.type === 'get-active-sessions') { - // Get all currently active sessions - const activeSessions = { - claude: getActiveClaudeSDKSessions(), - cursor: getActiveCursorSessions(), - codex: getActiveCodexSessions(), - gemini: getActiveGeminiSessions() - }; - writer.send({ - type: 'active-sessions', - sessions: activeSessions - }); - } - } catch (error) { - console.error('[ERROR] Chat WebSocket error:', error.message); - writer.send({ - type: 'error', - error: error.message - }); - } - }); - - ws.on('close', () => { - console.log('๐Ÿ”Œ Chat client disconnected'); - // Remove from connected clients - connectedClients.delete(ws); - }); -} - -// Handle shell WebSocket connections -function handleShellConnection(ws) { - console.log('๐Ÿš Shell client connected'); - let shellProcess = null; - let ptySessionKey = null; - let urlDetectionBuffer = ''; - const announcedAuthUrls = new Set(); - - ws.on('message', async (message) => { - try { - const data = JSON.parse(message); - console.log('๐Ÿ“จ Shell message received:', data.type); - - if (data.type === 'init') { - const projectPath = data.projectPath || process.cwd(); - const sessionId = data.sessionId; - const hasSession = data.hasSession; - const provider = data.provider || 'claude'; - const initialCommand = data.initialCommand; - const isPlainShell = data.isPlainShell || (!!initialCommand && !hasSession) || provider === 'plain-shell'; - urlDetectionBuffer = ''; - announcedAuthUrls.clear(); - - // Login commands (Claude/Cursor auth) should never reuse cached sessions - const isLoginCommand = initialCommand && ( - initialCommand.includes('setup-token') || - initialCommand.includes('cursor-agent login') || - initialCommand.includes('auth login') - ); - - // Include command hash in session key so different commands get separate sessions - const commandSuffix = isPlainShell && initialCommand - ? `_cmd_${Buffer.from(initialCommand).toString('base64').slice(0, 16)}` - : ''; - ptySessionKey = `${projectPath}_${sessionId || 'default'}${commandSuffix}`; - - // Kill any existing login session before starting fresh - if (isLoginCommand) { - const oldSession = ptySessionsMap.get(ptySessionKey); - if (oldSession) { - console.log('๐Ÿงน Cleaning up existing login session:', ptySessionKey); - if (oldSession.timeoutId) clearTimeout(oldSession.timeoutId); - if (oldSession.pty && oldSession.pty.kill) oldSession.pty.kill(); - ptySessionsMap.delete(ptySessionKey); - } - } - - const existingSession = isLoginCommand ? null : ptySessionsMap.get(ptySessionKey); - if (existingSession) { - console.log('โ™ป๏ธ Reconnecting to existing PTY session:', ptySessionKey); - shellProcess = existingSession.pty; - - clearTimeout(existingSession.timeoutId); - - ws.send(JSON.stringify({ - type: 'output', - data: `\x1b[36m[Reconnected to existing session]\x1b[0m\r\n` - })); - - if (existingSession.buffer && existingSession.buffer.length > 0) { - console.log(`๐Ÿ“œ Sending ${existingSession.buffer.length} buffered messages`); - existingSession.buffer.forEach(bufferedData => { - ws.send(JSON.stringify({ - type: 'output', - data: bufferedData - })); - }); - } - - existingSession.ws = ws; - - return; - } - - console.log('[INFO] Starting shell in:', projectPath); - console.log('๐Ÿ“‹ Session info:', hasSession ? `Resume session ${sessionId}` : (isPlainShell ? 'Plain shell mode' : 'New session')); - console.log('๐Ÿค– Provider:', isPlainShell ? 'plain-shell' : provider); - if (initialCommand) { - console.log('โšก Initial command:', initialCommand); - } - - // First send a welcome message - let welcomeMsg; - if (isPlainShell) { - welcomeMsg = `\x1b[36mStarting terminal in: ${projectPath}\x1b[0m\r\n`; - } else { - const providerName = provider === 'cursor' ? 'Cursor' : (provider === 'codex' ? 'Codex' : (provider === 'gemini' ? 'Gemini' : 'Claude')); - welcomeMsg = hasSession ? - `\x1b[36mResuming ${providerName} session ${sessionId} in: ${projectPath}\x1b[0m\r\n` : - `\x1b[36mStarting new ${providerName} session in: ${projectPath}\x1b[0m\r\n`; - } - - ws.send(JSON.stringify({ - type: 'output', - data: welcomeMsg - })); - - try { - // Validate projectPath โ€” resolve to absolute and verify it exists - const resolvedProjectPath = path.resolve(projectPath); - try { - const stats = fs.statSync(resolvedProjectPath); - if (!stats.isDirectory()) { - throw new Error('Not a directory'); - } - } catch (pathErr) { - ws.send(JSON.stringify({ type: 'error', message: 'Invalid project path' })); - return; - } - - // Validate sessionId โ€” only allow safe characters - const safeSessionIdPattern = /^[a-zA-Z0-9_.\-:]+$/; - if (sessionId && !safeSessionIdPattern.test(sessionId)) { - ws.send(JSON.stringify({ type: 'error', message: 'Invalid session ID' })); - return; - } - - // Build shell command โ€” use cwd for project path (never interpolate into shell string) - let shellCommand; - if (isPlainShell) { - // Plain shell mode - run the initial command in the project directory - shellCommand = initialCommand; - } else if (provider === 'cursor') { - if (hasSession && sessionId) { - shellCommand = `cursor-agent --resume="${sessionId}"`; - } else { - shellCommand = 'cursor-agent'; - } - } else if (provider === 'codex') { - // Use codex command; attempt to resume and fall back to a new session when the resume fails. - if (hasSession && sessionId) { - if (os.platform() === 'win32') { - // PowerShell syntax for fallback - shellCommand = `codex resume "${sessionId}"; if ($LASTEXITCODE -ne 0) { codex }`; - } else { - shellCommand = `codex resume "${sessionId}" || codex`; - } - } else { - shellCommand = 'codex'; - } - } else if (provider === 'gemini') { - const command = initialCommand || 'gemini'; - let resumeId = sessionId; - if (hasSession && sessionId) { - try { - // Gemini CLI enforces its own native session IDs, unlike other agents that accept arbitrary string names. - // The UI only knows about its internal generated `sessionId` (e.g. gemini_1234). - // We must fetch the mapping from the backend session manager to pass the native `cliSessionId` to the shell. - const sess = sessionManager.getSession(sessionId); - if (sess && sess.cliSessionId) { - resumeId = sess.cliSessionId; - // Validate the looked-up CLI session ID too - if (!safeSessionIdPattern.test(resumeId)) { - resumeId = null; - } - } - } catch (err) { - console.error('Failed to get Gemini CLI session ID:', err); - } - } - - if (hasSession && resumeId) { - shellCommand = `${command} --resume "${resumeId}"`; - } else { - shellCommand = command; - } - } else { - // Claude (default provider) - const command = initialCommand || 'claude'; - if (hasSession && sessionId) { - if (os.platform() === 'win32') { - shellCommand = `claude --resume "${sessionId}"; if ($LASTEXITCODE -ne 0) { claude }`; - } else { - shellCommand = `claude --resume "${sessionId}" || claude`; - } - } else { - shellCommand = command; - } - } - - console.log('๐Ÿ”ง Executing shell command:', shellCommand); - - // Use appropriate shell based on platform - const shell = os.platform() === 'win32' ? 'powershell.exe' : 'bash'; - const shellArgs = os.platform() === 'win32' ? ['-Command', shellCommand] : ['-c', shellCommand]; - - // Use terminal dimensions from client if provided, otherwise use defaults - const termCols = data.cols || 80; - const termRows = data.rows || 24; - console.log('๐Ÿ“ Using terminal dimensions:', termCols, 'x', termRows); - - shellProcess = pty.spawn(shell, shellArgs, { - name: 'xterm-256color', - cols: termCols, - rows: termRows, - cwd: resolvedProjectPath, - env: { - ...process.env, - TERM: 'xterm-256color', - COLORTERM: 'truecolor', - FORCE_COLOR: '3' - } - }); - - console.log('๐ŸŸข Shell process started with PTY, PID:', shellProcess.pid); - - ptySessionsMap.set(ptySessionKey, { - pty: shellProcess, - ws: ws, - buffer: [], - timeoutId: null, - projectPath, - sessionId - }); - - // Handle data output - shellProcess.onData((data) => { - const session = ptySessionsMap.get(ptySessionKey); - if (!session) return; - - if (session.buffer.length < 5000) { - session.buffer.push(data); - } else { - session.buffer.shift(); - session.buffer.push(data); - } - - if (session.ws && session.ws.readyState === WebSocket.OPEN) { - let outputData = data; - - const cleanChunk = stripAnsiSequences(data); - urlDetectionBuffer = `${urlDetectionBuffer}${cleanChunk}`.slice(-SHELL_URL_PARSE_BUFFER_LIMIT); - - outputData = outputData.replace( - /OPEN_URL:\s*(https?:\/\/[^\s\x1b\x07]+)/g, - '[INFO] Opening in browser: $1' - ); - - const emitAuthUrl = (detectedUrl, autoOpen = false) => { - const normalizedUrl = normalizeDetectedUrl(detectedUrl); - if (!normalizedUrl) return; - - const isNewUrl = !announcedAuthUrls.has(normalizedUrl); - if (isNewUrl) { - announcedAuthUrls.add(normalizedUrl); - session.ws.send(JSON.stringify({ - type: 'auth_url', - url: normalizedUrl, - autoOpen - })); - } - - }; - - const normalizedDetectedUrls = extractUrlsFromText(urlDetectionBuffer) - .map((url) => normalizeDetectedUrl(url)) - .filter(Boolean); - - // Prefer the most complete URL if shorter prefix variants are also present. - const dedupedDetectedUrls = Array.from(new Set(normalizedDetectedUrls)).filter((url, _, urls) => - !urls.some((otherUrl) => otherUrl !== url && otherUrl.startsWith(url)) - ); - - dedupedDetectedUrls.forEach((url) => emitAuthUrl(url, false)); - - if (shouldAutoOpenUrlFromOutput(cleanChunk) && dedupedDetectedUrls.length > 0) { - const bestUrl = dedupedDetectedUrls.reduce((longest, current) => - current.length > longest.length ? current : longest - ); - emitAuthUrl(bestUrl, true); - } - - // Send regular output - session.ws.send(JSON.stringify({ - type: 'output', - data: outputData - })); - } - }); - - // Handle process exit - shellProcess.onExit((exitCode) => { - console.log('๐Ÿ”š Shell process exited with code:', exitCode.exitCode, 'signal:', exitCode.signal); - const session = ptySessionsMap.get(ptySessionKey); - if (session && session.ws && session.ws.readyState === WebSocket.OPEN) { - session.ws.send(JSON.stringify({ - type: 'output', - data: `\r\n\x1b[33mProcess exited with code ${exitCode.exitCode}${exitCode.signal ? ` (${exitCode.signal})` : ''}\x1b[0m\r\n` - })); - } - if (session && session.timeoutId) { - clearTimeout(session.timeoutId); - } - ptySessionsMap.delete(ptySessionKey); - shellProcess = null; - }); - - } catch (spawnError) { - console.error('[ERROR] Error spawning process:', spawnError); - ws.send(JSON.stringify({ - type: 'output', - data: `\r\n\x1b[31mError: ${spawnError.message}\x1b[0m\r\n` - })); - } - - } else if (data.type === 'input') { - // Send input to shell process - if (shellProcess && shellProcess.write) { - try { - shellProcess.write(data.data); - } catch (error) { - console.error('Error writing to shell:', error); - } - } else { - console.warn('No active shell process to send input to'); - } - } else if (data.type === 'resize') { - // Handle terminal resize - if (shellProcess && shellProcess.resize) { - console.log('Terminal resize requested:', data.cols, 'x', data.rows); - shellProcess.resize(data.cols, data.rows); - } - } - } catch (error) { - console.error('[ERROR] Shell WebSocket error:', error.message); - if (ws.readyState === WebSocket.OPEN) { - ws.send(JSON.stringify({ - type: 'output', - data: `\r\n\x1b[31mError: ${error.message}\x1b[0m\r\n` - })); - } - } - }); - - ws.on('close', () => { - console.log('๐Ÿ”Œ Shell client disconnected'); - - if (ptySessionKey) { - const session = ptySessionsMap.get(ptySessionKey); - if (session) { - console.log('โณ PTY session kept alive, will timeout in 30 minutes:', ptySessionKey); - session.ws = null; - - session.timeoutId = setTimeout(() => { - console.log('โฐ PTY session timeout, killing process:', ptySessionKey); - if (session.pty && session.pty.kill) { - session.pty.kill(); - } - ptySessionsMap.delete(ptySessionKey); - }, PTY_SESSION_TIMEOUT); - } - } - }); - - ws.on('error', (error) => { - console.error('[ERROR] Shell WebSocket error:', error); - }); -} // Image upload endpoint. Accepts the DB-assigned `projectId` (not a folder name) // but the current implementation doesn't need to touch the project directory, // so we just leave the param rename for consistency with the rest of the API. diff --git a/server/modules/projects/services/projects.service.ts b/server/modules/projects/services/projects.service.ts index 62d6aee7..d7227952 100644 --- a/server/modules/projects/services/projects.service.ts +++ b/server/modules/projects/services/projects.service.ts @@ -3,8 +3,9 @@ import path from 'node:path'; import { projectsDb, sessionsDb } from '@/modules/database/index.js'; import { sessionSynchronizerService } from '@/modules/providers/index.js'; +import { WS_OPEN_STATE, connectedClients } from '@/modules/websocket/index.js'; +import type { RealtimeClientConnection } from '@/shared/types.js'; import { findAppRoot, getModuleDir } from '@/utils/runtime-paths.js'; -import { connectedClients } from '@/index.js'; type SessionSummary = { id: string; @@ -183,8 +184,8 @@ function broadcastProgress(progress: ProgressUpdate) { ...progress, }); - connectedClients.forEach((client: any) => { - if (client.readyState === WebSocket.OPEN) { + connectedClients.forEach((client: RealtimeClientConnection) => { + if (client.readyState === WS_OPEN_STATE) { client.send(message); } }); diff --git a/server/modules/providers/services/sessions-watcher.service.ts b/server/modules/providers/services/sessions-watcher.service.ts index 911f6e64..4e2ef422 100644 --- a/server/modules/providers/services/sessions-watcher.service.ts +++ b/server/modules/providers/services/sessions-watcher.service.ts @@ -5,9 +5,9 @@ import { promises as fsPromises } from 'node:fs'; import chokidar, { type FSWatcher } from 'chokidar'; import { sessionSynchronizerService } from '@/modules/providers/services/session-synchronizer.service.js'; +import { WS_OPEN_STATE, connectedClients } from '@/modules/websocket/index.js'; import type { LLMProvider } from '@/shared/types.js'; import { getProjectsWithSessions } from '@/modules/projects/index.js'; -import { connectedClients } from '@/index.js'; type WatcherEventType = 'add' | 'change'; @@ -45,7 +45,6 @@ const WATCHER_IGNORED_PATTERNS = [ ]; const watchers: FSWatcher[] = []; -const WS_OPEN_STATE = 1; /** * Filters watcher events to provider-specific session artifact file types. diff --git a/server/modules/websocket/README.md b/server/modules/websocket/README.md new file mode 100644 index 00000000..76d8e7b1 --- /dev/null +++ b/server/modules/websocket/README.md @@ -0,0 +1,267 @@ +# WebSocket Module + +This module owns the server-side WebSocket gateway used by: + +1. Chat streaming (`/ws`) +2. Interactive terminal sessions (`/shell`) +3. Plugin WebSocket passthrough (`/plugin-ws/:pluginName`) + +It is intentionally structured as **small services** plus a **barrel export** in `index.ts`. + +## Public API + +`server/modules/websocket/index.ts` exports: + +1. `createWebSocketServer(server, dependencies)` +Creates and wires the shared `ws` server. +2. `connectedClients` and `WS_OPEN_STATE` +Shared chat client registry and open-state constant used by other modules. + +## Why Dependency Injection Is Used + +The module receives runtime-specific functions from `server/index.js` instead of importing legacy runtime files directly. + +Benefits: + +1. Keeps module boundaries clean (`server/modules/*` architecture rule). +2. Makes each service easier to test in isolation. +3. Keeps WebSocket transport concerns separate from provider runtime concerns. + +## Service Map + +| File | Responsibility | +|---|---| +| `services/websocket-server.service.ts` | Creates `WebSocketServer`, binds `verifyClient`, routes connection by pathname | +| `services/websocket-auth.service.ts` | Authenticates upgrade requests and attaches `request.user` | +| `services/chat-websocket.service.ts` | Handles `/ws` chat protocol and provider command/session control messages | +| `services/shell-websocket.service.ts` | Handles `/shell` PTY lifecycle, reconnect buffering, auth URL detection | +| `services/plugin-websocket-proxy.service.ts` | Bridges client socket to plugin socket | +| `services/websocket-writer.service.ts` | Adapts raw WebSocket to writer interface (`send`, `setSessionId`, `getSessionId`) | +| `services/websocket-state.service.ts` | Holds shared chat client set and open-state constant | + +## High-Level Architecture + +```mermaid +flowchart LR + A[HTTP Server] --> B[createWebSocketServer] + B --> C[verifyWebSocketClient] + B --> D{Pathname} + D -->|/ws| E[handleChatConnection] + D -->|/shell| F[handleShellConnection] + D -->|/plugin-ws/:name| G[handlePluginWsProxy] + D -->|other| H[close()] + + E --> I[connectedClients Set] + E --> J[WebSocketWriter] + F --> K[ptySessionsMap] + G --> L[Upstream Plugin ws://127.0.0.1:port/ws] + + I --> M[projects.service broadcastProgress] + I --> N[sessions-watcher.service projects_updated] +``` + +## Connection Handshake + Routing + +```mermaid +sequenceDiagram + participant Client + participant WSS as WebSocketServer + participant Auth as verifyWebSocketClient + participant Router as connection router + participant Chat as /ws handler + participant Shell as /shell handler + participant Proxy as /plugin-ws handler + + Client->>WSS: Upgrade Request + WSS->>Auth: verifyClient(info) + alt Platform mode + Auth->>Auth: authenticateWebSocket(null) + Auth->>Auth: attach request.user + else OSS mode + Auth->>Auth: read token from ?token or Authorization + Auth->>Auth: authenticateWebSocket(token) + Auth->>Auth: attach request.user + end + + alt Auth failed + Auth-->>WSS: false (reject handshake) + else Auth ok + Auth-->>WSS: true + WSS->>Router: on("connection", ws, request) + alt pathname == /ws + Router->>Chat: handleChatConnection(ws, request, deps.chat) + else pathname == /shell + Router->>Shell: handleShellConnection(ws, deps.shell) + else pathname startsWith /plugin-ws/ + Router->>Proxy: handlePluginWsProxy(ws, pathname, getPluginPort) + else unknown + Router->>Router: ws.close() + end + end +``` + +## `/ws` Chat Flow + +When a chat socket connects: + +1. Add socket to `connectedClients`. +2. Build `WebSocketWriter` (captures `userId` from authenticated request). +3. Parse each incoming message with `parseIncomingJsonObject`. +4. Dispatch by `data.type`. +5. On close, remove socket from `connectedClients`. + +### Chat Message Dispatch + +```mermaid +flowchart TD + A[Incoming WS message] --> B[parseIncomingJsonObject] + B -->|invalid| C[send {type:error}] + B -->|ok| D{data.type} + + D -->|claude-command| E[queryClaudeSDK] + D -->|cursor-command| F[spawnCursor] + D -->|codex-command| G[queryCodex] + D -->|gemini-command| H[spawnGemini] + D -->|cursor-resume| I[spawnCursor resume] + D -->|abort-session| J[abort by provider] + D -->|claude-permission-response| K[resolveToolApproval] + D -->|cursor-abort| L[abortCursorSession] + D -->|check-session-status| M[is*SessionActive + optional reconnectSessionWriter] + D -->|get-pending-permissions| N[getPendingApprovalsForSession] + D -->|get-active-sessions| O[getActive*Sessions] +``` + +### Chat Notes + +1. `abort-session` returns a normalized `complete` message with `aborted: true`. +2. `check-session-status` returns `{ type: "session-status", isProcessing }`. +3. Claude status checks can reconnect output stream to the new socket via `reconnectSessionWriter`. + +## `/shell` Terminal Flow + +The shell handler manages persistent PTY sessions keyed by: + +`_[_cmd_]` + +This enables reconnect behavior and isolates command-specific plain-shell sessions. + +### Shell Lifecycle + +```mermaid +stateDiagram-v2 + [*] --> WaitingInit + WaitingInit --> ValidateInit: message.type == init + ValidateInit --> ReconnectExisting: session key exists and not login reset + ValidateInit --> SpawnNewPTY: valid path + valid sessionId + ValidateInit --> EmitError: invalid payload/path/sessionId + + ReconnectExisting --> Running: attach ws, replay buffer + SpawnNewPTY --> Running: pty.spawn + wire onData/onExit + + Running --> Running: input -> pty.write + Running --> Running: resize -> pty.resize + Running --> Running: onData -> buffer + output + auth_url detection + Running --> Exited: onExit + Running --> Detached: ws close + + Detached --> Running: reconnect before timeout + Detached --> Killed: timeout reached -> pty.kill + Exited --> [*] + Killed --> [*] + EmitError --> WaitingInit +``` + +### Shell Behaviors in Detail + +1. `init`: +Reads `projectPath`, `sessionId`, `provider`, `hasSession`, `initialCommand`, `isPlainShell`. +2. Login reset: +For login-like commands, existing keyed PTY session is killed and recreated. +3. Validation: +Path must exist and be a directory; `sessionId` must match safe pattern. +4. Command build: +Provider-specific command construction with resume semantics. +5. PTY output buffering: +Stores up to 5000 chunks for replay on reconnect. +6. URL detection: +Strips ANSI, accumulates text buffer, extracts URLs, emits `auth_url` once per normalized URL, supports `autoOpen`. +7. Close behavior: +Socket disconnect does not instantly kill PTY; session is kept alive and terminated on timeout. + +## `/plugin-ws/:pluginName` Proxy Flow + +```mermaid +sequenceDiagram + participant Client + participant Proxy as handlePluginWsProxy + participant PM as getPluginPort + participant Upstream as Plugin WS + + Client->>Proxy: Connect /plugin-ws/:name + Proxy->>Proxy: Validate pluginName regex + alt Invalid name + Proxy-->>Client: close(4400, "Invalid plugin name") + else Valid + Proxy->>PM: getPluginPort(name) + alt Plugin not running + Proxy-->>Client: close(4404, "Plugin not running") + else Port found + Proxy->>Upstream: new WebSocket(ws://127.0.0.1:port/ws) + Client-->>Upstream: relay messages bidirectionally + Upstream-->>Client: relay messages bidirectionally + Upstream-->>Client: close propagation + Client-->>Upstream: close propagation + Upstream-->>Client: close(4502, "Upstream error") on upstream error + end + end +``` + +## Shared Client Registry and Broadcasts + +Only chat sockets (`/ws`) are tracked in `connectedClients`. + +That shared set is consumed by: + +1. `modules/projects/services/projects.service.ts` +Broadcasts `loading_progress` while project snapshots are being built. +2. `modules/providers/services/sessions-watcher.service.ts` +Broadcasts `projects_updated` when provider session artifacts change. + +This design centralizes cross-module realtime fanout without requiring route-local references to WebSocket internals. + +## Writer Adapter (`WebSocketWriter`) + +`WebSocketWriter` normalizes chat transport behavior to match existing writer-style interfaces used elsewhere. + +Methods: + +1. `send(data)` +JSON-serializes and sends only if socket is open. +2. `setSessionId(sessionId)` / `getSessionId()` +Supports provider session bookkeeping and resume flows. +3. `updateWebSocket(newRawWs)` +Allows active session stream redirection on reconnect. + +## Error Handling and Close Codes + +Current explicit close codes in this module: + +1. `4400`: Invalid plugin name +2. `4404`: Plugin not running +3. `4502`: Upstream plugin WebSocket error + +Other errors: + +1. Chat handler catches and emits `{ type: "error", error }`. +2. Shell handler catches and writes terminal-visible error output. +3. Unknown websocket paths are closed immediately. + +## Extending This Module + +To add a new websocket route: + +1. Add a new handler service under `services/`. +2. Extend `WebSocketServerDependencies` in `websocket-server.service.ts` if needed. +3. Add a new pathname branch in the router. +4. Wire dependency injection from `server/index.js`. +5. Keep `index.ts` as barrel-only export surface. diff --git a/server/modules/websocket/index.ts b/server/modules/websocket/index.ts new file mode 100644 index 00000000..da65ee82 --- /dev/null +++ b/server/modules/websocket/index.ts @@ -0,0 +1,2 @@ +export { WS_OPEN_STATE, connectedClients } from './services/websocket-state.service.js'; +export { createWebSocketServer } from './services/websocket-server.service.js'; diff --git a/server/modules/websocket/services/chat-websocket.service.ts b/server/modules/websocket/services/chat-websocket.service.ts new file mode 100644 index 00000000..95fabe55 --- /dev/null +++ b/server/modules/websocket/services/chat-websocket.service.ts @@ -0,0 +1,271 @@ +import type { WebSocket } from 'ws'; + +import { connectedClients } from '@/modules/websocket/services/websocket-state.service.js'; +import { WebSocketWriter } from '@/modules/websocket/services/websocket-writer.service.js'; +import type { + AnyRecord, + AuthenticatedWebSocketRequest, + LLMProvider, +} from '@/shared/types.js'; +import { createNormalizedMessage, parseIncomingJsonObject } from '@/shared/utils.js'; + +type ChatIncomingMessage = AnyRecord & { + type?: string; + command?: string; + options?: AnyRecord; + provider?: string; + sessionId?: string; + requestId?: string; + allow?: unknown; + updatedInput?: unknown; + message?: unknown; + rememberEntry?: unknown; +}; + +const DEFAULT_PROVIDER: LLMProvider = 'claude'; + +type ChatWebSocketDependencies = { + queryClaudeSDK: (command: string, options: unknown, writer: WebSocketWriter) => Promise; + spawnCursor: (command: string, options: unknown, writer: WebSocketWriter) => Promise; + queryCodex: (command: string, options: unknown, writer: WebSocketWriter) => Promise; + spawnGemini: (command: string, options: unknown, writer: WebSocketWriter) => Promise; + abortClaudeSDKSession: (sessionId: string) => Promise; + abortCursorSession: (sessionId: string) => boolean; + abortCodexSession: (sessionId: string) => boolean; + abortGeminiSession: (sessionId: string) => boolean; + resolveToolApproval: ( + requestId: string, + payload: { + allow: boolean; + updatedInput?: unknown; + message?: string; + rememberEntry?: unknown; + } + ) => void; + isClaudeSDKSessionActive: (sessionId: string) => boolean; + isCursorSessionActive: (sessionId: string) => boolean; + isCodexSessionActive: (sessionId: string) => boolean; + isGeminiSessionActive: (sessionId: string) => boolean; + reconnectSessionWriter: (sessionId: string, ws: WebSocket) => boolean; + getPendingApprovalsForSession: (sessionId: string) => unknown[]; + getActiveClaudeSDKSessions: () => unknown; + getActiveCursorSessions: () => unknown; + getActiveCodexSessions: () => unknown; + getActiveGeminiSessions: () => unknown; +}; + +/** + * Normalizes potentially invalid provider names coming from websocket payloads. + */ +function readProvider(value: unknown): LLMProvider { + if (value === 'claude' || value === 'cursor' || value === 'codex' || value === 'gemini') { + return value; + } + + return DEFAULT_PROVIDER; +} + +/** + * Extracts the authenticated request user id in the formats currently produced + * by platform and OSS auth code paths. + */ +function readRequestUserId( + request: AuthenticatedWebSocketRequest | undefined +): string | number | null { + const user = request?.user; + if (!user) { + return null; + } + + if (typeof user.id === 'string' || typeof user.id === 'number') { + return user.id; + } + + if (typeof user.userId === 'string' || typeof user.userId === 'number') { + return user.userId; + } + + return null; +} + +/** + * Handles authenticated chat websocket messages used by the main chat panel. + */ +export function handleChatConnection( + ws: WebSocket, + request: AuthenticatedWebSocketRequest, + dependencies: ChatWebSocketDependencies +): void { + console.log('[INFO] Chat WebSocket connected'); + connectedClients.add(ws); + + const writer = new WebSocketWriter(ws, readRequestUserId(request)); + + ws.on('message', async (rawMessage) => { + try { + const parsed = parseIncomingJsonObject(rawMessage); + if (!parsed) { + throw new Error('Invalid websocket payload'); + } + + const data = parsed as ChatIncomingMessage; + const messageType = data.type; + if (!messageType) { + throw new Error('Message type is required'); + } + + if (messageType === 'claude-command') { + await dependencies.queryClaudeSDK(data.command ?? '', data.options, writer); + return; + } + + if (messageType === 'cursor-command') { + await dependencies.spawnCursor(data.command ?? '', data.options, writer); + return; + } + + if (messageType === 'codex-command') { + await dependencies.queryCodex(data.command ?? '', data.options, writer); + return; + } + + if (messageType === 'gemini-command') { + await dependencies.spawnGemini(data.command ?? '', data.options, writer); + return; + } + + if (messageType === 'cursor-resume') { + await dependencies.spawnCursor( + '', + { + sessionId: data.sessionId, + resume: true, + cwd: data.options?.cwd, + }, + writer + ); + return; + } + + if (messageType === 'abort-session') { + const provider = readProvider(data.provider); + const sessionId = typeof data.sessionId === 'string' ? data.sessionId : ''; + let success = false; + + if (provider === 'cursor') { + success = dependencies.abortCursorSession(sessionId); + } else if (provider === 'codex') { + success = dependencies.abortCodexSession(sessionId); + } else if (provider === 'gemini') { + success = dependencies.abortGeminiSession(sessionId); + } else { + success = await dependencies.abortClaudeSDKSession(sessionId); + } + + writer.send( + createNormalizedMessage({ + kind: 'complete', + exitCode: success ? 0 : 1, + aborted: true, + success, + sessionId, + provider, + }) + ); + return; + } + + if (messageType === 'claude-permission-response') { + if (typeof data.requestId === 'string' && data.requestId.length > 0) { + dependencies.resolveToolApproval(data.requestId, { + allow: Boolean(data.allow), + updatedInput: data.updatedInput, + message: typeof data.message === 'string' ? data.message : undefined, + rememberEntry: data.rememberEntry, + }); + } + return; + } + + if (messageType === 'cursor-abort') { + const sessionId = typeof data.sessionId === 'string' ? data.sessionId : ''; + const success = dependencies.abortCursorSession(sessionId); + writer.send( + createNormalizedMessage({ + kind: 'complete', + exitCode: success ? 0 : 1, + aborted: true, + success, + sessionId, + provider: 'cursor', + }) + ); + return; + } + + if (messageType === 'check-session-status') { + const provider = readProvider(data.provider); + const sessionId = typeof data.sessionId === 'string' ? data.sessionId : ''; + let isActive = false; + + if (provider === 'cursor') { + isActive = dependencies.isCursorSessionActive(sessionId); + } else if (provider === 'codex') { + isActive = dependencies.isCodexSessionActive(sessionId); + } else if (provider === 'gemini') { + isActive = dependencies.isGeminiSessionActive(sessionId); + } else { + isActive = dependencies.isClaudeSDKSessionActive(sessionId); + if (isActive) { + dependencies.reconnectSessionWriter(sessionId, ws); + } + } + + writer.send({ + type: 'session-status', + sessionId, + provider, + isProcessing: isActive, + }); + return; + } + + if (messageType === 'get-pending-permissions') { + const sessionId = typeof data.sessionId === 'string' ? data.sessionId : ''; + if (sessionId && dependencies.isClaudeSDKSessionActive(sessionId)) { + const pending = dependencies.getPendingApprovalsForSession(sessionId); + writer.send({ + type: 'pending-permissions-response', + sessionId, + data: pending, + }); + } + return; + } + + if (messageType === 'get-active-sessions') { + writer.send({ + type: 'active-sessions', + sessions: { + claude: dependencies.getActiveClaudeSDKSessions(), + cursor: dependencies.getActiveCursorSessions(), + codex: dependencies.getActiveCodexSessions(), + gemini: dependencies.getActiveGeminiSessions(), + }, + }); + } + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + console.error('[ERROR] Chat WebSocket error:', message); + writer.send({ + type: 'error', + error: message, + }); + } + }); + + ws.on('close', () => { + console.log('[INFO] Chat client disconnected'); + connectedClients.delete(ws); + }); +} diff --git a/server/modules/websocket/services/plugin-websocket-proxy.service.ts b/server/modules/websocket/services/plugin-websocket-proxy.service.ts new file mode 100644 index 00000000..491fd540 --- /dev/null +++ b/server/modules/websocket/services/plugin-websocket-proxy.service.ts @@ -0,0 +1,65 @@ +import { WebSocket } from 'ws'; + +/** + * Proxies an authenticated client websocket to a plugin websocket endpoint. + */ +export function handlePluginWsProxy( + clientWs: WebSocket, + pathname: string, + getPluginPort: (pluginName: string) => number | null +): void { + const pluginName = pathname.replace('/plugin-ws/', ''); + if (!pluginName || /[^a-zA-Z0-9_-]/.test(pluginName)) { + clientWs.close(4400, 'Invalid plugin name'); + return; + } + + const port = getPluginPort(pluginName); + if (!port) { + clientWs.close(4404, 'Plugin not running'); + return; + } + + const upstream = new WebSocket(`ws://127.0.0.1:${port}/ws`); + + upstream.on('open', () => { + console.log(`[Plugins] WS proxy connected to "${pluginName}" on port ${port}`); + }); + + upstream.on('message', (data) => { + if (clientWs.readyState === WebSocket.OPEN) { + clientWs.send(data); + } + }); + + clientWs.on('message', (data) => { + if (upstream.readyState === WebSocket.OPEN) { + upstream.send(data); + } + }); + + upstream.on('close', () => { + if (clientWs.readyState === WebSocket.OPEN) { + clientWs.close(); + } + }); + + clientWs.on('close', () => { + if (upstream.readyState === WebSocket.OPEN) { + upstream.close(); + } + }); + + upstream.on('error', (error) => { + console.error(`[Plugins] WS proxy error for "${pluginName}":`, error.message); + if (clientWs.readyState === WebSocket.OPEN) { + clientWs.close(4502, 'Upstream error'); + } + }); + + clientWs.on('error', () => { + if (upstream.readyState === WebSocket.OPEN) { + upstream.close(); + } + }); +} diff --git a/server/modules/websocket/services/shell-websocket.service.ts b/server/modules/websocket/services/shell-websocket.service.ts new file mode 100644 index 00000000..9bf7046b --- /dev/null +++ b/server/modules/websocket/services/shell-websocket.service.ts @@ -0,0 +1,453 @@ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; + +import pty, { type IPty } from 'node-pty'; +import { WebSocket, type RawData } from 'ws'; + +import { parseIncomingJsonObject } from '@/shared/utils.js'; + +type ShellIncomingMessage = { + type?: string; + data?: string; + cols?: number; + rows?: number; + projectPath?: string; + sessionId?: string; + hasSession?: boolean; + provider?: string; + initialCommand?: string; + isPlainShell?: boolean; +}; + +type PtySessionEntry = { + pty: IPty; + ws: WebSocket | null; + buffer: string[]; + timeoutId: NodeJS.Timeout | null; + projectPath: string; + sessionId: string | null; +}; + +const ptySessionsMap = new Map(); +const PTY_SESSION_TIMEOUT = 30 * 60 * 1000; +const SHELL_URL_PARSE_BUFFER_LIMIT = 32768; + +type ShellWebSocketDependencies = { + getSessionById: (sessionId: string) => { cliSessionId?: string } | null | undefined; + stripAnsiSequences: (content: string) => string; + normalizeDetectedUrl: (url: string) => string | null; + extractUrlsFromText: (content: string) => string[]; + shouldAutoOpenUrlFromOutput: (content: string) => boolean; +}; + +/** + * Reads a string field from untyped payloads and falls back when absent. + */ +function readString(value: unknown, fallback = ''): string { + return typeof value === 'string' ? value : fallback; +} + +/** + * Reads a boolean field from untyped payloads and falls back when absent. + */ +function readBoolean(value: unknown, fallback = false): boolean { + return typeof value === 'boolean' ? value : fallback; +} + +/** + * Reads a finite number field from untyped payloads and falls back when absent. + */ +function readNumber(value: unknown, fallback: number): number { + return typeof value === 'number' && Number.isFinite(value) ? value : fallback; +} + +/** + * Parses incoming websocket shell messages and keeps processing safe when + * malformed payloads are received. + */ +function parseShellMessage(rawMessage: RawData): ShellIncomingMessage | null { + const payload = parseIncomingJsonObject(rawMessage); + if (!payload) { + return null; + } + + return payload as ShellIncomingMessage; +} + +/** + * Resolves provider command line for plain shell and agent-backed shell modes. + */ +function buildShellCommand( + message: ShellIncomingMessage, + dependencies: ShellWebSocketDependencies +): string { + const hasSession = readBoolean(message.hasSession); + const sessionId = readString(message.sessionId); + const initialCommand = readString(message.initialCommand); + const provider = readString(message.provider, 'claude'); + const safeSessionIdPattern = /^[a-zA-Z0-9_.\-:]+$/; + const isPlainShell = + readBoolean(message.isPlainShell) || + (!!initialCommand && !hasSession) || + provider === 'plain-shell'; + + if (isPlainShell) { + return initialCommand; + } + + if (provider === 'cursor') { + if (hasSession && sessionId) { + return `cursor-agent --resume="${sessionId}"`; + } + return 'cursor-agent'; + } + + if (provider === 'codex') { + if (hasSession && sessionId) { + if (os.platform() === 'win32') { + return `codex resume "${sessionId}"; if ($LASTEXITCODE -ne 0) { codex }`; + } + return `codex resume "${sessionId}" || codex`; + } + return 'codex'; + } + + if (provider === 'gemini') { + const command = initialCommand || 'gemini'; + let resumeId = sessionId; + if (hasSession && sessionId) { + try { + const existingSession = dependencies.getSessionById(sessionId); + if (existingSession && existingSession.cliSessionId) { + resumeId = existingSession.cliSessionId; + if (!safeSessionIdPattern.test(resumeId)) { + resumeId = ''; + } + } + } catch (error) { + console.error('Failed to get Gemini CLI session ID:', error); + } + } + + if (hasSession && resumeId) { + return `${command} --resume "${resumeId}"`; + } + return command; + } + + const command = initialCommand || 'claude'; + if (hasSession && sessionId) { + if (os.platform() === 'win32') { + return `claude --resume "${sessionId}"; if ($LASTEXITCODE -ne 0) { claude }`; + } + return `claude --resume "${sessionId}" || claude`; + } + return command; +} + +/** + * Handles websocket connections used by the standalone shell terminal UI. + */ +export function handleShellConnection( + ws: WebSocket, + dependencies: ShellWebSocketDependencies +): void { + console.log('[INFO] Shell websocket connected'); + + let shellProcess: IPty | null = null; + let ptySessionKey: string | null = null; + let urlDetectionBuffer = ''; + const announcedAuthUrls = new Set(); + + ws.on('message', async (rawMessage) => { + try { + const data = parseShellMessage(rawMessage); + if (!data?.type) { + throw new Error('Invalid websocket payload'); + } + + if (data.type === 'init') { + const projectPath = readString(data.projectPath, process.cwd()); + const sessionId = readString(data.sessionId) || null; + const hasSession = readBoolean(data.hasSession); + const provider = readString(data.provider, 'claude'); + const initialCommand = readString(data.initialCommand); + const isPlainShell = + readBoolean(data.isPlainShell) || + (!!initialCommand && !hasSession) || + provider === 'plain-shell'; + + urlDetectionBuffer = ''; + announcedAuthUrls.clear(); + + const isLoginCommand = + !!initialCommand && + (initialCommand.includes('setup-token') || + initialCommand.includes('cursor-agent login') || + initialCommand.includes('auth login')); + + const commandSuffix = + isPlainShell && initialCommand + ? `_cmd_${Buffer.from(initialCommand).toString('base64').slice(0, 16)}` + : ''; + ptySessionKey = `${projectPath}_${sessionId ?? 'default'}${commandSuffix}`; + + if (isLoginCommand) { + const oldSession = ptySessionsMap.get(ptySessionKey); + if (oldSession) { + if (oldSession.timeoutId) { + clearTimeout(oldSession.timeoutId); + } + oldSession.pty.kill(); + ptySessionsMap.delete(ptySessionKey); + } + } + + const existingSession = isLoginCommand ? null : ptySessionsMap.get(ptySessionKey); + if (existingSession) { + shellProcess = existingSession.pty; + if (existingSession.timeoutId) { + clearTimeout(existingSession.timeoutId); + } + + ws.send( + JSON.stringify({ + type: 'output', + data: '\x1b[36m[Reconnected to existing session]\x1b[0m\r\n', + }) + ); + + if (existingSession.buffer.length > 0) { + existingSession.buffer.forEach((bufferedData) => { + ws.send( + JSON.stringify({ + type: 'output', + data: bufferedData, + }) + ); + }); + } + + existingSession.ws = ws; + return; + } + + const resolvedProjectPath = path.resolve(projectPath); + try { + const stats = fs.statSync(resolvedProjectPath); + if (!stats.isDirectory()) { + throw new Error('Not a directory'); + } + } catch { + ws.send(JSON.stringify({ type: 'error', message: 'Invalid project path' })); + return; + } + + const safeSessionIdPattern = /^[a-zA-Z0-9_.\-:]+$/; + if (sessionId && !safeSessionIdPattern.test(sessionId)) { + ws.send(JSON.stringify({ type: 'error', message: 'Invalid session ID' })); + return; + } + + const shellCommand = buildShellCommand(data, dependencies); + const shell = os.platform() === 'win32' ? 'powershell.exe' : 'bash'; + const shellArgs = + os.platform() === 'win32' ? ['-Command', shellCommand] : ['-c', shellCommand]; + const termCols = readNumber(data.cols, 80); + const termRows = readNumber(data.rows, 24); + + shellProcess = pty.spawn(shell, shellArgs, { + name: 'xterm-256color', + cols: termCols, + rows: termRows, + cwd: resolvedProjectPath, + env: { + ...process.env, + TERM: 'xterm-256color', + COLORTERM: 'truecolor', + FORCE_COLOR: '3', + }, + }); + + ptySessionsMap.set(ptySessionKey, { + pty: shellProcess, + ws, + buffer: [], + timeoutId: null, + projectPath, + sessionId, + }); + + shellProcess.onData((chunk) => { + if (!ptySessionKey) { + return; + } + + const session = ptySessionsMap.get(ptySessionKey); + if (!session) { + return; + } + + if (session.buffer.length < 5000) { + session.buffer.push(chunk); + } else { + session.buffer.shift(); + session.buffer.push(chunk); + } + + if (session.ws && session.ws.readyState === WebSocket.OPEN) { + let outputData = chunk; + const cleanChunk = dependencies.stripAnsiSequences(chunk); + urlDetectionBuffer = `${urlDetectionBuffer}${cleanChunk}`.slice(-SHELL_URL_PARSE_BUFFER_LIMIT); + + outputData = outputData.replace( + /OPEN_URL:\s*(https?:\/\/[^\s\x1b\x07]+)/g, + '[INFO] Opening in browser: $1' + ); + + const emitAuthUrl = (detectedUrl: string, autoOpen = false) => { + const normalizedUrl = dependencies.normalizeDetectedUrl(detectedUrl); + if (!normalizedUrl) { + return; + } + + const isNewUrl = !announcedAuthUrls.has(normalizedUrl); + if (isNewUrl) { + announcedAuthUrls.add(normalizedUrl); + session.ws?.send( + JSON.stringify({ + type: 'auth_url', + url: normalizedUrl, + autoOpen, + }) + ); + } + }; + + const normalizedDetectedUrls = dependencies.extractUrlsFromText(urlDetectionBuffer) + .map((url) => dependencies.normalizeDetectedUrl(url)) + .filter((url): url is string => Boolean(url)); + + const dedupedDetectedUrls = Array.from(new Set(normalizedDetectedUrls)).filter( + (url, _, urls) => + !urls.some((otherUrl) => otherUrl !== url && otherUrl.startsWith(url)) + ); + + dedupedDetectedUrls.forEach((url) => emitAuthUrl(url, false)); + + if ( + dependencies.shouldAutoOpenUrlFromOutput(cleanChunk) && + dedupedDetectedUrls.length > 0 + ) { + const bestUrl = dedupedDetectedUrls.reduce((longest, current) => + current.length > longest.length ? current : longest + ); + emitAuthUrl(bestUrl, true); + } + + session.ws.send( + JSON.stringify({ + type: 'output', + data: outputData, + }) + ); + } + }); + + shellProcess.onExit((exitCode) => { + if (!ptySessionKey) { + return; + } + + const session = ptySessionsMap.get(ptySessionKey); + if (session && session.ws && session.ws.readyState === WebSocket.OPEN) { + session.ws.send( + JSON.stringify({ + type: 'output', + data: `\r\n\x1b[33mProcess exited with code ${exitCode.exitCode}${ + exitCode.signal != null ? ` (${exitCode.signal})` : '' + }\x1b[0m\r\n`, + }) + ); + } + + if (session?.timeoutId) { + clearTimeout(session.timeoutId); + } + + ptySessionsMap.delete(ptySessionKey); + shellProcess = null; + }); + + let welcomeMsg = `\x1b[36mStarting terminal in: ${projectPath}\x1b[0m\r\n`; + if (!isPlainShell) { + const providerName = + provider === 'cursor' + ? 'Cursor' + : provider === 'codex' + ? 'Codex' + : provider === 'gemini' + ? 'Gemini' + : 'Claude'; + welcomeMsg = hasSession + ? `\x1b[36mResuming ${providerName} session ${sessionId} in: ${projectPath}\x1b[0m\r\n` + : `\x1b[36mStarting new ${providerName} session in: ${projectPath}\x1b[0m\r\n`; + } + + ws.send( + JSON.stringify({ + type: 'output', + data: welcomeMsg, + }) + ); + return; + } + + if (data.type === 'input') { + if (shellProcess) { + shellProcess.write(readString(data.data)); + } + return; + } + + if (data.type === 'resize') { + if (shellProcess) { + shellProcess.resize(readNumber(data.cols, 80), readNumber(data.rows, 24)); + } + } + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + console.error('[ERROR] Shell WebSocket error:', message); + if (ws.readyState === WebSocket.OPEN) { + ws.send( + JSON.stringify({ + type: 'output', + data: `\r\n\x1b[31mError: ${message}\x1b[0m\r\n`, + }) + ); + } + } + }); + + ws.on('close', () => { + if (!ptySessionKey) { + return; + } + + const session = ptySessionsMap.get(ptySessionKey); + if (!session) { + return; + } + + session.ws = null; + session.timeoutId = setTimeout(() => { + session.pty.kill(); + ptySessionsMap.delete(ptySessionKey as string); + }, PTY_SESSION_TIMEOUT); + }); + + ws.on('error', (error) => { + console.error('[ERROR] Shell WebSocket error:', error); + }); +} diff --git a/server/modules/websocket/services/websocket-auth.service.ts b/server/modules/websocket/services/websocket-auth.service.ts new file mode 100644 index 00000000..bd689d5d --- /dev/null +++ b/server/modules/websocket/services/websocket-auth.service.ts @@ -0,0 +1,54 @@ +import type { VerifyClientCallbackSync } from 'ws'; + +import type { AuthenticatedWebSocketRequest } from '@/shared/types.js'; + +type WebSocketAuthDependencies = { + isPlatform: boolean; + authenticateWebSocket: (token: string | null) => { + id?: string | number; + userId?: string | number; + username?: string; + [key: string]: unknown; + } | null; +}; + +/** + * Authenticates websocket upgrade requests before the `connection` handler runs. + */ +export function verifyWebSocketClient( + info: Parameters>[0], + dependencies: WebSocketAuthDependencies +): boolean { + const request = info.req as AuthenticatedWebSocketRequest; + console.log('WebSocket connection attempt to:', request.url); + + // Platform mode: use the first DB user and skip token checks. + if (dependencies.isPlatform) { + const user = dependencies.authenticateWebSocket(null); + if (!user) { + console.log('[WARN] Platform mode: No user found in database'); + return false; + } + + request.user = user; + console.log('[OK] Platform mode WebSocket authenticated for user:', user.username); + return true; + } + + // OSS mode: read JWT from query string first, then Authorization header. + const upgradeUrl = new URL(request.url ?? '/', 'http://localhost'); + const token = + upgradeUrl.searchParams.get('token') ?? + request.headers.authorization?.split(' ')[1] ?? + null; + + const user = dependencies.authenticateWebSocket(token); + if (!user) { + console.log('[WARN] WebSocket authentication failed'); + return false; + } + + request.user = user; + console.log('[OK] WebSocket authenticated for user:', user.username); + return true; +} diff --git a/server/modules/websocket/services/websocket-server.service.ts b/server/modules/websocket/services/websocket-server.service.ts new file mode 100644 index 00000000..7e5c12e4 --- /dev/null +++ b/server/modules/websocket/services/websocket-server.service.ts @@ -0,0 +1,58 @@ +import type { Server as HttpServer } from 'node:http'; + +import { WebSocketServer, type VerifyClientCallbackSync } from 'ws'; + +import { handleChatConnection } from '@/modules/websocket/services/chat-websocket.service.js'; +import { verifyWebSocketClient } from '@/modules/websocket/services/websocket-auth.service.js'; +import { handlePluginWsProxy } from '@/modules/websocket/services/plugin-websocket-proxy.service.js'; +import { handleShellConnection } from '@/modules/websocket/services/shell-websocket.service.js'; +import type { AuthenticatedWebSocketRequest } from '@/shared/types.js'; + +type WebSocketServerDependencies = { + verifyClient: Parameters[1]; + chat: Parameters[2]; + shell: Parameters[1]; + getPluginPort: Parameters[2]; +}; + +/** + * Creates and wires the server-wide websocket gateway used for chat, shell, and + * plugin proxy routes. + */ +export function createWebSocketServer( + server: HttpServer, + dependencies: WebSocketServerDependencies +): WebSocketServer { + const wss = new WebSocketServer({ + server, + verifyClient: (( + info: Parameters>[0] + ) => verifyWebSocketClient(info, dependencies.verifyClient)), + }); + + wss.on('connection', (ws, request) => { + const incomingRequest = request as AuthenticatedWebSocketRequest; + const url = incomingRequest.url ?? '/'; + const pathname = new URL(url, 'http://localhost').pathname; + + if (pathname === '/shell') { + handleShellConnection(ws, dependencies.shell); + return; + } + + if (pathname === '/ws') { + handleChatConnection(ws, incomingRequest, dependencies.chat); + return; + } + + if (pathname.startsWith('/plugin-ws/')) { + handlePluginWsProxy(ws, pathname, dependencies.getPluginPort); + return; + } + + console.log('[WARN] Unknown WebSocket path:', pathname); + ws.close(); + }); + + return wss; +} diff --git a/server/modules/websocket/services/websocket-state.service.ts b/server/modules/websocket/services/websocket-state.service.ts new file mode 100644 index 00000000..3cffce24 --- /dev/null +++ b/server/modules/websocket/services/websocket-state.service.ts @@ -0,0 +1,16 @@ +import type { RealtimeClientConnection } from '@/shared/types.js'; + +/** + * Numeric readyState for an open WebSocket connection. + * + * We keep this in module state so services that broadcast updates do not need + * to import `ws` directly just to compare open/closed state. + */ +export const WS_OPEN_STATE = 1; + +/** + * Shared registry of active chat WebSocket connections. + * + * Project/session services publish realtime updates by iterating this set. + */ +export const connectedClients = new Set(); diff --git a/server/modules/websocket/services/websocket-writer.service.ts b/server/modules/websocket/services/websocket-writer.service.ts new file mode 100644 index 00000000..af307ad6 --- /dev/null +++ b/server/modules/websocket/services/websocket-writer.service.ts @@ -0,0 +1,38 @@ +import { WS_OPEN_STATE } from '@/modules/websocket/services/websocket-state.service.js'; +import type { RealtimeClientConnection } from '@/shared/types.js'; + +/** + * Thin transport adapter that gives WebSocket connections the same interface as + * SSE writers used by API routes (`send`, `setSessionId`, `getSessionId`). + */ +export class WebSocketWriter { + ws: RealtimeClientConnection; + sessionId: string | null; + userId: string | number | null; + isWebSocketWriter: boolean; + + constructor(ws: RealtimeClientConnection, userId: string | number | null = null) { + this.ws = ws; + this.sessionId = null; + this.userId = userId; + this.isWebSocketWriter = true; + } + + send(data: unknown): void { + if (this.ws.readyState === WS_OPEN_STATE) { + this.ws.send(JSON.stringify(data)); + } + } + + updateWebSocket(newRawWs: RealtimeClientConnection): void { + this.ws = newRawWs; + } + + setSessionId(sessionId: string): void { + this.sessionId = sessionId; + } + + getSessionId(): string | null { + return this.sessionId; + } +} diff --git a/server/shared/types.ts b/server/shared/types.ts index c826021e..fc0150fa 100644 --- a/server/shared/types.ts +++ b/server/shared/types.ts @@ -1,3 +1,5 @@ +import type { IncomingMessage } from 'node:http'; + //----------------- HTTP RESPONSE SHAPES ------------ /** * Canonical success envelope used by backend APIs that return a structured payload. @@ -18,6 +20,43 @@ export type ApiSuccessShape = { */ export type AnyRecord = Record; +// --------------------------- +//----------------- WEBSOCKET TRANSPORT TYPES ------------ +/** + * Minimal websocket client contract used by backend broadcaster services. + * + * Any transport object added to `connectedClients` must implement these two + * members so shared services can safely send JSON strings and check whether the + * socket is still open before broadcasting. + */ +export type RealtimeClientConnection = { + readyState: number; + send(data: string): void; +}; + +/** + * Authenticated user payload attached to websocket upgrade requests. + * + * Platform and OSS auth flows currently use either `id` or `userId`; both are + * represented here so websocket handlers can resolve a stable writer user id. + */ +export type AuthenticatedWebSocketUser = { + id?: string | number; + userId?: string | number; + username?: string; + [key: string]: unknown; +}; + +/** + * HTTP upgrade request shape after websocket authentication succeeds. + * + * `verifyClient` populates `request.user` with the authenticated payload, and + * downstream websocket handlers rely on this extended request type. + */ +export type AuthenticatedWebSocketRequest = IncomingMessage & { + user?: AuthenticatedWebSocketUser; +}; + // --------------------------- //----------------- PROVIDER MESSAGE MODEL ------------ /** diff --git a/server/shared/utils.ts b/server/shared/utils.ts index 01459a2f..390d78fd 100644 --- a/server/shared/utils.ts +++ b/server/shared/utils.ts @@ -182,6 +182,62 @@ export const readStringRecord = (value: unknown): Record | undef return Object.keys(normalized).length > 0 ? normalized : undefined; }; +// --------------------------- +//----------------- WEBSOCKET PAYLOAD PARSING UTILITIES ------------ +/** + * Parses one websocket message payload into a plain JSON object record. + * + * Use this in realtime handlers that receive raw websocket payloads as `string`, + * `Buffer`, `ArrayBuffer`, or chunk arrays. The helper converts supported + * payload formats to UTF-8 text, parses JSON, and returns only object payloads. + * Primitive/array/invalid payloads return `null` so callers can handle bad input + * without throwing from deeply nested message handlers. + */ +export const parseIncomingJsonObject = (payload: unknown): AnyRecord | null => { + let text: string | null = null; + + if (typeof payload === 'string') { + text = payload; + } else if (Buffer.isBuffer(payload)) { + text = payload.toString('utf8'); + } else if (payload instanceof ArrayBuffer) { + text = Buffer.from(payload).toString('utf8'); + } else if (Array.isArray(payload)) { + const buffers = payload + .map((entry) => { + if (Buffer.isBuffer(entry)) { + return entry; + } + + if (entry instanceof ArrayBuffer) { + return Buffer.from(entry); + } + + if (ArrayBuffer.isView(entry)) { + return Buffer.from(entry.buffer, entry.byteOffset, entry.byteLength); + } + + return null; + }) + .filter((entry): entry is Buffer => entry !== null); + + if (buffers.length > 0) { + text = Buffer.concat(buffers).toString('utf8'); + } + } + + if (typeof text !== 'string' || text.trim().length === 0) { + return null; + } + + try { + const parsed = JSON.parse(text) as unknown; + return readObjectRecord(parsed); + } catch { + return null; + } +}; + /** * Reads a JSON config file and guarantees a plain object result. *