브리지와 리모트 컨트롤

Environments API가 작업을 전달할 때는 불투명한 secret 필드를 함께 붙입니다. 이것은 base64url로 인코딩된 JSON 블롭입니다. bridge/workSecret.ts의 decodeWorkSecret()가 이를 풀어냅니다.

export function decodeWorkSecret(secret: string): WorkSecret {
  const json = Buffer.from(secret, 'base64url').toString('utf-8')
  const parsed: unknown = jsonParse(json)
  // version === 1, session_ingress_token, api_base_url을 검증
  return parsed as WorkSecret
}

type WorkSecret = {
  version: 1
  session_ingress_token: string   // CCR worker 엔드포인트용 JWT
  api_base_url: string
  sources: Array<{ type: string; git_info?: ... }>
  auth: Array<{ type: string; token: string }>
  use_code_sessions?: boolean    // 서버가 결정하는 CCR v2 선택자
}

session_ingress_token은 worker 계층 작업을 승인하는 짧은 수명의 JWT입니다. 이 자리에 OAuth 토큰을 대신 쓸 수는 없습니다. CCR은 JWT의 session_id와 role=worker 클레임을 직접 검증합니다.

v1/v2 분기는 모든 인증 검사를 통과한 뒤 bridge/initReplBridge.ts에서 결정됩니다.

// tengu_bridge_repl_v2는 REPL에 env-less (v2) 경로를 활성화한다.
// perpetual=true이면 v1로 되돌아간다. bridge-pointer는 아직 v2에 연결되지 않았다.
if (isEnvLessBridgeEnabled() && !perpetual) {
  const versionError = await checkEnvLessBridgeMinVersion()
  if (versionError) {
    onStateChange?.('failed', 'run `claude update` to upgrade')
    return null
  }
  const { initEnvLessBridgeCore } = await import('./remoteBridgeCore.js')
  return initEnvLessBridgeCore({ baseUrl, orgUUID, title, ... })
}

// v1 경로: env 기반 register/poll/ack/heartbeat
return initBridgeCore({ dir, machineName, branch, gitRepoUrl, ... })

버전 하한은 서로 독립적입니다. tengu_bridge_min_version은 v1을, tengu_bridge_repl_v2_config.min_version은 v2를 제어합니다. 둘 다 GrowthBook 동적 설정이므로, 운영팀은 하한을 0.0.0으로 낮춰 롤백할 수 있습니다.

CCR v2 호환 계층은 ID를 둘로 나눕니다. 인프라 엔드포인트는 cse_* ID를 내주고, claude.ai 프런트엔드는 session_* 경로를 사용합니다. 둘은 접두사만 다를 뿐 같은 UUID입니다.

// bridge/sessionIdCompat.ts
export function toCompatSessionId(id: string): string {
  if (!id.startsWith('cse_')) return id
  if (_isCseShimEnabled && !_isCseShimEnabled()) return id
  return 'session_' + id.slice('cse_'.length)
}

export function toInfraSessionId(id: string): string {
  if (!id.startsWith('session_')) return id
  return 'cse_' + id.slice('session_'.length)
}

// sameSessionId()는 접두사를 무시해서, poll 루프가
// 호환 게이트가 켜졌을 때 자기 세션을 '외부 세션'으로 거부하지 않게 한다
export function sameSessionId(a: string, b: string): boolean {
  const aBody = a.slice(a.lastIndexOf('_') + 1)
  const bBody = b.slice(b.lastIndexOf('_') + 1)
  return aBody.length >= 4 && aBody === bBody
}

isCseShimEnabled 킬 스위치는 setCseShimGate()로 주입되며, Agent SDK 번들에 GrowthBook을 import하지 않기 위한 장치입니다.

createV1ReplTransport()는 HybridTransport를 감싸는 얇은 패스스루 래퍼입니다. HybridTransport는 인바운드 메시지를 위해 Session-Ingress로 WebSocket을 열고, 아웃바운드는 HTTP POST를 사용합니다. v1은 SSE 시퀀스 번호를 쓰지 않습니다. 재생은 서버 측 커서가 처리합니다.

export function createV1ReplTransport(
  hybrid: HybridTransport,
): ReplBridgeTransport {
  return {
    write: msg => hybrid.write(msg),
    writeBatch: msgs => hybrid.writeBatch(msgs),
    close: () => hybrid.close(),
    isConnectedStatus: () => hybrid.isConnectedStatus(),
    getLastSequenceNum: () => 0,     // WS 재생은 SSE 시퀀스 번호와 다르다
    reportState: () => {},              // no-op
    reportMetadata: () => {},
    reportDelivery: () => {},
    flush: () => Promise.resolve(),     // 각 POST는 write마다 await된다
    // ... 그 밖의 패스스루들
  }
}

