Merge remote-tracking branch 'origin/salvage/ssh-restore-required-token' into adhoc/ssh-sweep-combined

This commit is contained in:
Neil
2026-09-01 16:58:20 -07:00
22 changed files with 743 additions and 135 deletions
+7 -2
View File
@@ -1,5 +1,10 @@
export const SSH_SESSION_EXPIRED_ERROR = 'SSH_SESSION_EXPIRED'
export const SSH_PTY_IDENTITY_MISMATCH_ERROR = 'SSH_PTY_IDENTITY_MISMATCH'
import { SSH_PTY_IDENTITY_MISMATCH_ERROR } from '../../shared/ssh-pty-failure-tokens'
export {
SSH_PTY_IDENTITY_MISMATCH_ERROR,
SSH_SESSION_EXPIRED_ERROR,
SSH_SOURCE_RESTORE_REQUIRED_ERROR
} from '../../shared/ssh-pty-failure-tokens'
export function isSshPtyNotFoundError(error: unknown): boolean {
const message = error instanceof Error ? error.message : String(error)
@@ -1,5 +1,5 @@
import { describe, expect, it, vi } from 'vitest'
import { SSH_SESSION_EXPIRED_ERROR } from './ssh-pty-errors'
import { SSH_SOURCE_RESTORE_REQUIRED_ERROR } from './ssh-pty-errors'
import { SshPtyProvider } from './ssh-pty-provider'
describe('SSH PTY provider session reattach incarnation', () => {
@@ -30,7 +30,7 @@ describe('SSH PTY provider session reattach incarnation', () => {
)
})
it('fails closed when generic reattach requires source restoration', async () => {
it('fails closed after bounded generic reattach restore retries', async () => {
const mux = {
request: vi.fn().mockResolvedValue({
incarnationId: 'incarnation-reattached',
@@ -45,7 +45,9 @@ describe('SSH PTY provider session reattach incarnation', () => {
const provider = new SshPtyProvider('conn-1', mux as never)
await expect(provider.spawn({ cols: 80, rows: 24, sessionId: 'pty-old' })).rejects.toThrow(
`${SSH_SESSION_EXPIRED_ERROR}: pty-old`
`${SSH_SOURCE_RESTORE_REQUIRED_ERROR}: pty-old`
)
// The retry is what repairs a retired delivery; without it the pane needs a respawn to recover.
expect(mux.request).toHaveBeenCalledTimes(3)
})
})
+1 -1
View File
@@ -14,10 +14,10 @@ import { spawnFreshSshPty } from './ssh-agent-session-create-operation'
import { mapSshPtyProcessList } from './ssh-agent-session-process-list'
import {
requestSshPtyAttach,
reattachSshPtySessionForSpawn,
type PtySourceRecoveryRequest,
type SshPtyAttachResult
} from './ssh-pty-session-reattach'
import { reattachSshPtySessionForSpawn } from './ssh-pty-session-reattach-for-spawn'
import { buildSshPtySpawnRequest } from './ssh-pty-spawn-request'
import { SshPtySpawnExitRaceTracker } from './ssh-pty-spawn-exit-race'
import { SshAgentSessionCapabilities } from './ssh-agent-session-capabilities'
@@ -2,6 +2,7 @@ import { describe, expect, it, vi } from 'vitest'
import {
SSH_PTY_IDENTITY_MISMATCH_ERROR,
SSH_SESSION_EXPIRED_ERROR,
SSH_SOURCE_RESTORE_REQUIRED_ERROR,
isSshPtyAbsentFromRelayError
} from './ssh-pty-errors'
import { SshPtyProvider } from './ssh-pty-provider'
@@ -73,6 +74,9 @@ describe('SSH PTY relay absence verdict', () => {
)
expect(isSshPtyAbsentFromRelayError(rejection)).toBe(false)
expect((rejection as Error).message).toContain(SSH_SESSION_EXPIRED_ERROR)
// restoreRequired is a SUCCESSFUL answer proving the PTY live, so it must not carry the one
// token that authorises a respawn.
expect((rejection as Error).message).toBe(`${SSH_SOURCE_RESTORE_REQUIRED_ERROR}: pty-1`)
expect((rejection as Error).message).not.toContain(SSH_SESSION_EXPIRED_ERROR)
})
})
@@ -0,0 +1,47 @@
import { toRelaySshPtyId } from './ssh-pty-id'
import { SSH_SOURCE_RESTORE_REQUIRED_ERROR } from './ssh-pty-errors'
import type { PtySpawnResult } from './types'
import {
reattachSshPtySessionWithExitFence,
type SshPtyReattachResult
} from './ssh-pty-session-reattach'
/**
* The full reattach path a spawn takes when it carries a sessionId: fence the exit race, reject a
* session whose source the relay could not restore, and commit or roll back the source-activation
* lease.
*
* Lives beside `reattachSshPtySession` rather than in SshPtyProvider.spawn so the lease's commit
* and rollback stay in one place — a caller that only wrapped the fence could return without
* committing and silently leak the activation.
*/
export async function reattachSshPtySessionForSpawn(
args: Parameters<typeof reattachSshPtySessionWithExitFence>[0] & {
acceptLivePty: (relayPtyId: string) => void
}
): Promise<PtySpawnResult> {
let result: SshPtyReattachResult | undefined
try {
result = await reattachSshPtySessionWithExitFence(args)
if (result.sourceRecovery?.status === 'restoreRequired') {
// A restoreRequired answer is a SUCCESSFUL RPC that proved the PTY alive — only its output
// delivery was retired. Never raise SSH_SESSION_EXPIRED here: that token authorises the
// renderer to retire the pane binding and cold-start a fresh agent over the same worktree,
// orphaning live remote work (docs/reference/ssh-execution-boundary.md).
throw new Error(
`${SSH_SOURCE_RESTORE_REQUIRED_ERROR}: ${toRelaySshPtyId(args.connectionId, result.id)}`
)
}
args.acceptLivePty(result.id)
result.sourceActivationLease?.commit()
const {
sourceActivationLease: _lease,
sourceRecovery: _sourceRecovery,
...spawnResult
} = result
return spawnResult
} catch (error) {
result?.sourceActivationLease?.rollback()
throw error
}
}
@@ -0,0 +1,87 @@
import { describe, expect, it, vi } from 'vitest'
import { reattachSshPtySession } from './ssh-pty-session-reattach'
const restoreRequired = {
incarnationId: 'incarnation-live',
sourceRecovery: { status: 'restoreRequired', reason: 'checkpointUnavailable' }
}
describe('SSH PTY session reattach restore retry', () => {
it('reattaches after restoreRequired and returns replay from the live PTY', async () => {
const request = vi
.fn()
.mockResolvedValueOnce(restoreRequired)
.mockResolvedValueOnce({ incarnationId: 'incarnation-live', replay: 'scrollback' })
const result = await reattachSshPtySession({
mux: { request } as never,
connectionId: 'conn-1',
sessionId: 'pty-live',
options: { cols: 80, rows: 24, sessionId: 'pty-live' }
})
expect(request).toHaveBeenCalledTimes(2)
expect(result).toMatchObject({
id: 'ssh:conn-1@@pty-live',
isReattach: true,
replay: 'scrollback',
incarnationId: 'incarnation-live'
})
expect(result.sourceRecovery).toBeUndefined()
})
it('rolls back a provisional source lease before retrying', async () => {
const sourceActivation = {
status: 'pending',
clientGeneration: 1,
ownerGeneration: 1,
ptyIncarnation: 'incarnation-live',
deliveryToken: 'token-1',
checkpointSourceEndSu: 0,
recoveryEndSu: 0
}
const request = vi
.fn()
.mockResolvedValueOnce({ ...restoreRequired, sourceActivation })
.mockResolvedValueOnce({ incarnationId: 'incarnation-live', replay: 'scrollback' })
const rollback = vi.fn().mockResolvedValue(true)
const installSourceActivation = vi.fn().mockReturnValue({ commit: vi.fn(), rollback })
const result = await reattachSshPtySession({
mux: { request } as never,
connectionId: 'conn-1',
sessionId: 'pty-live',
options: { cols: 80, rows: 24, sessionId: 'pty-live' },
installSourceActivation
})
expect(rollback).toHaveBeenCalledOnce()
expect(result.replay).toBe('scrollback')
})
it('stops before retrying when stale-delivery cancellation is unconfirmed', async () => {
const sourceActivation = {
status: 'pending',
clientGeneration: 1,
ownerGeneration: 1,
ptyIncarnation: 'incarnation-live',
deliveryToken: 'token-1',
checkpointSourceEndSu: 0,
recoveryEndSu: 0
}
const request = vi.fn().mockResolvedValue({ ...restoreRequired, sourceActivation })
const rollback = vi.fn().mockResolvedValue(false)
const result = await reattachSshPtySession({
mux: { request } as never,
connectionId: 'conn-1',
sessionId: 'pty-live',
options: { cols: 80, rows: 24, sessionId: 'pty-live' },
installSourceActivation: vi.fn().mockReturnValue({ commit: vi.fn(), rollback })
})
expect(request).toHaveBeenCalledOnce()
expect(rollback).toHaveBeenCalledOnce()
expect(result.sourceRecovery).toMatchObject({ status: 'restoreRequired' })
})
})
+47 -57
View File
@@ -28,7 +28,7 @@ export type SshPtyAttachResult = {
sourceActivationLease?: SshPtyReceivingActivationLease
}
type SshPtyReattachResult = PtySpawnResult & {
export type SshPtyReattachResult = PtySpawnResult & {
sourceRecovery?: PtySourceRecoveryResult
sourceActivationLease?: SshPtyReceivingActivationLease
}
@@ -172,6 +172,8 @@ function sameSourceActivation(
export type { PtySourceRecoveryRequest }
const SSH_PTY_RESTORE_REQUIRED_ATTACH_ATTEMPTS = 3
export async function reattachSshPtySession(args: {
mux: SshChannelMultiplexer
connectionId: string
@@ -189,27 +191,51 @@ export async function reattachSshPtySession(args: {
// Why: expected pane identity prevents a reused relay id from attaching the wrong shell.
const expectedPaneKey = args.options.paneKey ?? args.options.env?.ORCA_PANE_KEY
const expectedTabId = args.options.tabId ?? args.options.env?.ORCA_TAB_ID
const attachResult = await requestSshPtyAttach({
mux: args.mux,
relayPtyId: relaySessionId,
params: {
id: relaySessionId,
cols: args.options.cols,
rows: args.options.rows,
suppressReplayNotification: true,
// A reattach always paints into a NEW terminal: a reconnect bumps tab.generation, which is
// the pane's React key, so TerminalPane remounts and the old xterm is disposed with its
// buffer. Without this the relay sees a delivery still open under our unchanged client id,
// answers "you already have this", and the pane stays blank until new output arrives.
requireReplay: true,
...(expectedPaneKey ? { expectedPaneKey } : {}),
...(expectedTabId ? { expectedTabId } : {})
},
installSourceActivation: args.installSourceActivation,
rememberPtyIncarnation: args.rememberPtyIncarnation
})
let attachResult: SshPtyAttachResult
for (let attempt = 1; ; attempt += 1) {
attachResult = await requestSshPtyAttach({
mux: args.mux,
relayPtyId: relaySessionId,
params: {
id: relaySessionId,
cols: args.options.cols,
rows: args.options.rows,
suppressReplayNotification: true,
// A reattach always paints into a NEW terminal: a reconnect bumps tab.generation, which is
// the pane's React key, so TerminalPane remounts and the old xterm is disposed with its
// buffer. Without this the relay sees a delivery still open under our unchanged client id,
// answers "you already have this", and the pane stays blank until new output arrives.
requireReplay: true,
...(expectedPaneKey ? { expectedPaneKey } : {}),
...(expectedTabId ? { expectedTabId } : {})
},
installSourceActivation: args.installSourceActivation,
rememberPtyIncarnation: args.rememberPtyIncarnation
})
if (
attachResult.sourceRecovery?.status !== 'restoreRequired' ||
attempt >= SSH_PTY_RESTORE_REQUIRED_ATTACH_ATTEMPTS
) {
break
}
// restoreRequired proves the PTY is live and only its delivery was retired, so retrying the
// attach is the repair. Cancel the provisional delivery first, or the retry races the lease
// it is meant to replace; if cancellation cannot be confirmed, stop rather than stack leases.
const staleDeliveryCanceled = attachResult.sourceActivationLease
? await attachResult.sourceActivationLease.rollback()
: true
if (!staleDeliveryCanceled) {
console.warn(
`[ssh-pty] pty.attach for ${args.sessionId} could not confirm stale delivery cancellation; failing closed`
)
break
}
console.warn(
`[ssh-pty] pty.attach for ${args.sessionId} requires restore (${attachResult.sourceRecovery.reason}), retrying attach ${attempt + 1}/${SSH_PTY_RESTORE_REQUIRED_ATTACH_ATTEMPTS}`
)
}
console.warn(
`[ssh-pty] pty.attach succeeded for ${args.sessionId}, replay=${!!attachResult.replay}`
`[ssh-pty] pty.attach ${attachResult.sourceRecovery?.status === 'restoreRequired' ? 'still requires restore' : 'succeeded'} for ${args.sessionId}, replay=${!!attachResult.replay}`
)
return {
id: toAppSshPtyId(args.connectionId, relaySessionId),
@@ -267,39 +293,3 @@ export async function reattachSshPtySessionWithExitFence(
args.exitRaceTracker.finish(operation)
}
}
/**
* The full reattach path a spawn takes when it carries a sessionId: fence the
* exit race, reject a session the relay can no longer restore, and commit or
* roll back the source-activation lease.
*
* Lives here rather than in SshPtyProvider.spawn so the lease's commit and
* rollback stay in one place — a caller that only wrapped the fence could
* return without committing and silently leak the activation.
*/
export async function reattachSshPtySessionForSpawn(
args: Parameters<typeof reattachSshPtySessionWithExitFence>[0] & {
acceptLivePty: (relayPtyId: string) => void
}
): Promise<PtySpawnResult> {
let result: SshPtyReattachResult | undefined
try {
result = await reattachSshPtySessionWithExitFence(args)
if (result.sourceRecovery?.status === 'restoreRequired') {
throw new Error(
`${SSH_SESSION_EXPIRED_ERROR}: ${toRelaySshPtyId(args.connectionId, result.id)}`
)
}
args.acceptLivePty(result.id)
result.sourceActivationLease?.commit()
const {
sourceActivationLease: _lease,
sourceRecovery: _sourceRecovery,
...spawnResult
} = result
return spawnResult
} catch (error) {
result?.sourceActivationLease?.rollback()
throw error
}
}
+2
View File
@@ -9,6 +9,8 @@ import type { DispatcherClientWriter, SinkWriteSettlement } from './dispatcher-c
export type RequestContext = {
clientId: number
/** Transport incarnation of the client id; changes when the primary socket is replaced. */
transportGeneration?: number
isStale: () => boolean
signal?: AbortSignal
sessionIdentity?: RelayClientSessionIdentity
+2
View File
@@ -104,6 +104,7 @@ export abstract class RelayDispatcherRpcRouting extends RelayDispatcherFrameCode
}
const context: RequestContext = {
clientId: client.id,
transportGeneration: gen,
isStale: () =>
client.generation !== gen || !this.clients.has(client.id) || abortController.signal.aborted,
signal: abortController.signal,
@@ -179,6 +180,7 @@ export abstract class RelayDispatcherRpcRouting extends RelayDispatcherFrameCode
const gen = client.generation
handler(notif.params ?? {}, {
clientId: client.id,
transportGeneration: gen,
isStale: () => client.generation !== gen || !this.clients.has(client.id),
sessionIdentity: client.sessionIdentity,
onResponseSettled: () => {
@@ -0,0 +1,42 @@
import type { PtySourceDeliveryIdentity } from '../shared/pty-source-credit-contract'
import type { PtySourceRecoveryCheckpoint } from '../shared/pty-source-recovery-contract'
import type { PtySourceReceivingActivation } from '../shared/pty-source-receiving-activation'
export type RelayPtySourceDeliveryRecord = {
clientId: number
/** Transport incarnation of clientId; undefined preserves legacy harness behavior. */
clientTransportGeneration?: number
identity: PtySourceDeliveryIdentity
sourceActivation: PtySourceReceivingActivation
displayEnd: number
activating: boolean
activationRecoveryRequest: PtySourceRecoveryCheckpoint | null
sealed: boolean
legacyExitAccepted: boolean
sourceExitState: 'idle' | 'pending' | 'published'
sending: boolean
turnFrames: number
turnSourceSu: number
turnScheduled: boolean
sendWaiters: Set<() => void>
recoveryCheckpointSourceEndSu: number | null
recoveryEndSu: number | null
recoveryCompletionPending: boolean
restoreRequired: boolean
rotationPending: boolean
}
/**
* Match a delivery's owner including its socket incarnation. Keying on clientId alone lets a
* re-bound primary channel act on — or inherit — a delivery bound to the previous, dead sink;
* the dispatcher's own publication ledger has always keyed on `${id}:${generation}`.
*/
export function sameClientTransport(
record: { clientId: number; clientTransportGeneration?: number },
context: { clientId: number; transportGeneration?: number }
): boolean {
return (
record.clientId === context.clientId &&
record.clientTransportGeneration === context.transportGeneration
)
}
+51 -39
View File
@@ -7,12 +7,17 @@ import type { PtySourceReceivingActivation } from '../shared/pty-source-receivin
import {
createPtySourceReceivingActivation,
pendingPtySourceRecoveryResult,
registerCanceledPtySourceRetirement,
registerPtySourceActivationSettlement,
samePtySourceRecoveryRequest
} from './relay-pty-source-activation'
import {
publishPtySourceRestoreRequired,
requirePtySourceRestore,
type PtySourceRestoreDeps
} from './relay-pty-source-restore-required'
import {
RelayPtySourceSendScheduler,
sameClientTransport,
type RelayPtySourceDeliveryRecord,
type RelayPtySourcePublicationCounters
} from './relay-pty-source-send-scheduler'
@@ -65,18 +70,30 @@ export class RelayPtySourcePublication {
context: RequestContext | undefined,
recovery?: PtySourceRecoveryRequest
): false | 'opened' | 'rotated' | 'existing' | PtySourceRecoveryResult {
if (!context?.onResponseSettled) {
this.sender.releaseRotationFence(this.deliveries.get(id))
let current = this.deliveries.get(id)
// Every bail-out below releases or retires `current`, but a superseded request may find the
// delivery its own replacement opened. Releasing that fence resumes a send the replacement is
// still rotating; retiring it blanks the pane that owns it. So each acts only on a record this
// transport still owns. Safe because pty.attach arms the fence (waitForPendingSend) and
// releases it here in the same handler: a record that changed underneath was rotated away, and
// its fence died with it.
const ownsCurrent =
context !== undefined && current !== undefined ? sameClientTransport(current, context) : false
if (!context?.onResponseSettled || context.isStale()) {
if (ownsCurrent) {
this.sender.releaseRotationFence(current)
}
return false
}
const mode = this.session.deliveryMode(context.clientId)
let current = this.deliveries.get(id)
if (mode === 'unadmitted' || mode === 'subscriber') {
this.sender.releaseRotationFence(current)
if (ownsCurrent) {
this.sender.releaseRotationFence(current)
}
return false
}
if (mode === 'legacy-owner') {
if (current) {
if (current && ownsCurrent) {
this.session.cancelDelivery(current.identity, 'source-credit-disabled')
this.sender.wakeSendWaiters(current)
this.deliveries.delete(id)
@@ -85,7 +102,8 @@ export class RelayPtySourcePublication {
return false
}
if (
current?.clientId === context.clientId &&
current !== undefined &&
sameClientTransport(current, context) &&
!current.restoreRequired &&
current.sourceExitState !== 'pending' &&
this.deliveryClosedUnderRecord(current)
@@ -96,11 +114,16 @@ export class RelayPtySourcePublication {
this.onCapacity(id)
current = undefined
}
if (current?.clientId === context.clientId) {
if (current !== undefined && sameClientTransport(current, context)) {
this.sender.releaseRotationFence(current)
if (current.activating && current.activationRecoveryRequest) {
if (!samePtySourceRecoveryRequest(current.activationRecoveryRequest, recovery)) {
return this.publishRestoreRequired(id, context, 'checkpointUnavailable')
return publishPtySourceRestoreRequired(
this.restoreDeps,
id,
context,
'checkpointUnavailable'
)
}
this.registerActivationSettlement(id, current, context)
return pendingPtySourceRecoveryResult(current)
@@ -113,7 +136,7 @@ export class RelayPtySourcePublication {
let recoveryEndSu: number | null = null
let recoveryWasSealed = false
if (!current && recovery) {
return this.publishRestoreRequired(id, context, 'deliveryUnavailable')
return publishPtySourceRestoreRequired(this.restoreDeps, id, context, 'deliveryUnavailable')
}
if (current) {
try {
@@ -127,7 +150,13 @@ export class RelayPtySourcePublication {
recovery.ownerGeneration !== current.identity.ownerGeneration ||
recovery.ptyIncarnation !== current.identity.ptyIncarnation
) {
return this.requireRestore(id, current, context, 'checkpointUnavailable')
return requirePtySourceRestore(
this.restoreDeps,
id,
current,
context,
'checkpointUnavailable'
)
}
const rotation = this.session.rotateDelivery(
current.identity,
@@ -141,7 +170,8 @@ export class RelayPtySourcePublication {
recoveryWasSealed = snapshot.state === 'sealed-unsettled'
this.counters.rotated++
} catch (error) {
return this.requireRestore(
return requirePtySourceRestore(
this.restoreDeps,
id,
current,
context,
@@ -162,6 +192,7 @@ export class RelayPtySourcePublication {
const activationRecoveryEndSu = recoveryEndSu ?? activationSnapshot.receivedEndSu
const record: RelayPtySourceDeliveryRecord = {
clientId: context.clientId,
clientTransportGeneration: context.transportGeneration,
identity,
sourceActivation: createPtySourceReceivingActivation(
identity,
@@ -295,32 +326,13 @@ export class RelayPtySourcePublication {
})
}
private requireRestore(
id: string,
current: RelayPtySourceDeliveryRecord,
context: RequestContext,
reason: string
): Readonly<{ status: 'restoreRequired'; reason: string }> {
this.session.cancelDelivery(current.identity, `recovery-${reason}`)
current.restoreRequired = true
current.activating = false
this.sender.wakeSendWaiters(current)
registerCanceledPtySourceRetirement(current, context, this.deliveries, this.onCapacity)
return this.publishRestoreRequired(id, context, reason)
}
private publishRestoreRequired(
id: string,
context: RequestContext,
reason: string
): Readonly<{ status: 'restoreRequired'; reason: string }> {
const result = Object.freeze({ status: 'restoreRequired' as const, reason })
context.onResponseSettled?.((settlement) => {
if (settlement.ok) {
this.dispatcher.notifyClient(context.clientId, 'pty.restoreRequired', { id, reason })
}
})
this.onCapacity(id)
return result
private get restoreDeps(): PtySourceRestoreDeps {
return {
dispatcher: this.dispatcher,
onCapacity: this.onCapacity,
session: this.session,
sender: this.sender,
deliveries: this.deliveries
}
}
}
@@ -0,0 +1,50 @@
import type { RelayDispatcher, RequestContext } from './dispatcher'
import type { RelayPtySourceDeliveryRecord } from './relay-pty-source-delivery-record'
import { registerCanceledPtySourceRetirement } from './relay-pty-source-activation'
import type { RelayPtySourceSendScheduler } from './relay-pty-source-send-scheduler'
import type { SshPtyConsumerSessionAdapter } from './ssh-pty-consumer-session-adapter'
export type PtySourceRestoreDeps = Readonly<{
dispatcher: RelayDispatcher
onCapacity: (id: string) => void
session: SshPtyConsumerSessionAdapter
sender: RelayPtySourceSendScheduler
deliveries: Map<string, RelayPtySourceDeliveryRecord>
}>
type RestoreRequired = Readonly<{ status: 'restoreRequired'; reason: string }>
/**
* The PTY is live and only its delivery is unusable, so retire the delivery and tell the client to
* open a new one. Never an absence verdict — see docs/reference/ssh-execution-boundary.md.
*/
export function requirePtySourceRestore(
deps: PtySourceRestoreDeps,
id: string,
current: RelayPtySourceDeliveryRecord,
context: RequestContext,
reason: string
): RestoreRequired {
deps.session.cancelDelivery(current.identity, `recovery-${reason}`)
current.restoreRequired = true
current.activating = false
deps.sender.wakeSendWaiters(current)
registerCanceledPtySourceRetirement(current, context, deps.deliveries, deps.onCapacity)
return publishPtySourceRestoreRequired(deps, id, context, reason)
}
export function publishPtySourceRestoreRequired(
deps: PtySourceRestoreDeps,
id: string,
context: RequestContext,
reason: string
): RestoreRequired {
const result = Object.freeze({ status: 'restoreRequired' as const, reason })
context.onResponseSettled?.((settlement) => {
if (settlement.ok) {
deps.dispatcher.notifyClient(context.clientId, 'pty.restoreRequired', { id, reason })
}
})
deps.onCapacity(id)
return result
}
+5 -28
View File
@@ -1,9 +1,8 @@
import type {
PtySourceDeliveryIdentity,
PtySourceDeliverySnapshot
} from '../shared/pty-source-credit-contract'
import type { PtySourceRecoveryCheckpoint } from '../shared/pty-source-recovery-contract'
import type { PtySourceReceivingActivation } from '../shared/pty-source-receiving-activation'
import type { PtySourceDeliverySnapshot } from '../shared/pty-source-credit-contract'
import type { RelayPtySourceDeliveryRecord } from './relay-pty-source-delivery-record'
export { sameClientTransport } from './relay-pty-source-delivery-record'
export type { RelayPtySourceDeliveryRecord } from './relay-pty-source-delivery-record'
import type { RelayDispatcher, SinkWriteSettlement } from './dispatcher'
import {
PTY_SOURCE_SCHEDULER_MAX_FRAMES,
@@ -12,28 +11,6 @@ import {
import type { SshPtyConsumerSessionAdapter } from './ssh-pty-consumer-session-adapter'
import { completePtySourceRecovery } from './relay-pty-source-recovery-completion'
export type RelayPtySourceDeliveryRecord = {
clientId: number
identity: PtySourceDeliveryIdentity
sourceActivation: PtySourceReceivingActivation
displayEnd: number
activating: boolean
activationRecoveryRequest: PtySourceRecoveryCheckpoint | null
sealed: boolean
legacyExitAccepted: boolean
sourceExitState: 'idle' | 'pending' | 'published'
sending: boolean
turnFrames: number
turnSourceSu: number
turnScheduled: boolean
sendWaiters: Set<() => void>
recoveryCheckpointSourceEndSu: number | null
recoveryEndSu: number | null
recoveryCompletionPending: boolean
restoreRequired: boolean
rotationPending: boolean
}
export type RelayPtySourcePublicationCounters = {
opened: number
rotated: number
@@ -0,0 +1,214 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import {
RelayDispatcher,
type RelayClientSessionIdentity,
type RequestContext,
type SinkWriteSettlement
} from './dispatcher'
import { encodeJsonRpcFrame, MessageType } from './protocol'
import { RelayPtySourcePublication } from './relay-pty-source-publication'
import { SshPtyConsumerSessionAdapter } from './ssh-pty-consumer-session-adapter'
const endpointIdentity: RelayClientSessionIdentity = {
principal: 'endpoint-principal',
authenticated: true,
allowSessionOwner: true,
authenticationKind: 'endpoint-credential'
}
function requestFrame(id: number, method: string, params: Record<string, unknown>): Buffer {
return encodeJsonRpcFrame({ jsonrpc: '2.0', id, method, params }, id, 0)
}
function responseResult(buffer: Buffer): Record<string, unknown> | null {
if (buffer[0] !== MessageType.Regular) {
return null
}
const length = buffer.readUInt32BE(9)
const message = JSON.parse(buffer.subarray(13, 13 + length).toString('utf8'))
return message.id === undefined ? null : (message.result ?? null)
}
async function flushRequests(): Promise<void> {
await new Promise((resolve) => setImmediate(resolve))
}
describe('PTY source activation across a transport change', () => {
let dispatcher: RelayDispatcher | null = null
afterEach(() => {
dispatcher?.dispose()
dispatcher = null
})
async function createHarness() {
const settlements: ((result: SinkWriteSettlement) => void)[] = []
const writes: Buffer[] = []
dispatcher = new RelayDispatcher(
(data, onSettled) => {
writes.push(Buffer.from(data))
onSettled({ ok: true })
return true
},
{ supportsWriteCallback: true },
endpointIdentity
)
let publication: RelayPtySourcePublication
const adapter = new SshPtyConsumerSessionAdapter(dispatcher, 'build-a', undefined, (id) =>
publication.onCreditAvailable(id)
)
publication = new RelayPtySourcePublication(dispatcher, adapter, () => {})
dispatcher.feed(
requestFrame(1, 'pty.openClient', {
protocolVersion: 1,
clientInstanceId: 'client-1',
requestedRole: 'session-owner',
capabilities: { outputFlowControl: { versions: [1], requestedWindowSu: 4 } }
})
)
await flushRequests()
return { adapter, publication, settlements, writes }
}
function contextOn(
transportGeneration: number | undefined,
settlements: unknown[]
): RequestContext {
return {
clientId: 1,
...(transportGeneration === undefined ? {} : { transportGeneration }),
isStale: () => false,
sessionIdentity: endpointIdentity,
onResponseSettled: (callback) => settlements.push(callback)
}
}
it('short-circuits same-client activation on the same transport', async () => {
const { publication, settlements } = await createHarness()
expect(publication.activate('pty-1', 'incarnation-1', contextOn(0, settlements))).toBe('opened')
;(settlements[0] as (result: SinkWriteSettlement) => void)({ ok: true })
expect(publication.activate('pty-1', 'incarnation-1', contextOn(0, settlements))).toBe(
'existing'
)
})
it('does not claim existing when the client returns on a new transport', async () => {
const { publication, settlements } = await createHarness()
expect(publication.activate('pty-1', 'incarnation-1', contextOn(0, settlements))).toBe('opened')
;(settlements[0] as (result: SinkWriteSettlement) => void)({ ok: true })
expect(publication.activate('pty-1', 'incarnation-1', contextOn(1, settlements))).not.toBe(
'existing'
)
})
it('keeps id-only behavior for callers without transport generations', async () => {
const { publication, settlements } = await createHarness()
expect(publication.activate('pty-1', 'incarnation-1', contextOn(undefined, settlements))).toBe(
'opened'
)
;(settlements[0] as (result: SinkWriteSettlement) => void)({ ok: true })
expect(publication.activate('pty-1', 'incarnation-1', contextOn(undefined, settlements))).toBe(
'existing'
)
})
it('ignores a stale activation before it can cancel the replacement delivery', async () => {
const { publication, adapter, settlements } = await createHarness()
expect(publication.activate('pty-1', 'incarnation-1', contextOn(1, settlements))).toBe('opened')
settlements[0]({ ok: true })
const cancelDelivery = vi.spyOn(adapter, 'cancelDelivery')
const staleSettlements: ((result: SinkWriteSettlement) => void)[] = []
const staleContext: RequestContext = {
...contextOn(0, staleSettlements),
isStale: () => true
}
expect(
publication.activate('pty-1', 'incarnation-2', staleContext, {
status: 'checkpoint',
clientGeneration: 999,
ownerGeneration: 999,
ptyIncarnation: 'stale-incarnation',
deliveryToken: 'stale-token',
acceptedSourceEndSu: 0
})
).toBe(false)
expect(cancelDelivery).not.toHaveBeenCalled()
expect(staleSettlements).toHaveLength(0)
})
it('does not release a replacement fence from a stale activation', async () => {
const { publication, settlements, writes } = await createHarness()
expect(publication.activate('pty-1', 'incarnation-1', contextOn(0, settlements))).toBe('opened')
settlements[0]({ ok: true })
const activation = publication.receivingActivation('pty-1', 1)!
const ownerGrant = writes.map(responseResult).find((result) => result?.ownerLease)!
// The old transport arms its rotation fence while waiting for a checkpoint-safe send.
await expect(publication.waitForPendingSend('pty-1')).resolves.toBe(true)
const replacementWrites: Buffer[] = []
const replacementClientId = dispatcher!.attachClient(
(data, onSettled) => {
replacementWrites.push(Buffer.from(data))
onSettled({ ok: true })
return true
},
{ supportsWriteCallback: true },
endpointIdentity
)
dispatcher!.feedClient(
replacementClientId,
requestFrame(2, 'pty.openClient', {
protocolVersion: 1,
clientInstanceId: 'client-1',
requestedRole: 'session-owner',
resume: {
ownerGeneration: ownerGrant.ownerGeneration,
ownerLease: ownerGrant.ownerLease
},
capabilities: { outputFlowControl: { versions: [1], requestedWindowSu: 4 } }
})
)
await flushRequests()
const replacementSettlements: ((result: SinkWriteSettlement) => void)[] = []
const recovery = {
status: 'checkpoint' as const,
clientGeneration: activation.clientGeneration,
ownerGeneration: activation.ownerGeneration,
ptyIncarnation: activation.ptyIncarnation,
deliveryToken: activation.deliveryToken,
acceptedSourceEndSu: 0
}
expect(
publication.activate(
'pty-1',
'incarnation-1',
{
...contextOn(0, replacementSettlements),
clientId: replacementClientId
},
recovery
)
).toMatchObject({ status: 'pending' })
replacementSettlements[0]({ ok: true })
// Arm the replacement fence, then let the superseded transport's activation resume.
// isStale() stays false on purpose: the original client is still attached at its original
// generation, so production would report it live and execution reaches the delivery-mode
// bail-outs rather than the stale early-out. Hard-coding true here hid that path entirely.
await expect(publication.waitForPendingSend('pty-1')).resolves.toBe(true)
expect(publication.activate('pty-1', 'incarnation-1', contextOn(0, []), recovery)).toBe(false)
expect(publication.publish('pty-1', { data: 'replacement-output' }, false)).toBe(false)
// Release the replacement fence through its owning transport so this test leaves no parked work.
expect(
publication.activate('pty-1', 'incarnation-1', {
...contextOn(0, []),
clientId: replacementClientId
})
).toBe('existing')
expect(replacementWrites.length).toBeGreaterThan(0)
})
})
@@ -148,6 +148,16 @@ describe('humanizeTerminalError', () => {
expect(humanized).not.toContain('exited')
})
// A live PTY whose delivery was retired must never get the "open a new terminal" copy: acting on
// that abandons a running agent on the host.
it('describes a retired output source as reconnecting, not as a lost session', () => {
const humanized = humanizeTerminalError('SSH_SOURCE_RESTORE_REQUIRED: ssh:conn@@pty-7')
expect(humanized).not.toContain('SSH_SOURCE_RESTORE_REQUIRED')
expect(humanized).not.toContain('ssh:conn@@pty-7')
expect(humanized).not.toContain('Open a new terminal')
expect(humanized).toContain('still running')
})
it('replaces only the unreattachable line in an aggregated error', () => {
const humanized = humanizeTerminalError('Paste failed.\nSSH_SESSION_EXPIRED: orca:2f1c@@pty-7')
expect(humanized.startsWith('Paste failed.\n')).toBe(true)
@@ -179,6 +189,10 @@ describe('isExplainedTerminalError', () => {
).toBe(true)
})
it('suppresses the issue link while a live session restores its output source', () => {
expect(isExplainedTerminalError('SSH_SOURCE_RESTORE_REQUIRED: ssh:conn@@pty-7')).toBe(true)
})
it('suppresses the issue link for a session the host cannot reattach', () => {
expect(isExplainedTerminalError('SSH_SESSION_EXPIRED: orca:2f1c@@pty-7')).toBe(true)
expect(
@@ -2,6 +2,7 @@ import { useEffect, useState } from 'react'
import { translate } from '@/i18n/i18n'
import { resolveClientEnvironmentFooter } from '@/lib/client-environment-info'
import { hasClientEnvironmentFooter } from '../../../../shared/client-environment-info'
import { SSH_SOURCE_RESTORE_REQUIRED_ERROR } from '../../../../shared/ssh-pty-failure-tokens'
const SSH_PREFIX = 'SSH connection is not active'
// Produced by pty-connection.ts reportError() when a PTY reattach can't reach its SSH host.
@@ -32,6 +33,12 @@ const UNREATTACHABLE_SESSION_SOURCES = [
'SSH_SESSION_EXPIRED:[ \\t]*\\S*(?:[ \\t]+SSH_PTY_IDENTITY_MISMATCH)?',
'PTY "[^"\\r\\n]*" not found(?: \\(identity mismatch\\))?'
]
// The relay answered and proved the shell is still running — only its output delivery was retired.
// Deliberately NOT one of the sources above: that copy tells the user to open a new terminal, which
// here would abandon a live agent. Same lastIndex hazard, so keep test and replace forms separate.
const SOURCE_RESTORE_REQUIRED_SOURCE = `${SSH_SOURCE_RESTORE_REQUIRED_ERROR}(?::[ \\t]*\\S*)?`
const SOURCE_RESTORE_REQUIRED_PATTERN = new RegExp(SOURCE_RESTORE_REQUIRED_SOURCE)
const SOURCE_RESTORE_REQUIRED_REPLACE_PATTERN = new RegExp(SOURCE_RESTORE_REQUIRED_SOURCE, 'g')
const UNREATTACHABLE_SESSION_PATTERNS = UNREATTACHABLE_SESSION_SOURCES.map(
(source) => new RegExp(source)
)
@@ -74,6 +81,7 @@ export function isExplainedTerminalError(error: string): boolean {
(line) =>
TERMINAL_HOST_GONE_PATTERN.test(line) ||
LEGACY_TERMINAL_HOST_GONE_PATTERN.test(line) ||
SOURCE_RESTORE_REQUIRED_PATTERN.test(line) ||
UNREATTACHABLE_SESSION_PATTERNS.some((pattern) => pattern.test(line))
)
}
@@ -102,6 +110,12 @@ export function humanizeTerminalError(error: string): string {
)
)
}
humanized = humanized.replace(SOURCE_RESTORE_REQUIRED_REPLACE_PATTERN, () =>
translate(
'auto.components.terminal.pane.TerminalErrorToast.sourceRestoring',
'Reconnecting this terminal — its output is being re-established. The session is still running.'
)
)
humanized = humanizeUnreattachableSession(humanized)
if (!isExplainedTerminalError(humanized)) {
return humanized
@@ -1,4 +1,5 @@
import { isRuntimeOwnedSshTargetId } from '../../../../shared/execution-host'
import { SSH_SESSION_EXPIRED_ERROR } from '../../../../shared/ssh-pty-failure-tokens'
import { extractIpcErrorMessage } from '@/lib/ipc-error'
import { ensurePtyDispatcher } from './pty-dispatcher'
import {
@@ -15,7 +16,6 @@ import type { IpcPtySessionHandlers } from './ipc-pty-session-handlers'
import { spawnIpcPty } from './ipc-pty-spawn-request'
import type { IpcPtyTransportOptions, PtyConnectResult, PtyTransport } from './pty-transport-types'
const SSH_SESSION_EXPIRED_ERROR = 'SSH_SESSION_EXPIRED'
const SSH_PTY_CONNECTION_MISMATCH_MARKER = 'belongs to SSH connection'
type PtyConnectOptions = Parameters<PtyTransport['connect']>[0]
@@ -282,6 +282,128 @@ describe('connectPanePty', () => {
)
})
// Positive control for the test below. Without it that assertion is untestable: it would pass on a
// build where nothing can fresh-spawn at all. This pins that THIS harness does respawn on the one
// token that authorises it, so "did not respawn" there is a real result.
it('fresh-spawns when the host proves the reattach session is gone', async () => {
const { connectPanePty } = await import('./pty-connection')
const restoredPtyId = toAppSshPtyId('target-a', 'pty-gone-session')
const transport = createMockTransport()
transport.connect.mockImplementationOnce(async (options) => {
options.callbacks?.onError?.(`SSH_SESSION_EXPIRED: target-a@@pty-gone-session`)
return undefined
})
transportFactoryQueue.push(transport)
const pendingRetry = {
attemptId: 'attempt-gone-session',
authority: {
targetId: 'target-a',
providerEpoch: 'epoch-1',
connectionGeneration: 3
},
tabGeneration: 7,
startedAt: 1
}
const settleDirectSshPaneRetry = vi.fn()
mockStoreState = {
...mockStoreState,
tabsByWorktree: {
'wt-1': [{ id: 'tab-1', ptyId: restoredPtyId, generation: 7 }]
},
ptyIdsByTabId: { 'tab-1': [restoredPtyId] },
repos: [{ id: 'repo1', connectionId: 'target-a', displayName: 'orca' }],
sshConnectionStates: new Map([
[
'target-a',
{
targetId: 'target-a',
status: 'connected',
providerEpoch: 'epoch-1',
connectionGeneration: 3
}
]
]),
directSshPaneRetryByTabId: { 'tab-1': pendingRetry },
settleDirectSshPaneRetry
}
const deps = createDeps({
restoredLeafId: LEAF_1,
restoredPtyIdByLeafId: { [LEAF_1]: restoredPtyId }
})
connectPanePty(createPane(1) as never, createManager(1) as never, deps as never)
await flushAsyncTicks(16)
expect(deps.clearTabPtyId).toHaveBeenCalledWith('tab-1', restoredPtyId)
})
it('does not fresh-spawn when reattach only needs its live source restored', async () => {
const { connectPanePty } = await import('./pty-connection')
const restoredPtyId = toAppSshPtyId('target-a', 'pty-live-source')
const transport = createMockTransport()
transport.connect.mockImplementationOnce(async (options) => {
options.callbacks?.onError?.('SSH_SOURCE_RESTORE_REQUIRED: target-a@@pty-live-source')
return undefined
})
transportFactoryQueue.push(transport)
const pendingRetry = {
attemptId: 'attempt-live-source-restore',
authority: {
targetId: 'target-a',
providerEpoch: 'epoch-1',
connectionGeneration: 3
},
tabGeneration: 7,
startedAt: 1
}
const settleDirectSshPaneRetry = vi.fn()
mockStoreState = {
...mockStoreState,
tabsByWorktree: {
'wt-1': [{ id: 'tab-1', ptyId: restoredPtyId, generation: 7 }]
},
ptyIdsByTabId: { 'tab-1': [restoredPtyId] },
repos: [{ id: 'repo1', connectionId: 'target-a', displayName: 'orca' }],
sshConnectionStates: new Map([
[
'target-a',
{
targetId: 'target-a',
status: 'connected',
providerEpoch: 'epoch-1',
connectionGeneration: 3
}
]
]),
directSshPaneRetryByTabId: { 'tab-1': pendingRetry },
settleDirectSshPaneRetry
}
const deps = createDeps({
restoredLeafId: LEAF_1,
restoredPtyIdByLeafId: { [LEAF_1]: restoredPtyId }
})
connectPanePty(createPane(1) as never, createManager(1) as never, deps as never)
await flushAsyncTicks(16)
expect(transport.connect).toHaveBeenCalledExactlyOnceWith(
expect.objectContaining({ sessionId: restoredPtyId })
)
// The exact call the positive control above proves this harness makes on SSH_SESSION_EXPIRED.
expect(deps.clearTabPtyId).not.toHaveBeenCalled()
expect(transport.connect).not.toHaveBeenCalledWith(
expect.not.objectContaining({ sessionId: restoredPtyId })
)
expect(settleDirectSshPaneRetry).toHaveBeenCalledWith({
status: 'failed',
tabId: 'tab-1',
attemptId: pendingRetry.attemptId,
authority: pendingRetry.authority,
tabGeneration: pendingRetry.tabGeneration
})
expect(window.api.pty.kill).not.toHaveBeenCalled()
})
// Why both directions: #15166 dropped #14844's reconnect model-paint gate and pinned the
// degraded relay-only paint here. The gate's contract is that the relay wins only when the
// replay SHOWS the app left the alternate screen, so pin the veto and its absence together.
@@ -2,7 +2,7 @@ import { e2eConfig } from '@/lib/e2e-config'
export const pendingSpawnByPaneKey = new Map<string, Promise<string | null>>()
export const pendingSpawnGenerationByPaneKey = new Map<string, number>()
export const SSH_SESSION_EXPIRED_ERROR = 'SSH_SESSION_EXPIRED'
export { SSH_SESSION_EXPIRED_ERROR } from '../../../../../shared/ssh-pty-failure-tokens'
// Why: relay requests expire at 30s; leave one second for their fallback before re-arming locally.
export const DIRECT_SSH_PANE_RETRY_SETTLEMENT_TIMEOUT_MS = 31_000
export const REMOTE_PTY_ID_PREFIX = 'remote:'
@@ -20,6 +20,7 @@ import type {
RuntimeTerminalSend
} from '../../../../shared/runtime-types'
import { TERMINAL_CREATE_IDEMPOTENCY_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version'
import { SSH_SESSION_EXPIRED_ERROR } from '../../../../shared/ssh-pty-failure-tokens'
import { agentResumeHostAuthorityCapability } from '../../runtime/agent-resume-host-authority-capability'
import {
isTerminalInputTooLargeWithDeferredMeasurement,
@@ -116,7 +117,6 @@ type RemoteAgentSessionLaunchResult =
| RuntimeEnsureAgentSessionResult
| RuntimeCreateAgentSessionResult
| { terminal: RuntimeTerminalCreate; disposition?: undefined }
const SSH_SESSION_EXPIRED_ERROR = 'SSH_SESSION_EXPIRED'
function isRemoteTerminalStaleMessage(message: string): boolean {
return message.includes('terminal_handle_stale')
+2 -1
View File
@@ -3072,7 +3072,8 @@
"cc6d997c65": "Restart the terminal daemon from here to clear stale daemon state.",
"7ee11bc0db": "Orca couldn't confirm whether this terminal's previous session is still running, so it left the session untouched. Reopen this pane to retry.",
"e16012e31e": "The terminal daemon that owned this session exited, so the session and its scrollback could not be recovered. Open a new terminal to continue.",
"sessionUnavailable": "Orca couldn't reattach to this pane's terminal session on the host. Open a new terminal to continue."
"sessionUnavailable": "Orca couldn't reattach to this pane's terminal session on the host. Open a new terminal to continue.",
"sourceRestoring": "Reconnecting this terminal — its output is being re-established. The session is still running."
},
"TerminalProcessExitOverlay": {
"capacityTitle": "Git Bash console limit reached",
+23
View File
@@ -0,0 +1,23 @@
/**
* Tokens a failed SSH PTY reattach can carry, shared by the main-process provider that throws them
* and the renderer consumers that decide what a pane may do next.
*
* They are deliberately disjoint strings: every consumer matches with `includes()`, so a token that
* embedded another would silently inherit its authority.
*/
/**
* A reachable relay answered for this exact PTY id and the session is gone. The only token that
* authorises replacing the pane's PTY (see docs/reference/ssh-execution-boundary.md — `exited`
* requires positive evidence of absence from the host that owns the process).
*/
export const SSH_SESSION_EXPIRED_ERROR = 'SSH_SESSION_EXPIRED'
/** The relay found the PTY, but it is bound to a different pane identity. */
export const SSH_PTY_IDENTITY_MISMATCH_ERROR = 'SSH_PTY_IDENTITY_MISMATCH'
/**
* The relay proved the PTY is live and only retired its output delivery. Never respawn on this:
* the shell, its agent, and its work are all still running on the host.
*/
export const SSH_SOURCE_RESTORE_REQUIRED_ERROR = 'SSH_SOURCE_RESTORE_REQUIRED'