Bridge & Remote Control

When the Environments API delivers work, it attaches an opaque secret field that is a base64url-encoded JSON blob. decodeWorkSecret() in bridge/workSecret.ts unpacks it:

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

type WorkSecret = {
  version: 1
  session_ingress_token: string   // JWT for CCR worker endpoints
  api_base_url: string
  sources: Array<{ type: string; git_info?: ... }>
  auth: Array<{ type: string; token: string }>
  use_code_sessions?: boolean    // server-driven CCR v2 selector
}

The session_ingress_token is the short-lived JWT that authorises worker-tier operations. OAuth tokens cannot be used in their place — CCR validates the JWT's session_id and role=worker claims directly.

The v1/v2 branch is resolved in bridge/initReplBridge.ts after all auth checks pass:

// tengu_bridge_repl_v2 enables env-less (v2) path for the REPL.
// perpetual=true falls back to v1 — bridge-pointer not yet wired to 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 path: env-based register/poll/ack/heartbeat
return initBridgeCore({ dir, machineName, branch, gitRepoUrl, ... })

There are independent version floors: tengu_bridge_min_version gates v1; tengu_bridge_repl_v2_config.min_version gates v2. Both are GrowthBook dynamic configs — ops can roll back by lowering the floor to 0.0.0.

The CCR v2 compat layer creates a split: infrastructure endpoints hand out cse_* IDs, while the claude.ai frontend routes on session_*. Both are the same UUID with a different prefix.

// 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() ignores prefix so the poll loop doesn't
// reject its own session as "foreign" when the compat gate is on
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
}

The isCseShimEnabled kill switch is injected via setCseShimGate() to avoid importing GrowthBook into the Agent SDK bundle.

createV1ReplTransport() is a thin pass-through that wraps HybridTransport. HybridTransport opens a WebSocket to Session-Ingress for inbound messages and uses HTTP POST for outbound. v1 never uses SSE sequence numbers — the server-side cursor handles replay.

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 replay != SSE seq nums
    reportState: () => {},              // no-op
    reportMetadata: () => {},
    reportDelivery: () => {},
    flush: () => Promise.resolve(),     // POSTs are awaited per-write
    // ... other pass-throughs
  }
}

The v2 transport is asymmetric: reads come over SSE (Server-Sent Events); writes go through CCRClient which posts to /worker/events via SerialBatchEventUploader. This split is intentional — the inbound SSE stream can pause while CCRClient's heartbeat and write path stay alive.

export async function createV2ReplTransport(opts: {
  sessionUrl: string
  ingressToken: string
  sessionId: string
  initialSequenceNum?: number  // resume from this SSE seq on reconnect
  epoch?: number               // skip registerWorker if provided by /bridge
  getAuthToken?: () => string | undefined  // multi-session safe
  outboundOnly?: boolean       // skip SSE read (mirror mode)
}): Promise<ReplBridgeTransport> {

  // registerWorker returns worker_epoch — required for CCRClient
  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 from server: our epoch was superseded by another worker
      ccr.close(); sse.close()
      onCloseCb?.(4090)  // poll-loop recovery code
      throw new Error('epoch superseded')
    },
  })
  // ACK 'processed' immediately alongside 'received' to prevent
  // phantom prompt floods on daemon restart (CC-1263)
  sse.setOnEvent(event => {
    ccr.reportDelivery(event.event_id, 'received')
    ccr.reportDelivery(event.event_id, 'processed')
  })
  return { write: msg => ccr.writeEvent(msg), ... }
}

When a bridge session starts, it POST-flushes the historical conversation. Any new messages that arrive during that flush would interleave with history on the server. FlushGate queues them until flush completes:

class FlushGate<T> {
  start(): void            // mark flush in-progress, enqueue() starts queuing
  end(): T[]               // flush done; return queued items for draining
  enqueue(...items: T[]): boolean  // true if active (queued), false if pass-through
  drop(): number           // discard queue (transport closed permanently)
  deactivate(): void       // transport swapped — new one will drain
}

deactivate() is called when the transport is replaced (e.g., reconnect after env loss) — items are preserved for the new transport's end() call.

// 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 coordinates the client side for CCR-hosted sessions that the local CLI is viewing. It receives permission requests from CCR via WebSocket and holds them until the user responds:

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 {
      // Unsupported subtype: send error response immediately so server doesn't hang
      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 }),
        },
      },
    })
  }
}

