Something went wrong. Try again.
source dump of claude code forked from oppi.li/claude-code
Something went wrong. Try again.
16 kB · 461 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462/** * Shared transport-layer helpers for bridge message handling. * * Extracted from replBridge.ts so both the env-based core (initBridgeCore) * and the env-less core (initEnvLessBridgeCore) can use the same ingress * parsing, control-request handling, and echo-dedup machinery. * * Everything here is pure — no closure over bridge-specific state. All * collaborators (transport, sessionId, UUID sets, callbacks) are passed * as params. */
import { randomUUID } from 'crypto'import type { SDKMessage } from '../entrypoints/agentSdkTypes.js'import type { SDKControlRequest, SDKControlResponse,} from '../entrypoints/sdk/controlTypes.js'import type { SDKResultSuccess } from '../entrypoints/sdk/coreTypes.js'import { logEvent } from '../services/analytics/index.js'import { EMPTY_USAGE } from '../services/api/emptyUsage.js'import type { Message } from '../types/message.js'import { normalizeControlMessageKeys } from '../utils/controlMessageCompat.js'import { logForDebugging } from '../utils/debug.js'import { stripDisplayTagsAllowEmpty } from '../utils/displayTags.js'import { errorMessage } from '../utils/errors.js'import type { PermissionMode } from '../utils/permissions/PermissionMode.js'import { jsonParse } from '../utils/slowOperations.js'import type { ReplBridgeTransport } from './replBridgeTransport.js'
// ─── Type guards ─────────────────────────────────────────────────────────────
/** Type predicate for parsed WebSocket messages. SDKMessage is a * discriminated union on `type` — validating the discriminant is * sufficient for the predicate; callers narrow further via the union. */export function isSDKMessage(value: unknown): value is SDKMessage { return ( value !== null && typeof value === 'object' && 'type' in value && typeof value.type === 'string' )}
/** Type predicate for control_response messages from the server. */export function isSDKControlResponse( value: unknown,): value is SDKControlResponse { return ( value !== null && typeof value === 'object' && 'type' in value && value.type === 'control_response' && 'response' in value )}
/** Type predicate for control_request messages from the server. */export function isSDKControlRequest( value: unknown,): value is SDKControlRequest { return ( value !== null && typeof value === 'object' && 'type' in value && value.type === 'control_request' && 'request_id' in value && 'request' in value )}
/** * True for message types that should be forwarded to the bridge transport. * The server only wants user/assistant turns and slash-command system events; * everything else (tool_result, progress, etc.) is internal REPL chatter. */export function isEligibleBridgeMessage(m: Message): boolean { // Virtual messages (REPL inner calls) are display-only — bridge/SDK // consumers see the REPL tool_use/result which summarizes the work. if ((m.type === 'user' || m.type === 'assistant') && m.isVirtual) { return false } return ( m.type === 'user' || m.type === 'assistant' || (m.type === 'system' && m.subtype === 'local_command') )}
/** * Extract title-worthy text from a Message for onUserMessage. Returns * undefined for messages that shouldn't title the session: non-user, meta * (nudges), tool results, compact summaries, non-human origins (task * notifications, channel messages), or pure display-tag content * (<ide_opened_file>, <session-start-hook>, etc.). * * Synthetic interrupts ([Request interrupted by user]) are NOT filtered here — * isSyntheticMessage lives in messages.ts (heavy import, pulls command * registry). The initialMessages path in initReplBridge checks it; the * writeMessages path reaching an interrupt as the *first* message is * implausible (an interrupt implies a prior prompt already flowed through). */export function extractTitleText(m: Message): string | undefined { if (m.type !== 'user' || m.isMeta || m.toolUseResult || m.isCompactSummary) return undefined if (m.origin && m.origin.kind !== 'human') return undefined const content = m.message.content let raw: string | undefined if (typeof content === 'string') { raw = content } else { for (const block of content) { if (block.type === 'text') { raw = block.text break } } } if (!raw) return undefined const clean = stripDisplayTagsAllowEmpty(raw) return clean || undefined}
// ─── Ingress routing ─────────────────────────────────────────────────────────
/** * Parse an ingress WebSocket message and route it to the appropriate handler. * Ignores messages whose UUID is in recentPostedUUIDs (echoes of what we sent) * or in recentInboundUUIDs (re-deliveries we've already forwarded — e.g. * server replayed history after a transport swap lost the seq-num cursor). */export function handleIngressMessage( data: string, recentPostedUUIDs: BoundedUUIDSet, recentInboundUUIDs: BoundedUUIDSet, onInboundMessage: ((msg: SDKMessage) => void | Promise<void>) | undefined, onPermissionResponse?: ((response: SDKControlResponse) => void) | undefined, onControlRequest?: ((request: SDKControlRequest) => void) | undefined,): void { try { const parsed: unknown = normalizeControlMessageKeys(jsonParse(data))
// control_response is not an SDKMessage — check before the type guard if (isSDKControlResponse(parsed)) { logForDebugging('[bridge:repl] Ingress message type=control_response') onPermissionResponse?.(parsed) return }
// control_request from the server (initialize, set_model, can_use_tool). // Must respond promptly or the server kills the WS (~10-14s timeout). if (isSDKControlRequest(parsed)) { logForDebugging( `[bridge:repl] Inbound control_request subtype=${parsed.request.subtype}`, ) onControlRequest?.(parsed) return }
if (!isSDKMessage(parsed)) return
// Check for UUID to detect echoes of our own messages const uuid = 'uuid' in parsed && typeof parsed.uuid === 'string' ? parsed.uuid : undefined
if (uuid && recentPostedUUIDs.has(uuid)) { logForDebugging( `[bridge:repl] Ignoring echo: type=${parsed.type} uuid=${uuid}`, ) return }
// Defensive dedup: drop inbound prompts we've already forwarded. The // SSE seq-num carryover (lastTransportSequenceNum) is the primary fix // for history-replay; this catches edge cases where that negotiation // fails (server ignores from_sequence_num, transport died before // receiving any frames, etc). if (uuid && recentInboundUUIDs.has(uuid)) { logForDebugging( `[bridge:repl] Ignoring re-delivered inbound: type=${parsed.type} uuid=${uuid}`, ) return }
logForDebugging( `[bridge:repl] Ingress message type=${parsed.type}${uuid ? ` uuid=${uuid}` : ''}`, )
if (parsed.type === 'user') { if (uuid) recentInboundUUIDs.add(uuid) logEvent('tengu_bridge_message_received', { is_repl: true, }) // Fire-and-forget — handler may be async (attachment resolution). void onInboundMessage?.(parsed) } else { logForDebugging( `[bridge:repl] Ignoring non-user inbound message: type=${parsed.type}`, ) } } catch (err) { logForDebugging( `[bridge:repl] Failed to parse ingress message: ${errorMessage(err)}`, ) }}
// ─── Server-initiated control requests ───────────────────────────────────────
export type ServerControlRequestHandlers = { transport: ReplBridgeTransport | null sessionId: string /** * When true, all mutable requests (interrupt, set_model, set_permission_mode, * set_max_thinking_tokens) reply with an error instead of false-success. * initialize still replies success — the server kills the connection otherwise. * Used by the outbound-only bridge mode and the SDK's /bridge subpath so claude.ai sees a * proper error instead of "action succeeded but nothing happened locally". */ outboundOnly?: boolean onInterrupt?: () => void onSetModel?: (model: string | undefined) => void onSetMaxThinkingTokens?: (maxTokens: number | null) => void onSetPermissionMode?: ( mode: PermissionMode, ) => { ok: true } | { ok: false; error: string }}
const OUTBOUND_ONLY_ERROR = 'This session is outbound-only. Enable Remote Control locally to allow inbound control.'
/** * Respond to inbound control_request messages from the server. The server * sends these for session lifecycle events (initialize, set_model) and * for turn-level coordination (interrupt, set_max_thinking_tokens). If we * don't respond, the server hangs and kills the WS after ~10-14s. * * Previously a closure inside initBridgeCore's onWorkReceived; now takes * collaborators as params so both cores can use it. */export function handleServerControlRequest( request: SDKControlRequest, handlers: ServerControlRequestHandlers,): void { const { transport, sessionId, outboundOnly, onInterrupt, onSetModel, onSetMaxThinkingTokens, onSetPermissionMode, } = handlers if (!transport) { logForDebugging( '[bridge:repl] Cannot respond to control_request: transport not configured', ) return }
let response: SDKControlResponse
// Outbound-only: reply error for mutable requests so claude.ai doesn't show // false success. initialize must still succeed (server kills the connection // if it doesn't — see comment above). if (outboundOnly && request.request.subtype !== 'initialize') { response = { type: 'control_response', response: { subtype: 'error', request_id: request.request_id, error: OUTBOUND_ONLY_ERROR, }, } const event = { ...response, session_id: sessionId } void transport.write(event) logForDebugging( `[bridge:repl] Rejected ${request.request.subtype} (outbound-only) request_id=${request.request_id}`, ) return }
switch (request.request.subtype) { case 'initialize': // Respond with minimal capabilities — the REPL handles // commands, models, and account info itself. response = { type: 'control_response', response: { subtype: 'success', request_id: request.request_id, response: { commands: [], output_style: 'normal', available_output_styles: ['normal'], models: [], account: {}, pid: process.pid, }, }, } break
case 'set_model': onSetModel?.(request.request.model) response = { type: 'control_response', response: { subtype: 'success', request_id: request.request_id, }, } break
case 'set_max_thinking_tokens': onSetMaxThinkingTokens?.(request.request.max_thinking_tokens) response = { type: 'control_response', response: { subtype: 'success', request_id: request.request_id, }, } break
case 'set_permission_mode': { // The callback returns a policy verdict so we can send an error // control_response without importing isAutoModeGateEnabled / // isBypassPermissionsModeDisabled here (bootstrap-isolation). If no // callback is registered (daemon context, which doesn't wire this — // see daemonBridge.ts), return an error verdict rather than a silent // false-success: the mode is never actually applied in that context, // so success would lie to the client. const verdict = onSetPermissionMode?.(request.request.mode) ?? { ok: false, error: 'set_permission_mode is not supported in this context (onSetPermissionMode callback not registered)', } if (verdict.ok) { response = { type: 'control_response', response: { subtype: 'success', request_id: request.request_id, }, } } else { response = { type: 'control_response', response: { subtype: 'error', request_id: request.request_id, error: verdict.error, }, } } break }
case 'interrupt': onInterrupt?.() response = { type: 'control_response', response: { subtype: 'success', request_id: request.request_id, }, } break
default: // Unknown subtype — respond with error so the server doesn't // hang waiting for a reply that never comes. response = { type: 'control_response', response: { subtype: 'error', request_id: request.request_id, error: `REPL bridge does not handle control_request subtype: ${request.request.subtype}`, }, } }
const event = { ...response, session_id: sessionId } void transport.write(event) logForDebugging( `[bridge:repl] Sent control_response for ${request.request.subtype} request_id=${request.request_id} result=${response.response.subtype}`, )}
// ─── Result message (for session archival on teardown) ───────────────────────
/** * Build a minimal `SDKResultSuccess` message for session archival. * The server needs this event before a WS close to trigger archival. */export function makeResultMessage(sessionId: string): SDKResultSuccess { return { type: 'result', subtype: 'success', duration_ms: 0, duration_api_ms: 0, is_error: false, num_turns: 0, result: '', stop_reason: null, total_cost_usd: 0, usage: { ...EMPTY_USAGE }, modelUsage: {}, permission_denials: [], session_id: sessionId, uuid: randomUUID(), }}
// ─── BoundedUUIDSet (echo-dedup ring buffer) ─────────────────────────────────
/** * FIFO-bounded set backed by a circular buffer. Evicts the oldest entry * when capacity is reached, keeping memory usage constant at O(capacity). * * Messages are added in chronological order, so evicted entries are always * the oldest. The caller relies on external ordering (the hook's * lastWrittenIndexRef) as the primary dedup — this set is a secondary * safety net for echo filtering and race-condition dedup. */export class BoundedUUIDSet { private readonly capacity: number private readonly ring: (string | undefined)[] private readonly set = new Set<string>() private writeIdx = 0
constructor(capacity: number) { this.capacity = capacity this.ring = new Array<string | undefined>(capacity) }
add(uuid: string): void { if (this.set.has(uuid)) return // Evict the entry at the current write position (if occupied) const evicted = this.ring[this.writeIdx] if (evicted !== undefined) { this.set.delete(evicted) } this.ring[this.writeIdx] = uuid this.set.add(uuid) this.writeIdx = (this.writeIdx + 1) % this.capacity }
has(uuid: string): boolean { return this.set.has(uuid) }
clear(): void { this.set.clear() this.ring.fill(undefined) this.writeIdx = 0 }}