Bridge 与 Remote

当环境 API 交付工作时，它会附加一个不透明的 secret 字段是一个 base64url 编码的 JSON blob。 decodeWorkSecret() in bridge/workSecret.ts 解压它：

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 是授权工作层操作的短暂 JWT。 OAuth 令牌不能在其位置使用 - CCR 验证 JWT session_id and role=worker 直接索赔。

v1/v2 分支已解决 bridge/initReplBridge.ts 所有身份验证检查通过后：

// 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, ... })

有独立版本楼层： tengu_bridge_min_version 门 v1； tengu_bridge_repl_v2_config.min_version 盖茨 v2. 两者都是 GrowthBook 动态配置 - 操作可以通过将下限降低到 0.0.0.

CCR v2 兼容层造成分裂：基础设施端点分发 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() 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 终止开关通过注入 setCseShimGate() 以避免将 GrowthBook 导入 Agent SDK 捆绑包。

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

v2 传输是不对称的： reads 过来SSE（服务器发送事件）； writes 经过 CCRClient 哪个帖子到 /worker/events via SerialBatchEventUploader。 这种分割是有意为之的——入站 SSE 流可以暂停，而 CCRClient 的心跳和写入路径保持活动状态。

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), ... }
}

当桥接会话启动时，它会 POST 刷新历史对话。 在该刷新期间到达的任何新消息都将与服务器上的历史记录交错。 FlushGate 将它们排队直到刷新完成：

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() 当传输被替换时被调用（例如，环境丢失后重新连接）——项目被保留用于新的传输 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 协调本地 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 {
      // 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 }),
        },
      },
    })
  }
}

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,
      }],
      // 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
}

之前五项检查必须全部通过 initReplBridge proceeds:

1. 运行时门: isBridgeEnabledBlocking() — 需要 tengu_ccr_bridge GrowthBook 标志和 claude.ai OAuth 订阅（无 Bedrock/Vertex/API 密钥身份验证）。
2. OAuth 存在令牌: getBridgeAccessToken() 必须返回一个值。
3. 组织方针: isPolicyAllowed('allow_remote_control') — 企业管理员可以为所有组织成员禁用 RC。
4. 代币新鲜度：主动刷新 + 如果过期且不可刷新则跳过（避免保证 401 循环）。
5. 最低版本: checkBridgeMinVersion() 对于 v1， checkEnvLessBridgeMinVersion() 对于 v2 — 操作员可以强制整个机队进行升级。

如果任何一扇门出现故障， onStateChange?.('failed', reason) 被调用并且函数返回 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 的本机 WebSocket 通过标头传递身份验证； 节点使用 ws 具有相同身份验证标头的包（订阅端点不需要连接后身份验证消息）。

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
}

连接错误和一般轮询错误具有独立的退避预算。 连接错误（注册/WebSocket 失败）在 10 分钟后放弃。 一般错误（工作轮询中的 HTTP 500）也会在 10 分钟后放弃——服务器是会话活跃度的权威。

// 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
}

活动工作项按照 GrowthBook 配置的时间间隔进行检测。 心跳使用会话入口 JWT（不是 OAuth） SessionIngressAuth — 轻量级的无数据库 JWT 验证。 在 401/403（JWT 过期）上，网桥调用 reconnectSession 重新排队工作，以便下一次轮询提供新的凭据：

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)
      }
    }
  }
}

主动令牌刷新调度程序也会在到期前 5 分钟触发。 v1 会话直接接收新的 OAuth 令牌； v2 会话经过 reconnectSession 因为 CCR 工作端点拒绝 OAuth 令牌。