v2 전송은 비대칭입니다. 읽기는 SSE (Server-Sent Events)로 들어오고, 쓰기는 CCRClient를 통해 SerialBatchEventUploader가 /worker/events로 POST합니다. 이 분리는 의도된 설계입니다. 인바운드 SSE 스트림이 잠시 멈춰도 CCRClient의 heartbeat와 쓰기 경로는 살아 있을 수 있습니다.

export async function createV2ReplTransport(opts: {
  sessionUrl: string
  ingressToken: string
  sessionId: string
  initialSequenceNum?: number  // 재연결 시 이 SSE 시퀀스부터 재개
  epoch?: number               // /bridge가 제공했다면 registerWorker 생략
  getAuthToken?: () => string | undefined  // 멀티 세션에서도 안전
  outboundOnly?: boolean       // SSE 읽기 생략 (미러 모드)
}): Promise<ReplBridgeTransport> {

  // registerWorker는 CCRClient에 필요한 worker_epoch를 반환한다
  const epoch = opts.epoch ?? (await registerWorker(sessionUrl, ingressToken))

  const sse = new SSETransport(sseUrl, {}, sessionId, ...)
  const ccr = new CCRClient(sse, new URL(sessionUrl), {
    getAuthHeaders,
    onEpochMismatch: () => {
      // 서버의 409 응답: 다른 worker가 우리 epoch를 대체했다
      ccr.close(); sse.close()
      onCloseCb?.(4090)  // poll 루프 복구 코드
      throw new Error('epoch superseded')
    },
  })
  // daemon 재시작 때 유령 프롬프트가 쏟아지는 일을 막으려면
  // 'received'와 함께 'processed'도 즉시 ACK해야 한다 (CC-1263)
  sse.setOnEvent(event => {
    ccr.reportDelivery(event.event_id, 'received')
    ccr.reportDelivery(event.event_id, 'processed')
  })
  return { write: msg => ccr.writeEvent(msg), ... }
}

브리지 세션이 시작되면 과거 대화를 POST로 한 번에 flush합니다. 이때 새 메시지가 들어오면 서버에서는 이력과 뒤섞일 수 있습니다. FlushGate는 flush가 끝날 때까지 그것들을 큐에 넣어 둡니다.

class FlushGate<T> {
  start(): void            // flush 진행 중으로 표시, enqueue()가 큐잉을 시작
  end(): T[]               // flush 완료, 비워 낼 큐 항목들을 반환
  enqueue(...items: T[]): boolean  // active면 true(큐에 저장), 아니면 false(그대로 통과)
  drop(): number           // 큐를 폐기 (transport가 영구적으로 닫힘)
  deactivate(): void       // transport가 교체됨, 새 transport가 이를 비운다
}

deactivate()는 transport가 교체될 때 호출됩니다. 예를 들어 env loss 뒤 재연결되는 경우입니다. 큐에 쌓인 항목은 새 transport의 end() 호출까지 보존됩니다.

// bridge/bridgePermissionCallbacks.ts
type BridgePermissionResponse = {
  behavior: 'allow' | 'deny'
  updatedInput?: Record<string, unknown>
  updatedPermissions?: PermissionUpdate[]
  message?: string
}

function isBridgePermissionResponse(value: unknown): value is BridgePermissionResponse {
  if (!value || typeof value !== 'object') return false
  return (
    'behavior' in value &&
    (value.behavior === 'allow' || value.behavior === 'deny')
  )
}

remote/RemoteSessionManager.ts는 로컬 CLI가 바라보고 있는 CCR 호스팅 세션의 클라이언트 측을 조정합니다. WebSocket을 통해 CCR에서 권한 요청을 받고, 사용자가 응답할 때까지 그것을 붙잡아 둡니다.

class RemoteSessionManager {
  private pendingPermissionRequests = new Map<string, SDKControlPermissionRequest>()

  private handleControlRequest(req: SDKControlRequest): void {
    const { request_id, request: inner } = req
    if (inner.subtype === 'can_use_tool') {
      this.pendingPermissionRequests.set(request_id, inner)
      this.callbacks.onPermissionRequest(inner, request_id)
    } else {
      // 지원하지 않는 subtype이면 서버가 멈추지 않도록 즉시 에러 응답을 보낸다
      this.websocket?.sendControlResponse({ type: 'control_response', response: {
        subtype: 'error', request_id, error: 'Unsupported subtype'
      }})
    }
  }

  respondToPermissionRequest(requestId: string, result: RemotePermissionResponse): void {
    this.pendingPermissionRequests.delete(requestId)
    this.websocket?.sendControlResponse({
      type: 'control_response',
      response: {
        subtype: 'success', request_id: requestId,
        response: {
          behavior: result.behavior,
          ...(result.behavior === 'allow'
            ? { updatedInput: result.updatedInput }
            : { message: result.message }),
        },
      },
    })
  }
}

