🎯 Use case
This file lives under “bridge/”, which covers the bridge between the UI/shell and the agent (IPC, REPL hooks, permissions, session glue). On the API surface it exposes ReplBridgeHandle, BridgeState, BridgeCoreParams, BridgeCoreHandle, and initBridgeCore (and more) — mainly types, interfaces, or factory objects. Dependencies touch crypto. It composes internal code from bridgeApi, types, utils, services, and bridgeMessaging (relative imports).
Generated from folder role, exports, dependency roots, and inline comments — not hand-reviewed for every path.
🧠 Inline summary
// biome-ignore-all assist/source/organizeImports: ANT-ONLY import markers must not be reordered import { randomUUID } from 'crypto' import { createBridgeApiClient, BridgeFatalError,
📤 Exports (heuristic)
ReplBridgeHandleBridgeStateBridgeCoreParamsBridgeCoreHandleinitBridgeCorestartWorkPollLoopPOLL_ERROR_INITIAL_DELAY_MSPOLL_ERROR_MAX_DELAY_MSPOLL_ERROR_GIVE_UP_MS
📚 External import roots
Package roots from from "…" (relative paths omitted).
crypto
🖥️ Source preview
// biome-ignore-all assist/source/organizeImports: ANT-ONLY import markers must not be reordered
import { randomUUID } from 'crypto'
import {
createBridgeApiClient,
BridgeFatalError,
isExpiredErrorType,
isSuppressible403,
} from './bridgeApi.js'
import type { BridgeConfig, BridgeApiClient } from './types.js'
import { logForDebugging } from '../utils/debug.js'
import { logForDiagnosticsNoPII } from '../utils/diagLogs.js'
import {
type AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
logEvent,
} from '../services/analytics/index.js'
import { registerCleanup } from '../utils/cleanupRegistry.js'
import {
handleIngressMessage,
handleServerControlRequest,
makeResultMessage,
isEligibleBridgeMessage,
extractTitleText,
BoundedUUIDSet,
} from './bridgeMessaging.js'
import {
decodeWorkSecret,
buildSdkUrl,
buildCCRv2SdkUrl,
sameSessionId,
} from './workSecret.js'
import { toCompatSessionId, toInfraSessionId } from './sessionIdCompat.js'
import { updateSessionBridgeId } from '../utils/concurrentSessions.js'
import { getTrustedDeviceToken } from './trustedDevice.js'
import { HybridTransport } from '../cli/transports/HybridTransport.js'
import {
type ReplBridgeTransport,
createV1ReplTransport,
createV2ReplTransport,
} from './replBridgeTransport.js'
import { updateSessionIngressAuthToken } from '../utils/sessionIngressAuth.js'
import { isEnvTruthy, isInProtectedNamespace } from '../utils/envUtils.js'
import { validateBridgeId } from './bridgeApi.js'
import {
describeAxiosError,
extractHttpStatus,
logBridgeSkip,
} from './debugUtils.js'
import type { Message } from '../types/message.js'
import type { SDKMessage } from '../entrypoints/agentSdkTypes.js'
import type { PermissionMode } from '../utils/permissions/PermissionMode.js'
import type {
SDKControlRequest,
SDKControlResponse,
} from '../entrypoints/sdk/controlTypes.js'
import { createCapacityWake, type CapacitySignal } from './capacityWake.js'
import { FlushGate } from './flushGate.js'
import {
DEFAULT_POLL_CONFIG,
type PollIntervalConfig,
} from './pollConfigDefaults.js'
import { errorMessage } from '../utils/errors.js'
import { sleep } from '../utils/sleep.js'
import {
wrapApiForFaultInjection,
registerBridgeDebugHandle,
clearBridgeDebugHandle,
injectBridgeFault,
} from './bridgeDebug.js'
export type ReplBridgeHandle = {
bridgeSessionId: string
environmentId: string
sessionIngressUrl: string
writeMessages(messages: Message[]): void
writeSdkMessages(messages: SDKMessage[]): void
sendControlRequest(request: SDKControlRequest): void
sendControlResponse(response: SDKControlResponse): void
sendControlCancelRequest(requestId: string): void
sendResult(): void
teardown(): Promise<void>
}
export type BridgeState = 'ready' | 'connected' | 'reconnecting' | 'failed'
/**
* Explicit-param input to initBridgeCore. Everything initReplBridge reads
* from bootstrap state (cwd, session ID, git, OAuth) becomes a field here.
* A daemon caller (Agent SDK, PR 4) that never runs main.tsx fills these
* in itself.
*/
export type BridgeCoreParams = {
dir: string
machineName: string
branch: string
gitRepoUrl: string | null
title: string
baseUrl: string
sessionIngressUrl: string
/**
* Opaque string sent as metadata.worker_type. Use BridgeWorkerType for
* the two CLI-originated values; daemon callers may send any string the
* backend recognizes (it's just a filter key on the web side).
*/
workerType: string
getAccessToken: () => string | undefined
/**
* POST /v1/sessions. Injected because `createSession.ts` lazy-loads
* `auth.ts`/`model.ts`/`oauth/client.ts` and `bun --outfile` inlines
* dynamic imports — the lazy-load doesn't help, the whole REPL tree ends
* up in the Agent SDK bundle.
*
* REPL wrapper passes `createBridgeSession` from `createSession.ts`.
* Daemon wrapper passes `createBridgeSessionLean` from `sessionApi.ts`
* (HTTP-only, orgUUID+model supplied by the daemon caller).
*
* Receives `gitRepoUrl`+`branch` so the REPL wrapper can build the git
* source/outcome for claude.ai's session card. Daemon ignores them.
*/
createSession: (opts: {
environmentId: string
title: string
gitRepoUrl: string | null
branch: string
signal: AbortSignal
}) => Promise<string | null>
/**
* POST /v1/sessions/{id}/archive. Same injection rationale. Best-effort;
* the callback MUST NOT throw.
*/
archiveSession: (sessionId: string) => Promise<void>
/**
* Invoked on reconnect-after-env-lost to refresh the title. REPL wrapper
* reads session storage (picks up /rename); daemon returns the static
* title. Defaults to () => title.
*/
getCurrentTitle?: () => string
/**
* Converts internal Message[] → SDKMessage[] for writeMessages() and the
* initial-flush/drain paths. REPL wrapper passes the real toSDKMessages
* from utils/messages/mappers.ts. Daemon callers that only use
* writeSdkMessages() and pass no initialMessages can omit this — those
* code paths are unreachable.
*
* Injected rather than imported because mappers.ts transitively pulls in
* src/commands.ts via messages.ts → api.ts → prompts.ts, dragging the
* entire command registry + React tree into the Agent SDK bundle.
*/
toSDKMessages?: (messages: Message[]) => SDKMessage[]
/**
* OAuth 401 refresh handler passed to createBridgeApiClient. REPL wrapper
* passes handleOAuth401Error; daemon passes its AuthManager's handler.
* Injected because utils/auth.ts transitively pulls in the command
* registry via config.ts → file.ts → permissions/filesystem.ts →
* sessionStorage.ts → commands.ts.
*/
onAuth401?: (staleAccessToken: string) => Promise<boolean>
/**
* Poll interval config getter for the work-poll heartbeat loop. REPL
* wrapper passes the GrowthBook-backed getPollIntervalConfig (allows ops
* to live-tune poll rates fleet-wide). Daemon passes a static config
* with a 60s heartbeat (5× headroom under the 300s work-lease TTL).
* Injected because growthbook.ts transitively pulls in the command
* registry via the same config.ts chain.
*/
getPollIntervalConfig?: () => PollIntervalConfig
/**
* Max initial messages to replay on connect. REPL wrapper reads from the
* tengu_bridge_initial_history_cap GrowthBook flag. Daemon passes no
* initialMessages so this is never read. Default 200 matches the flag
* default.
*/
initialHistoryCap?: number
// Same REPL-flush machinery as InitBridgeOptions — daemon omits these.
initialMessages?: Message[]
previouslyFlushedUUIDs?: Set<string>
onInboundMessage?: (msg: SDKMessage) => void
onPermissionResponse?: (response: SDKControlResponse) => void
onInterrupt?: () => void
onSetModel?: (model: string | undefined) => void
onSetMaxThinkingTokens?: (maxTokens: number | null) => void
/**
* Returns a policy verdict so this module can emit an error control_response
* without importing the policy checks itself (bootstrap-isolation constraint).
* The callback must guard `auto` (isAutoModeGateEnabled) and
* `bypassPermissions` (isBypassPermissionsModeDisabled AND
* isBypassPermissionsModeAvailable) BEFORE calling transitionPermissionMode —
* that function's internal auto-gate check is a defensive throw, not a
* graceful guard, and its side-effect order is setAutoModeActive(true) then
* throw, which corrupts the 3-way invariant documented in src/CLAUDE.md if
* the callback lets the throw escape here.
*/
onSetPermissionMode?: (
mode: PermissionMode,
) => { ok: true } | { ok: false; error: string }
onStateChange?: (state: BridgeState, detail?: string) => void
/**
* Fires on each real user message to flow through writeMessages() until
* the callback returns true (done). Mirrors remoteBridgeCore.ts's
* onUserMessage so the REPL bridge can derive a session title from early
* prompts when none was set at init time (e.g. user runs /remote-control
* on an empty conversation, then types). Tool-result wrappers, meta
* messages, and display-tag-only messages are skipped. Receives
* currentSessionId so the wrapper can PATCH the title without a closure
* dance to reach the not-yet-returned handle. The caller owns the
* derive-at-count-1-and-3 policy; the transport just keeps calling until
* told to stop. Not fired for the writeSdkMessages daemon path (daemon
* sets its own title at init). Distinct from SessionSpawnOpts's
* onFirstUserMessage (spawn-bridge, PR #21250), which stays fire-once.
*/
onUserMessage?: (text: string, sessionId: string) => boolean
/** See InitBridgeOptions.perpetual. */
perpetual?: boolean
/**
* Seeds lastTransportSequenceNum — the SSE event-stream high-water mark
* that's carried across transport swaps within one process. Daemon callers
* pass the value they persisted at shutdown so the FIRST SSE connect of a
* fresh process sends from_sequence_num and the server doesn't replay full
* history. REPL callers omit (fresh session each run → 0 is correct).
*/
initialSSESequenceNum?: number
}
/**
* Superset of ReplBridgeHandle. Adds getSSESequenceNum for daemon callers
* that persist the SSE seq-num across process restarts and pass it back as
* initialSSESequenceNum on the next start.
*/
export type BridgeCoreHandle = ReplBridgeHandle & {
/**
* Current SSE sequence-number high-water mark. Updates as transports
* swap. Daemon callers persist this on shutdown and pass it back as
* initialSSESequenceNum on next start.
*/
getSSESequenceNum(): number
}
/**
* Poll error recovery constants. When the work poll starts failing (e.g.
* server 500s), we use exponential backoff and give up after this timeout.
* This is deliberately long — the server is the authority on when a session
* is truly dead. As long as the server accepts our poll, we keep waiting
* for it to re-dispatch the work item.
*/
const POLL_ERROR_INITIAL_DELAY_MS = 2_000
const POLL_ERROR_MAX_DELAY_MS = 60_000
const POLL_ERROR_GIVE_UP_MS = 15 * 60 * 1000
// Monotonically increasing counter for distinguishing init calls in logs
let initSequence = 0
/**
* Bootstrap-free core: env registration → session creation → poll loop →
* ingress WS → teardown. Reads nothing from bootstrap/state or
* sessionStorage — all context comes from params. Caller (initReplBridge
* below, or a daemon in PR 4) has already passed entitlement gates and
* gathered git/auth/title.
*
* Returns null on registration or session-creation failure.
*/
export async function initBridgeCore(
params: BridgeCoreParams,
): Promise<BridgeCoreHandle | null> {
const {
dir,
machineName,
branch,
gitRepoUrl,
title,
baseUrl,
sessionIngressUrl,
workerType,
getAccessToken,
createSession,
archiveSession,
getCurrentTitle = () => title,
toSDKMessages = () => {
throw new Error(
'BridgeCoreParams.toSDKMessages not provided. Pass it if you use writeMessages() or initialMessages — daemon callers that only use writeSdkMessages() never hit this path.',
)
},
onAuth401,
getPollIntervalConfig = () => DEFAULT_POLL_CONFIG,
initialHistoryCap = 200,
initialMessages,
previouslyFlushedUUIDs,
onInboundMessage,
onPermissionResponse,
onInterrupt,
onSetModel,
onSetMaxThinkingTokens,
onSetPermissionMode,
onStateChange,
onUserMessage,
perpetual,
initialSSESequenceNum = 0,
} = params
const seq = ++initSequence
// bridgePointer import hoisted: perpetual mode reads it before register;
// non-perpetual writes it after session create; both use clear at teardown.
const { writeBridgePointer, clearBridgePointer, readBridgePointer } =
await import('./bridgePointer.js')
// Perpetual mode: read the crash-recovery pointer and treat it as prior
// state. The pointer is written unconditionally after session create
// (crash-recovery for all sessions); perpetual mode just skips the
// teardown clear so it survives clean exits too. Only reuse 'repl'
// pointers — a crashed standalone bridge (`claude remote-control`)
// writes source:'standalone' with a different workerType.
const rawPrior = perpetual ? await readBridgePointer(dir) : null
const prior = rawPrior?.source === 'repl' ? rawPrior : null
logForDebugging(
`[bridge:repl] initBridgeCore #${seq} starting (initialMessages=${initialMessages?.length ?? 0}${prior ? ` perpetual prior=env:${prior.environmentId}` : ''})`,
)
// 5. Register bridge environment
const rawApi = createBridgeApiClient({
baseUrl,
getAccessToken,
runnerVersion: MACRO.VERSION,
onDebug: logForDebugging,
onAuth401,
getTrustedDeviceToken,
})
// Ant-only: interpose so /bridge-kick can inject poll/register/heartbeat
// failures. Zero cost in external builds (rawApi passes through unchanged).
const api =
process.env.USER_TYPE === 'ant' ? wrapApiForFaultInjection(rawApi) : rawApi
const bridgeConfig: BridgeConfig = {
dir,
machineName,
branch,
gitRepoUrl,
maxSessions: 1,
spawnMode: 'single-session',
verbose: false,
sandbox: false,
bridgeId: randomUUID(),
workerType,
environmentId: randomUUID(),
reuseEnvironmentId: prior?.environmentId,
apiBaseUrl: baseUrl,
sessionIngressUrl,
}
let environmentId: string
let environmentSecret: string
try {
const reg = await api.registerBridgeEnvironment(bridgeConfig)
environmentId = reg.environment_id
environmentSecret = reg.environment_secret
} catch (err) {
logBridgeSkip(
'registration_failed',
`[bridge:repl] Environment registration failed: ${errorMessage(err)}`,
)
// Stale pointer may be the cause (expired/deleted env) — clear it so
// the next start doesn't retry the same dead ID.
if (prior) {
await clearBridgePointer(dir)
}
onStateChange?.('failed', errorMessage(err))
return null
}
logForDebugging(`[bridge:repl] Environment registered: ${environmentId}`)
logForDiagnosticsNoPII('info', 'bridge_repl_env_registered')
logEvent('tengu_bridge_repl_env_registered', {})
/**
* Reconnect-in-place: if the just-registered environmentId matches what
* was requested, call reconnectSession to force-stop stale workers and
* re-queue the session. Used at init (perpetual mode — env is alive but
* idle after clean teardown) and in doReconnect() Strategy 1 (env lost
* then resurrected). Returns true on success; caller falls back to
* fresh session creation on false.
*/
async function tryReconnectInPlace(
requestedEnvId: string,
sessionId: string,
): Promise<boolean> {
if (environmentId !== requestedEnvId) {
logForDebugging(
`[bridge:repl] Env mismatch (requested ${requestedEnvId}, got ${environmentId}) — cannot reconnect in place`,
)
return false
}
// The pointer stores what createBridgeSession returned (session_*,
// compat/convert.go:41). /bridge/reconnect is an environments-layer
// endpoint — once the server's ccr_v2_compat_enabled gate is on it
// looks sessions up by their infra tag (cse_*) and returns "Session
// not found" for the session_* costume. We don't know the gate state
// pre-poll, so try both; the re-tag is a no-op if the ID is already
// cse_* (doReconnect Strategy 1 path — currentSessionId never mutates
// to cse_* but future-proof the check).
const infraId = toInfraSessionId(sessionId)
const candidates =
infraId === sessionId ? [sessionId] : [sessionId, infraId]
for (const id of candidates) {
try {
await api.reconnectSession(environmentId, id)
logForDebugging(
`[bridge:repl] Reconnected session ${id} in place on env ${environmentId}`,
)
return true
} catch (err) {
logForDebugging(
`[bridge:repl] reconnectSession(${id}) failed: ${errorMessage(err)}`,
)
}
}
logForDebugging(
'[bridge:repl] reconnectSession exhausted — falling through to fresh session',
)
return false
}
// Perpetual init: env is alive but has no queued work after clean
// teardown. reconnectSession re-queues it. doReconnect() has the same
// call but only fires on poll 404 (env dead);
// here the env is alive but idle.
const reusedPriorSession = prior
? await tryReconnectInPlace(prior.environmentId, prior.sessionId)
: false
if (prior && !reusedPriorSession) {
await clearBridgePointer(dir)
}
// 6. Create session on the bridge. Initial messages are NOT included as
// session creation events because those use STREAM_ONLY persistence and
// are published before the CCR UI subscribes, so they get lost. Instead,
// initial messages are flushed via the ingress WebSocket once it connects.
// Mutable session ID — updated when the environment+session pair is
// re-created after a connection loss.
let currentSessionId: string
if (reusedPriorSession && prior) {
currentSessionId = prior.sessionId
logForDebugging(
`[bridge:repl] Perpetual session reused: ${currentSessionId}`,
)
// Server already has all initialMessages from the prior CLI run. Mark
// them as previously-flushed so the initial flush filter excludes them
// (previouslyFlushedUUIDs is a fresh Set on every CLI start). Duplicate
// UUIDs cause the server to kill the WebSocket.
if (initialMessages && previouslyFlushedUUIDs) {
for (const msg of initialMessages) {
previouslyFlushedUUIDs.add(msg.uuid)
}
}
} else {
const createdSessionId = await createSession({
environmentId,
title,
gitRepoUrl,
branch,
signal: AbortSignal.timeout(15_000),
})
if (!createdSessionId) {
logForDebugging(
'[bridge:repl] Session creation failed, deregistering environment',
)
logEvent('tengu_bridge_repl_session_failed', {})
await api.deregisterEnvironment(environmentId).catch(() => {})
onStateChange?.('failed', 'Session creation failed')
return null
}
currentSessionId = createdSessionId
logForDebugging(`[bridge:repl] Session created: ${currentSessionId}`)
}
// Crash-recovery pointer: written now so a kill -9 at any point after
// this leaves a recoverable trail. Cleared in teardown (non-perpetual)
// or left alone (perpetual mode — pointer survives clean exit too).
// `claude remote-control --continue` from the same directory will detect
// it and offer to resume.
await writeBridgePointer(dir, {
sessionId: currentSessionId,
environmentId,
source: 'repl',
})
logForDiagnosticsNoPII('info', 'bridge_repl_session_created')
logEvent('tengu_bridge_repl_started', {
has_initial_messages: !!(initialMessages && initialMessages.length > 0),
inProtectedNamespace: isInProtectedNamespace(),
})
// UUIDs of initial messages. Used for dedup in writeMessages to avoid
// re-sending messages that were already flushed on WebSocket open.
const initialMessageUUIDs = new Set<string>()
if (initialMessages) {
for (const msg of initialMessages) {
initialMessageUUIDs.add(msg.uuid)
}
}
// Bounded ring buffer of UUIDs for messages we've already sent to the
// server via the ingress WebSocket. Serves two purposes:
// 1. Echo filtering — ignore our own messages bouncing back on the WS.
// 2. Secondary dedup in writeMessages — catch race conditions where
// the hook's index-based tracking isn't sufficient.
//
// Seeded with initialMessageUUIDs so that when the server echoes back
// the initial conversation context over the ingress WebSocket, those
// messages are recognized as echoes and not re-injected into the REPL.
//
// Capacity of 2000 covers well over any realistic echo window (echoes
// arrive within milliseconds) and any messages that might be re-encountered
// after compaction. The hook's lastWrittenIndexRef is the primary dedup;
// this is a safety net.
const recentPostedUUIDs = new BoundedUUIDSet(2000)
for (const uuid of initialMessageUUIDs) {
recentPostedUUIDs.add(uuid)
}
// Bounded set of INBOUND prompt UUIDs we've already forwarded to the REPL.
// Defensive dedup for when the server re-delivers prompts (seq-num
// negotiation failure, server edge cases, transport swap races). The
// seq-num carryover below is the primary fix; this is the safety net.
const recentInboundUUIDs = new BoundedUUIDSet(2000)
// 7. Start poll loop for work items — this is what makes the session
// "live" on claude.ai. When a user types there, the backend dispatches
// a work item to our environment. We poll for it, get the ingress token,
// and connect the ingress WebSocket.
//
// The poll loop keeps running: when work arrives it connects the ingress
// WebSocket, and if the WebSocket drops unexpectedly (code != 1000) it
// resumes polling to get a fresh ingress token and reconnect.
const pollController = new AbortController()
// Adapter over either HybridTransport (v1: WS reads + POST writes to
// Session-Ingress) or SSETransport+CCRClient (v2: SSE reads + POST
// writes to CCR /worker/*). The v1/v2 choice is made in onWorkReceived:
// server-driven via secret.use_code_sessions, with CLAUDE_BRIDGE_USE_CCR_V2
// as an ant-dev override.
let transport: ReplBridgeTransport | null = null
// Bumped on every onWorkReceived. Captured in createV2ReplTransport's .then()
// closure to detect stale resolutions: if two calls race while transport is
// null, both registerWorker() (bumping server epoch), and whichever resolves
// SECOND is the correct one — but the transport !== null check gets this
// backwards (first-to-resolve installs, second discards). The generation
// counter catches it independent of transport state.
let v2Generation = 0
// SSE sequence-number high-water mark carried across transport swaps.
// Without this, each new SSETransport starts at 0, sends no
// from_sequence_num / Last-Event-ID on its first connect, and the server
// replays the entire session event history — every prompt ever sent
// re-delivered as fresh inbound messages on every onWorkReceived.
//
// Seed only when we actually reconnected the prior session. If
// `reusedPriorSession` is false we fell through to `createSession()` —
// the caller's persisted seq-num belongs to a dead session and applying
// it to the fresh stream (starting at 1) silently drops events. Same
// hazard as doReconnect Strategy 2; same fix as the reset there.
let lastTransportSequenceNum = reusedPriorSession ? initialSSESequenceNum : 0
// Track the current work ID so teardown can call stopWork
let currentWorkId: string | null = null
// Session ingress JWT for the current work item — used for heartbeat auth.
let currentIngressToken: string | null = null
// Signal to wake the at-capacity sleep early when the transport is lost,
// so the poll loop immediately switches back to fast polling for new work.
const capacityWake = createCapacityWake(pollController.signal)
const wakePollLoop = capacityWake.wake
const capacitySignal = capacityWake.signal
// Gates message writes during the initial flush to prevent ordering
// races where new messages arrive at the server interleaved with history.
const flushGate = new FlushGate<Message>()
// Latch for onUserMessage — flips true when the callback returns true
// (policy says "done deriving"). If no callback, skip scanning entirely
// (daemon path — no title derivation needed).
let userMessageCallbackDone = !onUserMessage
// Shared counter for environment re-creations, used by both
// onEnvironmentLost and the abnormal-close handler.
const MAX_ENVIRONMENT_RECREATIONS = 3
let environmentRecreations = 0
let reconnectPromise: Promise<boolean> | null = null
/**
* Recover from onEnvironmentLost (poll returned 404 — env was reaped
* server-side). Tries two strategies in order:
*
* 1. Reconnect-in-place: idempotent re-register with reuseEnvironmentId
* → if the backend returns the same env ID, call reconnectSession()
* to re-queue the existing session. currentSessionId stays the same;
* the URL on the user's phone stays valid; previouslyFlushedUUIDs is
* preserved so history isn't re-sent.
*
* 2. Fresh session fallback: if the backend returns a different env ID
* (original TTL-expired, e.g. laptop slept >4h) or reconnectSession()
* throws, archive the old session and create a new one on the
* now-registered env. Old behavior before #20460 primitives landed.
*
* Uses a promise-based reentrancy guard so concurrent callers share the
* same reconnection attempt.
*/
async function reconnectEnvironmentWithSession(): Promise<boolean> {
if (reconnectPromise) {
return reconnectPromise
}
reconnectPromise = doReconnect()
try {
return await reconnectPromise
} finally {
reconnectPromise = null
}
}
async function doReconnect(): Promise<boolean> {
environmentRecreations++
// Invalidate any in-flight v2 handshake — the environment is being
// recreated, so a stale transport arriving post-reconnect would be
// pointed at a dead session.
v2Generation++
logForDebugging(
`[bridge:repl] Reconnecting after env lost (attempt ${environmentRecreations}/${MAX_ENVIRONMENT_RECREATIONS})`,
)
if (environmentRecreations > MAX_ENVIRONMENT_RECREATIONS) {
logForDebugging(
`[bridge:repl] Environment reconnect limit reached (${MAX_ENVIRONMENT_RECREATIONS}), giving up`,
)
return false
}
// Close the stale transport. Capture seq BEFORE close — if Strategy 1
// (tryReconnectInPlace) succeeds we keep the SAME session, and the
// next transport must resume where this one left off, not replay from
// the last transport-swap checkpoint.
if (transport) {
const seq = transport.getLastSequenceNum()
if (seq > lastTransportSequenceNum) {
lastTransportSequenceNum = seq
}
transport.close()
transport = null
}
// Transport is gone — wake the poll loop out of its at-capacity
// heartbeat sleep so it can fast-poll for re-dispatched work.
wakePollLoop()
// Reset flush gate so writeMessages() hits the !transport guard
// instead of silently queuing into a dead buffer.
flushGate.drop()
// Release the current work item (force=false — we may want the session
// back). Best-effort: the env is probably gone, so this likely 404s.
if (currentWorkId) {
const workIdBeingCleared = currentWorkId
await api
.stopWork(environmentId, workIdBeingCleared, false)
.catch(() => {})
// When doReconnect runs concurrently with the poll loop (ws_closed
// handler case — void-called, unlike the awaited onEnvironmentLost
// path), onWorkReceived can fire during the stopWork await and set
// a fresh currentWorkId. If it did, the poll loop has already
// recovered on its own — defer to it rather than proceeding to
// archiveSession, which would destroy the session its new
// transport is connected to.
if (currentWorkId !== workIdBeingCleared) {
logForDebugging(
'[bridge:repl] Poll loop recovered during stopWork await — deferring to it',
)
environmentRecreations = 0
return true
}
currentWorkId = null
currentIngressToken = null
}
// Bail out if teardown started while we were awaiting
if (pollController.signal.aborted) {
logForDebugging('[bridge:repl] Reconnect aborted by teardown')
return false
}
// Strategy 1: idempotent re-register with the server-issued env ID.
// If the backend resurrects the same env (fresh secret), we can
// reconnect the existing session. If it hands back a different ID, the
// original env is truly gone and we fall through to a fresh session.
const requestedEnvId = environmentId
bridgeConfig.reuseEnvironmentId = requestedEnvId
try {
const reg = await api.registerBridgeEnvironment(bridgeConfig)
environmentId = reg.environment_id
environmentSecret = reg.environment_secret
} catch (err) {
bridgeConfig.reuseEnvironmentId = undefined
logForDebugging(
`[bridge:repl] Environment re-registration failed: ${errorMessage(err)}`,
)
return false
}
// Clear before any await — a stale value would poison the next fresh
// registration if doReconnect runs again.
bridgeConfig.reuseEnvironmentId = undefined
logForDebugging(
`[bridge:repl] Re-registered: requested=${requestedEnvId} got=${environmentId}`,
)
// Bail out if teardown started while we were registering
if (pollController.signal.aborted) {
logForDebugging(
'[bridge:repl] Reconnect aborted after env registration, cleaning up',
)
await api.deregisterEnvironment(environmentId).catch(() => {})
return false
}
// Same race as above, narrower window: poll loop may have set up a
// transport during the registerBridgeEnvironment await. Bail before
// tryReconnectInPlace/archiveSession kill it server-side.
if (transport !== null) {
logForDebugging(
'[bridge:repl] Poll loop recovered during registerBridgeEnvironment await — deferring to it',
)
environmentRecreations = 0
return true
}
// Strategy 1: same helper as perpetual init. currentSessionId stays
// the same on success; URL on mobile/web stays valid;
// previouslyFlushedUUIDs preserved (no re-flush).
if (await tryReconnectInPlace(requestedEnvId, currentSessionId)) {
logEvent('tengu_bridge_repl_reconnected_in_place', {})
environmentRecreations = 0
return true
}
// Env differs → TTL-expired/reaped; or reconnect failed.
// Don't deregister — we have a fresh secret for this env either way.
if (environmentId !== requestedEnvId) {
logEvent('tengu_bridge_repl_env_expired_fresh_session', {})
}
// Strategy 2: fresh session on the now-registered environment.
// Archive the old session first — it's orphaned (bound to a dead env,
// or reconnectSession rejected it). Don't deregister the env — we just
// got a fresh secret for it and are about to use it.
await archiveSession(currentSessionId)
// Bail out if teardown started while we were archiving
if (pollController.signal.aborted) {
logForDebugging(
'[bridge:repl] Reconnect aborted after archive, cleaning up',
)
await api.deregisterEnvironment(environmentId).catch(() => {})
return false
}
// Re-read the current title in case the user renamed the session.
// REPL wrapper reads session storage; daemon wrapper returns the
// original title (nothing to refresh).
const currentTitle = getCurrentTitle()
// Create a new session on the now-registered environment
const newSessionId = await createSession({
environmentId,
title: currentTitle,
gitRepoUrl,
branch,
signal: AbortSignal.timeout(15_000),
})
if (!newSessionId) {
logForDebugging(
'[bridge:repl] Session creation failed during reconnection',
)
return false
}
// Bail out if teardown started during session creation (up to 15s)
if (pollController.signal.aborted) {
logForDebugging(
'[bridge:repl] Reconnect aborted after session creation, cleaning up',
)
await archiveSession(newSessionId)
return false
}
currentSessionId = newSessionId
// Re-publish to the PID file so peer dedup (peerRegistry.ts) picks up the
// new ID — setReplBridgeHandle only fires at init/teardown, not reconnect.
void updateSessionBridgeId(toCompatSessionId(newSessionId)).catch(() => {})
// Reset per-session transport state IMMEDIATELY after the session swap,
// before any await. If this runs after `await writeBridgePointer` below,
// there's a window where handle.bridgeSessionId already returns session B
// but getSSESequenceNum() still returns session A's seq — a daemon
// persistState() in that window writes {bridgeSessionId: B, seq: OLD_A},
// which PASSES the session-ID validation check and defeats it entirely.
//
// The SSE seq-num is scoped to the session's event stream — carrying it
// over leaves the transport's lastSequenceNum stuck high (seq only
// advances when received > last), and its next internal reconnect would
// send from_sequence_num=OLD_SEQ against a stream starting at 1 → all
// events in the gap silently dropped. Inbound UUID dedup is also
// session-scoped.
lastTransportSequenceNum = 0
recentInboundUUIDs.clear()
// Title derivation is session-scoped too: if the user typed during the
// createSession await above, the callback fired against the OLD archived
// session ID (PATCH lost) and the new session got `currentTitle` captured
// BEFORE they typed. Reset so the next prompt can re-derive. Self-
// correcting: if the caller's policy is already done (explicit title or
// count ≥ 3), it returns true on the first post-reset call and re-latches.
userMessageCallbackDone = !onUserMessage
logForDebugging(`[bridge:repl] Re-created session: ${currentSessionId}`)
// Rewrite the crash-recovery pointer with the new IDs so a crash after
// this point resumes the right session. (The reconnect-in-place path
// above doesn't touch the pointer — same session, same env.)
await writeBridgePointer(dir, {
sessionId: currentSessionId,
environmentId,
source: 'repl',
})
// Clear flushed UUIDs so initial messages are re-sent to the new session.
// UUIDs are scoped per-session on the server, so re-flushing is safe.
previouslyFlushedUUIDs?.clear()
// Reset the counter so independent reconnections hours apart don't
// exhaust the limit — it guards against rapid consecutive failures,
// not lifetime total.
environmentRecreations = 0
return true
}
// Helper: get the current OAuth access token for session ingress auth.
// Unlike the JWT path, OAuth tokens are refreshed by the standard OAuth
// flow — no proactive scheduler needed.
function getOAuthToken(): string | undefined {
return getAccessToken()
}
// Drain any messages that were queued during the initial flush.
// Called after writeBatch completes (or fails) so queued messages
// are sent in order after the historical messages.
function drainFlushGate(): void {
const msgs = flushGate.end()
if (msgs.length === 0) return
if (!transport) {
logForDebugging(
`[bridge:repl] Cannot drain ${msgs.length} pending message(s): no transport`,
)
return
}
for (const msg of msgs) {
recentPostedUUIDs.add(msg.uuid)
}
const sdkMessages = toSDKMessages(msgs)
const events = sdkMessages.map(sdkMsg => ({
...sdkMsg,
session_id: currentSessionId,
}))
logForDebugging(
`[bridge:repl] Drained ${msgs.length} pending message(s) after flush`,
)
void transport.writeBatch(events)
}
// Teardown reference — set after definition below. All callers are async
// callbacks that run after assignment, so the reference is always valid.
let doTeardownImpl: (() => Promise<void>) | null = null
function triggerTeardown(): void {
void doTeardownImpl?.()
}
/**
* Body of the transport's setOnClose callback, hoisted to initBridgeCore
* scope so /bridge-kick can fire it directly. setOnClose wraps this with
* a stale-transport guard; debugFireClose calls it bare.
*
* With autoReconnect:true, this only fires on: clean close (1000),
* permanent server rejection (4001/1002/4003), or 10-min budget
* exhaustion. Transient drops are retried internally by the transport.
*/
function handleTransportPermanentClose(closeCode: number | undefined): void {
logForDebugging(
`[bridge:repl] Transport permanently closed: code=${closeCode}`,
)
logEvent('tengu_bridge_repl_ws_closed', {
code: closeCode,
})
// Capture SSE seq high-water mark before nulling. When called from
// setOnClose the guard guarantees transport !== null; when fired from
// /bridge-kick it may already be null (e.g. fired twice) — skip.
if (transport) {
const closedSeq = transport.getLastSequenceNum()
if (closedSeq > lastTransportSequenceNum) {
lastTransportSequenceNum = closedSeq
}
transport = null
}
// Transport is gone — wake the poll loop out of its at-capacity
// heartbeat sleep so it's fast-polling by the time the reconnect
// below completes and the server re-queues work.
wakePollLoop()
// Reset flush state so writeMessages() hits the !transport guard
// (with a warning log) instead of silently queuing into a buffer
// that will never be drained. Unlike onWorkReceived (which
// preserves pending messages for the new transport), onClose is
// a permanent close — no new transport will drain these.
const dropped = flushGate.drop()
if (dropped > 0) {
logForDebugging(
`[bridge:repl] Dropping ${dropped} pending message(s) on transport close (code=${closeCode})`,
{ level: 'warn' },
)
}
if (closeCode === 1000) {
// Clean close — session ended normally. Tear down the bridge.
onStateChange?.('failed', 'session ended')
pollController.abort()
triggerTeardown()
return
}
// Transport reconnect budget exhausted or permanent server
// rejection. By this point the env has usually been reaped
// server-side (BQ 2026-03-12: ~98% of ws_closed never recover
// via poll alone). stopWork(force=false) can't re-dispatch work
// from an archived env; reconnectEnvironmentWithSession can
// re-activate it via POST /bridge/reconnect, or fall through
// to a fresh session if the env is truly gone. The poll loop
// (already woken above) picks up the re-queued work once
// doReconnect completes.
onStateChange?.(
'reconnecting',
`Remote Control connection lost (code ${closeCode})`,
)
logForDebugging(
`[bridge:repl] Transport reconnect budget exhausted (code=${closeCode}), attempting env reconnect`,
)
void reconnectEnvironmentWithSession().then(success => {
if (success) return
// doReconnect has four abort-check return-false sites for
// teardown-in-progress. Don't pollute the BQ failure signal
// or double-teardown when the user just quit.
if (pollController.signal.aborted) return
// doReconnect returns false (never throws) on genuine failure.
// The dangerous case: registerBridgeEnvironment succeeded (so
// environmentId now points at a fresh valid env) but
// createSession failed — poll loop would poll a sessionless
// env getting null work with no errors, never hitting any
// give-up path. Tear down explicitly.
logForDebugging(
'[bridge:repl] reconnectEnvironmentWithSession resolved false — tearing down',
)
logEvent('tengu_bridge_repl_reconnect_failed', {
close_code: closeCode,
})
onStateChange?.('failed', 'reconnection failed')
triggerTeardown()
})
}
// Ant-only: SIGUSR2 → force doReconnect() for manual testing. Skips the
// ~30s poll wait — fire-and-observe in the debug log immediately.
// Windows has no USR signals; `process.on` would throw there.
let sigusr2Handler: (() => void) | undefined
if (process.env.USER_TYPE === 'ant' && process.platform !== 'win32') {
sigusr2Handler = () => {
logForDebugging(
'[bridge:repl] SIGUSR2 received — forcing doReconnect() for testing',
)
void reconnectEnvironmentWithSession()
}
process.on('SIGUSR2', sigusr2Handler)
}
// Ant-only: /bridge-kick fault injection. handleTransportPermanentClose
// is defined below and assigned into this slot so the slash command can
// invoke it directly — the real setOnClose callback is buried inside
// wireTransport which is itself inside onWorkReceived.
let debugFireClose: ((code: number) => void) | null = null
if (process.env.USER_TYPE === 'ant') {
registerBridgeDebugHandle({
fireClose: code => {
if (!debugFireClose) {
logForDebugging('[bridge:debug] fireClose: no transport wired yet')
return
}
logForDebugging(`[bridge:debug] fireClose(${code}) — injecting`)
debugFireClose(code)
},
forceReconnect: () => {
logForDebugging('[bridge:debug] forceReconnect — injecting')
void reconnectEnvironmentWithSession()
},
injectFault: injectBridgeFault,
wakePollLoop,
describe: () =>
`env=${environmentId} session=${currentSessionId} transport=${transport?.getStateLabel() ?? 'null'} workId=${currentWorkId ?? 'null'}`,
})
}
const pollOpts = {
api,
getCredentials: () => ({ environmentId, environmentSecret }),
signal: pollController.signal,
getPollIntervalConfig,
onStateChange,
getWsState: () => transport?.getStateLabel() ?? 'null',
// REPL bridge is single-session: having any transport == at capacity.
// No need to check isConnectedStatus() — even while the transport is
// auto-reconnecting internally (up to 10 min), poll is heartbeat-only.
isAtCapacity: () => transport !== null,
capacitySignal,
onFatalError: triggerTeardown,
getHeartbeatInfo: () => {
if (!currentWorkId || !currentIngressToken) {
return null
}
return {
environmentId,
workId: currentWorkId,
sessionToken: currentIngressToken,
}
},
// Work-item JWT expired (or work gone). The transport is useless —
// SSE reconnects and CCR writes use the same stale token. Without
// this callback the poll loop would do a 10-min at-capacity backoff,
// during which the work lease (300s TTL) expires and the server stops
// forwarding prompts → ~25-min dead window observed in daemon logs.
// Kill the transport + work state so isAtCapacity()=false; the loop
// fast-polls and picks up the server's re-dispatched work in seconds.
onHeartbeatFatal: (err: BridgeFatalError) => {
logForDebugging(
`[bridge:repl] heartbeatWork fatal (status=${err.status}) — tearing down work item for fast re-dispatch`,
)
if (transport) {
const seq = transport.getLastSequenceNum()
if (seq > lastTransportSequenceNum) {
lastTransportSequenceNum = seq
}
transport.close()
transport = null
}
flushGate.drop()
// force=false → server re-queues. Likely already expired, but
// idempotent and makes re-dispatch immediate if not.
if (currentWorkId) {
void api
.stopWork(environmentId, currentWorkId, false)
.catch((e: unknown) => {
logForDebugging(
`[bridge:repl] stopWork after heartbeat fatal: ${errorMessage(e)}`,
)
})
}
currentWorkId = null
currentIngressToken = null
wakePollLoop()
onStateChange?.(
'reconnecting',
'Work item lease expired, fetching fresh token',
)
},
async onEnvironmentLost() {
const success = await reconnectEnvironmentWithSession()
if (!success) {
return null
}
return { environmentId, environmentSecret }
},
onWorkReceived: (
workSessionId: string,
ingressToken: string,
workId: string,
serverUseCcrV2: boolean,
) => {
// When new work arrives while a transport is already open, the
// server has decided to re-dispatch (e.g. token rotation, server
// restart). Close the existing transport and reconnect — discarding
// the work causes a stuck 'reconnecting' state if the old WS dies
// shortly after (the server won't re-dispatch a work item it
// already delivered).
// ingressToken (JWT) is stored for heartbeat auth (both v1 and v2).
// Transport auth diverges — see the v1/v2 split below.
if (transport?.isConnectedStatus()) {
logForDebugging(
`[bridge:repl] Work received while transport connected, replacing with fresh token (workId=${workId})`,
)
}
logForDebugging(
`[bridge:repl] Work received: workId=${workId} workSessionId=${workSessionId} currentSessionId=${currentSessionId} match=${sameSessionId(workSessionId, currentSessionId)}`,
)
// Refresh the crash-recovery pointer's mtime. Staleness checks file
// mtime (not embedded timestamp) so this re-write bumps the clock —
// a 5h+ session that crashes still has a fresh pointer. Fires once
// per work dispatch (infrequent — bounded by user message rate).
void writeBridgePointer(dir, {
sessionId: currentSessionId,
environmentId,
source: 'repl',
})
// Reject foreign session IDs — the server shouldn't assign sessions
// from other environments. Since we create env+session as a pair,
// a mismatch indicates an unexpected server-side reassignment.
//
// Compare by underlying UUID, not by tagged-ID prefix. When CCR
// v2's compat layer serves the session, createBridgeSession gets
// session_* from the v1-facing API (compat/convert.go:41) but the
// infrastructure layer delivers cse_* in the work queue
// (container_manager.go:129). Same UUID, different tag.
if (!sameSessionId(workSessionId, currentSessionId)) {
logForDebugging(
`[bridge:repl] Rejecting foreign session: expected=${currentSessionId} got=${workSessionId}`,
)
return
}
currentWorkId = workId
currentIngressToken = ingressToken
// Server decides per-session (secret.use_code_sessions from the work
// secret, threaded through runWorkPollLoop). The env var is an ant-dev
// override for forcing v2 before the server flag is on for your user —
// requires ccr_v2_compat_enabled server-side or registerWorker 404s.
//
// Kept separate from CLAUDE_CODE_USE_CCR_V2 (the child-SDK transport
// selector set by sessionRunner/environment-manager) to avoid the
// inheritance hazard in spawn mode where the parent's orchestrator
// var would leak into a v1 child.
const useCcrV2 =
serverUseCcrV2 || isEnvTruthy(process.env.CLAUDE_BRIDGE_USE_CCR_V2)
// Auth is the one place v1 and v2 diverge hard:
//
// - v1 (Session-Ingress): accepts OAuth OR JWT. We prefer OAuth
// because the standard OAuth refresh flow handles expiry — no
// separate JWT refresh scheduler needed.
//
// - v2 (CCR /worker/*): REQUIRES the JWT. register_worker.go:32
// validates the session_id claim, which OAuth tokens don't carry.
// The JWT from the work secret has both that claim and the worker
// role (environment_auth.py:856). JWT refresh: when it expires the
// server re-dispatches work with a fresh one, and onWorkReceived
// fires again. createV2ReplTransport stores it via
// updateSessionIngressAuthToken() before touching the network.
let v1OauthToken: string | undefined
if (!useCcrV2) {
v1OauthToken = getOAuthToken()
if (!v1OauthToken) {
logForDebugging(
'[bridge:repl] No OAuth token available for session ingress, skipping work',
)
return
}
updateSessionIngressAuthToken(v1OauthToken)
}
logEvent('tengu_bridge_repl_work_received', {})
// Close the previous transport. Nullify BEFORE calling close() so
// the close callback doesn't treat the programmatic close as
// "session ended normally" and trigger a full teardown.
if (transport) {
const oldTransport = transport
transport = null
// Capture the SSE sequence high-water mark so the next transport
// resumes the stream instead of replaying from seq 0. Use max() —
// a transport that died early (never received any frames) would
// otherwise reset a non-zero mark back to 0.
const oldSeq = oldTransport.getLastSequenceNum()
if (oldSeq > lastTransportSequenceNum) {
lastTransportSequenceNum = oldSeq
}
oldTransport.close()
}
// Reset flush state — the old flush (if any) is no longer relevant.
// Preserve pending messages so they're drained after the new
// transport's flush completes (the hook has already advanced its
// lastWrittenIndex and won't re-send them).
flushGate.deactivate()
// Closure adapter over the shared handleServerControlRequest —
// captures transport/currentSessionId so the transport.setOnData
// callback below doesn't need to thread them through.
const onServerControlRequest = (request: SDKControlRequest): void =>
handleServerControlRequest(request, {
transport,
sessionId: currentSessionId,
onInterrupt,
onSetModel,
onSetMaxThinkingTokens,
onSetPermissionMode,
})
let initialFlushDone = false
// Wire callbacks onto a freshly constructed transport and connect.
// Extracted so the (sync) v1 and (async) v2 construction paths can
// share the identical callback + flush machinery.
const wireTransport = (newTransport: ReplBridgeTransport): void => {
transport = newTransport
newTransport.setOnConnect(() => {
// Guard: if transport was replaced by a newer onWorkReceived call
// while the WS was connecting, ignore this stale callback.
if (transport !== newTransport) return
logForDebugging('[bridge:repl] Ingress transport connected')
logEvent('tengu_bridge_repl_ws_connected', {})
// Update the env var with the latest OAuth token so POST writes
// (which read via getSessionIngressAuthToken()) use a fresh token.
// v2 skips this — createV2ReplTransport already stored the JWT,
// and overwriting it with OAuth would break subsequent /worker/*
// requests (session_id claim check).
if (!useCcrV2) {
const freshToken = getOAuthToken()
if (freshToken) {
updateSessionIngressAuthToken(freshToken)
}
}
// Reset teardownStarted so future teardowns are not blocked.
teardownStarted = false
// Flush initial messages only on first connect, not on every
// WS reconnection. Re-flushing would cause duplicate messages.
// IMPORTANT: onStateChange('connected') is deferred until the
// flush completes. This prevents writeMessages() from sending
// new messages that could arrive at the server interleaved with
// the historical messages, and delays the web UI from showing
// the session as active until history is persisted.
if (
!initialFlushDone &&
initialMessages &&
initialMessages.length > 0
) {
initialFlushDone = true
// Cap the initial flush to the most recent N messages. The full
// history is UI-only (model doesn't see it) and large replays cause
// slow session-ingress persistence (each event is a threadstore write)
// plus elevated Firestore pressure. A 0 or negative cap disables it.
const historyCap = initialHistoryCap
const eligibleMessages = initialMessages.filter(
m =>
isEligibleBridgeMessage(m) &&
!previouslyFlushedUUIDs?.has(m.uuid),
)
const cappedMessages =
historyCap > 0 && eligibleMessages.length > historyCap
? eligibleMessages.slice(-historyCap)
: eligibleMessages
if (cappedMessages.length < eligibleMessages.length) {
logForDebugging(
`[bridge:repl] Capped initial flush: ${eligibleMessages.length} -> ${cappedMessages.length} (cap=${historyCap})`,
)
logEvent('tengu_bridge_repl_history_capped', {
eligible_count: eligibleMessages.length,
capped_count: cappedMessages.length,
})
}
const sdkMessages = toSDKMessages(cappedMessages)
if (sdkMessages.length > 0) {
logForDebugging(
`[bridge:repl] Flushing ${sdkMessages.length} initial message(s) via transport`,
)
const events = sdkMessages.map(sdkMsg => ({
...sdkMsg,
session_id: currentSessionId,
}))
const dropsBefore = newTransport.droppedBatchCount
void newTransport
.writeBatch(events)
.then(() => {
// If any batch was dropped during this flush (SI down for
// maxConsecutiveFailures attempts), flush() still resolved
// normally but the events were NOT delivered. Don't mark
// UUIDs as flushed — keep them eligible for re-send on the
// next onWorkReceived (JWT refresh re-dispatch, line ~1144).
if (newTransport.droppedBatchCount > dropsBefore) {
logForDebugging(
`[bridge:repl] Initial flush dropped ${newTransport.droppedBatchCount - dropsBefore} batch(es) — not marking ${sdkMessages.length} UUID(s) as flushed`,
)
return
}
if (previouslyFlushedUUIDs) {
for (const sdkMsg of sdkMessages) {
if (sdkMsg.uuid) {
previouslyFlushedUUIDs.add(sdkMsg.uuid)
}
}
}
})
.catch(e =>
logForDebugging(`[bridge:repl] Initial flush failed: ${e}`),
)
.finally(() => {
// Guard: if transport was replaced during the flush,
// don't signal connected or drain — the new transport
// owns the lifecycle now.
if (transport !== newTransport) return
drainFlushGate()
onStateChange?.('connected')
})
} else {
// All initial messages were already flushed (filtered by
// previouslyFlushedUUIDs). No flush POST needed — clear
// the flag and signal connected immediately. This is the
// first connect for this transport (inside !initialFlushDone),
// so no flush POST is in-flight — the flag was set before
// connect() and must be cleared here.
drainFlushGate()
onStateChange?.('connected')
}
} else if (!flushGate.active) {
// No initial messages or already flushed on first connect.
// WS auto-reconnect path — only signal connected if no flush
// POST is in-flight. If one is, .finally() owns the lifecycle.
onStateChange?.('connected')
}
})
newTransport.setOnData(data => {
handleIngressMessage(
data,
recentPostedUUIDs,
recentInboundUUIDs,
onInboundMessage,
onPermissionResponse,
onServerControlRequest,
)
})
// Body lives at initBridgeCore scope so /bridge-kick can call it
// directly via debugFireClose. All referenced closures (transport,
// wakePollLoop, flushGate, reconnectEnvironmentWithSession, etc.)
// are already at that scope. The only lexical dependency on
// wireTransport was `newTransport.getLastSequenceNum()` — but after
// the guard below passes we know transport === newTransport.
debugFireClose = handleTransportPermanentClose
newTransport.setOnClose(closeCode => {
// Guard: if transport was replaced, ignore stale close.
if (transport !== newTransport) return
handleTransportPermanentClose(closeCode)
})
// Start the flush gate before connect() to cover the WS handshake
// window. Between transport assignment and setOnConnect firing,
// writeMessages() could send messages via HTTP POST before the
// initial flush starts. Starting the gate here ensures those
// calls are queued. If there are no initial messages, the gate
// stays inactive.
if (
!initialFlushDone &&
initialMessages &&
initialMessages.length > 0
) {
flushGate.start()
}
newTransport.connect()
} // end wireTransport
// Bump unconditionally — ANY new transport (v1 or v2) invalidates an
// in-flight v2 handshake. Also bumped in doReconnect().
v2Generation++
if (useCcrV2) {
// workSessionId is the cse_* form (infrastructure-layer ID from the
// work queue), which is what /v1/code/sessions/{id}/worker/* wants.
// The session_* form (currentSessionId) is NOT usable here —
// handler/convert.go:30 validates TagCodeSession.
const sessionUrl = buildCCRv2SdkUrl(baseUrl, workSessionId)
const thisGen = v2Generation
logForDebugging(
`[bridge:repl] CCR v2: sessionUrl=${sessionUrl} session=${workSessionId} gen=${thisGen}`,
)
void createV2ReplTransport({
sessionUrl,
ingressToken,
sessionId: workSessionId,
initialSequenceNum: lastTransportSequenceNum,
}).then(
t => {
// Teardown started while registerWorker was in flight. Teardown
// saw transport === null and skipped close(); installing now
// would leak CCRClient heartbeat timers and reset
// teardownStarted via wireTransport's side effects.
if (pollController.signal.aborted) {
t.close()
return
}
// onWorkReceived may have fired again while registerWorker()
// was in flight (server re-dispatch with a fresh JWT). The
// transport !== null check alone gets the race wrong when BOTH
// attempts saw transport === null — it keeps the first resolver
// (stale epoch) and discards the second (correct epoch). The
// generation check catches it regardless of transport state.
if (thisGen !== v2Generation) {
logForDebugging(
`[bridge:repl] CCR v2: discarding stale handshake gen=${thisGen} current=${v2Generation}`,
)
t.close()
return
}
wireTransport(t)
},
(err: unknown) => {
logForDebugging(
`[bridge:repl] CCR v2: createV2ReplTransport failed: ${errorMessage(err)}`,
{ level: 'error' },
)
logEvent('tengu_bridge_repl_ccr_v2_init_failed', {})
// If a newer attempt is in flight or already succeeded, don't
// touch its work item — our failure is irrelevant.
if (thisGen !== v2Generation) return
// Release the work item so the server re-dispatches immediately
// instead of waiting for its own timeout. currentWorkId was set
// above; without this, the session looks stuck to the user.
if (currentWorkId) {
void api
.stopWork(environmentId, currentWorkId, false)
.catch((e: unknown) => {
logForDebugging(
`[bridge:repl] stopWork after v2 init failure: ${errorMessage(e)}`,
)
})
currentWorkId = null
currentIngressToken = null
}
wakePollLoop()
},
)
} else {
// v1: HybridTransport (WS reads + POST writes to Session-Ingress).
// autoReconnect is true (default) — when the WS dies, the transport
// reconnects automatically with exponential backoff. POST writes
// continue during reconnection (they use getSessionIngressAuthToken()
// independently of WS state). The poll loop remains as a secondary
// fallback if the reconnect budget is exhausted (10 min).
//
// Auth: uses OAuth tokens directly instead of the JWT from the work
// secret. refreshHeaders picks up the latest OAuth token on each
// WS reconnect attempt.
const wsUrl = buildSdkUrl(sessionIngressUrl, workSessionId)
logForDebugging(`[bridge:repl] Ingress URL: ${wsUrl}`)
logForDebugging(
`[bridge:repl] Creating HybridTransport: session=${workSessionId}`,
)
// v1OauthToken was validated non-null above (we'd have returned early).
const oauthToken = v1OauthToken ?? ''
wireTransport(
createV1ReplTransport(
new HybridTransport(
new URL(wsUrl),
{
Authorization: `Bearer ${oauthToken}`,
'anthropic-version': '2023-06-01',
},
workSessionId,
() => ({
Authorization: `Bearer ${getOAuthToken() ?? oauthToken}`,
'anthropic-version': '2023-06-01',
}),
// Cap retries so a persistently-failing session-ingress can't
// pin the uploader drain loop for the lifetime of the bridge.
// 50 attempts ≈ 20 min (15s POST timeout + 8s backoff + jitter
// per cycle at steady state). Bridge-only — 1P keeps indefinite.
{
maxConsecutiveFailures: 50,
isBridge: true,
onBatchDropped: () => {
onStateChange?.(
'reconnecting',
'Lost sync with Remote Control — events could not be delivered',
)
// SI has been down ~20 min. Wake the poll loop so that when
// SI recovers, next poll → onWorkReceived → fresh transport
// → initial flush succeeds → onStateChange('connected') at
// ~line 1420. Without this, state stays 'reconnecting' even
// after SI recovers — daemon.ts:437 denies all permissions,
// useReplBridge.ts:311 keeps replBridgeSessionActive=false.
// If the env was archived during the outage, poll 404 →
// onEnvironmentLost recovery path handles it.
wakePollLoop()
},
},
),
),
)
}
},
}
void startWorkPollLoop(pollOpts)
// Perpetual mode: hourly mtime refresh of the crash-recovery pointer.
// The onWorkReceived refresh only fires per user prompt — a
// daemon idle for >4h would have a stale pointer, and the next restart
// would clear it (readBridgePointer TTL check) → fresh session. The
// standalone bridge (bridgeMain.ts) has an identical hourly timer.
const pointerRefreshTimer = perpetual
? setInterval(() => {
// doReconnect() reassigns currentSessionId/environmentId non-
// atomically (env at ~:634, session at ~:719, awaits in between).
// If this timer fires in that window, its fire-and-forget write can
// race with (and overwrite) doReconnect's own pointer write at ~:740,
// leaving the pointer at the now-archived old session. doReconnect
// writes the pointer itself, so skipping here is free.
if (reconnectPromise) return
void writeBridgePointer(dir, {
sessionId: currentSessionId,
environmentId,
source: 'repl',
})
}, 60 * 60_000)
: null
pointerRefreshTimer?.unref?.()
// Push a silent keep_alive frame on a fixed interval so upstream proxies
// and the session-ingress layer don't GC an otherwise-idle remote control
// session. The keep_alive type is filtered before reaching any client UI
// (Query.ts drops it; web/iOS/Android never see it in their message loop).
// Interval comes from GrowthBook (tengu_bridge_poll_interval_config
// session_keepalive_interval_v2_ms, default 120s); 0 = disabled.
const keepAliveIntervalMs =
getPollIntervalConfig().session_keepalive_interval_v2_ms
const keepAliveTimer =
keepAliveIntervalMs > 0
? setInterval(() => {
if (!transport) return
logForDebugging('[bridge:repl] keep_alive sent')
void transport.write({ type: 'keep_alive' }).catch((err: unknown) => {
logForDebugging(
`[bridge:repl] keep_alive write failed: ${errorMessage(err)}`,
)
})
}, keepAliveIntervalMs)
: null
keepAliveTimer?.unref?.()
// Shared teardown sequence used by both cleanup registration and
// the explicit teardown() method on the returned handle.
let teardownStarted = false
doTeardownImpl = async (): Promise<void> => {
if (teardownStarted) {
logForDebugging(
`[bridge:repl] Teardown already in progress, skipping duplicate call env=${environmentId} session=${currentSessionId}`,
)
return
}
teardownStarted = true
const teardownStart = Date.now()
logForDebugging(
`[bridge:repl] Teardown starting: env=${environmentId} session=${currentSessionId} workId=${currentWorkId ?? 'none'} transportState=${transport?.getStateLabel() ?? 'null'}`,
)
if (pointerRefreshTimer !== null) {
clearInterval(pointerRefreshTimer)
}
if (keepAliveTimer !== null) {
clearInterval(keepAliveTimer)
}
if (sigusr2Handler) {
process.off('SIGUSR2', sigusr2Handler)
}
if (process.env.USER_TYPE === 'ant') {
clearBridgeDebugHandle()
debugFireClose = null
}
pollController.abort()
logForDebugging('[bridge:repl] Teardown: poll loop aborted')
// Capture the live transport's seq BEFORE close() — close() is sync
// (just aborts the SSE fetch) and does NOT invoke onClose, so the
// setOnClose capture path never runs for explicit teardown.
// Without this, getSSESequenceNum() after teardown returns the stale
// lastTransportSequenceNum (captured at the last transport swap), and
// daemon callers persisting that value lose all events since then.
if (transport) {
const finalSeq = transport.getLastSequenceNum()
if (finalSeq > lastTransportSequenceNum) {
lastTransportSequenceNum = finalSeq
}
}
if (perpetual) {
// Perpetual teardown is LOCAL-ONLY — do not send result, do not call
// stopWork, do not close the transport. All of those signal the
// server (and any mobile/attach subscribers) that the session is
// ending. Instead: stop polling, let the socket die with the
// process; the backend times the work-item lease back to pending on
// its own (TTL 300s). Next daemon start reads the pointer and
// reconnectSession re-queues work.
transport = null
flushGate.drop()
// Refresh the pointer mtime so that sessions lasting longer than
// BRIDGE_POINTER_TTL_MS (4h) don't appear stale on next start.
await writeBridgePointer(dir, {
sessionId: currentSessionId,
environmentId,
source: 'repl',
})
logForDebugging(
`[bridge:repl] Teardown (perpetual): leaving env=${environmentId} session=${currentSessionId} alive on server, duration=${Date.now() - teardownStart}ms`,
)
return
}
// Fire the result message, then archive, THEN close. transport.write()
// only enqueues (SerialBatchEventUploader resolves on buffer-add); the
// stopWork/archive latency (~200-500ms) is the drain window for the
// result POST. Closing BEFORE archive meant relying on HybridTransport's
// void-ed 3s grace period, which nothing awaits — forceExit can kill the
// socket mid-POST. Same reorder as remoteBridgeCore.ts teardown (#22803).
const teardownTransport = transport
transport = null
flushGate.drop()
if (teardownTransport) {
void teardownTransport.write(makeResultMessage(currentSessionId))
}
const stopWorkP = currentWorkId
? api
.stopWork(environmentId, currentWorkId, true)
.then(() => {
logForDebugging('[bridge:repl] Teardown: stopWork completed')
})
.catch((err: unknown) => {
logForDebugging(
`[bridge:repl] Teardown stopWork failed: ${errorMessage(err)}`,
)
})
: Promise.resolve()
// Run stopWork and archiveSession in parallel. gracefulShutdown.ts:407
// races runCleanupFunctions() against 2s (NOT the 5s outer failsafe),
// so archive is capped at 1.5s at the injection site to stay under budget.
// archiveSession is contractually no-throw; the injected implementations
// log their own success/failure internally.
await Promise.all([stopWorkP, archiveSession(currentSessionId)])
teardownTransport?.close()
logForDebugging('[bridge:repl] Teardown: transport closed')
await api.deregisterEnvironment(environmentId).catch((err: unknown) => {
logForDebugging(
`[bridge:repl] Teardown deregister failed: ${errorMessage(err)}`,
)
})
// Clear the crash-recovery pointer — explicit disconnect or clean REPL
// exit means the user is done with this session. Crash/kill-9 never
// reaches this line, leaving the pointer for next-launch recovery.
await clearBridgePointer(dir)
logForDebugging(
`[bridge:repl] Teardown complete: env=${environmentId} duration=${Date.now() - teardownStart}ms`,
)
}
// 8. Register cleanup for graceful shutdown
const unregister = registerCleanup(() => doTeardownImpl?.())
logForDebugging(
`[bridge:repl] Ready: env=${environmentId} session=${currentSessionId}`,
)
onStateChange?.('ready')
return {
get bridgeSessionId() {
return currentSessionId
},
get environmentId() {
return environmentId
},
getSSESequenceNum() {
// lastTransportSequenceNum only updates when a transport is CLOSED
// (captured at swap/onClose). During normal operation the CURRENT
// transport's live seq isn't reflected there. Merge both so callers
// (e.g. daemon persistState()) get the actual high-water mark.
const live = transport?.getLastSequenceNum() ?? 0
return Math.max(lastTransportSequenceNum, live)
},
sessionIngressUrl,
writeMessages(messages) {
// Filter to user/assistant messages that haven't already been sent.
// Two layers of dedup:
// - initialMessageUUIDs: messages sent as session creation events
// - recentPostedUUIDs: messages recently sent via POST
const filtered = messages.filter(
m =>
isEligibleBridgeMessage(m) &&
!initialMessageUUIDs.has(m.uuid) &&
!recentPostedUUIDs.has(m.uuid),
)
if (filtered.length === 0) return
// Fire onUserMessage for title derivation. Scan before the flushGate
// check — prompts are title-worthy even if they queue behind the
// initial history flush. Keeps calling on every title-worthy message
// until the callback returns true; the caller owns the policy.
if (!userMessageCallbackDone) {
for (const m of filtered) {
const text = extractTitleText(m)
if (text !== undefined && onUserMessage?.(text, currentSessionId)) {
userMessageCallbackDone = true
break
}
}
}
// Queue messages while the initial flush is in progress to prevent
// them from arriving at the server interleaved with history.
if (flushGate.enqueue(...filtered)) {
logForDebugging(
`[bridge:repl] Queued ${filtered.length} message(s) during initial flush`,
)
return
}
if (!transport) {
const types = filtered.map(m => m.type).join(',')
logForDebugging(
`[bridge:repl] Transport not configured, dropping ${filtered.length} message(s) [${types}] for session=${currentSessionId}`,
{ level: 'warn' },
)
return
}
// Track in the bounded ring buffer for echo filtering and dedup.
for (const msg of filtered) {
recentPostedUUIDs.add(msg.uuid)
}
logForDebugging(
`[bridge:repl] Sending ${filtered.length} message(s) via transport`,
)
// Convert to SDK format and send via HTTP POST (HybridTransport).
// The web UI receives them via the subscribe WebSocket.
const sdkMessages = toSDKMessages(filtered)
const events = sdkMessages.map(sdkMsg => ({
...sdkMsg,
session_id: currentSessionId,
}))
void transport.writeBatch(events)
},
writeSdkMessages(messages) {
// Daemon path: query() already yields SDKMessage, skip conversion.
// Still run echo dedup (server bounces writes back on the WS).
// No initialMessageUUIDs filter — daemon has no initial messages.
// No flushGate — daemon never starts it (no initial flush).
const filtered = messages.filter(
m => !m.uuid || !recentPostedUUIDs.has(m.uuid),
)
if (filtered.length === 0) return
if (!transport) {
logForDebugging(
`[bridge:repl] Transport not configured, dropping ${filtered.length} SDK message(s) for session=${currentSessionId}`,
{ level: 'warn' },
)
return
}
for (const msg of filtered) {
if (msg.uuid) recentPostedUUIDs.add(msg.uuid)
}
const events = filtered.map(m => ({ ...m, session_id: currentSessionId }))
void transport.writeBatch(events)
},
sendControlRequest(request: SDKControlRequest) {
if (!transport) {
logForDebugging(
'[bridge:repl] Transport not configured, skipping control_request',
)
return
}
const event = { ...request, session_id: currentSessionId }
void transport.write(event)
logForDebugging(
`[bridge:repl] Sent control_request request_id=${request.request_id}`,
)
},
sendControlResponse(response: SDKControlResponse) {
if (!transport) {
logForDebugging(
'[bridge:repl] Transport not configured, skipping control_response',
)
return
}
const event = { ...response, session_id: currentSessionId }
void transport.write(event)
logForDebugging('[bridge:repl] Sent control_response')
},
sendControlCancelRequest(requestId: string) {
if (!transport) {
logForDebugging(
'[bridge:repl] Transport not configured, skipping control_cancel_request',
)
return
}
const event = {
type: 'control_cancel_request' as const,
request_id: requestId,
session_id: currentSessionId,
}
void transport.write(event)
logForDebugging(
`[bridge:repl] Sent control_cancel_request request_id=${requestId}`,
)
},
sendResult() {
if (!transport) {
logForDebugging(
`[bridge:repl] sendResult: skipping, transport not configured session=${currentSessionId}`,
)
return
}
void transport.write(makeResultMessage(currentSessionId))
logForDebugging(
`[bridge:repl] Sent result for session=${currentSessionId}`,
)
},
async teardown() {
unregister()
await doTeardownImpl?.()
logForDebugging('[bridge:repl] Torn down')
logEvent('tengu_bridge_repl_teardown', {})
},
}
}
/**
* Persistent poll loop for work items. Runs in the background for the
* lifetime of the bridge connection.
*
* When a work item arrives, acknowledges it and calls onWorkReceived
* with the session ID and ingress token (which connects the ingress
* WebSocket). Then continues polling — the server will dispatch a new
* work item if the ingress WebSocket drops, allowing automatic
* reconnection without tearing down the bridge.
*/
async function startWorkPollLoop({
api,
getCredentials,
signal,
onStateChange,
onWorkReceived,
onEnvironmentLost,
getWsState,
isAtCapacity,
capacitySignal,
onFatalError,
getPollIntervalConfig = () => DEFAULT_POLL_CONFIG,
getHeartbeatInfo,
onHeartbeatFatal,
}: {
api: BridgeApiClient
getCredentials: () => { environmentId: string; environmentSecret: string }
signal: AbortSignal
onStateChange?: (state: BridgeState, detail?: string) => void
onWorkReceived: (
sessionId: string,
ingressToken: string,
workId: string,
useCodeSessions: boolean,
) => void
/** Called when the environment has been deleted. Returns new credentials or null. */
onEnvironmentLost?: () => Promise<{
environmentId: string
environmentSecret: string
} | null>
/** Returns the current WebSocket readyState label for diagnostic logging. */
getWsState?: () => string
/**
* Returns true when the caller cannot accept new work (transport already
* connected). When true, the loop polls at the configured at-capacity
* interval as a heartbeat only. Server-side BRIDGE_LAST_POLL_TTL is
* 4 hours — anything shorter than that is sufficient for liveness.
*/
isAtCapacity?: () => boolean
/**
* Produces a signal that aborts when capacity frees up (transport lost),
* merged with the loop signal. Used to interrupt the at-capacity sleep
* so recovery polling starts immediately.
*/
capacitySignal?: () => CapacitySignal
/** Called on unrecoverable errors (e.g. server-side expiry) to trigger full teardown. */
onFatalError?: () => void
/** Poll interval config getter — defaults to DEFAULT_POLL_CONFIG. */
getPollIntervalConfig?: () => PollIntervalConfig
/**
* Returns the current work ID and session ingress token for heartbeat.
* When null, heartbeat is not possible (no active work item).
*/
getHeartbeatInfo?: () => {
environmentId: string
workId: string
sessionToken: string
} | null
/**
* Called when heartbeatWork throws BridgeFatalError (401/403/404/410 —
* JWT expired or work item gone). Caller should tear down the transport
* + work state so isAtCapacity() flips to false and the loop fast-polls
* for the server's re-dispatched work item. When provided, the loop
* SKIPS the at-capacity backoff sleep (which would otherwise cause a
* ~10-minute dead window before recovery). When omitted, falls back to
* the backoff sleep to avoid a tight poll+heartbeat loop.
*/
onHeartbeatFatal?: (err: BridgeFatalError) => void
}): Promise<void> {
const MAX_ENVIRONMENT_RECREATIONS = 3
logForDebugging(
`[bridge:repl] Starting work poll loop for env=${getCredentials().environmentId}`,
)
let consecutiveErrors = 0
let firstErrorTime: number | null = null
let lastPollErrorTime: number | null = null
let environmentRecreations = 0
// Set when the at-capacity sleep overruns its deadline by a large margin
// (process suspension). Consumed at the top of the next iteration to
// force one fast-poll cycle — isAtCapacity() is `transport !== null`,
// which stays true while the transport auto-reconnects, so the poll
// loop would otherwise go straight back to a 10-minute sleep on a
// transport that may be pointed at a dead socket.
let suspensionDetected = false
while (!signal.aborted) {
// Capture credentials outside try so the catch block can detect
// whether a concurrent reconnection replaced the environment.
const { environmentId: envId, environmentSecret: envSecret } =
getCredentials()
const pollConfig = getPollIntervalConfig()
try {
const work = await api.pollForWork(
envId,
envSecret,
signal,
pollConfig.reclaim_older_than_ms,
)
// A successful poll proves the env is genuinely healthy — reset the
// env-loss counter so events hours apart each start fresh. Outside
// the state-change guard below because onEnvLost's success path
// already emits 'ready'; emitting again here would be a duplicate.
// (onEnvLost returning creds does NOT reset this — that would break
// oscillation protection when the new env immediately dies.)
environmentRecreations = 0
// Reset error tracking on successful poll
if (consecutiveErrors > 0) {
logForDebugging(
`[bridge:repl] Poll recovered after ${consecutiveErrors} consecutive error(s)`,
)
consecutiveErrors = 0
firstErrorTime = null
lastPollErrorTime = null
onStateChange?.('ready')
}
if (!work) {
// Read-and-clear: after a detected suspension, skip the at-capacity
// branch exactly once. The pollForWork above already refreshed the
// server's BRIDGE_LAST_POLL_TTL; this fast cycle gives any
// re-dispatched work item a chance to land before we go back under.
const skipAtCapacityOnce = suspensionDetected
suspensionDetected = false
if (isAtCapacity?.() && capacitySignal && !skipAtCapacityOnce) {
const atCapMs = pollConfig.poll_interval_ms_at_capacity
// Heartbeat loops WITHOUT polling. When at-capacity polling is also
// enabled (atCapMs > 0), the loop tracks a deadline and breaks out
// to poll at that interval — heartbeat and poll compose instead of
// one suppressing the other. Breaks out when:
// - Poll deadline reached (atCapMs > 0 only)
// - Auth fails (JWT expired → poll refreshes tokens)
// - Capacity wake fires (transport lost → poll for new work)
// - Heartbeat config disabled (GrowthBook update)
// - Loop aborted (shutdown)
if (
pollConfig.non_exclusive_heartbeat_interval_ms > 0 &&
getHeartbeatInfo
) {
logEvent('tengu_bridge_heartbeat_mode_entered', {
heartbeat_interval_ms:
pollConfig.non_exclusive_heartbeat_interval_ms,
})
// Deadline computed once at entry — GB updates to atCapMs don't
// shift an in-flight deadline (next entry picks up the new value).
const pollDeadline = atCapMs > 0 ? Date.now() + atCapMs : null
let needsBackoff = false
let hbCycles = 0
while (
!signal.aborted &&
isAtCapacity() &&
(pollDeadline === null || Date.now() < pollDeadline)
) {
const hbConfig = getPollIntervalConfig()
if (hbConfig.non_exclusive_heartbeat_interval_ms <= 0) break
const info = getHeartbeatInfo()
if (!info) break
// Capture capacity signal BEFORE the async heartbeat call so
// a transport loss during the HTTP request is caught by the
// subsequent sleep.
const cap = capacitySignal()
try {
await api.heartbeatWork(
info.environmentId,
info.workId,
info.sessionToken,
)
} catch (err) {
logForDebugging(
`[bridge:repl:heartbeat] Failed: ${errorMessage(err)}`,
)
if (err instanceof BridgeFatalError) {
cap.cleanup()
logEvent('tengu_bridge_heartbeat_error', {
status:
err.status as unknown as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
error_type: (err.status === 401 || err.status === 403
? 'auth_failed'
: 'fatal') as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
})
// JWT expired (401/403) or work item gone (404/410).
// Either way the current transport is dead — SSE
// reconnects and CCR writes will fail on the same
// stale token. If the caller gave us a recovery hook,
// tear down work state and skip backoff: isAtCapacity()
// flips to false, next outer-loop iteration fast-polls
// for the server's re-dispatched work item. Without
// the hook, backoff to avoid tight poll+heartbeat loop.
if (onHeartbeatFatal) {
onHeartbeatFatal(err)
logForDebugging(
`[bridge:repl:heartbeat] Fatal (status=${err.status}), work state cleared — fast-polling for re-dispatch`,
)
} else {
needsBackoff = true
}
break
}
}
hbCycles++
await sleep(
hbConfig.non_exclusive_heartbeat_interval_ms,
cap.signal,
)
cap.cleanup()
}
const exitReason = needsBackoff
? 'error'
: signal.aborted
? 'shutdown'
: !isAtCapacity()
? 'capacity_changed'
: pollDeadline !== null && Date.now() >= pollDeadline
? 'poll_due'
: 'config_disabled'
logEvent('tengu_bridge_heartbeat_mode_exited', {
reason:
exitReason as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
heartbeat_cycles: hbCycles,
})
// On auth_failed or fatal, backoff before polling to avoid a
// tight poll+heartbeat loop. Fall through to the shared sleep
// below — it's the same capacitySignal-wrapped sleep the legacy
// path uses, and both need the suspension-overrun check.
if (!needsBackoff) {
if (exitReason === 'poll_due') {
// bridgeApi throttles empty-poll logs (EMPTY_POLL_LOG_INTERVAL=100)
// so the once-per-10min poll_due poll is invisible at counter=2.
// Log it here so verification runs see both endpoints in the debug log.
logForDebugging(
`[bridge:repl] Heartbeat poll_due after ${hbCycles} cycles — falling through to pollForWork`,
)
}
continue
}
}
// At-capacity sleep — reached by both the legacy path (heartbeat
// disabled) and the heartbeat-backoff path (needsBackoff=true).
// Merged so the suspension detector covers both; previously the
// backoff path had no overrun check and could go straight back
// under for 10 min after a laptop wake. Use atCapMs when enabled,
// else the heartbeat interval as a floor (guaranteed > 0 on the
// backoff path) so heartbeat-only configs don't tight-loop.
const sleepMs =
atCapMs > 0
? atCapMs
: pollConfig.non_exclusive_heartbeat_interval_ms
if (sleepMs > 0) {
const cap = capacitySignal()
const sleepStart = Date.now()
await sleep(sleepMs, cap.signal)
cap.cleanup()
// Process-suspension detector. A setTimeout overshooting its
// deadline by 60s means the process was suspended (laptop lid,
// SIGSTOP, VM pause) — even a pathological GC pause is seconds,
// not minutes. Early aborts (wakePollLoop → cap.signal) produce
// overrun < 0 and fall through. Note: this only catches sleeps
// that outlast their deadline; WebSocketTransport's ping
// interval (10s granularity) is the primary detector for shorter
// suspensions. This is the backstop for when that detector isn't
// running (transport mid-reconnect, interval stopped).
const overrun = Date.now() - sleepStart - sleepMs
if (overrun > 60_000) {
logForDebugging(
`[bridge:repl] At-capacity sleep overran by ${Math.round(overrun / 1000)}s — process suspension detected, forcing one fast-poll cycle`,
)
logEvent('tengu_bridge_repl_suspension_detected', {
overrun_ms: overrun,
})
suspensionDetected = true
}
}
} else {
await sleep(pollConfig.poll_interval_ms_not_at_capacity, signal)
}
continue
}
// Decode before type dispatch — need the JWT for the explicit ack.
let secret
try {
secret = decodeWorkSecret(work.secret)
} catch (err) {
logForDebugging(
`[bridge:repl] Failed to decode work secret: ${errorMessage(err)}`,
)
logEvent('tengu_bridge_repl_work_secret_failed', {})
// Can't ack (needs the JWT we failed to decode). stopWork uses OAuth.
// Prevents XAUTOCLAIM re-delivering this poisoned item every cycle.
await api.stopWork(envId, work.id, false).catch(() => {})
continue
}
// Explicitly acknowledge to prevent redelivery. Non-fatal on failure:
// server re-delivers, and the onWorkReceived callback handles dedup.
logForDebugging(`[bridge:repl] Acknowledging workId=${work.id}`)
try {
await api.acknowledgeWork(envId, work.id, secret.session_ingress_token)
} catch (err) {
logForDebugging(
`[bridge:repl] Acknowledge failed workId=${work.id}: ${errorMessage(err)}`,
)
}
if (work.data.type === 'healthcheck') {
logForDebugging('[bridge:repl] Healthcheck received')
continue
}
if (work.data.type === 'session') {
const workSessionId = work.data.id
try {
validateBridgeId(workSessionId, 'session_id')
} catch {
logForDebugging(
`[bridge:repl] Invalid session_id in work: ${workSessionId}`,
)
continue
}
onWorkReceived(
workSessionId,
secret.session_ingress_token,
work.id,
secret.use_code_sessions === true,
)
logForDebugging('[bridge:repl] Work accepted, continuing poll loop')
}
} catch (err) {
if (signal.aborted) break
// Detect permanent "environment deleted" error — no amount of
// retrying will recover. Re-register a new environment instead.
// Checked BEFORE the generic BridgeFatalError bail. pollForWork uses
// validateStatus: s => s < 500, so 404 is always wrapped into a
// BridgeFatalError by handleErrorStatus() — never an axios-shaped
// error. The poll endpoint's only path param is the env ID; 404
// unambiguously means env-gone (no-work is a 200 with null body).
// The server sends error.type='not_found_error' (standard Anthropic
// API shape), not a bridge-specific string — but status===404 is
// the real signal and survives body-shape changes.
if (
err instanceof BridgeFatalError &&
err.status === 404 &&
onEnvironmentLost
) {
// If credentials have already been refreshed by a concurrent
// reconnection (e.g. WS close handler), the stale poll's error
// is expected — skip onEnvironmentLost and retry with fresh creds.
const currentEnvId = getCredentials().environmentId
if (envId !== currentEnvId) {
logForDebugging(
`[bridge:repl] Stale poll error for old env=${envId}, current env=${currentEnvId} — skipping onEnvironmentLost`,
)
consecutiveErrors = 0
firstErrorTime = null
continue
}
environmentRecreations++
logForDebugging(
`[bridge:repl] Environment deleted, attempting re-registration (attempt ${environmentRecreations}/${MAX_ENVIRONMENT_RECREATIONS})`,
)
logEvent('tengu_bridge_repl_env_lost', {
attempt: environmentRecreations,
} as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS)
if (environmentRecreations > MAX_ENVIRONMENT_RECREATIONS) {
logForDebugging(
`[bridge:repl] Environment re-registration limit reached (${MAX_ENVIRONMENT_RECREATIONS}), giving up`,
)
onStateChange?.(
'failed',
'Environment deleted and re-registration limit reached',
)
onFatalError?.()
break
}
onStateChange?.('reconnecting', 'environment lost, recreating session')
const newCreds = await onEnvironmentLost()
// doReconnect() makes several sequential network calls (1-5s).
// If the user triggered teardown during that window, its internal
// abort checks return false — but we need to re-check here to
// avoid emitting a spurious 'failed' + onFatalError() during
// graceful shutdown.
if (signal.aborted) break
if (newCreds) {
// Credentials are updated in the outer scope via
// reconnectEnvironmentWithSession — getCredentials() will
// return the fresh values on the next poll iteration.
// Do NOT reset environmentRecreations here — onEnvLost returning
// creds only proves we tried to fix it, not that the env is
// healthy. A successful poll (above) is the reset point; if the
// new env immediately dies again we still want the limit to fire.
consecutiveErrors = 0
firstErrorTime = null
onStateChange?.('ready')
logForDebugging(
`[bridge:repl] Re-registered environment: ${newCreds.environmentId}`,
)
continue
}
onStateChange?.(
'failed',
'Environment deleted and re-registration failed',
)
onFatalError?.()
break
}
// Fatal errors (401/403/404/410) — no point retrying
if (err instanceof BridgeFatalError) {
const isExpiry = isExpiredErrorType(err.errorType)
const isSuppressible = isSuppressible403(err)
logForDebugging(
`[bridge:repl] Fatal poll error: ${err.message} (status=${err.status}, type=${err.errorType ?? 'unknown'})${isSuppressible ? ' (suppressed)' : ''}`,
)
logEvent('tengu_bridge_repl_fatal_error', {
status: err.status,
error_type:
err.errorType as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
})
logForDiagnosticsNoPII(
isExpiry ? 'info' : 'error',
'bridge_repl_fatal_error',
{ status: err.status, error_type: err.errorType },
)
// Cosmetic 403 errors (e.g., external_poll_sessions scope,
// environments:manage permission) — suppress user-visible error
// but always trigger teardown so cleanup runs.
if (!isSuppressible) {
onStateChange?.(
'failed',
isExpiry
? 'session expired · /remote-control to reconnect'
: err.message,
)
}
// Always trigger teardown — matches bridgeMain.ts where fatalExit=true
// is unconditional and post-loop cleanup always runs.
onFatalError?.()
break
}
const now = Date.now()
// Detect system sleep/wake: if the gap since the last poll error
// greatly exceeds the max backoff delay, the machine likely slept.
// Reset error tracking so we retry with a fresh budget instead of
// immediately giving up.
if (
lastPollErrorTime !== null &&
now - lastPollErrorTime > POLL_ERROR_MAX_DELAY_MS * 2
) {
logForDebugging(
`[bridge:repl] Detected system sleep (${Math.round((now - lastPollErrorTime) / 1000)}s gap), resetting poll error budget`,
)
logForDiagnosticsNoPII('info', 'bridge_repl_poll_sleep_detected', {
gapMs: now - lastPollErrorTime,
})
consecutiveErrors = 0
firstErrorTime = null
}
lastPollErrorTime = now
consecutiveErrors++
if (firstErrorTime === null) {
firstErrorTime = now
}
const elapsed = now - firstErrorTime
const httpStatus = extractHttpStatus(err)
const errMsg = describeAxiosError(err)
const wsLabel = getWsState?.() ?? 'unknown'
logForDebugging(
`[bridge:repl] Poll error (attempt ${consecutiveErrors}, elapsed ${Math.round(elapsed / 1000)}s, ws=${wsLabel}): ${errMsg}`,
)
logEvent('tengu_bridge_repl_poll_error', {
status: httpStatus,
consecutiveErrors,
elapsedMs: elapsed,
} as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS)
// Only transition to 'reconnecting' on the first error — stay
// there until a successful poll (avoid flickering the UI state).
if (consecutiveErrors === 1) {
onStateChange?.('reconnecting', errMsg)
}
// Give up after continuous failures
if (elapsed >= POLL_ERROR_GIVE_UP_MS) {
logForDebugging(
`[bridge:repl] Poll failures exceeded ${POLL_ERROR_GIVE_UP_MS / 1000}s (${consecutiveErrors} errors), giving up`,
)
logForDiagnosticsNoPII('info', 'bridge_repl_poll_give_up')
logEvent('tengu_bridge_repl_poll_give_up', {
consecutiveErrors,
elapsedMs: elapsed,
lastStatus: httpStatus,
} as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS)
onStateChange?.('failed', 'connection to server lost')
break
}
// Exponential backoff: 2s → 4s → 8s → 16s → 32s → 60s (cap)
const backoff = Math.min(
POLL_ERROR_INITIAL_DELAY_MS * 2 ** (consecutiveErrors - 1),
POLL_ERROR_MAX_DELAY_MS,
)
// The poll_due heartbeat-loop exit leaves a healthy lease exposed to
// this backoff path. Heartbeat before each sleep so /poll outages
// (the VerifyEnvironmentSecretAuth DB path heartbeat was introduced to
// avoid) don't kill the 300s lease TTL.
if (getPollIntervalConfig().non_exclusive_heartbeat_interval_ms > 0) {
const info = getHeartbeatInfo?.()
if (info) {
try {
await api.heartbeatWork(
info.environmentId,
info.workId,
info.sessionToken,
)
} catch {
// Best-effort — if heartbeat also fails the lease dies, same as
// pre-poll_due behavior (where the only heartbeat-loop exits were
// ones where the lease was already dying).
}
}
}
await sleep(backoff, signal)
}
}
logForDebugging(
`[bridge:repl] Work poll loop ended (aborted=${signal.aborted}) env=${getCredentials().environmentId}`,
)
}
// Exported for testing only
export {
startWorkPollLoop as _startWorkPollLoopForTesting,
POLL_ERROR_INITIAL_DELAY_MS as _POLL_ERROR_INITIAL_DELAY_MS_ForTesting,
POLL_ERROR_MAX_DELAY_MS as _POLL_ERROR_MAX_DELAY_MS_ForTesting,
POLL_ERROR_GIVE_UP_MS as _POLL_ERROR_GIVE_UP_MS_ForTesting,
}