mirror of
https://github.com/stablyai/orca.git
synced 2026-10-27 08:02:51 +00:00
fix(ssh): quarantine unverifiable reattach evidence
This commit is contained in:
@@ -114,7 +114,7 @@ describe('subscribeSshPtyNotifications', () => {
|
||||
incarnationId: 'incarnation-1'
|
||||
})
|
||||
|
||||
expect(recordExit).toHaveBeenCalledWith('pty-1', 'incarnation-1')
|
||||
expect(recordExit).toHaveBeenCalledWith('pty-1', 'incarnation-1', expect.any(Function))
|
||||
expect(livePtyIds.has('ssh:conn@@pty-1')).toBe(true)
|
||||
expect(onExit).toHaveBeenCalledWith({
|
||||
id: 'ssh:conn@@pty-1',
|
||||
|
||||
@@ -40,7 +40,7 @@ export function subscribeSshPtyNotifications(args: {
|
||||
rejectedDataListeners?: Set<SshPtyDataCallback>
|
||||
replayListeners: Set<SshPtyReplayCallback>
|
||||
exitListeners: Set<SshPtyExitCallback>
|
||||
recordExit: (relayPtyId: string, incarnationId: unknown) => void
|
||||
recordExit: (relayPtyId: string, incarnationId: unknown, publish?: () => void) => boolean | void
|
||||
providerGeneration: number
|
||||
resolvePtyIncarnation: (relayPtyId: string, incarnationId?: unknown) => string
|
||||
peekPtyIncarnation: (relayPtyId: string) => string | undefined
|
||||
@@ -160,19 +160,23 @@ export function subscribeSshPtyNotifications(args: {
|
||||
}
|
||||
const relayPtyId = params.id
|
||||
if (method === 'pty.exit') {
|
||||
const id = args.toAppPtyId(relayPtyId)
|
||||
const ptyIncarnation = args.resolvePtyIncarnation(relayPtyId, params.incarnationId)
|
||||
args.recordExit(relayPtyId, params.incarnationId)
|
||||
for (const listener of args.exitListeners) {
|
||||
listener({
|
||||
id,
|
||||
code: params.code as number,
|
||||
providerGeneration: args.providerGeneration,
|
||||
ptyIncarnation,
|
||||
...(isPtyIncarnationId(params.incarnationId)
|
||||
? { incarnationId: params.incarnationId }
|
||||
: {})
|
||||
})
|
||||
const publish = (): void => {
|
||||
const id = args.toAppPtyId(relayPtyId)
|
||||
const ptyIncarnation = args.resolvePtyIncarnation(relayPtyId, params.incarnationId)
|
||||
for (const listener of args.exitListeners) {
|
||||
listener({
|
||||
id,
|
||||
code: params.code as number,
|
||||
providerGeneration: args.providerGeneration,
|
||||
ptyIncarnation,
|
||||
...(isPtyIncarnationId(params.incarnationId)
|
||||
? { incarnationId: params.incarnationId }
|
||||
: {})
|
||||
})
|
||||
}
|
||||
}
|
||||
if (!args.recordExit(relayPtyId, params.incarnationId, publish)) {
|
||||
publish()
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
@@ -139,6 +139,8 @@ it('does not classify a legacy exit as current while reattach resolves its incar
|
||||
isDisposed: vi.fn().mockReturnValue(false)
|
||||
}
|
||||
const provider = new SshPtyProvider('conn-1', mux as never)
|
||||
const exitListener = vi.fn()
|
||||
provider.onExit(exitListener)
|
||||
let resolveAttach: (() => void) | undefined
|
||||
mux.request.mockImplementation((method: string, _params, options) => {
|
||||
if (method !== 'pty.attach') {
|
||||
@@ -161,6 +163,7 @@ it('does not classify a legacy exit as current while reattach resolves its incar
|
||||
await vi.waitFor(() => expect(resolveAttach).toBeTypeOf('function'))
|
||||
const notify = mux.onNotification.mock.calls[0]?.[0]
|
||||
notify?.('pty.exit', { id: 'pty-existing', code: 0 })
|
||||
expect(exitListener).not.toHaveBeenCalled()
|
||||
resolveAttach?.()
|
||||
|
||||
await expect(spawn).resolves.toMatchObject({
|
||||
@@ -168,7 +171,21 @@ it('does not classify a legacy exit as current while reattach resolves its incar
|
||||
incarnationId: 'incarnation-current',
|
||||
isReattach: true
|
||||
})
|
||||
expect(exitListener).not.toHaveBeenCalled()
|
||||
await expect(provider.probePtyLiveness('ssh:conn-1@@pty-existing')).resolves.toBe(true)
|
||||
|
||||
notify?.('pty.exit', {
|
||||
id: 'pty-existing',
|
||||
code: 7,
|
||||
incarnationId: 'incarnation-current'
|
||||
})
|
||||
expect(exitListener).toHaveBeenCalledExactlyOnceWith({
|
||||
id: 'ssh:conn-1@@pty-existing',
|
||||
code: 7,
|
||||
incarnationId: 'incarnation-current',
|
||||
providerGeneration: expect.any(Number),
|
||||
ptyIncarnation: 'incarnation-current'
|
||||
})
|
||||
})
|
||||
|
||||
it('returns a provisional source activation lease to reconnect authority', async () => {
|
||||
|
||||
@@ -28,7 +28,11 @@ export class SshPtyProviderOutputState {
|
||||
args: {
|
||||
mux: SshChannelMultiplexer
|
||||
toAppPtyId: (id: string) => string
|
||||
recordExit: (relayPtyId: string, incarnationId: unknown) => void
|
||||
recordExit: (
|
||||
relayPtyId: string,
|
||||
incarnationId: unknown,
|
||||
publish?: () => void
|
||||
) => boolean | void
|
||||
}
|
||||
) {
|
||||
this.subscription = subscribeSshPtyNotifications({
|
||||
@@ -41,9 +45,7 @@ export class SshPtyProviderOutputState {
|
||||
resolvePtyIncarnation: (relayPtyId, incarnationId) =>
|
||||
this.resolvePtyIncarnation(relayPtyId, incarnationId),
|
||||
peekPtyIncarnation: (relayPtyId) => this.incarnationByRelayPtyId.get(relayPtyId),
|
||||
recordExit: (relayPtyId, incarnationId) => {
|
||||
args.recordExit(relayPtyId, incarnationId)
|
||||
}
|
||||
recordExit: args.recordExit
|
||||
})
|
||||
}
|
||||
|
||||
@@ -125,6 +127,12 @@ export class SshPtyProviderOutputState {
|
||||
}
|
||||
}
|
||||
|
||||
acceptPtyIncarnation(relayPtyId: string, incarnationId: unknown): void {
|
||||
if (typeof incarnationId === 'string' && incarnationId.length > 0) {
|
||||
this.incarnationByRelayPtyId.set(relayPtyId, incarnationId)
|
||||
}
|
||||
}
|
||||
|
||||
acceptExit(relayPtyId: string): void {
|
||||
this.subscription?.acceptExit(relayPtyId)
|
||||
this.incarnationByRelayPtyId.delete(relayPtyId)
|
||||
|
||||
@@ -3,6 +3,43 @@ import { SSH_PTY_RESTORE_REQUIRED_ERROR } from './ssh-pty-errors'
|
||||
import { SshPtyProvider } from './ssh-pty-provider'
|
||||
|
||||
describe('SSH PTY provider session reattach incarnation', () => {
|
||||
it('replaces provisional exit identity with the authoritative attach incarnation', async () => {
|
||||
let notify: ((method: string, params: Record<string, unknown>) => void) | undefined
|
||||
let resolveAttach: ((value: { incarnationId: string }) => void) | undefined
|
||||
const mux = {
|
||||
request: vi.fn(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
resolveAttach = resolve
|
||||
})
|
||||
),
|
||||
notify: vi.fn(),
|
||||
onNotification: vi.fn(
|
||||
(callback: (method: string, params: Record<string, unknown>) => void) => {
|
||||
notify = callback
|
||||
return vi.fn()
|
||||
}
|
||||
)
|
||||
}
|
||||
const provider = new SshPtyProvider('conn-1', mux as never)
|
||||
const onData = vi.fn()
|
||||
provider.onData(onData)
|
||||
const attach = provider.attachForReconnect('ssh:conn-1@@pty-old')
|
||||
await vi.waitFor(() => expect(resolveAttach).toBeTypeOf('function'))
|
||||
|
||||
notify?.('pty.exit', { id: 'pty-old', code: 0 })
|
||||
resolveAttach?.({ incarnationId: 'incarnation-reattached' })
|
||||
await attach
|
||||
notify?.('pty.data', { id: 'pty-old', data: 'current-output' })
|
||||
|
||||
expect(onData).toHaveBeenCalledWith({
|
||||
id: 'ssh:conn-1@@pty-old',
|
||||
data: 'current-output',
|
||||
providerGeneration: expect.any(Number),
|
||||
ptyIncarnation: 'incarnation-reattached'
|
||||
})
|
||||
})
|
||||
|
||||
it('remembers the authoritative incarnation before a legacy exit arrives', async () => {
|
||||
let notify: ((method: string, params: Record<string, unknown>) => void) | undefined
|
||||
const mux = {
|
||||
|
||||
@@ -3,6 +3,7 @@ import { SshPtyProvider } from './ssh-pty-provider'
|
||||
import { POWERLEVEL10K_WIZARD_DISABLE_ENV } from '../pty/powerlevel10k-wizard-env'
|
||||
import { PTY_STARTUP_INGRESS_VERSION } from '../../shared/pty-startup-ingress'
|
||||
import { AGENT_SESSION_EXECUTION_OWNER_PROTOCOL_VERSION } from '../../shared/agent-session-host-authority'
|
||||
import { SSH_PTY_LIVENESS_UNVERIFIABLE_ERROR } from './ssh-pty-errors'
|
||||
|
||||
type MockMultiplexer = {
|
||||
request: ReturnType<typeof vi.fn>
|
||||
@@ -681,11 +682,11 @@ describe('SshPtyProvider', () => {
|
||||
expect(mux.request).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('preserves transient reattach failures for retry handling', async () => {
|
||||
it('classifies transient reattach failures as unverifiable', async () => {
|
||||
mux.request.mockRejectedValueOnce(new Error('SSH connection lost, reconnecting...'))
|
||||
|
||||
await expect(provider.spawn({ cols: 80, rows: 24, sessionId: 'pty-old' })).rejects.toThrow(
|
||||
'SSH connection lost, reconnecting...'
|
||||
`${SSH_PTY_LIVENESS_UNVERIFIABLE_ERROR}: pty-old`
|
||||
)
|
||||
|
||||
expect(mux.request).toHaveBeenCalledTimes(1)
|
||||
|
||||
@@ -59,9 +59,8 @@ export class SshPtyProvider implements IPtyProvider {
|
||||
this.outputState = new SshPtyProviderOutputState(providerGeneration, {
|
||||
mux,
|
||||
toAppPtyId: (id) => this.toAppPtyId(id),
|
||||
recordExit: (relayPtyId, incarnationId) => {
|
||||
this.spawnExitRaces.recordExit(relayPtyId, incarnationId)
|
||||
}
|
||||
recordExit: (relayPtyId, incarnationId, publish) =>
|
||||
this.spawnExitRaces.recordExit(relayPtyId, incarnationId, publish)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -101,7 +100,7 @@ export class SshPtyProvider implements IPtyProvider {
|
||||
installSourceActivation: (relayPtyId, activation) =>
|
||||
this.outputState.installReceivingActivation(relayPtyId, activation),
|
||||
rememberPtyIncarnation: (relayPtyId, incarnationId) =>
|
||||
this.outputState.rememberPtyIncarnation(relayPtyId, incarnationId),
|
||||
this.outputState.acceptPtyIncarnation(relayPtyId, incarnationId),
|
||||
acceptLivePty: (relayPtyId) => this.acceptLivePty(relayPtyId),
|
||||
acceptUnverifiablePty: (relayPtyId) => this.acceptUnverifiablePty(relayPtyId),
|
||||
acceptExitedPty: (relayPtyId) => this.acceptExitedPty(relayPtyId)
|
||||
@@ -130,7 +129,7 @@ export class SshPtyProvider implements IPtyProvider {
|
||||
installSourceActivation: (id, activation) =>
|
||||
this.outputState.installReceivingActivation(id, activation),
|
||||
rememberPtyIncarnation: (id, incarnation) =>
|
||||
this.outputState.rememberPtyIncarnation(id, incarnation),
|
||||
this.outputState.acceptPtyIncarnation(id, incarnation),
|
||||
acceptLivePty: (id) => this.acceptLivePty(id),
|
||||
toAppPtyId: this.toAppPtyId
|
||||
})
|
||||
@@ -160,7 +159,7 @@ export class SshPtyProvider implements IPtyProvider {
|
||||
installSourceActivation: (ptyId, activation) =>
|
||||
this.outputState.installReceivingActivation(ptyId, activation),
|
||||
rememberPtyIncarnation: (ptyId, incarnationId) =>
|
||||
this.outputState.rememberPtyIncarnation(ptyId, incarnationId)
|
||||
this.outputState.acceptPtyIncarnation(ptyId, incarnationId)
|
||||
})
|
||||
this.acceptLivePty(id)
|
||||
}
|
||||
@@ -190,7 +189,7 @@ export class SshPtyProvider implements IPtyProvider {
|
||||
installSourceActivation: (ptyId, activation) =>
|
||||
this.outputState.installReceivingActivation(ptyId, activation),
|
||||
rememberPtyIncarnation: (ptyId, incarnationId) =>
|
||||
this.outputState.rememberPtyIncarnation(ptyId, incarnationId)
|
||||
this.outputState.acceptPtyIncarnation(ptyId, incarnationId)
|
||||
})
|
||||
this.acceptLivePty(id)
|
||||
return result
|
||||
@@ -290,7 +289,7 @@ export class SshPtyProvider implements IPtyProvider {
|
||||
for (const process of processes) {
|
||||
this.acceptLivePty(process.id)
|
||||
const relayPtyId = this.toRelayPtyId(process.id)
|
||||
this.outputState.rememberPtyIncarnation(relayPtyId, process.incarnationId)
|
||||
this.outputState.acceptPtyIncarnation(relayPtyId, process.incarnationId)
|
||||
}
|
||||
return processes
|
||||
}
|
||||
|
||||
@@ -61,6 +61,20 @@ describe('SSH PTY reattach when the relay requires source restoration', () => {
|
||||
await expect(provider.probePtyLiveness('ssh:conn-1@@pty-unprobed')).resolves.toBeNull()
|
||||
})
|
||||
|
||||
it('reports unverifiable when the initial attach loses contact', async () => {
|
||||
const mux = createMockMux()
|
||||
mux.request.mockRejectedValueOnce(new Error('Multiplexer disposed'))
|
||||
const provider = new SshPtyProvider('conn-1', mux as never)
|
||||
|
||||
const message = await spawnError(provider)
|
||||
|
||||
expect(message).toContain(SSH_PTY_LIVENESS_UNVERIFIABLE_ERROR)
|
||||
expect(message).not.toContain(SSH_SESSION_EXPIRED_ERROR)
|
||||
expect(provider.hasPty('ssh:conn-1@@pty-old')).toBe(false)
|
||||
await expect(provider.probePtyLiveness('ssh:conn-1@@pty-old')).resolves.toBeNull()
|
||||
expect(mux.request.mock.calls.filter((call) => call[0] === 'pty.attach')).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('keeps a reconnecting PTY unverifiable after an authoritative empty inventory', async () => {
|
||||
const id = 'ssh:conn-1@@pty-reconnecting'
|
||||
const mux = createMockMux()
|
||||
|
||||
@@ -7,6 +7,8 @@ import {
|
||||
SSH_SESSION_EXPIRED_ERROR,
|
||||
isSshPtyExitedEvidenceError,
|
||||
isSshPtyIdentityMismatchError,
|
||||
isSshPtyLivenessUnverifiableError,
|
||||
isSshPtyRestoreRequiredError,
|
||||
isSshPtyNotFoundError
|
||||
} from './ssh-pty-errors'
|
||||
import { toAppSshPtyId, toRelaySshPtyId } from './ssh-pty-id'
|
||||
@@ -237,7 +239,7 @@ export async function reattachSshPtySessionWithExitFence(
|
||||
exitRaceTracker: SshPtySpawnExitRaceTracker
|
||||
}
|
||||
): Promise<SshPtyReattachResult> {
|
||||
const operation = args.exitRaceTracker.begin()
|
||||
const operation = args.exitRaceTracker.begin(toRelaySshPtyId(args.connectionId, args.sessionId))
|
||||
let result: SshPtyReattachResult | undefined
|
||||
try {
|
||||
result = await reattachSshPtySession(args)
|
||||
@@ -276,19 +278,7 @@ export async function reattachSshPtySessionForSpawn(
|
||||
if (unresumable.sourceActivationLease) {
|
||||
await unresumable.sourceActivationLease.rollback()
|
||||
}
|
||||
try {
|
||||
result = await reattachSshPtySessionWithExitFence(args)
|
||||
} catch (error) {
|
||||
if (isSshPtyIdentityMismatchError(error) || isSshPtyExitedEvidenceError(error)) {
|
||||
throw error
|
||||
}
|
||||
// Why: losing contact during verification is unverifiable, not live or exited.
|
||||
args.acceptUnverifiablePty(unresumable.id)
|
||||
throw new Error(
|
||||
`${SSH_PTY_LIVENESS_UNVERIFIABLE_ERROR}: ${toRelaySshPtyId(args.connectionId, unresumable.id)}`,
|
||||
{ cause: error }
|
||||
)
|
||||
}
|
||||
result = await reattachSshPtySessionWithExitFence(args)
|
||||
if (result.sourceRecovery?.status === 'restoreRequired') {
|
||||
args.acceptLivePty(result.id)
|
||||
throw new Error(
|
||||
@@ -304,6 +294,17 @@ export async function reattachSshPtySessionForSpawn(
|
||||
await result?.sourceActivationLease?.rollback()
|
||||
if (isSshPtyExitedEvidenceError(error)) {
|
||||
args.acceptExitedPty(result?.id ?? toAppSshPtyId(args.connectionId, args.sessionId))
|
||||
} else if (
|
||||
!isSshPtyIdentityMismatchError(error) &&
|
||||
!isSshPtyRestoreRequiredError(error) &&
|
||||
!isSshPtyLivenessUnverifiableError(error)
|
||||
) {
|
||||
const id = result?.id ?? toAppSshPtyId(args.connectionId, args.sessionId)
|
||||
args.acceptUnverifiablePty(id)
|
||||
throw new Error(
|
||||
`${SSH_PTY_LIVENESS_UNVERIFIABLE_ERROR}: ${toRelaySshPtyId(args.connectionId, id)}`,
|
||||
{ cause: error }
|
||||
)
|
||||
}
|
||||
throw error
|
||||
}
|
||||
|
||||
@@ -1,36 +1,55 @@
|
||||
import { isPtyIncarnationId, type PtyIncarnationId } from '../../shared/pty-incarnation'
|
||||
|
||||
type PendingSshPtySpawn = {
|
||||
exits: { relayPtyId: string; incarnationId?: PtyIncarnationId }[]
|
||||
relayPtyId?: string
|
||||
exits: {
|
||||
relayPtyId: string
|
||||
incarnationId?: PtyIncarnationId
|
||||
publish?: () => void
|
||||
}[]
|
||||
}
|
||||
|
||||
export class SshPtySpawnExitRaceTracker {
|
||||
private pending = new Set<PendingSshPtySpawn>()
|
||||
|
||||
begin(): PendingSshPtySpawn {
|
||||
const operation = { exits: [] }
|
||||
begin(relayPtyId?: string): PendingSshPtySpawn {
|
||||
const operation = { ...(relayPtyId ? { relayPtyId } : {}), exits: [] }
|
||||
this.pending.add(operation)
|
||||
return operation
|
||||
}
|
||||
|
||||
recordExit(relayPtyId: string, incarnationId: unknown): void {
|
||||
recordExit(relayPtyId: string, incarnationId: unknown, publish?: () => void): boolean {
|
||||
let quarantined = false
|
||||
let published = false
|
||||
const publishOnce = (): void => {
|
||||
if (!published) {
|
||||
published = true
|
||||
publish?.()
|
||||
}
|
||||
}
|
||||
for (const operation of this.pending) {
|
||||
const quarantine = operation.relayPtyId === relayPtyId && publish !== undefined
|
||||
operation.exits.push({
|
||||
relayPtyId,
|
||||
...(isPtyIncarnationId(incarnationId) ? { incarnationId } : {})
|
||||
...(isPtyIncarnationId(incarnationId) ? { incarnationId } : {}),
|
||||
...(quarantine ? { publish: publishOnce } : {})
|
||||
})
|
||||
quarantined ||= quarantine
|
||||
}
|
||||
return quarantined
|
||||
}
|
||||
|
||||
didMatchingExitArrive(
|
||||
operation: PendingSshPtySpawn,
|
||||
result: { id: string; incarnationId?: PtyIncarnationId }
|
||||
): boolean {
|
||||
return operation.exits.some(
|
||||
const matchingExit = operation.exits.find(
|
||||
(exit) =>
|
||||
exit.relayPtyId === result.id &&
|
||||
(!result.incarnationId || exit.incarnationId === result.incarnationId)
|
||||
)
|
||||
matchingExit?.publish?.()
|
||||
return matchingExit !== undefined
|
||||
}
|
||||
|
||||
finish(operation: PendingSshPtySpawn): void {
|
||||
|
||||
@@ -133,7 +133,7 @@ describe('SSH relay PTY incarnation exits', () => {
|
||||
providerGeneration: 31,
|
||||
acceptExitedPty
|
||||
} as never)
|
||||
vi.mocked(isCurrentPtyExit).mockReturnValueOnce(false)
|
||||
vi.mocked(isCurrentPtyExit).mockReturnValueOnce(true).mockReturnValueOnce(false)
|
||||
|
||||
onExit({
|
||||
id: 'ssh:target-1@@pty-reused',
|
||||
|
||||
@@ -565,6 +565,7 @@ describe('SshRelaySession reconnect incarnation ordering', () => {
|
||||
const sourceActivationLease = { commit: vi.fn(), rollback: vi.fn() }
|
||||
vi.mocked(getSshPtyProvider).mockReturnValue({
|
||||
attachForReconnect: vi.fn().mockImplementation(async () => {
|
||||
emitExitDuringAttach({ id: APP_PTY_ID, code: 0 })
|
||||
emitExitDuringAttach({
|
||||
id: APP_PTY_ID,
|
||||
code: 0,
|
||||
|
||||
@@ -2215,6 +2215,9 @@ export class SshRelaySession {
|
||||
}
|
||||
|
||||
private async acceptPtyExit(payload: SshPtyExitPayload): Promise<void> {
|
||||
if (!isCurrentPtyExit(payload)) {
|
||||
return
|
||||
}
|
||||
await acceptSshPtyOutputExit({
|
||||
id: payload.id,
|
||||
code: payload.code,
|
||||
@@ -2389,11 +2392,9 @@ export class SshRelaySession {
|
||||
if (!shouldContinue()) {
|
||||
return
|
||||
}
|
||||
const exitDuringAttach = pendingReattach.exits.find(
|
||||
(exit) =>
|
||||
!exit.incarnationId ||
|
||||
!attachResult.incarnationId ||
|
||||
exit.incarnationId === attachResult.incarnationId
|
||||
const exitDuringAttach = this.findExactPendingExit(
|
||||
pendingReattach,
|
||||
attachResult.incarnationId
|
||||
)
|
||||
if (exitDuringAttach && !recoveryRequest) {
|
||||
if (attachResult.incarnationId) {
|
||||
@@ -2507,11 +2508,9 @@ export class SshRelaySession {
|
||||
}
|
||||
pendingReattach.livePassthrough = true
|
||||
}
|
||||
const exitAfterActivation = pendingReattach.exits.find(
|
||||
(exit) =>
|
||||
!exit.incarnationId ||
|
||||
!attachResult.incarnationId ||
|
||||
exit.incarnationId === attachResult.incarnationId
|
||||
const exitAfterActivation = this.findExactPendingExit(
|
||||
pendingReattach,
|
||||
attachResult.incarnationId
|
||||
)
|
||||
if (exitAfterActivation) {
|
||||
await this.acceptPtyExit(exitAfterActivation)
|
||||
|
||||
Reference in New Issue
Block a user