REPL의 권한 UI는 도구 사용 블록이 들어 있는 실제 AssistantMessage를 기대합니다. 그런데 권한 요청이 리모트 CCR 컨테이너에서 오면 로컬에는 그런 메시지가 없습니다. remote/remotePermissionBridge.ts가 이를 만들어 냅니다.

export function createSyntheticAssistantMessage(
  request: SDKControlPermissionRequest,
  requestId: string,
): AssistantMessage {
  return {
    type: 'assistant',
    uuid: randomUUID(),
    message: {
      id: `remote-${requestId}`,
      type: 'message',
      role: 'assistant',
      content: [{
        type: 'tool_use',
        id: request.tool_use_id,
        name: request.tool_name,
        input: request.input,
      }],
      // usage가 0인 스텁 필드들 ...
    } as AssistantMessage['message'],
    requestId: undefined,
    timestamp: new Date().toISOString(),
  }
}

// 로컬에서 알지 못하는 도구를 위한 처리, 예: 리모트 컨테이너의 MCP 도구
export function createToolStub(toolName: string): Tool {
  return {
    name: toolName,
    isEnabled: () => true,
    needsPermissions: () => true,
    // 표시용으로 처음 3개의 input key:value 쌍을 렌더링한다
    call: async () => ({ data: '' }),
  } as unknown as Tool
}

// commands/bridge/bridge.tsx (compiled)
function BridgeToggle({ onDone, name }) {
  const replBridgeConnected = useAppState(s => s.replBridgeConnected)
  const replBridgeEnabled   = useAppState(s => s.replBridgeEnabled)
  const [showDisconnectDialog, setShow] = useState(false)

  useEffect(() => {
    if ((replBridgeConnected || replBridgeEnabled) && !replBridgeOutboundOnly) {
      setShow(true)  // 이미 연결되어 있음, 대화상자를 표시
      return
    }
    (async () => {
      const error = await checkBridgePrerequisites()
      if (error) { onDone(error, { display: 'system' }); return }
      setAppState(prev => ({
        ...prev,
        replBridgeEnabled: true,
        replBridgeExplicit: true,
        replBridgeOutboundOnly: false,
        replBridgeInitialName: name,
      }))
      onDone('Remote Control connecting…', { display: 'system' })
    })()
  }, [])  // 마운트 시 한 번만 실행
}

initReplBridge가 진행되기 전에 다섯 가지 검사가 모두 통과해야 합니다.

1. 런타임 게이트: isBridgeEnabledBlocking(), tengu_ccr_bridge GrowthBook 플래그와 claude.ai OAuth 구독이 모두 필요합니다. Bedrock/Vertex/API-key 인증은 허용되지 않습니다.
2. OAuth 토큰 존재: getBridgeAccessToken()이 값을 반환해야 합니다.
3. 조직 정책: isPolicyAllowed('allow_remote_control'), 엔터프라이즈 관리자는 조직 전체 구성원의 RC를 끌 수 있습니다.
4. 토큰 신선도: 선제 갱신을 수행하고, 만료됐고 갱신도 불가능하다면 건너뜁니다. 확정적인 401 루프를 피하기 위함입니다.
5. 최소 버전: v1은 checkBridgeMinVersion(), v2는 checkEnvLessBridgeMinVersion()으로 검사합니다. 운영팀은 전체 배포군에 강제 업그레이드를 걸 수 있습니다.

어느 하나라도 실패하면 onStateChange?.('failed', reason)가 호출되고 함수는 null을 반환합니다.

const RECONNECT_DELAY_MS = 2000
const MAX_RECONNECT_ATTEMPTS = 5
const MAX_SESSION_NOT_FOUND_RETRIES = 3  // 4001은 compaction 중 일시적일 수 있다

const PERMANENT_CLOSE_CODES = new Set([
  4003,  // unauthorized, 즉시 중단
])

private handleClose(closeCode: number): void {
  if (PERMANENT_CLOSE_CODES.has(closeCode)) {
    this.callbacks.onClose?.()
    return
  }
  if (closeCode === 4001) {
    // session not found, 선형 backoff로 최대 3번 재시도
    this.sessionNotFoundRetries++
    if (this.sessionNotFoundRetries > MAX_SESSION_NOT_FOUND_RETRIES) {
      this.callbacks.onClose?.()
      return
    }
    this.scheduleReconnect(RECONNECT_DELAY_MS * this.sessionNotFoundRetries, ...)
    return
  }
  if (previousState === 'connected' && this.reconnectAttempts < MAX_RECONNECT_ATTEMPTS) {
    this.reconnectAttempts++
    this.scheduleReconnect(RECONNECT_DELAY_MS, ...)
  }
}

