mirror of
https://github.com/stablyai/orca.git
synced 2026-09-30 16:02:56 +00:00
Merge remote-tracking branch 'origin/fix/reclaim-connection-scope-consistency' into adhoc/ssh-sweep-combined
This commit is contained in:
@@ -66,6 +66,50 @@ function createRuntime(provider?: {
|
||||
return runtime
|
||||
}
|
||||
|
||||
// Why: an SSH-backed workspace whose spawn response was lost — the leak in #17929.
|
||||
function installRemoteReclaimHarness(
|
||||
runtime: OrcaRuntimeService,
|
||||
listProcesses: ReturnType<typeof vi.fn>
|
||||
): void {
|
||||
const handleByPtyId = new Map<string, string>()
|
||||
Object.assign(runtime, {
|
||||
ptyController: { listProcesses },
|
||||
resolveTerminalWorkspaceLaunchScope: vi.fn(async () => ({
|
||||
id: 'worktree-1',
|
||||
path: '/remote/worktree-1',
|
||||
connectionId: 'ssh-1'
|
||||
})),
|
||||
executionOwnerSupportsAgentSessionOperation: vi.fn(async () => true),
|
||||
markWorkspaceTrustedForAgent: vi.fn(async () => {}),
|
||||
adoptControllerTerminalHandle: vi.fn((ptyId: string, handle: string) => {
|
||||
handleByPtyId.set(ptyId, handle)
|
||||
}),
|
||||
recordPtyWorktree: vi.fn((ptyId: string, worktreeId: string, state: { title?: string }) => ({
|
||||
ptyId,
|
||||
worktreeId,
|
||||
title: state.title ?? null
|
||||
})),
|
||||
issuePtyHandle: vi.fn((pty: { ptyId: string }) => handleByPtyId.get(pty.ptyId))
|
||||
})
|
||||
}
|
||||
|
||||
async function fenceRemoteAgentSessionSpawn(runtime: OrcaRuntimeService) {
|
||||
const failure = Object.assign(new Error('execution_owner_unavailable'), {
|
||||
agentSessionOperationOutcome: 'unknown' as const
|
||||
})
|
||||
const createTerminal = vi
|
||||
.spyOn(runtime, 'createTerminal')
|
||||
.mockImplementation(async (_worktree, opts) => {
|
||||
opts?.onPtySpawnCommitted?.()
|
||||
throw failure
|
||||
})
|
||||
const id = operationId()
|
||||
await expect(runtime.createAgentSession(request(id), { clientId: 'device-a' })).rejects.toThrow(
|
||||
failure.message
|
||||
)
|
||||
return { createTerminal, failure, id }
|
||||
}
|
||||
|
||||
describe('agent-session create operation ledger', () => {
|
||||
it('selects legacy before trust, spawn, or ledger state for an old daemon', async () => {
|
||||
const provider = {
|
||||
@@ -291,6 +335,81 @@ describe('agent-session create operation ledger', () => {
|
||||
expect(createTerminal).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('reclaims a fenced remote spawn the host is still holding', async () => {
|
||||
const runtime = createRuntime()
|
||||
const listProcesses = vi.fn(async () => [] as never[])
|
||||
installRemoteReclaimHarness(runtime, listProcesses)
|
||||
const { createTerminal, id, failure } = await fenceRemoteAgentSessionSpawn(runtime)
|
||||
const orphanHandle = createTerminal.mock.calls[0]?.[1]?.preAllocatedHandle as string
|
||||
listProcesses.mockResolvedValue([
|
||||
{
|
||||
id: 'ssh-1:pty2:e:1',
|
||||
cwd: '/remote/worktree-1',
|
||||
title: 'codex',
|
||||
worktreeId: 'worktree-1',
|
||||
terminalHandle: orphanHandle
|
||||
}
|
||||
] as never)
|
||||
|
||||
await expect(
|
||||
runtime.createAgentSession(request(id), { clientId: 'device-a' })
|
||||
).resolves.toMatchObject({
|
||||
disposition: 'replayed',
|
||||
terminal: { handle: orphanHandle, ptyId: 'ssh-1:pty2:e:1', worktreeId: 'worktree-1' }
|
||||
})
|
||||
expect(listProcesses).toHaveBeenCalledWith('ssh-1')
|
||||
expect(createTerminal).toHaveBeenCalledOnce()
|
||||
expect(failure.message).toBe('execution_owner_unavailable')
|
||||
})
|
||||
|
||||
it('replays the fenced failure when host inventory proves the spawn is gone', async () => {
|
||||
const runtime = createRuntime()
|
||||
const listProcesses = vi.fn(async () => [] as never[])
|
||||
installRemoteReclaimHarness(runtime, listProcesses)
|
||||
const { createTerminal, id, failure } = await fenceRemoteAgentSessionSpawn(runtime)
|
||||
|
||||
await expect(runtime.createAgentSession(request(id), { clientId: 'device-a' })).rejects.toThrow(
|
||||
failure.message
|
||||
)
|
||||
expect(listProcesses).toHaveBeenCalledWith('ssh-1')
|
||||
expect(createTerminal).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('replays the fenced failure when the remote host cannot answer', async () => {
|
||||
const runtime = createRuntime()
|
||||
const listProcesses = vi.fn(async () => {
|
||||
throw new Error('relay offline')
|
||||
})
|
||||
installRemoteReclaimHarness(runtime, listProcesses)
|
||||
const { createTerminal, id, failure } = await fenceRemoteAgentSessionSpawn(runtime)
|
||||
|
||||
await expect(runtime.createAgentSession(request(id), { clientId: 'device-a' })).rejects.toThrow(
|
||||
failure.message
|
||||
)
|
||||
expect(createTerminal).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('refuses to adopt a same-handle PTY that belongs to another workspace', async () => {
|
||||
const runtime = createRuntime()
|
||||
const listProcesses = vi.fn(async () => [] as never[])
|
||||
installRemoteReclaimHarness(runtime, listProcesses)
|
||||
const { createTerminal, id, failure } = await fenceRemoteAgentSessionSpawn(runtime)
|
||||
listProcesses.mockResolvedValue([
|
||||
{
|
||||
id: 'ssh-1:pty2:e:9',
|
||||
cwd: '/remote/worktree-2',
|
||||
title: 'codex',
|
||||
worktreeId: 'worktree-2',
|
||||
terminalHandle: createTerminal.mock.calls[0]?.[1]?.preAllocatedHandle
|
||||
}
|
||||
] as never)
|
||||
|
||||
await expect(runtime.createAgentSession(request(id), { clientId: 'device-a' })).rejects.toThrow(
|
||||
failure.message
|
||||
)
|
||||
expect(createTerminal).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('retains a replay fence when the provider reports an unknown spawn outcome', async () => {
|
||||
const runtime = createRuntime()
|
||||
const failure = Object.assign(new Error('cleanup could not prove exit'), {
|
||||
|
||||
@@ -24,6 +24,10 @@ import {
|
||||
} from '../../shared/tui-agent-launch-defaults'
|
||||
import { buildAgentDraftLaunchPlan, buildAgentStartupPlan } from '../../shared/tui-agent-startup'
|
||||
import type { RuntimeTerminalCreate } from '../../shared/runtime-types'
|
||||
import type {
|
||||
AgentSessionCreateOperation,
|
||||
AgentSessionCreateReclaimIdentity
|
||||
} from './runtime-terminal-contracts'
|
||||
import {
|
||||
deterministicAgentSessionUuid,
|
||||
isAgentSessionOperationOutcomeUnknown
|
||||
@@ -72,7 +76,16 @@ export class OrcaRuntimeWithCreateAgentSession extends OrcaRuntimeWithGetAgentSe
|
||||
if (existing.fingerprint !== requestFingerprint) {
|
||||
throw new Error('agent_session_operation_conflict')
|
||||
}
|
||||
const replayed = await existing.promise
|
||||
let replayed: RuntimeCreateAgentSessionResult
|
||||
try {
|
||||
replayed = await existing.promise
|
||||
} catch (error) {
|
||||
const reclaimed = await this.reclaimFencedAgentSessionSpawn(existing.reclaim.identity)
|
||||
if (!reclaimed) {
|
||||
throw error
|
||||
}
|
||||
return { terminal: reclaimed, disposition: 'replayed' }
|
||||
}
|
||||
return { ...replayed, disposition: 'replayed' }
|
||||
}
|
||||
if (now - operationTimestamp > AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS) {
|
||||
@@ -96,6 +109,7 @@ export class OrcaRuntimeWithCreateAgentSession extends OrcaRuntimeWithGetAgentSe
|
||||
throw new Error('agent_session_operation_capacity')
|
||||
}
|
||||
let retainReplayFence = false
|
||||
const reclaim: AgentSessionCreateOperation['reclaim'] = {}
|
||||
const operation = (async (): Promise<RuntimeCreateAgentSessionResult> => {
|
||||
// Why: reserve the client operation before any async preflight so concurrent retries cannot
|
||||
// both observe an empty ledger and reach the execution owner independently.
|
||||
@@ -187,6 +201,13 @@ export class OrcaRuntimeWithCreateAgentSession extends OrcaRuntimeWithGetAgentSe
|
||||
const operationLeafId =
|
||||
request.placement?.leafId ?? deterministicAgentSessionUuid(`${executionOperationId}:leaf`)
|
||||
const operationHandle = `term_${deterministicAgentSessionUuid(`${executionOperationId}:handle`)}`
|
||||
// Why: recorded before dispatch — this handle is exported into the PTY as
|
||||
// ORCA_TERMINAL_HANDLE, so it is the only name a lost spawn can be re-found by.
|
||||
reclaim.identity = {
|
||||
worktreeId: workspace.id,
|
||||
connectionId: workspace.connectionId ?? null,
|
||||
terminalHandle: operationHandle
|
||||
}
|
||||
try {
|
||||
terminal = await this.createTerminal(`id:${workspace.id}`, {
|
||||
command: startup.launchCommand,
|
||||
@@ -216,7 +237,8 @@ export class OrcaRuntimeWithCreateAgentSession extends OrcaRuntimeWithGetAgentSe
|
||||
})()
|
||||
this.agentSessionCreateOperations.set(operationKey, {
|
||||
fingerprint: requestFingerprint,
|
||||
promise: operation
|
||||
promise: operation,
|
||||
reclaim
|
||||
})
|
||||
const expireOperation = (): void => {
|
||||
const expiresAt = Math.max(now, operationTimestamp) + AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS
|
||||
@@ -245,4 +267,25 @@ export class OrcaRuntimeWithCreateAgentSession extends OrcaRuntimeWithGetAgentSe
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
// Why: the host may still hold the PTY this operation launched. Adoption-only —
|
||||
// this never spawns and never kills, so an unreachable or silent host just replays
|
||||
// the original failure instead of authorising anything.
|
||||
private async reclaimFencedAgentSessionSpawn(
|
||||
identity: AgentSessionCreateReclaimIdentity | undefined
|
||||
): Promise<RuntimeTerminalCreate | null> {
|
||||
if (!identity) {
|
||||
return null
|
||||
}
|
||||
try {
|
||||
return await this.reconcileRemoteTerminalCreate(
|
||||
identity.worktreeId,
|
||||
identity.terminalHandle,
|
||||
identity.connectionId
|
||||
)
|
||||
} catch {
|
||||
// Unverifiable or ambiguous inventory is never evidence the PTY exited.
|
||||
return null
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -40,7 +40,14 @@ export class OrcaRuntimeWithTerminalCreateDeduplication extends OrcaRuntimeWithC
|
||||
clientMutationId,
|
||||
async () => {
|
||||
if (reconcileExisting) {
|
||||
const adopted = await this.reconcileRemoteTerminalCreate(workspace.id, preAllocatedHandle)
|
||||
const adopted = await this.reconcileRemoteTerminalCreate(
|
||||
workspace.id,
|
||||
preAllocatedHandle,
|
||||
// Why: an unreachable SSH host vanishes from the aggregate listing, which would read
|
||||
// as absence and respawn over live remote work. Local/folder workspaces have no
|
||||
// connection and keep the aggregate listing.
|
||||
workspace.connectionId ?? null
|
||||
)
|
||||
if (adopted) {
|
||||
return adopted
|
||||
}
|
||||
@@ -52,13 +59,16 @@ export class OrcaRuntimeWithTerminalCreateDeduplication extends OrcaRuntimeWithC
|
||||
|
||||
protected async reconcileRemoteTerminalCreate(
|
||||
worktreeId: string,
|
||||
terminalHandle: string
|
||||
terminalHandle: string,
|
||||
// Why: an aggregate listing drops a non-answering SSH host silently, which would read as
|
||||
// absence. Scoping to the owning host makes an unreachable relay throw instead.
|
||||
connectionId?: string | null
|
||||
): Promise<RuntimeTerminalCreate | null> {
|
||||
if (!this.ptyController?.listProcesses) {
|
||||
throw new Error('runtime_unavailable')
|
||||
}
|
||||
const listed = await withTimeoutResult(
|
||||
this.ptyController.listProcesses(),
|
||||
this.ptyController.listProcesses(connectionId),
|
||||
PTY_CONTROLLER_LIST_TIMEOUT_MS
|
||||
)
|
||||
if (!listed.ok) {
|
||||
|
||||
@@ -10,14 +10,18 @@ type CreateRun = (
|
||||
preAllocatedHandle: string | undefined
|
||||
) => Promise<RuntimeTerminalCreate>
|
||||
|
||||
function createRuntimeForDedupe(listProcesses = vi.fn(async (): Promise<PtyProcessInfo[]> => [])) {
|
||||
function createRuntimeForDedupe(
|
||||
listProcesses = vi.fn(async (): Promise<PtyProcessInfo[]> => []),
|
||||
scope: { connectionId?: string | null } = {}
|
||||
) {
|
||||
const handleByPtyId = new Map<string, string>()
|
||||
const runtime = Object.create(OrcaRuntimeService.prototype) as OrcaRuntimeService
|
||||
Object.assign(runtime, {
|
||||
terminalCreateIdempotency: new RemoteRuntimeTerminalCreateIdempotency(),
|
||||
ptyController: { listProcesses },
|
||||
resolveTerminalWorkspaceLaunchScope: vi.fn(async (selector: string) => ({
|
||||
id: selector.startsWith('id:') ? selector.slice(3) : selector
|
||||
id: selector.startsWith('id:') ? selector.slice(3) : selector,
|
||||
...scope
|
||||
})),
|
||||
adoptControllerTerminalHandle: vi.fn((ptyId: string, handle: string) => {
|
||||
handleByPtyId.set(ptyId, handle)
|
||||
@@ -253,3 +257,126 @@ describe('terminal create idempotency', () => {
|
||||
).resolves.toEqual(createdTerminal('terminal-2'))
|
||||
})
|
||||
})
|
||||
|
||||
// Mirrors listProcessesFromRuntimeController: `undefined` aggregates every provider and
|
||||
// silently drops a non-answering SSH host, `null` is local-only, a string is host-scoped
|
||||
// and rethrows the host's failure.
|
||||
function createHostScopedInventory(hosts: {
|
||||
local?: PtyProcessInfo[]
|
||||
ssh?: Record<string, PtyProcessInfo[] | 'unreachable'>
|
||||
}) {
|
||||
const local = hosts.local ?? []
|
||||
const ssh = hosts.ssh ?? {}
|
||||
return vi.fn(async (connectionId?: string | null): Promise<PtyProcessInfo[]> => {
|
||||
if (connectionId === null) {
|
||||
return local
|
||||
}
|
||||
if (typeof connectionId === 'string') {
|
||||
const host = ssh[connectionId]
|
||||
if (host === undefined || host === 'unreachable') {
|
||||
throw new Error('ssh relay did not answer')
|
||||
}
|
||||
return host
|
||||
}
|
||||
return [
|
||||
...local,
|
||||
...Object.values(ssh)
|
||||
.filter((sessions): sessions is PtyProcessInfo[] => sessions !== 'unreachable')
|
||||
.flat()
|
||||
]
|
||||
})
|
||||
}
|
||||
|
||||
function remoteSession(handle: string | undefined, worktreeId = 'worktree-1'): PtyProcessInfo {
|
||||
return {
|
||||
id: `${worktreeId}@@session-a`,
|
||||
cwd: '/remote/workspace',
|
||||
title: 'claude',
|
||||
worktreeId,
|
||||
...(handle ? { terminalHandle: handle } : {})
|
||||
}
|
||||
}
|
||||
|
||||
describe('terminal create reconciliation scopes inventory to the owning execution host', () => {
|
||||
it('reports runtime_unavailable instead of spawning a duplicate when the owning relay cannot answer', async () => {
|
||||
const listProcesses = createHostScopedInventory({
|
||||
// The first create's shell is alive on ssh-1; the relay simply cannot be asked about it.
|
||||
ssh: { 'ssh-1': 'unreachable', 'ssh-2': [remoteSession(undefined, 'worktree-9')] }
|
||||
})
|
||||
const { runtime } = createRuntimeForDedupe(listProcesses, { connectionId: 'ssh-1' })
|
||||
const create = vi.fn<CreateRun>()
|
||||
|
||||
await expect(
|
||||
runtime.dedupeTerminalCreate('device-a', 'id:worktree-1', 'mutation-1', true, create)
|
||||
).rejects.toThrow('runtime_unavailable')
|
||||
expect(create).not.toHaveBeenCalled()
|
||||
expect(listProcesses).toHaveBeenCalledWith('ssh-1')
|
||||
})
|
||||
|
||||
it('adopts the original PTY from the owning host listing', async () => {
|
||||
const handle = deriveRemoteRuntimeTerminalCreateHandle('device-a', 'worktree-1', 'mutation-1')
|
||||
const listProcesses = createHostScopedInventory({ ssh: { 'ssh-1': [remoteSession(handle)] } })
|
||||
const { runtime } = createRuntimeForDedupe(listProcesses, { connectionId: 'ssh-1' })
|
||||
const create = vi.fn<CreateRun>()
|
||||
|
||||
await expect(
|
||||
runtime.dedupeTerminalCreate('device-a', 'id:worktree-1', 'mutation-1', true, create)
|
||||
).resolves.toMatchObject({ handle, ptyId: 'worktree-1@@session-a' })
|
||||
expect(create).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('still creates a fresh terminal when the owning host authoritatively lacks the handle', async () => {
|
||||
const listProcesses = createHostScopedInventory({
|
||||
ssh: { 'ssh-1': [remoteSession(undefined, 'worktree-other')] }
|
||||
})
|
||||
const { runtime } = createRuntimeForDedupe(listProcesses, { connectionId: 'ssh-1' })
|
||||
const create = vi.fn<CreateRun>(async (_selector, handle) =>
|
||||
createdTerminal(handle ?? 'missing')
|
||||
)
|
||||
|
||||
const result = await runtime.dedupeTerminalCreate(
|
||||
'device-a',
|
||||
'id:worktree-1',
|
||||
'mutation-1',
|
||||
true,
|
||||
create
|
||||
)
|
||||
|
||||
expect(create).toHaveBeenCalledWith('id:worktree-1', result.handle)
|
||||
expect(listProcesses).toHaveBeenCalledWith('ssh-1')
|
||||
})
|
||||
|
||||
it('scopes the listing to the local host for a workspace with no connection', async () => {
|
||||
const handle = deriveRemoteRuntimeTerminalCreateHandle('device-a', 'worktree-1', 'mutation-1')
|
||||
const listProcesses = createHostScopedInventory({
|
||||
local: [{ ...remoteSession(handle), cwd: '/local/workspace', title: 'pwsh' }]
|
||||
})
|
||||
const { runtime } = createRuntimeForDedupe(listProcesses, { connectionId: null })
|
||||
const create = vi.fn<CreateRun>()
|
||||
|
||||
await expect(
|
||||
runtime.dedupeTerminalCreate('device-a', 'id:worktree-1', 'mutation-1', true, create)
|
||||
).resolves.toMatchObject({ handle, ptyId: 'worktree-1@@session-a' })
|
||||
expect(listProcesses).toHaveBeenCalledWith(null)
|
||||
expect(create).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('scopes the listing to the local host for a folder workspace with no connection', async () => {
|
||||
const listProcesses = createHostScopedInventory({})
|
||||
const { runtime } = createRuntimeForDedupe(listProcesses, { connectionId: null })
|
||||
const create = vi.fn<CreateRun>(async (_selector, handle) =>
|
||||
createdTerminal(handle ?? 'missing', 'folder:folder-1')
|
||||
)
|
||||
|
||||
const result = await runtime.dedupeTerminalCreate(
|
||||
'device-a',
|
||||
'id:folder:folder-1',
|
||||
'mutation-1',
|
||||
true,
|
||||
create
|
||||
)
|
||||
|
||||
expect(create).toHaveBeenCalledWith('id:folder:folder-1', result.handle)
|
||||
expect(listProcesses).toHaveBeenCalledWith(null)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -54,9 +54,19 @@ export type TerminalCreateOptions = {
|
||||
deferMobileSessionPublish?: boolean
|
||||
}
|
||||
|
||||
/** Identity a fenced spawn can be re-found by in the execution host's own inventory. */
|
||||
export type AgentSessionCreateReclaimIdentity = {
|
||||
worktreeId: string
|
||||
connectionId: string | null
|
||||
terminalHandle: string
|
||||
}
|
||||
|
||||
export type AgentSessionCreateOperation = {
|
||||
fingerprint: string
|
||||
promise: Promise<RuntimeCreateAgentSessionResult>
|
||||
// Why: a lost pty.spawn response leaves the host holding a live PTY the client
|
||||
// never named; this is the name it was launched under, so a replay can adopt it.
|
||||
reclaim: { identity?: AgentSessionCreateReclaimIdentity }
|
||||
}
|
||||
|
||||
export type PtyForegroundAgentRefresh = {
|
||||
|
||||
Reference in New Issue
Block a user