The REPL's permission UI expects a real AssistantMessage with the tool-use block. When a permission request comes from a remote CCR container, no such message exists locally. remote/remotePermissionBridge.ts fabricates one:

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,
      }],
      // zero-usage stub fields ...
    } as AssistantMessage['message'],
    requestId: undefined,
    timestamp: new Date().toISOString(),
  }
}

// For tools not known locally (e.g. MCP tools on the remote container):
export function createToolStub(toolName: string): Tool {
  return {
    name: toolName,
    isEnabled: () => true,
    needsPermissions: () => true,
    // ... renders first 3 input key:value pairs for display
    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)  // already connected → show dialog
      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' })
    })()
  }, [])  // fires once on mount
}

Five checks must all pass before initReplBridge proceeds:

1. Runtime gate: isBridgeEnabledBlocking() — requires tengu_ccr_bridge GrowthBook flag AND a claude.ai OAuth subscription (no Bedrock/Vertex/API-key auth).
2. OAuth token present: getBridgeAccessToken() must return a value.
3. Organization policy: isPolicyAllowed('allow_remote_control') — enterprise admins can disable RC for all org members.
4. Token freshness: proactive refresh + skip if expired-and-unrefreshable (avoids guaranteed 401 loops).
5. Min version: checkBridgeMinVersion() for v1, checkEnvLessBridgeMinVersion() for v2 — ops can force upgrades fleet-wide.

If any gate fails, onStateChange?.('failed', reason) is called and the function returns null.

const RECONNECT_DELAY_MS = 2000
const MAX_RECONNECT_ATTEMPTS = 5
const MAX_SESSION_NOT_FOUND_RETRIES = 3  // 4001 can be transient during compaction

const PERMANENT_CLOSE_CODES = new Set([
  4003,  // unauthorized — stop immediately
])

private handleClose(closeCode: number): void {
  if (PERMANENT_CLOSE_CODES.has(closeCode)) {
    this.callbacks.onClose?.()
    return
  }
  if (closeCode === 4001) {
    // session not found — retry up to 3 times with linear backoff
    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's native WebSocket passes auth via headers; Node uses the ws package with the same auth headers (no post-connect auth message needed for the subscribe endpoint).

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':
      // Only show errors — success results are noise in multi-turn
      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' → banner
      if (msg.subtype === 'compact_boundary') return { ... }  // marks compaction point
      return { type: 'ignored' }

    case 'tool_progress':  return { type: 'message', message: convertToolProgressMessage(msg) }
    case 'user':          return { type: 'ignored' }  // already added locally by REPL
    case 'auth_status':   return { type: 'ignored' }
    default:             return { type: 'ignored' }  // forward-compat: unknown types silently dropped
  }
}

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

// Sleep detection: if a poll tick is delayed by >2× the cap,
// the machine probably slept — reset error budget and reconnect immediately
function pollSleepDetectionThresholdMs(b: BackoffConfig): number {
  return b.connCapMs * 2  // 240_000ms — above max backoff cap
}

Connection errors and general poll errors have independent backoff budgets. Connection errors (registration/WebSocket failures) give up at 10 minutes. General errors (HTTP 500 on work poll) also give up at 10 minutes — the server is the authority on session liveness.

// bridgeMain.ts — activeSessions map tracks all running sessions
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>>()

// Per-session timeout watchdog (default 24h)
const DEFAULT_SESSION_TIMEOUT_MS = 24 * 60 * 60 * 1000

// SessionHandle exposes kill()/forceKill(), writeStdin(), activities ring buffer
type SessionHandle = {
  sessionId: string
  done: Promise<SessionDoneStatus>
  kill(): void
  forceKill(): void
  activities: SessionActivity[]      // ring buffer (last 10)
  currentActivity: SessionActivity | null
  accessToken: string
  lastStderr: string[]              // ring buffer (last 10 lines)
  writeStdin(data: string): void
  updateAccessToken(token: string): void
}

Active work items are heartbeated at a GrowthBook-configured interval. Heartbeats use the session ingress JWT (not OAuth) via SessionIngressAuth — a lightweight DB-free JWT validation. On 401/403 (JWT expired), the bridge calls reconnectSession to re-queue the work so the next poll delivers fresh credentials:

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 expired — re-dispatch so next poll delivers fresh token
        await api.reconnectSession(environmentId, sessionId)
      }
    }
  }
}

A proactive token refresh scheduler also fires 5 minutes before expiry. v1 sessions receive the new OAuth token directly; v2 sessions go through reconnectSession because CCR worker endpoints reject OAuth tokens.