Something went wrong. Try again.
source dump of claude code forked from oppi.li/claude-code
Something went wrong. Try again.
24 kB · 711 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712import axios, { type AxiosError } from 'axios'import type { StdoutMessage } from 'src/entrypoints/sdk/controlTypes.js'import { logForDebugging } from '../../utils/debug.js'import { logForDiagnosticsNoPII } from '../../utils/diagLogs.js'import { errorMessage } from '../../utils/errors.js'import { getSessionIngressAuthHeaders } from '../../utils/sessionIngressAuth.js'import { sleep } from '../../utils/sleep.js'import { jsonParse, jsonStringify } from '../../utils/slowOperations.js'import { getClaudeCodeUserAgent } from '../../utils/userAgent.js'import type { Transport } from './Transport.js'
// ---------------------------------------------------------------------------// Configuration// ---------------------------------------------------------------------------
const RECONNECT_BASE_DELAY_MS = 1000const RECONNECT_MAX_DELAY_MS = 30_000/** Time budget for reconnection attempts before giving up (10 minutes). */const RECONNECT_GIVE_UP_MS = 600_000/** Server sends keepalives every 15s; treat connection as dead after 45s of silence. */const LIVENESS_TIMEOUT_MS = 45_000
/** * HTTP status codes that indicate a permanent server-side rejection. * The transport transitions to 'closed' immediately without retrying. */const PERMANENT_HTTP_CODES = new Set([401, 403, 404])
// POST retry configuration (matches HybridTransport)const POST_MAX_RETRIES = 10const POST_BASE_DELAY_MS = 500const POST_MAX_DELAY_MS = 8000
/** Hoisted TextDecoder options to avoid per-chunk allocation in readStream. */const STREAM_DECODE_OPTS: TextDecodeOptions = { stream: true }
/** Hoisted axios validateStatus callback to avoid per-request closure allocation. */function alwaysValidStatus(): boolean { return true}
// ---------------------------------------------------------------------------// SSE Frame Parser// ---------------------------------------------------------------------------
type SSEFrame = { event?: string id?: string data?: string}
/** * Incrementally parse SSE frames from a text buffer. * Returns parsed frames and the remaining (incomplete) buffer. * * @internal exported for testing */export function parseSSEFrames(buffer: string): { frames: SSEFrame[] remaining: string} { const frames: SSEFrame[] = [] let pos = 0
// SSE frames are delimited by double newlines let idx: number while ((idx = buffer.indexOf('\n\n', pos)) !== -1) { const rawFrame = buffer.slice(pos, idx) pos = idx + 2
// Skip empty frames if (!rawFrame.trim()) continue
const frame: SSEFrame = {} let isComment = false
for (const line of rawFrame.split('\n')) { if (line.startsWith(':')) { // SSE comment (e.g., `:keepalive`) isComment = true continue }
const colonIdx = line.indexOf(':') if (colonIdx === -1) continue
const field = line.slice(0, colonIdx) // Per SSE spec, strip one leading space after colon if present const value = line[colonIdx + 1] === ' ' ? line.slice(colonIdx + 2) : line.slice(colonIdx + 1)
switch (field) { case 'event': frame.event = value break case 'id': frame.id = value break case 'data': // Per SSE spec, multiple data: lines are concatenated with \n frame.data = frame.data ? frame.data + '\n' + value : value break // Ignore other fields (retry:, etc.) } }
// Only emit frames that have data (or are pure comments which reset liveness) if (frame.data || isComment) { frames.push(frame) } }
return { frames, remaining: buffer.slice(pos) }}
// ---------------------------------------------------------------------------// Types// ---------------------------------------------------------------------------
type SSETransportState = | 'idle' | 'connected' | 'reconnecting' | 'closing' | 'closed'
/** * Payload for `event: client_event` frames, matching the StreamClientEvent * proto message in session_stream.proto. This is the only event type sent * to worker subscribers — delivery_update, session_update, ephemeral_event, * and catch_up_truncated are client-channel-only (see notifier.go and * event_stream.go SubscriberClient guard). */export type StreamClientEvent = { event_id: string sequence_num: number event_type: string source: string payload: Record<string, unknown> created_at: string}
// ---------------------------------------------------------------------------// SSETransport// ---------------------------------------------------------------------------
/** * Transport that uses SSE for reading and HTTP POST for writing. * * Reads events via Server-Sent Events from the CCR v2 event stream endpoint. * Writes events via HTTP POST with retry logic (same pattern as HybridTransport). * * Each `event: client_event` frame carries a StreamClientEvent proto JSON * directly in `data:`. The transport extracts `payload` and passes it to * `onData` as newline-delimited JSON for StructuredIO consumers. * * Supports automatic reconnection with exponential backoff and Last-Event-ID * for resumption after disconnection. */export class SSETransport implements Transport { private state: SSETransportState = 'idle' private onData?: (data: string) => void private onCloseCallback?: (closeCode?: number) => void private onEventCallback?: (event: StreamClientEvent) => void private headers: Record<string, string> private sessionId?: string private refreshHeaders?: () => Record<string, string> private readonly getAuthHeaders: () => Record<string, string>
// SSE connection state private abortController: AbortController | null = null private lastSequenceNum = 0 private seenSequenceNums = new Set<number>()
// Reconnection state private reconnectAttempts = 0 private reconnectStartTime: number | null = null private reconnectTimer: NodeJS.Timeout | null = null
// Liveness detection private livenessTimer: NodeJS.Timeout | null = null
// POST URL (derived from SSE URL) private postUrl: string
// Runtime epoch for CCR v2 event format
constructor( private readonly url: URL, headers: Record<string, string> = {}, sessionId?: string, refreshHeaders?: () => Record<string, string>, initialSequenceNum?: number, /** * Per-instance auth header source. Omit to read the process-wide * CLAUDE_CODE_SESSION_ACCESS_TOKEN (single-session callers). Required * for concurrent multi-session callers — the env-var path is a process * global and would stomp across sessions. */ getAuthHeaders?: () => Record<string, string>, ) { this.headers = headers this.sessionId = sessionId this.refreshHeaders = refreshHeaders this.getAuthHeaders = getAuthHeaders ?? getSessionIngressAuthHeaders this.postUrl = convertSSEUrlToPostUrl(url) // Seed with a caller-provided high-water mark so the first connect() // sends from_sequence_num / Last-Event-ID. Without this, a fresh // SSETransport always asks the server to replay from sequence 0 — // the entire session history on every transport swap. if (initialSequenceNum !== undefined && initialSequenceNum > 0) { this.lastSequenceNum = initialSequenceNum } logForDebugging(`SSETransport: SSE URL = ${url.href}`) logForDebugging(`SSETransport: POST URL = ${this.postUrl}`) logForDiagnosticsNoPII('info', 'cli_sse_transport_initialized') }
/** * High-water mark of sequence numbers seen on this stream. Callers that * recreate the transport (e.g. replBridge onWorkReceived) read this before * close() and pass it as `initialSequenceNum` to the next instance so the * server resumes from the right point instead of replaying everything. */ getLastSequenceNum(): number { return this.lastSequenceNum }
async connect(): Promise<void> { if (this.state !== 'idle' && this.state !== 'reconnecting') { logForDebugging( `SSETransport: Cannot connect, current state is ${this.state}`, { level: 'error' }, ) logForDiagnosticsNoPII('error', 'cli_sse_connect_failed') return }
this.state = 'reconnecting' const connectStartTime = Date.now()
// Build SSE URL with sequence number for resumption const sseUrl = new URL(this.url.href) if (this.lastSequenceNum > 0) { sseUrl.searchParams.set('from_sequence_num', String(this.lastSequenceNum)) }
// Build headers -- use fresh auth headers (supports Cookie for session keys). // Remove stale Authorization header from this.headers when Cookie auth is used, // since sending both confuses the auth interceptor. const authHeaders = this.getAuthHeaders() const headers: Record<string, string> = { ...this.headers, ...authHeaders, Accept: 'text/event-stream', 'anthropic-version': '2023-06-01', 'User-Agent': getClaudeCodeUserAgent(), } if (authHeaders['Cookie']) { delete headers['Authorization'] } if (this.lastSequenceNum > 0) { headers['Last-Event-ID'] = String(this.lastSequenceNum) }
logForDebugging(`SSETransport: Opening ${sseUrl.href}`) logForDiagnosticsNoPII('info', 'cli_sse_connect_opening')
this.abortController = new AbortController()
try { // eslint-disable-next-line eslint-plugin-n/no-unsupported-features/node-builtins const response = await fetch(sseUrl.href, { headers, signal: this.abortController.signal, })
if (!response.ok) { const isPermanent = PERMANENT_HTTP_CODES.has(response.status) logForDebugging( `SSETransport: HTTP ${response.status}${isPermanent ? ' (permanent)' : ''}`, { level: 'error' }, ) logForDiagnosticsNoPII('error', 'cli_sse_connect_http_error', { status: response.status, })
if (isPermanent) { this.state = 'closed' this.onCloseCallback?.(response.status) return }
this.handleConnectionError() return }
if (!response.body) { logForDebugging('SSETransport: No response body') this.handleConnectionError() return }
// Successfully connected const connectDuration = Date.now() - connectStartTime logForDebugging('SSETransport: Connected') logForDiagnosticsNoPII('info', 'cli_sse_connect_connected', { duration_ms: connectDuration, })
this.state = 'connected' this.reconnectAttempts = 0 this.reconnectStartTime = null this.resetLivenessTimer()
// Read the SSE stream await this.readStream(response.body) } catch (error) { if (this.abortController?.signal.aborted) { // Intentional close return }
logForDebugging( `SSETransport: Connection error: ${errorMessage(error)}`, { level: 'error' }, ) logForDiagnosticsNoPII('error', 'cli_sse_connect_error') this.handleConnectionError() } }
/** * Read and process the SSE stream body. */ // eslint-disable-next-line eslint-plugin-n/no-unsupported-features/node-builtins private async readStream(body: ReadableStream<Uint8Array>): Promise<void> { const reader = body.getReader() const decoder = new TextDecoder() let buffer = ''
try { while (true) { const { done, value } = await reader.read() if (done) break
buffer += decoder.decode(value, STREAM_DECODE_OPTS) const { frames, remaining } = parseSSEFrames(buffer) buffer = remaining
for (const frame of frames) { // Any frame (including keepalive comments) proves the connection is alive this.resetLivenessTimer()
if (frame.id) { const seqNum = parseInt(frame.id, 10) if (!isNaN(seqNum)) { if (this.seenSequenceNums.has(seqNum)) { logForDebugging( `SSETransport: DUPLICATE frame seq=${seqNum} (lastSequenceNum=${this.lastSequenceNum}, seenCount=${this.seenSequenceNums.size})`, { level: 'warn' }, ) logForDiagnosticsNoPII('warn', 'cli_sse_duplicate_sequence') } else { this.seenSequenceNums.add(seqNum) // Prevent unbounded growth: once we have many entries, prune // old sequence numbers that are well below the high-water mark. // Only sequence numbers near lastSequenceNum matter for dedup. if (this.seenSequenceNums.size > 1000) { const threshold = this.lastSequenceNum - 200 for (const s of this.seenSequenceNums) { if (s < threshold) { this.seenSequenceNums.delete(s) } } } } if (seqNum > this.lastSequenceNum) { this.lastSequenceNum = seqNum } } }
if (frame.event && frame.data) { this.handleSSEFrame(frame.event, frame.data) } else if (frame.data) { // data: without event: — server is emitting the old envelope format // or a bug. Log so incidents show as a signal instead of silent drops. logForDebugging( 'SSETransport: Frame has data: but no event: field — dropped', { level: 'warn' }, ) logForDiagnosticsNoPII('warn', 'cli_sse_frame_missing_event_field') } } } } catch (error) { if (this.abortController?.signal.aborted) return logForDebugging( `SSETransport: Stream read error: ${errorMessage(error)}`, { level: 'error' }, ) logForDiagnosticsNoPII('error', 'cli_sse_stream_read_error') } finally { reader.releaseLock() }
// Stream ended — reconnect unless we're closing if (this.state !== 'closing' && this.state !== 'closed') { logForDebugging('SSETransport: Stream ended, reconnecting') this.handleConnectionError() } }
/** * Handle a single SSE frame. The event: field names the variant; data: * carries the inner proto JSON directly (no envelope). * * Worker subscribers only receive client_event frames (see notifier.go) — * any other event type indicates a server-side change that CC doesn't yet * understand. Log a diagnostic so we notice in telemetry. */ private handleSSEFrame(eventType: string, data: string): void { if (eventType !== 'client_event') { logForDebugging( `SSETransport: Unexpected SSE event type '${eventType}' on worker stream`, { level: 'warn' }, ) logForDiagnosticsNoPII('warn', 'cli_sse_unexpected_event_type', { event_type: eventType, }) return }
let ev: StreamClientEvent try { ev = jsonParse(data) as StreamClientEvent } catch (error) { logForDebugging( `SSETransport: Failed to parse client_event data: ${errorMessage(error)}`, { level: 'error' }, ) return }
const payload = ev.payload if (payload && typeof payload === 'object' && 'type' in payload) { const sessionLabel = this.sessionId ? ` session=${this.sessionId}` : '' logForDebugging( `SSETransport: Event seq=${ev.sequence_num} event_id=${ev.event_id} event_type=${ev.event_type} payload_type=${String(payload.type)}${sessionLabel}`, ) logForDiagnosticsNoPII('info', 'cli_sse_message_received') // Pass the unwrapped payload as newline-delimited JSON, // matching the format that StructuredIO/WebSocketTransport consumers expect this.onData?.(jsonStringify(payload) + '\n') } else { logForDebugging( `SSETransport: Ignoring client_event with no type in payload: event_id=${ev.event_id}`, ) }
this.onEventCallback?.(ev) }
/** * Handle connection errors with exponential backoff and time budget. */ private handleConnectionError(): void { this.clearLivenessTimer()
if (this.state === 'closing' || this.state === 'closed') return
// Abort any in-flight SSE fetch this.abortController?.abort() this.abortController = null
const now = Date.now() if (!this.reconnectStartTime) { this.reconnectStartTime = now }
const elapsed = now - this.reconnectStartTime if (elapsed < RECONNECT_GIVE_UP_MS) { // Clear any existing timer if (this.reconnectTimer) { clearTimeout(this.reconnectTimer) this.reconnectTimer = null }
// Refresh headers before reconnecting if (this.refreshHeaders) { const freshHeaders = this.refreshHeaders() Object.assign(this.headers, freshHeaders) logForDebugging('SSETransport: Refreshed headers for reconnect') }
this.state = 'reconnecting' this.reconnectAttempts++
const baseDelay = Math.min( RECONNECT_BASE_DELAY_MS * Math.pow(2, this.reconnectAttempts - 1), RECONNECT_MAX_DELAY_MS, ) // Add ±25% jitter const delay = Math.max( 0, baseDelay + baseDelay * 0.25 * (2 * Math.random() - 1), )
logForDebugging( `SSETransport: Reconnecting in ${Math.round(delay)}ms (attempt ${this.reconnectAttempts}, ${Math.round(elapsed / 1000)}s elapsed)`, ) logForDiagnosticsNoPII('error', 'cli_sse_reconnect_attempt', { reconnectAttempts: this.reconnectAttempts, })
this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null void this.connect() }, delay) } else { logForDebugging( `SSETransport: Reconnection time budget exhausted after ${Math.round(elapsed / 1000)}s`, { level: 'error' }, ) logForDiagnosticsNoPII('error', 'cli_sse_reconnect_exhausted', { reconnectAttempts: this.reconnectAttempts, elapsedMs: elapsed, }) this.state = 'closed' this.onCloseCallback?.() } }
/** * Bound timeout callback. Hoisted from an inline closure so that * resetLivenessTimer (called per-frame) does not allocate a new closure * on every SSE frame. */ private readonly onLivenessTimeout = (): void => { this.livenessTimer = null logForDebugging('SSETransport: Liveness timeout, reconnecting', { level: 'error', }) logForDiagnosticsNoPII('error', 'cli_sse_liveness_timeout') this.abortController?.abort() this.handleConnectionError() }
/** * Reset the liveness timer. If no SSE frame arrives within the timeout, * treat the connection as dead and reconnect. */ private resetLivenessTimer(): void { this.clearLivenessTimer() this.livenessTimer = setTimeout(this.onLivenessTimeout, LIVENESS_TIMEOUT_MS) }
private clearLivenessTimer(): void { if (this.livenessTimer) { clearTimeout(this.livenessTimer) this.livenessTimer = null } }
// ----------------------------------------------------------------------- // Write (HTTP POST) — same pattern as HybridTransport // -----------------------------------------------------------------------
async write(message: StdoutMessage): Promise<void> { const authHeaders = this.getAuthHeaders() if (Object.keys(authHeaders).length === 0) { logForDebugging('SSETransport: No session token available for POST') logForDiagnosticsNoPII('warn', 'cli_sse_post_no_token') return }
const headers: Record<string, string> = { ...authHeaders, 'Content-Type': 'application/json', 'anthropic-version': '2023-06-01', 'User-Agent': getClaudeCodeUserAgent(), }
logForDebugging( `SSETransport: POST body keys=${Object.keys(message as Record<string, unknown>).join(',')}`, )
for (let attempt = 1; attempt <= POST_MAX_RETRIES; attempt++) { try { const response = await axios.post(this.postUrl, message, { headers, validateStatus: alwaysValidStatus, })
if (response.status === 200 || response.status === 201) { logForDebugging(`SSETransport: POST success type=${message.type}`) return }
logForDebugging( `SSETransport: POST ${response.status} body=${jsonStringify(response.data).slice(0, 200)}`, ) // 4xx errors (except 429) are permanent - don't retry if ( response.status >= 400 && response.status < 500 && response.status !== 429 ) { logForDebugging( `SSETransport: POST returned ${response.status} (client error), not retrying`, ) logForDiagnosticsNoPII('warn', 'cli_sse_post_client_error', { status: response.status, }) return }
// 429 or 5xx - retry logForDebugging( `SSETransport: POST returned ${response.status}, attempt ${attempt}/${POST_MAX_RETRIES}`, ) logForDiagnosticsNoPII('warn', 'cli_sse_post_retryable_error', { status: response.status, attempt, }) } catch (error) { const axiosError = error as AxiosError logForDebugging( `SSETransport: POST error: ${axiosError.message}, attempt ${attempt}/${POST_MAX_RETRIES}`, ) logForDiagnosticsNoPII('warn', 'cli_sse_post_network_error', { attempt, }) }
if (attempt === POST_MAX_RETRIES) { logForDebugging( `SSETransport: POST failed after ${POST_MAX_RETRIES} attempts, continuing`, ) logForDiagnosticsNoPII('warn', 'cli_sse_post_retries_exhausted') return }
const delayMs = Math.min( POST_BASE_DELAY_MS * Math.pow(2, attempt - 1), POST_MAX_DELAY_MS, ) await sleep(delayMs) } }
// ----------------------------------------------------------------------- // Transport interface // -----------------------------------------------------------------------
isConnectedStatus(): boolean { return this.state === 'connected' }
isClosedStatus(): boolean { return this.state === 'closed' }
setOnData(callback: (data: string) => void): void { this.onData = callback }
setOnClose(callback: (closeCode?: number) => void): void { this.onCloseCallback = callback }
setOnEvent(callback: (event: StreamClientEvent) => void): void { this.onEventCallback = callback }
close(): void { if (this.reconnectTimer) { clearTimeout(this.reconnectTimer) this.reconnectTimer = null } this.clearLivenessTimer()
this.state = 'closing' this.abortController?.abort() this.abortController = null }}
// ---------------------------------------------------------------------------// URL Conversion// ---------------------------------------------------------------------------
/** * Convert an SSE URL to the HTTP POST endpoint URL. * The SSE stream URL and POST URL share the same base; the POST endpoint * is at `/events` (without `/stream`). * * From: https://api.example.com/v2/session_ingress/session/<session_id>/events/stream * To: https://api.example.com/v2/session_ingress/session/<session_id>/events */function convertSSEUrlToPostUrl(sseUrl: URL): string { let pathname = sseUrl.pathname // Remove /stream suffix to get the POST events endpoint if (pathname.endsWith('/stream')) { pathname = pathname.slice(0, -'/stream'.length) } return `${sseUrl.protocol}//${sseUrl.host}${pathname}`}