From 4e7cec400fd2ef51fbaa6f79dc481e674b78fbd7 Mon Sep 17 00:00:00 2001 From: "kiloconnect[bot]" <240665456+kiloconnect[bot]@users.noreply.github.com> Date: Mon, 31 Aug 2026 07:53:39 +0000 Subject: [PATCH] (janitor/dead-code): remove orphaned SSEStreamProcessor in auto-fix-infra SSEStreamProcessor (services/auto-fix-infra/src/services/sse-stream-processor.ts) is dead code. It is never imported, never registered, and never exercised: - No import of the symbol `SSEStreamProcessor` or the path `sse-stream-processor` anywhere in the repo (source, tests, or config). - No test references it. No barrel re-exports it (src/services/ has no index.ts). The service entry point (src/index.ts) and fix-orchestrator.ts only import prompt-builder and cloud-agent-next-client. - The auto-fix Worker's actual cloud-agent client (cloud-agent-next-client.ts) uses HTTP fetch for prepare/initiate, not SSE. - Not referenced by wrangler.jsonc (main is src/index.ts; the only Durable Object is AutoFixOrchestrator), nor by any dynamic import, cron, queue, or RPC config. - The file has no imports and no module-level side effects (only type declarations and a class), so removal changes no initialization behavior. - Kilo-Org/kilocode (external API consumer) contains no reference. Verified with repo-wide grep, the service typecheck, lint, and unit tests. --- .../src/services/sse-stream-processor.ts | 204 ------------------ 1 file changed, 204 deletions(-) delete mode 100644 services/auto-fix-infra/src/services/sse-stream-processor.ts diff --git a/services/auto-fix-infra/src/services/sse-stream-processor.ts b/services/auto-fix-infra/src/services/sse-stream-processor.ts deleted file mode 100644 index 4b3799771d..0000000000 --- a/services/auto-fix-infra/src/services/sse-stream-processor.ts +++ /dev/null @@ -1,204 +0,0 @@ -/** - * SSEStreamProcessor - * - * Generic SSE (Server-Sent Events) stream processing service. - * Handles SSE parsing, buffer management, and error handling. - */ - -type StreamEventHandler = { - onSessionId?: (sessionId: string) => void; - onTextContent?: (text: string) => void; - onComplete?: () => void; - onError?: (error: Error) => void; -}; - -type StreamMetrics = { - totalEvents: number; - errorEvents: number; - parseErrors: number; - eventTypeCounts: Record; - startTime: number; - endTime?: number; -}; - -export class SSEStreamProcessor { - /** Maximum time to wait for SSE stream (20 minutes) */ - private static readonly STREAM_TIMEOUT_MS = 20 * 60 * 1000; - - /** - * Process an SSE stream with custom event handlers - */ - async processStream( - response: Response, - handlers: StreamEventHandler, - timeoutMs: number = SSEStreamProcessor.STREAM_TIMEOUT_MS - ): Promise { - // Add timeout protection - const timeoutPromise = new Promise((_, reject) => - setTimeout( - () => reject(new Error('SSE stream timeout - processing exceeded maximum time limit')), - timeoutMs - ) - ); - - await Promise.race([this.processStreamInternal(response, handlers), timeoutPromise]); - } - - /** - * Internal stream processing with metrics tracking - */ - private async processStreamInternal( - response: Response, - handlers: StreamEventHandler - ): Promise { - if (!response.body) { - throw new Error('No response body from stream'); - } - - const reader = (response.body as ReadableStream).getReader(); - const decoder = new TextDecoder(); - let buffer = ''; - - // Initialize metrics - const metrics: StreamMetrics = { - totalEvents: 0, - errorEvents: 0, - parseErrors: 0, - eventTypeCounts: {}, - startTime: Date.now(), - }; - - console.log('[SSEStreamProcessor] Starting stream processing'); - - try { - while (true) { - const { done, value } = await reader.read(); - - if (done) { - metrics.endTime = Date.now(); - const durationMs = metrics.endTime - metrics.startTime; - console.log('[SSEStreamProcessor] Stream ended naturally', { - ...metrics, - durationMs, - durationSeconds: Math.floor(durationMs / 1000), - }); - break; - } - - buffer += decoder.decode(value, { stream: true }); - - // Process complete lines - const lines = buffer.split('\n'); - buffer = lines.pop() || ''; - - for (const line of lines) { - if (line.startsWith('data: ')) { - const data = line.slice(6); - - if (data === '' || data === ':ping') { - continue; - } - - try { - const event: Record = JSON.parse(data) as Record; - metrics.totalEvents++; - - // Track event type counts - const eventType = - typeof event.streamEventType === 'string' ? event.streamEventType : 'unknown'; - metrics.eventTypeCounts[eventType] = (metrics.eventTypeCounts[eventType] || 0) + 1; - - // Extract sessionId from first event - if (handlers.onSessionId && typeof event.sessionId === 'string') { - handlers.onSessionId(event.sessionId); - } - - // Extract text content from kilocode events - const payload = event.payload as Record | undefined; - if (handlers.onTextContent && event.streamEventType === 'kilocode' && payload) { - if (typeof payload.content === 'string') { - handlers.onTextContent(payload.content); - } else if (payload.type === 'text' && typeof payload.text === 'string') { - handlers.onTextContent(payload.text); - } - } - // Also check for output events - else if ( - handlers.onTextContent && - event.streamEventType === 'output' && - typeof event.content === 'string' - ) { - handlers.onTextContent(event.content); - } - - // Handle completion event - if (event.streamEventType === 'complete') { - console.log('[SSEStreamProcessor] Stream completion event received', { - totalEvents: metrics.totalEvents, - }); - if (handlers.onComplete) { - handlers.onComplete(); - } - break; - } - - // Handle error event - // Note: cloud-agent SystemErrorEvent uses 'error' field, not 'message' - if (event.streamEventType === 'error') { - metrics.errorEvents++; - const errorDetail = - typeof event.error === 'string' - ? event.error - : typeof event.message === 'string' - ? event.message - : 'Unknown error'; - const error = new Error(`Stream error: ${errorDetail}`); - - // Log the error event details for debugging - console.warn('[SSEStreamProcessor] Error event received', { - message: event.message, - errorDetails: event.error, - eventNumber: metrics.totalEvents, - totalErrorEvents: metrics.errorEvents, - }); - - if (handlers.onError) { - handlers.onError(error); - } - - // Don't throw - error events are informational - // The stream should continue processing unless explicitly terminated - continue; - } - } catch (parseError) { - metrics.parseErrors++; - // Enhanced logging for parse errors - console.warn('[SSEStreamProcessor] Failed to parse SSE event', { - eventNumber: metrics.totalEvents + 1, - parseErrorCount: metrics.parseErrors, - dataLength: data.length, - dataPreview: data.slice(0, 100), - errorType: parseError?.constructor?.name, - errorMessage: parseError instanceof Error ? parseError.message : String(parseError), - }); - // Skip invalid JSON and continue processing - continue; - } - } - } - } - } finally { - reader.releaseLock(); - - // Final summary log - metrics.endTime = metrics.endTime || Date.now(); - const durationMs = metrics.endTime - metrics.startTime; - console.log('[SSEStreamProcessor] Stream processing complete', { - ...metrics, - durationMs, - durationSeconds: Math.floor(durationMs / 1000), - finalBufferSize: buffer.length, - }); - } - } -}