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>

dir: string

machineName: string

branch: string

gitRepoUrl: string | null

title: string

baseUrl: string

sessionIngressUrl: string

/**

workerType: string

getAccessToken: () => string | undefined

/**

createSession: (opts: {

environmentId: string

title: string

gitRepoUrl: string | null

branch: string

signal: AbortSignal

}) => Promise<string | null>

/**

archiveSession: (sessionId: string) => Promise<void>

/**

getCurrentTitle?: () => string

/**

toSDKMessages?: (messages: Message[]) => SDKMessage[]

/**

onAuth401?: (staleAccessToken: string) => Promise<boolean>

/**

getPollIntervalConfig?: () => PollIntervalConfig

/**

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

/**

onSetPermissionMode?: (

mode: PermissionMode,

) => { ok: true } | { ok: false; error: string }

onStateChange?: (state: BridgeState, detail?: string) => void

/**

onUserMessage?: (text: string, sessionId: string) => boolean

/** See InitBridgeOptions.perpetual. */

perpetual?: boolean

/**

initialSSESequenceNum?: number

/**

getSSESequenceNum(): number

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', {})

/**

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

/**

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?.()

}

/**

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

},