Bun의 네이티브 WebSocket은 헤더로 인증을 전달합니다. Node는 같은 인증 헤더를 쓰는 ws 패키지를 사용합니다. subscribe 엔드포인트에는 연결 후 인증 메시지가 따로 필요하지 않습니다.

function convertSDKMessage(msg: SDKMessage, opts?: ConvertOptions): ConvertedMessage {
  switch (msg.type) {
    case 'assistant':
      return { type: 'message', message: convertAssistantMessage(msg) }

    case 'stream_event':
      return { type: 'stream_event', event: convertStreamEvent(msg) }

    case 'result':
      // 에러만 보여 준다. 성공 결과는 멀티턴 대화에서 잡음이 된다
      return msg.subtype !== 'success'
        ? { type: 'message', message: convertResultMessage(msg) }
        : { type: 'ignored' }

    case 'system':
      if (msg.subtype === 'init')            return { type: 'message', message: convertInitMessage(msg) }
      if (msg.subtype === 'status')          return { ... }  // 'compacting'을 배너로 표시
      if (msg.subtype === 'compact_boundary') return { ... }  // compaction 지점을 표시
      return { type: 'ignored' }

    case 'tool_progress':  return { type: 'message', message: convertToolProgressMessage(msg) }
    case 'user':          return { type: 'ignored' }  // REPL이 이미 로컬에 추가했다
    case 'auth_status':   return { type: 'ignored' }
    default:             return { type: 'ignored' }  // forward-compat, 알 수 없는 타입은 조용히 버림
  }
}

const DEFAULT_BACKOFF: BackoffConfig = {
  connInitialMs:    2_000,
  connCapMs:      120_000,  // 2분
  connGiveUpMs:   600_000,  // 10분
  generalInitialMs:   500,
  generalCapMs:    30_000,
  generalGiveUpMs:600_000,
}

// Sleep 감지: poll tick이 cap보다 2배 넘게 늦어지면
// 기기가 잠들었다고 보고 에러 예산을 초기화한 뒤 즉시 재연결한다
function pollSleepDetectionThresholdMs(b: BackoffConfig): number {
  return b.connCapMs * 2  // 240_000ms, 최대 backoff cap보다 큼
}

연결 에러와 일반 poll 에러는 서로 독립된 backoff 예산을 가집니다. 연결 에러(등록/WebSocket 실패)는 10분에서 포기합니다. 일반 에러(work poll의 HTTP 500)도 10분에서 포기합니다. 세션 생존 여부의 최종 권한은 서버에 있습니다.

// bridgeMain.ts, activeSessions 맵이 실행 중인 모든 세션을 추적한다
const activeSessions = new Map<string, SessionHandle>()
const sessionStartTimes = new Map<string, number>()
const sessionIngressTokens = new Map<string, string>()
const sessionTimers = new Map<string, ReturnType<typeof setTimeout>>()

// 세션별 타임아웃 감시기 (기본값 24시간)
const DEFAULT_SESSION_TIMEOUT_MS = 24 * 60 * 60 * 1000

// SessionHandle은 kill()/forceKill(), writeStdin(), activities ring buffer를 노출한다
type SessionHandle = {
  sessionId: string
  done: Promise<SessionDoneStatus>
  kill(): void
  forceKill(): void
  activities: SessionActivity[]      // ring buffer (최근 10개)
  currentActivity: SessionActivity | null
  accessToken: string
  lastStderr: string[]              // ring buffer (최근 10줄)
  writeStdin(data: string): void
  updateAccessToken(token: string): void
}

활성 work item은 GrowthBook에서 설정한 간격으로 heartbeat를 보냅니다. heartbeat는 OAuth가 아니라 session ingress JWT를 사용하며, 이는 SessionIngressAuth를 통한 가벼운 DB 없는 JWT 검증입니다. 401/403이 나와 JWT가 만료되면 브리지는 reconnectSession을 호출해 작업을 다시 큐에 넣고, 다음 poll이 fresh credential을 전달하게 만듭니다.

async function heartbeatActiveWorkItems() {
  for (const [sessionId] of activeSessions) {
    const ingressToken = sessionIngressTokens.get(sessionId)
    try {
      await api.heartbeatWork(environmentId, workId, ingressToken)
    } catch (err) {
      if (err.status === 401 || err.status === 403) {
        // JWT 만료, 다음 poll이 fresh token을 주도록 다시 디스패치한다
        await api.reconnectSession(environmentId, sessionId)
      }
    }
  }
}

선제 토큰 갱신 스케줄러도 만료 5분 전에 동작합니다. v1 세션은 새 OAuth 토큰을 직접 받고, v2 세션은 CCR worker 엔드포인트가 OAuth 토큰을 거부하므로 reconnectSession 경로를 거칩니다.