mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
fix(remote-terminal): reattach a healthy remote pane promptly after a worktree switch (#29210)
* fix(remote-terminal): reattach a healthy remote pane promptly after a worktree switch (#27178) A remote pane whose stream ended while the host still reported its terminal ready waited the whole 15s inventory window before reusing it, and a third end within 60s demanded a replacement handle that never comes, parking recovery with no timer until the ~3 min deadline. Unknown-liveness outcomes on the attach and resubscribe paths also parked for online/resume events that a healthy link never fires. - Reuse a still-ready handle as soon as the host reports it, unless a terminal.send to it is unanswered (it may still come back stale). - Pace repeated ends on the same handle with growing delays instead of requiring a replacement; only ends within 10s of a reattach count. - Schedule unknown-liveness retries on the backoff ladder instead of parking; the existing deadline still latches Reconnect for a genuinely lost link. Co-Authored-By: Claude <noreply@anthropic.com> * fix(remote-terminal): fence an endless same-handle flap once its pacing ladder is spent Each paced reattach marked the pane healthy and reset the recovery deadline, so a handle that ended after every reattach retried forever without ever offering Reconnect. After the last rung it now falls back to the replacement wait, whose deadline latches Reconnect. Co-Authored-By: Claude <noreply@anthropic.com> --------- Co-authored-by: m4air <m4air@Mac.localdomain> Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -95,13 +95,15 @@ export class RemoteRuntimePtyRecoveryState {
|
||||
return this.phase !== 'disposed' && epoch === this.epoch
|
||||
}
|
||||
|
||||
schedule(epoch: number, retry: (epoch: number) => void): boolean {
|
||||
// Why the override: a caller pacing its own repeats (a flapping stream) outranks the per-epoch ladder.
|
||||
schedule(epoch: number, retry: (epoch: number) => void, delayOverrideMs?: number): boolean {
|
||||
if (!this.isCurrent(epoch)) {
|
||||
return false
|
||||
}
|
||||
this.clearRetryTimer()
|
||||
this.phase = 'backoff'
|
||||
const delayMs =
|
||||
delayOverrideMs ??
|
||||
REMOTE_RUNTIME_RECOVERY_DELAYS_MS[
|
||||
Math.min(this.attempt, REMOTE_RUNTIME_RECOVERY_DELAYS_MS.length - 1)
|
||||
]
|
||||
|
||||
+226
@@ -0,0 +1,226 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
createRemoteRuntimeTransportMocks,
|
||||
readyHostSessionInventoryResponse,
|
||||
type MultiplexSubscriptionCallbacks
|
||||
} from './remote-runtime-pty-transport-test-harness'
|
||||
import { REMOTE_RUNTIME_AUTO_RECOVERY_TIMEOUT_MS } from './remote-runtime-pty-recovery-state'
|
||||
import type { PtyTransport } from './pty-transport-types'
|
||||
|
||||
// These transport fixtures never read abandoned-launch ownership from the app store.
|
||||
const appStore = vi.hoisted(() => ({
|
||||
getState: vi.fn(() => {
|
||||
throw new Error('These transport fixtures must not read app-store ownership')
|
||||
})
|
||||
}))
|
||||
|
||||
vi.mock('@/store', () => ({ useAppStore: { getState: appStore.getState } }))
|
||||
|
||||
afterEach(() => {
|
||||
expect(appStore.getState).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
let subscriptionCallbacks: MultiplexSubscriptionCallbacks = null
|
||||
let resolvedPaneHandle = 'terminal-1'
|
||||
|
||||
const {
|
||||
runtimeCall,
|
||||
subscriptionSendBinary,
|
||||
latestSubscribePayload,
|
||||
subscribedTerminalHandles,
|
||||
emitSnapshot,
|
||||
resetRemoteRuntimeTransport
|
||||
} = createRemoteRuntimeTransportMocks({
|
||||
getCallbacks: () => subscriptionCallbacks,
|
||||
setCallbacks: (callbacks) => {
|
||||
subscriptionCallbacks = callbacks
|
||||
},
|
||||
getResolvedPaneHandle: () => resolvedPaneHandle,
|
||||
setResolvedPaneHandle: (handle) => {
|
||||
resolvedPaneHandle = handle
|
||||
}
|
||||
})
|
||||
|
||||
function hostSnapshot(status: 'pending-handle' | 'ready'): unknown {
|
||||
return {
|
||||
ok: true,
|
||||
result: {
|
||||
worktree: 'wt-1',
|
||||
publicationEpoch: 'epoch-1',
|
||||
snapshotVersion: 1,
|
||||
activeGroupId: null,
|
||||
activeTabId: 'host-tab-1::pane:1',
|
||||
activeTabType: 'terminal',
|
||||
tabs: [
|
||||
{
|
||||
type: 'terminal',
|
||||
id: 'host-tab-1::pane:1',
|
||||
parentTabId: 'host-tab-1',
|
||||
leafId: 'pane:1',
|
||||
title: 'Terminal',
|
||||
isActive: true,
|
||||
status,
|
||||
terminal: status === 'ready' ? 'terminal-1' : null
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function endLatestStream(): void {
|
||||
subscriptionCallbacks?.onResponse({
|
||||
ok: true,
|
||||
result: { type: 'end', streamId: latestSubscribePayload().streamId, code: 0 }
|
||||
})
|
||||
}
|
||||
|
||||
// #27178: a remote pane on a healthy link must not sit on "Reconnecting" after a worktree switch.
|
||||
describe('remote runtime PTY reattach on a healthy link', () => {
|
||||
beforeEach(() => {
|
||||
resetRemoteRuntimeTransport()
|
||||
vi.useFakeTimers()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
async function attachHostMirror(): Promise<PtyTransport> {
|
||||
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
|
||||
const transport = createRemoteRuntimePtyTransport('env-1', {
|
||||
worktreeId: 'wt-1',
|
||||
tabId: 'web-terminal-host-tab-1',
|
||||
leafId: 'pane:1'
|
||||
})
|
||||
transport.attach({ existingPtyId: 'remote:env-1@@terminal-1', callbacks: {} })
|
||||
await vi.waitFor(() => expect(subscriptionSendBinary).toHaveBeenCalled())
|
||||
emitSnapshot(latestSubscribePayload().streamId, 'initial')
|
||||
expect(transport.isConnected()).toBe(true)
|
||||
return transport
|
||||
}
|
||||
|
||||
it('reattaches a handle the host still reports ready within seconds of a stream end', async () => {
|
||||
const transport = await attachHostMirror()
|
||||
runtimeCall.mockImplementation(async (args: { method: string }) =>
|
||||
args.method === 'session.tabs.list' || args.method === 'session.tabs.activate'
|
||||
? readyHostSessionInventoryResponse('terminal-1')
|
||||
: { ok: true, result: {} }
|
||||
)
|
||||
|
||||
endLatestStream()
|
||||
await vi.advanceTimersByTimeAsync(2_000)
|
||||
|
||||
expect(subscribedTerminalHandles()).toEqual(['terminal-1', 'terminal-1'])
|
||||
emitSnapshot(latestSubscribePayload().streamId, 'reattached')
|
||||
expect(transport.isConnected()).toBe(true)
|
||||
expect(transport.getRecoveryState?.().phase).toBe('connected')
|
||||
transport.destroy?.()
|
||||
})
|
||||
|
||||
it('keeps reattaching a still-ready handle that ends repeatedly instead of parking', async () => {
|
||||
const transport = await attachHostMirror()
|
||||
runtimeCall.mockImplementation(async (args: { method: string }) =>
|
||||
args.method === 'session.tabs.list' || args.method === 'session.tabs.activate'
|
||||
? readyHostSessionInventoryResponse('terminal-1')
|
||||
: { ok: true, result: {} }
|
||||
)
|
||||
|
||||
for (let cycle = 0; cycle < 5; cycle += 1) {
|
||||
endLatestStream()
|
||||
await vi.advanceTimersByTimeAsync(10_000)
|
||||
expect(subscribedTerminalHandles()).toHaveLength(cycle + 2)
|
||||
emitSnapshot(latestSubscribePayload().streamId, `cycle ${cycle}`)
|
||||
expect(transport.isConnected()).toBe(true)
|
||||
expect(transport.getRecoveryState?.().phase).toBe('connected')
|
||||
}
|
||||
transport.destroy?.()
|
||||
})
|
||||
|
||||
it('retries a resubscribe whose host answered without a ready handle', async () => {
|
||||
const transport = await attachHostMirror()
|
||||
let hostReady = false
|
||||
runtimeCall.mockImplementation(async (args: { method: string }) =>
|
||||
args.method === 'session.tabs.list' || args.method === 'session.tabs.activate'
|
||||
? hostSnapshot(hostReady ? 'ready' : 'pending-handle')
|
||||
: { ok: true, result: {} }
|
||||
)
|
||||
|
||||
endLatestStream()
|
||||
await vi.advanceTimersByTimeAsync(16_000)
|
||||
expect(subscribedTerminalHandles()).toHaveLength(1)
|
||||
hostReady = true
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
|
||||
expect(subscribedTerminalHandles()).toEqual(['terminal-1', 'terminal-1'])
|
||||
transport.destroy?.()
|
||||
})
|
||||
|
||||
it('retries an attach whose host surface stayed pending past the bounded wait', async () => {
|
||||
let hostReady = false
|
||||
runtimeCall.mockImplementation(async (args: { method: string }) =>
|
||||
args.method === 'session.tabs.list' || args.method === 'session.tabs.activate'
|
||||
? hostSnapshot(hostReady ? 'ready' : 'pending-handle')
|
||||
: { ok: true, result: {} }
|
||||
)
|
||||
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
|
||||
const transport = createRemoteRuntimePtyTransport('env-1', {
|
||||
worktreeId: 'wt-1',
|
||||
tabId: 'web-terminal-host-tab-1',
|
||||
leafId: 'pane:1'
|
||||
})
|
||||
transport.attach({ existingPtyId: 'remote:env-1@@terminal-1', callbacks: {} })
|
||||
|
||||
await vi.advanceTimersByTimeAsync(16_000)
|
||||
expect(subscribedTerminalHandles()).toHaveLength(0)
|
||||
hostReady = true
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
|
||||
expect(subscribedTerminalHandles()).toEqual(['terminal-1'])
|
||||
transport.destroy?.()
|
||||
})
|
||||
|
||||
it('retries a resubscribe past repeated inventory failures once the host answers again', async () => {
|
||||
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
try {
|
||||
const transport = await attachHostMirror()
|
||||
let inventoryAvailable = false
|
||||
runtimeCall.mockImplementation(async (args: { method: string }) => {
|
||||
if (args.method === 'session.tabs.list' || args.method === 'session.tabs.activate') {
|
||||
if (!inventoryAvailable) {
|
||||
throw new Error('runtime command response timed out')
|
||||
}
|
||||
return readyHostSessionInventoryResponse('terminal-1')
|
||||
}
|
||||
return { ok: true, result: {} }
|
||||
})
|
||||
|
||||
endLatestStream()
|
||||
await vi.advanceTimersByTimeAsync(50_000)
|
||||
expect(subscribedTerminalHandles()).toHaveLength(1)
|
||||
inventoryAvailable = true
|
||||
await vi.advanceTimersByTimeAsync(35_000)
|
||||
|
||||
expect(subscribedTerminalHandles()).toEqual(['terminal-1', 'terminal-1'])
|
||||
transport.destroy?.()
|
||||
} finally {
|
||||
warn.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it('still latches a pane whose host never produces a handle', async () => {
|
||||
const transport = await attachHostMirror()
|
||||
runtimeCall.mockImplementation(async (args: { method: string }) =>
|
||||
args.method === 'session.tabs.list' || args.method === 'session.tabs.activate'
|
||||
? hostSnapshot('pending-handle')
|
||||
: { ok: true, result: {} }
|
||||
)
|
||||
|
||||
endLatestStream()
|
||||
await vi.advanceTimersByTimeAsync(REMOTE_RUNTIME_AUTO_RECOVERY_TIMEOUT_MS + 1_000)
|
||||
|
||||
expect(transport.getRecoveryState?.().phase).toBe('disconnected')
|
||||
expect(transport.isConnected()).toBe(false)
|
||||
expect(transport.retryRecovery?.()).toBe(true)
|
||||
transport.destroy?.()
|
||||
})
|
||||
})
|
||||
+4
-3
@@ -129,8 +129,9 @@ describe('initial host-mirror attach against a surface published as pending-hand
|
||||
await vi.advanceTimersByTimeAsync(HOST_SURFACE_ATTACH_WINDOW_MS)
|
||||
await expect(connect).resolves.toBeUndefined()
|
||||
|
||||
expect(transport.getRecoveryState?.().phase).toBe('recovering')
|
||||
expect(recoveryStates.at(-1)?.phase).toBe('recovering')
|
||||
// Why backoff: the wait ended with unknown liveness, so a retry is armed rather than parked (#27178).
|
||||
expect(transport.getRecoveryState?.().phase).toBe('backoff')
|
||||
expect(recoveryStates.at(-1)?.phase).toBe('backoff')
|
||||
|
||||
await vi.advanceTimersByTimeAsync(REMOTE_RUNTIME_AUTO_RECOVERY_TIMEOUT_MS)
|
||||
|
||||
@@ -222,7 +223,7 @@ describe('initial host-mirror attach against a surface published as pending-hand
|
||||
|
||||
expect(onError).not.toHaveBeenCalled()
|
||||
expect(countCalls('session.tabs.activate')).toBeGreaterThan(1)
|
||||
expect(transport.getRecoveryState?.().phase).toBe('recovering')
|
||||
expect(transport.getRecoveryState?.().phase).toBe('backoff')
|
||||
})
|
||||
|
||||
it('never reads an empty tab inventory as a closed remote terminal', async () => {
|
||||
|
||||
+20
-67
@@ -45,8 +45,9 @@ describe('createRemoteRuntimePtyTransport', () => {
|
||||
resetRemoteRuntimeTransport()
|
||||
})
|
||||
|
||||
it('caps unavailable host inventory at two recovery windows', async () => {
|
||||
it('keeps polling an unavailable host inventory on the backoff ladder until the deadline', async () => {
|
||||
vi.useFakeTimers()
|
||||
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
try {
|
||||
resolvedPaneHandle = 'terminal-stable'
|
||||
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
|
||||
@@ -78,76 +79,25 @@ describe('createRemoteRuntimePtyTransport', () => {
|
||||
result: { type: 'end', streamId: oldStreamId, code: 0 }
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(32_000)
|
||||
|
||||
const callsAfterTwoWindows = hostListCalls
|
||||
expect(callsAfterTwoWindows).toBeGreaterThan(25)
|
||||
expect(callsAfterTwoWindows).toBeLessThan(40)
|
||||
// Why: a parked third window made no attempt for the rest of the deadline (#27178).
|
||||
await vi.advanceTimersByTimeAsync(20_000)
|
||||
expect(hostListCalls).toBe(callsAfterTwoWindows)
|
||||
expect(transport.getRecoveryState?.().phase).toBe('recovering')
|
||||
expect(hostListCalls).toBeGreaterThan(callsAfterTwoWindows)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(REMOTE_RUNTIME_AUTO_RECOVERY_TIMEOUT_MS)
|
||||
expect(transport.getRecoveryState?.().phase).toBe('disconnected')
|
||||
const callsAtDeadline = hostListCalls
|
||||
await vi.advanceTimersByTimeAsync(60_000)
|
||||
expect(hostListCalls).toBe(callsAtDeadline)
|
||||
expect(subscribedTerminalHandles()).toEqual(['terminal-stable'])
|
||||
transport.destroy?.()
|
||||
} finally {
|
||||
warn.mockRestore()
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('reattaches from prior ready evidence when the trailing inventory poll fails', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
resolvedPaneHandle = 'terminal-stable'
|
||||
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
|
||||
const onPtyExit = vi.fn()
|
||||
const transport = createRemoteRuntimePtyTransport('hub-env', {
|
||||
worktreeId: 'wt-1',
|
||||
tabId: 'web-terminal-host-tab-1',
|
||||
leafId: 'pane:1',
|
||||
onPtyExit
|
||||
})
|
||||
|
||||
transport.attach({
|
||||
existingPtyId: 'remote:hub-env@@terminal-stable',
|
||||
callbacks: {}
|
||||
})
|
||||
await vi.waitFor(() => expect(subscriptionSendBinary).toHaveBeenCalled())
|
||||
const oldStreamId = latestSubscribePayload().streamId
|
||||
emitSnapshot(oldStreamId, 'before trailing inventory failure')
|
||||
|
||||
let hostListCalls = 0
|
||||
runtimeCall.mockImplementation(async (args: { method: string }) => {
|
||||
if (args.method !== 'session.tabs.list') {
|
||||
return { ok: true, result: {} }
|
||||
}
|
||||
hostListCalls += 1
|
||||
if (hostListCalls === 1) {
|
||||
return readyHostSessionInventoryResponse('terminal-stable')
|
||||
}
|
||||
throw new Error('final inventory poll failed')
|
||||
})
|
||||
|
||||
subscriptionCallbacks?.onResponse({
|
||||
ok: true,
|
||||
result: { type: 'end', streamId: oldStreamId, code: 0 }
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(16_000)
|
||||
|
||||
expect(hostListCalls).toBeGreaterThan(1)
|
||||
expect(hostListCalls).toBeLessThan(25)
|
||||
expect(subscribedTerminalHandles()).toEqual(['terminal-stable', 'terminal-stable'])
|
||||
expect(onPtyExit).not.toHaveBeenCalled()
|
||||
expect(transport.isConnected()).toBe(false)
|
||||
emitSnapshot(latestSubscribePayload().streamId, 'reattached from ready evidence')
|
||||
expect(transport.isConnected()).toBe(true)
|
||||
transport.destroy?.()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('strengthens repeated same-handle end recovery until it disconnects', async () => {
|
||||
it('paces repeated same-handle stream ends instead of disconnecting', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
resolvedPaneHandle = 'terminal-flapping'
|
||||
@@ -171,31 +121,34 @@ describe('createRemoteRuntimePtyTransport', () => {
|
||||
emitSnapshot(latestSubscribePayload().streamId, 'initial stream')
|
||||
|
||||
for (let cycle = 0; cycle < 2; cycle += 1) {
|
||||
const endingStreamId = latestSubscribePayload().streamId
|
||||
subscriptionCallbacks?.onResponse({
|
||||
ok: true,
|
||||
result: { type: 'end', streamId: endingStreamId, code: 0 }
|
||||
result: { type: 'end', streamId: latestSubscribePayload().streamId, code: 0 }
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(16_000)
|
||||
await vi.advanceTimersByTimeAsync(1_000)
|
||||
expect(subscribedTerminalHandles()).toHaveLength(cycle + 2)
|
||||
emitSnapshot(latestSubscribePayload().streamId, `same handle cycle ${cycle}`)
|
||||
expect(transport.isConnected()).toBe(true)
|
||||
}
|
||||
|
||||
// Why: the third end inside the flap window waits out a backoff before reusing the handle.
|
||||
subscriptionCallbacks?.onResponse({
|
||||
ok: true,
|
||||
result: { type: 'end', streamId: latestSubscribePayload().streamId, code: 0 }
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(16_000)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(500)
|
||||
expect(subscribedTerminalHandles()).toHaveLength(3)
|
||||
expect(transport.getRecoveryState?.().phase).toBe('backoff')
|
||||
await vi.advanceTimersByTimeAsync(1_500)
|
||||
expect(subscribedTerminalHandles()).toEqual([
|
||||
'terminal-flapping',
|
||||
'terminal-flapping',
|
||||
'terminal-flapping',
|
||||
'terminal-flapping'
|
||||
])
|
||||
expect(transport.isConnected()).toBe(false)
|
||||
await vi.advanceTimersByTimeAsync(REMOTE_RUNTIME_AUTO_RECOVERY_TIMEOUT_MS)
|
||||
expect(transport.getRecoveryState?.().phase).toBe('disconnected')
|
||||
emitSnapshot(latestSubscribePayload().streamId, 'paced reattach')
|
||||
expect(transport.isConnected()).toBe(true)
|
||||
expect(transport.getRecoveryState?.().phase).toBe('connected')
|
||||
transport.destroy?.()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
|
||||
+22
-20
@@ -120,7 +120,7 @@ describe('createRemoteRuntimePtyTransport', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('requires replacement after repeated same-handle stream-end flaps', async () => {
|
||||
it('paces same-handle stream-end flaps, then latches Reconnect once the ladder is spent', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
resolvedPaneHandle = 'terminal-flapping'
|
||||
@@ -145,44 +145,36 @@ describe('createRemoteRuntimePtyTransport', () => {
|
||||
: Promise.resolve({ ok: true, result: {} })
|
||||
)
|
||||
|
||||
for (let cycle = 0; cycle < 2; cycle += 1) {
|
||||
// Why: flap delays grow to 30s, so 31s per cycle covers every rung.
|
||||
for (let cycle = 0; cycle < 8; cycle += 1) {
|
||||
const endedStreamId = latestSubscribePayload().streamId
|
||||
subscriptionCallbacks?.onResponse({
|
||||
ok: true,
|
||||
result: { type: 'end', streamId: endedStreamId, code: 0 }
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(16_000)
|
||||
await vi.advanceTimersByTimeAsync(31_000)
|
||||
expect(subscribedTerminalHandles()).toHaveLength(cycle + 2)
|
||||
expect(latestSubscribePayload().terminal).toBe('terminal-flapping')
|
||||
emitSnapshot(latestSubscribePayload().streamId, `same handle ${cycle + 1}`)
|
||||
expect(transport.isConnected()).toBe(true)
|
||||
}
|
||||
|
||||
const condemnedStreamId = latestSubscribePayload().streamId
|
||||
// Why: an endless flap resets the deadline on every reattach, so it must stop pacing and fence the handle.
|
||||
subscriptionCallbacks?.onResponse({
|
||||
ok: true,
|
||||
result: { type: 'end', streamId: condemnedStreamId, code: 0 }
|
||||
result: { type: 'end', streamId: latestSubscribePayload().streamId, code: 0 }
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(16_000)
|
||||
|
||||
expect(subscribedTerminalHandles()).toEqual([
|
||||
'terminal-flapping',
|
||||
'terminal-flapping',
|
||||
'terminal-flapping'
|
||||
])
|
||||
expect(transport.getPtyId()).toBe('remote:env-1@@terminal-flapping')
|
||||
expect(transport.isConnected()).toBe(false)
|
||||
expect(onPtyExit).not.toHaveBeenCalled()
|
||||
|
||||
await vi.advanceTimersByTimeAsync(REMOTE_RUNTIME_AUTO_RECOVERY_TIMEOUT_MS)
|
||||
await vi.advanceTimersByTimeAsync(REMOTE_RUNTIME_AUTO_RECOVERY_TIMEOUT_MS + 16_000)
|
||||
expect(subscribedTerminalHandles()).toHaveLength(9)
|
||||
expect(transport.getRecoveryState?.().phase).toBe('disconnected')
|
||||
expect(subscribedTerminalHandles()).toHaveLength(3)
|
||||
expect(transport.getPtyId()).toBe('remote:env-1@@terminal-flapping')
|
||||
expect(onPtyExit).not.toHaveBeenCalled()
|
||||
transport.destroy?.()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('reuses prior ready evidence when the trailing inventory poll fails', async () => {
|
||||
it('reuses prior ready evidence fenced by an unanswered send when the trailing inventory poll fails', async () => {
|
||||
vi.useFakeTimers()
|
||||
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
try {
|
||||
@@ -204,6 +196,9 @@ describe('createRemoteRuntimePtyTransport', () => {
|
||||
let hostListCalls = 0
|
||||
runtimeCall.mockImplementation(
|
||||
async (args: { method: string; timeoutMs?: number }): Promise<unknown> => {
|
||||
if (args.method === 'terminal.send') {
|
||||
return new Promise(() => {})
|
||||
}
|
||||
if (args.method !== 'session.tabs.list') {
|
||||
return { ok: true, result: {} }
|
||||
}
|
||||
@@ -215,6 +210,13 @@ describe('createRemoteRuntimePtyTransport', () => {
|
||||
}
|
||||
)
|
||||
|
||||
// Why: an unanswered send fences early same-handle reuse, so only the window's ready evidence reattaches.
|
||||
void transport.sendInputAccepted?.('unanswered', 'driving')
|
||||
await vi.waitFor(() =>
|
||||
expect(runtimeCall).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ method: 'terminal.send' })
|
||||
)
|
||||
)
|
||||
subscriptionCallbacks?.onResponse({
|
||||
ok: true,
|
||||
result: { type: 'end', streamId: oldStreamId, code: 0 }
|
||||
|
||||
@@ -116,12 +116,15 @@ const REMOTE_RUNTIME_MAX_PENDING_QUERY_REPLIES = 64
|
||||
const HOST_SESSION_ATTACH_POLL_MS = 150
|
||||
const HOST_SESSION_POLL_MAX_MS = 1_000
|
||||
const HOST_SESSION_ATTACH_TIMEOUT_MS = 15_000
|
||||
const HOST_SESSION_INVENTORY_MAX_WINDOWS_PER_RECOVERY = 2
|
||||
const HOST_SESSION_SAME_HANDLE_END_REUSE_LIMIT = 2
|
||||
// Why its own constant: this fences how long an end-then-reattach on the same handle still counts as
|
||||
// one recovery, which is unrelated to how long auto-recovery keeps retrying. It read the recovery
|
||||
// budget before that budget became a derived value, and must not drift with it.
|
||||
const HOST_SESSION_SAME_HANDLE_END_REUSE_WINDOW_MS = 60_000
|
||||
// a flap, which is unrelated to how long auto-recovery keeps retrying. Only an end soon after the
|
||||
// reattach is a flap; a pane revealed again later must not pay the flap backoff (#27178).
|
||||
const HOST_SESSION_SAME_HANDLE_END_REUSE_WINDOW_MS = 10_000
|
||||
// Why: a flapping handle the host still reports ready is reattached ever more slowly before it is fenced.
|
||||
const HOST_SESSION_SAME_HANDLE_FLAP_DELAYS_MS = [
|
||||
1_000, 2_000, 4_000, 8_000, 15_000, 30_000
|
||||
] as const
|
||||
const MAX_SURFACED_TERMINAL_ERRORS = 8
|
||||
const TERMINAL_CREATE_RETRY_DELAYS_MS = [250, 500, 1000, 2000, 4000, 8000, 15_000, 30_000] as const
|
||||
|
||||
@@ -132,11 +135,6 @@ type PersistedHostPaneResolution =
|
||||
| { kind: 'absent' }
|
||||
| { kind: 'unavailable' }
|
||||
|
||||
type HostSessionHandleWaitResult = {
|
||||
handle: string | null | undefined
|
||||
inventoryFailed: boolean
|
||||
}
|
||||
|
||||
function stricterReplacementPolicy(
|
||||
left: HostHandleReplacementPolicy,
|
||||
right: HostHandleReplacementPolicy
|
||||
@@ -268,8 +266,6 @@ export function createRemoteRuntimePtyTransport(
|
||||
// Why: a spent auto-recovery window is the evidence that licenses reattaching the fenced handle; explicit retries must not erase it.
|
||||
let autoRecoveryWindowSpent = false
|
||||
let settleHostSessionAttachRetry: ((retry: boolean) => void) | null = null
|
||||
let resubscribeInventoryEpoch: number | null = null
|
||||
let resubscribeInventoryWindows = 0
|
||||
let sameHandleEndReuseHandle: string | null = null
|
||||
let sameHandleEndReuseCount = 0
|
||||
let sameHandleEndReuseAttachedAt: number | null = null
|
||||
@@ -400,14 +396,6 @@ export function createRemoteRuntimePtyTransport(
|
||||
replacementPolicy
|
||||
)
|
||||
}
|
||||
const beginResubscribeInventoryWindow = (recoveryEpoch: number): number => {
|
||||
if (resubscribeInventoryEpoch !== recoveryEpoch) {
|
||||
resubscribeInventoryEpoch = recoveryEpoch
|
||||
resubscribeInventoryWindows = 0
|
||||
}
|
||||
resubscribeInventoryWindows += 1
|
||||
return resubscribeInventoryWindows
|
||||
}
|
||||
const resetSameHandleEndReuse = (): void => {
|
||||
sameHandleEndReuseHandle = null
|
||||
sameHandleEndReuseCount = 0
|
||||
@@ -421,18 +409,33 @@ export function createRemoteRuntimePtyTransport(
|
||||
sameHandleEndReuseCount += 1
|
||||
sameHandleEndReuseAttachedAt = Date.now()
|
||||
}
|
||||
const replacementPolicyAfterStreamEnd = (targetHandle: string): HostHandleReplacementPolicy => {
|
||||
// Why: an end proves only attachment loss, so a flap is paced, not parked — parking waited out the whole deadline (#27178).
|
||||
// null once the ladder is spent: each reattach resets the recovery deadline, so an endless flap must fall back to the
|
||||
// replacement wait, whose deadline latches Reconnect.
|
||||
const sameHandleEndFlapDelayMs = (targetHandle: string): number | null => {
|
||||
if (
|
||||
sameHandleEndReuseHandle !== targetHandle ||
|
||||
sameHandleEndReuseAttachedAt === null ||
|
||||
Date.now() - sameHandleEndReuseAttachedAt >= HOST_SESSION_SAME_HANDLE_END_REUSE_WINDOW_MS
|
||||
) {
|
||||
resetSameHandleEndReuse()
|
||||
return 'prefer-replacement'
|
||||
return 0
|
||||
}
|
||||
return sameHandleEndReuseCount >= HOST_SESSION_SAME_HANDLE_END_REUSE_LIMIT
|
||||
? 'require-replacement'
|
||||
: 'prefer-replacement'
|
||||
const flaps = sameHandleEndReuseCount - HOST_SESSION_SAME_HANDLE_END_REUSE_LIMIT
|
||||
return flaps < 0 ? 0 : (HOST_SESSION_SAME_HANDLE_FLAP_DELAYS_MS[flaps] ?? null)
|
||||
}
|
||||
// Why: an unanswered send can still come back terminal_handle_stale, so its handle is not yet safe to reuse.
|
||||
const terminalSendsInFlight = new Map<string, number>()
|
||||
const trackTerminalSend = <T>(targetHandle: string, request: Promise<T>): Promise<T> => {
|
||||
terminalSendsInFlight.set(targetHandle, (terminalSendsInFlight.get(targetHandle) ?? 0) + 1)
|
||||
return request.finally(() => {
|
||||
const remaining = (terminalSendsInFlight.get(targetHandle) ?? 1) - 1
|
||||
if (remaining > 0) {
|
||||
terminalSendsInFlight.set(targetHandle, remaining)
|
||||
} else {
|
||||
terminalSendsInFlight.delete(targetHandle)
|
||||
}
|
||||
})
|
||||
}
|
||||
const adoptExecutionMetadata = (terminal: {
|
||||
executionHostId?: ExecutionHostId
|
||||
@@ -831,7 +834,8 @@ export function createRemoteRuntimePtyTransport(
|
||||
|
||||
async function waitForHostSessionHandleWithRecovery(
|
||||
hostTabId: string,
|
||||
isCurrent: () => boolean
|
||||
isCurrent: () => boolean,
|
||||
outcome: { boundedWaitEnded: boolean }
|
||||
): Promise<string | undefined | false> {
|
||||
let recoveryEpoch = recovery.isActive ? recovery.currentEpoch : undefined
|
||||
while (isCurrent()) {
|
||||
@@ -840,6 +844,7 @@ export function createRemoteRuntimePtyTransport(
|
||||
if (!isCurrent()) {
|
||||
return undefined
|
||||
}
|
||||
outcome.boundedWaitEnded = hostHandle === undefined
|
||||
if (recoveryEpoch !== undefined) {
|
||||
if (!recovery.isCurrent(recoveryEpoch)) {
|
||||
return undefined
|
||||
@@ -867,9 +872,9 @@ export function createRemoteRuntimePtyTransport(
|
||||
previousHandle: string,
|
||||
replacementPolicy: HostHandleReplacementPolicy,
|
||||
recoveryEpoch: number
|
||||
): Promise<HostSessionHandleWaitResult> {
|
||||
): Promise<string | null | undefined> {
|
||||
if (!worktreeId) {
|
||||
return { handle: null, inventoryFailed: false }
|
||||
return null
|
||||
}
|
||||
const worktree = toRuntimeWorktreeSelector(worktreeId)
|
||||
const startedAt = Date.now()
|
||||
@@ -878,13 +883,13 @@ export function createRemoteRuntimePtyTransport(
|
||||
let nextRequest: 'activate' | 'list' = 'activate'
|
||||
let lastRequestError: unknown = null
|
||||
let lastReadyHandle: string | null = null
|
||||
const finishBoundedWait = (): HostSessionHandleWaitResult => {
|
||||
const finishBoundedWait = (): string | undefined => {
|
||||
const effectivePolicy = stricterReplacementPolicy(
|
||||
replacementPolicy,
|
||||
getRecoveryReplacementPolicy(previousHandle)
|
||||
)
|
||||
if (effectivePolicy === 'prefer-replacement' && lastReadyHandle) {
|
||||
return { handle: lastReadyHandle, inventoryFailed: false }
|
||||
return lastReadyHandle
|
||||
}
|
||||
if (lastRequestError) {
|
||||
console.warn(
|
||||
@@ -893,7 +898,7 @@ export function createRemoteRuntimePtyTransport(
|
||||
)
|
||||
}
|
||||
// Why: a bounded wait without removal evidence is unknown liveness; keep the pane for a later snapshot to reattach.
|
||||
return { handle: undefined, inventoryFailed: lastRequestError !== null }
|
||||
return undefined
|
||||
}
|
||||
while (
|
||||
!destroyed &&
|
||||
@@ -938,7 +943,16 @@ export function createRemoteRuntimePtyTransport(
|
||||
getRecoveryReplacementPolicy(previousHandle)
|
||||
)
|
||||
if (nextHandle && (effectivePolicy === 'reuse' || nextHandle !== previousHandle)) {
|
||||
return { handle: nextHandle, inventoryFailed: false }
|
||||
return nextHandle
|
||||
}
|
||||
// Why: a host still reporting the ended handle ready kept its PTY; waiting out the window for a
|
||||
// replacement held the banner up ~15s after every worktree switch (#27178).
|
||||
if (
|
||||
nextHandle === previousHandle &&
|
||||
effectivePolicy === 'prefer-replacement' &&
|
||||
!terminalSendsInFlight.has(previousHandle)
|
||||
) {
|
||||
return nextHandle
|
||||
}
|
||||
if (request === 'list') {
|
||||
if (hasHostSessionTerminalSurface(listed, hostTabId)) {
|
||||
@@ -947,7 +961,7 @@ export function createRemoteRuntimePtyTransport(
|
||||
nextRequest = 'activate'
|
||||
}
|
||||
} else if (hostSnapshotAffirmsWorktreeContents(listed)) {
|
||||
return { handle: null, inventoryFailed: false }
|
||||
return null
|
||||
}
|
||||
// Why: an unpublished frame from a relaunched host is unknown liveness, so keep asking.
|
||||
} else {
|
||||
@@ -970,7 +984,7 @@ export function createRemoteRuntimePtyTransport(
|
||||
await new Promise((resolve) => setTimeout(resolve, Math.min(pollMs, remainingMs)))
|
||||
pollMs = Math.min(pollMs * 2, HOST_SESSION_POLL_MAX_MS)
|
||||
}
|
||||
return { handle: undefined, inventoryFailed: false }
|
||||
return undefined
|
||||
}
|
||||
|
||||
// Why: the reconnect button and a parked external trigger revive a pane the same way — replay whichever entry point opened it.
|
||||
@@ -1040,19 +1054,29 @@ export function createRemoteRuntimePtyTransport(
|
||||
(expectedAttachGeneration === undefined || expectedAttachGeneration === attachGeneration) &&
|
||||
(expectedLifecycleEpoch === undefined || expectedLifecycleEpoch === lifecycleEpoch)
|
||||
const hostTabId = toHostSessionTabId(tabId)
|
||||
const hostHandle = await waitForHostSessionHandleWithRecovery(hostTabId, isCurrent)
|
||||
const waitOutcome = { boundedWaitEnded: false }
|
||||
const hostHandle = await waitForHostSessionHandleWithRecovery(hostTabId, isCurrent, waitOutcome)
|
||||
if (!isCurrent()) {
|
||||
return undefined
|
||||
}
|
||||
if (hostHandle === undefined) {
|
||||
connecting = false
|
||||
// Why: no handle and no removal evidence is unknown liveness, so own an epoch and keep an unarmed retry parked for the
|
||||
// banner and online/resume to fire — a silent 'connecting' renders no banner and retryRecovery() refuses to run.
|
||||
if (recovery.currentPhase !== 'disconnected') {
|
||||
const recoveryEpoch = recovery.isActive ? recovery.currentEpoch : recovery.begin()
|
||||
recovery.parkRetryForExternalTrigger(recoveryEpoch, () => {
|
||||
replayLastTransportEntryPoint()
|
||||
})
|
||||
// Why: no handle and no removal evidence is unknown liveness, so own an epoch — a silent 'connecting' renders no banner
|
||||
// and retryRecovery() refuses to run. Schedule, not park: a healthy link fires no online/resume, so a parked retry sat
|
||||
// out the whole deadline (#27178); schedule() still leaves it revivable once the deadline latches.
|
||||
const replayAttach = (): void => {
|
||||
replayLastTransportEntryPoint()
|
||||
}
|
||||
if (recovery.currentPhase === 'disconnected') {
|
||||
// Why: a bounded wait still in flight at the cutoff stays revivable; a settled single-shot retry wait does not.
|
||||
if (waitOutcome.boundedWaitEnded) {
|
||||
recovery.parkRetryAfterDeadline(replayAttach)
|
||||
}
|
||||
} else {
|
||||
recovery.schedule(
|
||||
recovery.isActive ? recovery.currentEpoch : recovery.begin(),
|
||||
replayAttach
|
||||
)
|
||||
}
|
||||
emitRecoveryState()
|
||||
return undefined
|
||||
@@ -1691,12 +1715,15 @@ export function createRemoteRuntimePtyTransport(
|
||||
return false
|
||||
}
|
||||
// Why: acknowledged sends order behind pending debounce text but must not collapse large paste back into one remote RPC.
|
||||
const result = await callRuntime<{ send: RuntimeTerminalSend }>('terminal.send', {
|
||||
terminal: targetHandle,
|
||||
text: chunk,
|
||||
client: { id: clientId, type: 'desktop' },
|
||||
...(desiredViewport ? { viewport: desiredViewport, claimViewport: true as const } : {})
|
||||
})
|
||||
const result = await trackTerminalSend(
|
||||
targetHandle,
|
||||
callRuntime<{ send: RuntimeTerminalSend }>('terminal.send', {
|
||||
terminal: targetHandle,
|
||||
text: chunk,
|
||||
client: { id: clientId, type: 'desktop' },
|
||||
...(desiredViewport ? { viewport: desiredViewport, claimViewport: true as const } : {})
|
||||
})
|
||||
)
|
||||
if (result.send.accepted !== true) {
|
||||
return false
|
||||
}
|
||||
@@ -1756,12 +1783,15 @@ export function createRemoteRuntimePtyTransport(
|
||||
queuePendingClaimInput(text, queryReply)
|
||||
return true
|
||||
}
|
||||
void callRuntime<{ send: RuntimeTerminalSend }>('terminal.send', {
|
||||
terminal: targetHandle,
|
||||
text,
|
||||
client: { id: clientId, type: 'desktop' },
|
||||
...(desiredViewport ? { viewport: desiredViewport, claimViewport: true as const } : {})
|
||||
})
|
||||
void trackTerminalSend(
|
||||
targetHandle,
|
||||
callRuntime<{ send: RuntimeTerminalSend }>('terminal.send', {
|
||||
terminal: targetHandle,
|
||||
text,
|
||||
client: { id: clientId, type: 'desktop' },
|
||||
...(desiredViewport ? { viewport: desiredViewport, claimViewport: true as const } : {})
|
||||
})
|
||||
)
|
||||
.then((result) => {
|
||||
if (
|
||||
connected &&
|
||||
@@ -2060,20 +2090,28 @@ export function createRemoteRuntimePtyTransport(
|
||||
): Promise<void> {
|
||||
if (tabId && isWebTerminalSurfaceTabId(tabId)) {
|
||||
const hostTabId = toHostSessionTabId(tabId)
|
||||
const inventoryWindow = beginResubscribeInventoryWindow(recoveryEpoch)
|
||||
const waitResult = await waitForResubscribeHostSessionHandle(
|
||||
const nextHandle = await waitForResubscribeHostSessionHandle(
|
||||
hostTabId,
|
||||
previousHandle,
|
||||
replacementPolicy,
|
||||
recoveryEpoch
|
||||
)
|
||||
const nextHandle = waitResult.handle
|
||||
if (
|
||||
destroyed ||
|
||||
!connected ||
|
||||
handle !== previousHandle ||
|
||||
!recovery.isCurrent(recoveryEpoch)
|
||||
) {
|
||||
if (destroyed || !connected || handle !== previousHandle) {
|
||||
return
|
||||
}
|
||||
const retryResubscribe = (nextEpoch: number): void => {
|
||||
// Why: a retry is a fresh attempt, not a repeated stale send, so it takes over this epoch's snapshot wait.
|
||||
clearPublishedHandleWait()
|
||||
scheduleResubscribeAfterTransportClose(
|
||||
handle ? getRecoveryReplacementPolicy(handle) : 'reuse',
|
||||
nextEpoch
|
||||
)
|
||||
}
|
||||
if (!recovery.isCurrent(recoveryEpoch)) {
|
||||
if (nextHandle === undefined && recovery.ownsEpoch(recoveryEpoch)) {
|
||||
// Why: an attempt still in flight when the deadline latched must stay revivable by online/resume.
|
||||
recovery.parkRetryAfterDeadline(retryResubscribe)
|
||||
}
|
||||
return
|
||||
}
|
||||
if (nextHandle === undefined) {
|
||||
@@ -2081,24 +2119,14 @@ export function createRemoteRuntimePtyTransport(
|
||||
replacementPolicy,
|
||||
getRecoveryReplacementPolicy(previousHandle)
|
||||
)
|
||||
if (
|
||||
effectivePolicy !== 'require-replacement' &&
|
||||
waitResult.inventoryFailed &&
|
||||
inventoryWindow < HOST_SESSION_INVENTORY_MAX_WINDOWS_PER_RECOVERY
|
||||
) {
|
||||
throw Object.assign(new Error('Remote runtime session inventory polling failed.'), {
|
||||
code: 'remote_runtime_unavailable'
|
||||
})
|
||||
if (effectivePolicy === 'require-replacement') {
|
||||
// Why: re-polling cannot mint a replacement for a stale handle; park for the host's published one or online/resume/reconnect.
|
||||
recovery.parkRetryForExternalTrigger(recoveryEpoch, retryResubscribe)
|
||||
return
|
||||
}
|
||||
// Why: liveness is unknown, so auto-retry stops here; keep an unarmed retry parked for online/resume/reconnect to fire.
|
||||
recovery.parkRetryForExternalTrigger(recoveryEpoch, (nextEpoch) => {
|
||||
// Why: an external trigger is a fresh attempt, not a repeated stale send, so it takes over this epoch's snapshot wait.
|
||||
clearPublishedHandleWait()
|
||||
scheduleResubscribeAfterTransportClose(
|
||||
handle ? getRecoveryReplacementPolicy(handle) : 'reuse',
|
||||
nextEpoch
|
||||
)
|
||||
})
|
||||
// Why: liveness is unknown, so keep retrying on the backoff ladder until the deadline — a link healthy
|
||||
// enough to answer fires no online/resume, so a parked retry sat out the whole deadline (#27178).
|
||||
recovery.schedule(recoveryEpoch, retryResubscribe)
|
||||
return
|
||||
}
|
||||
if (!nextHandle) {
|
||||
@@ -2250,7 +2278,10 @@ export function createRemoteRuntimePtyTransport(
|
||||
})
|
||||
}
|
||||
|
||||
function scheduleResubscribeAfterBackoff(): void {
|
||||
function scheduleResubscribeAfterBackoff(
|
||||
replacementPolicy: HostHandleReplacementPolicy = 'reuse',
|
||||
delayMs?: number
|
||||
): void {
|
||||
if (destroyed || !connected || !handle) {
|
||||
return
|
||||
}
|
||||
@@ -2261,9 +2292,13 @@ export function createRemoteRuntimePtyTransport(
|
||||
viewportBatcher.clear()
|
||||
clearPendingViewportClaim()
|
||||
}
|
||||
recovery.schedule(recoveryEpoch, (nextEpoch) => {
|
||||
scheduleResubscribeAfterTransportClose('reuse', nextEpoch)
|
||||
})
|
||||
recovery.schedule(
|
||||
recoveryEpoch,
|
||||
(nextEpoch) => {
|
||||
scheduleResubscribeAfterTransportClose(replacementPolicy, nextEpoch)
|
||||
},
|
||||
delayMs
|
||||
)
|
||||
}
|
||||
|
||||
async function subscribeToHandle(
|
||||
@@ -2390,10 +2425,15 @@ export function createRemoteRuntimePtyTransport(
|
||||
multiplexedStream = null
|
||||
multiplexedStreamHandle = null
|
||||
clearPendingViewportClaim()
|
||||
// Why: a legacy bare end or waiter failure proves only attachment loss; bounded same-handle reuse preserves the tab without looping forever.
|
||||
scheduleResubscribeAfterTransportClose(
|
||||
replacementPolicyAfterStreamEnd(subscribedHandle)
|
||||
)
|
||||
// Why: a legacy bare end or waiter failure proves only attachment loss; reuse the same handle, pacing flaps.
|
||||
const flapDelayMs = sameHandleEndFlapDelayMs(subscribedHandle)
|
||||
if (flapDelayMs === null) {
|
||||
scheduleResubscribeAfterTransportClose('require-replacement')
|
||||
} else if (flapDelayMs > 0) {
|
||||
scheduleResubscribeAfterBackoff('prefer-replacement', flapDelayMs)
|
||||
} else {
|
||||
scheduleResubscribeAfterTransportClose('prefer-replacement')
|
||||
}
|
||||
return
|
||||
}
|
||||
unregisterShutdownHandlers(subscribedPtyId)
|
||||
|
||||
Reference in New Issue
Block a user