mirror of
https://github.com/stablyai/orca.git
synced 2026-10-20 00:03:45 +00:00
fix(orchestration): close recovery review gaps
This commit is contained in:
@@ -156,4 +156,44 @@ describe('AgentHookServer authority evidence', () => {
|
||||
})
|
||||
).toBeNull()
|
||||
})
|
||||
|
||||
it('preserves hydrated authority when stale alias cleanup does not own the stable pane', async () => {
|
||||
const server = new AgentHookServer()
|
||||
servers.push(server)
|
||||
const launchToken = 'launch-before-restart'
|
||||
const launchTokenHash = createHash('sha256').update(launchToken).digest('hex')
|
||||
const hydrated = {
|
||||
paneKey: PANE_KEY,
|
||||
launchToken,
|
||||
tabId: 'tab-authority',
|
||||
worktreeId: 'repo::before',
|
||||
connectionId: 'ssh-target',
|
||||
payload: { state: 'working', prompt: 'before', agentType: 'codex' },
|
||||
receivedAt: 100,
|
||||
stateStartedAt: 100
|
||||
} satisfies AgentHookEventPayload & { receivedAt: number; stateStartedAt: number }
|
||||
server._getStateForTests().lastStatusByPaneKey.set(PANE_KEY, hydrated)
|
||||
server.registerPaneKeyAlias('tab-authority:0', PANE_KEY, 'old-pty')
|
||||
await server.start()
|
||||
server.ingestRemote(
|
||||
{
|
||||
paneKey: PANE_KEY,
|
||||
launchToken,
|
||||
tabId: 'tab-authority',
|
||||
worktreeId: 'repo::current',
|
||||
payload: { state: 'working', prompt: 'current', agentType: 'codex' }
|
||||
},
|
||||
'ssh-target'
|
||||
)
|
||||
|
||||
server.clearPaneKeyAliasesForPty('old-pty', { shouldClearStablePaneKey: () => false })
|
||||
|
||||
expect(
|
||||
server.attestCompatibilityAuthority({
|
||||
paneKey: PANE_KEY,
|
||||
launchTokenHash,
|
||||
connectionId: 'ssh-target'
|
||||
})
|
||||
).toEqual({ paneKey: PANE_KEY, source: 'current_hook' })
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1441,13 +1441,17 @@ export class AgentHookServer {
|
||||
const clearedStatusPaneKeys = new Set<string>()
|
||||
for (const [legacyPaneKey, entry] of this.legacyPaneKeyAliases) {
|
||||
if (entry.ptyId === ptyId) {
|
||||
this.revokeHydratedAuthorityForPaneKeys(new Set([legacyPaneKey, entry.stablePaneKey]))
|
||||
const shouldClearStablePaneKey =
|
||||
options?.shouldClearStablePaneKey?.(entry.stablePaneKey) ?? true
|
||||
const revokedPaneKeys = new Set([legacyPaneKey])
|
||||
if (shouldClearStablePaneKey) {
|
||||
revokedPaneKeys.add(entry.stablePaneKey)
|
||||
}
|
||||
this.revokeHydratedAuthorityForPaneKeys(revokedPaneKeys)
|
||||
this.legacyPaneKeyAliases.delete(legacyPaneKey)
|
||||
clearPaneCacheState(this.state, legacyPaneKey)
|
||||
this.currentAuthorityObservations.delete(legacyPaneKey)
|
||||
this.promptSentDedupeByPaneKey.delete(legacyPaneKey)
|
||||
const shouldClearStablePaneKey =
|
||||
options?.shouldClearStablePaneKey?.(entry.stablePaneKey) ?? true
|
||||
if (shouldClearStablePaneKey && this.state.lastStatusByPaneKey.has(entry.stablePaneKey)) {
|
||||
statusChanged = true
|
||||
clearedStatusPaneKeys.add(entry.stablePaneKey)
|
||||
|
||||
@@ -3691,6 +3691,21 @@ describe('Store', () => {
|
||||
expect(hostSession.lastVisitedAtByWorktreeId?.['r1::/path/wt1']).toBeUndefined()
|
||||
})
|
||||
|
||||
it('lists only persisted workspace-session host partitions', async () => {
|
||||
const store = await createStore()
|
||||
expect(store.getWorkspaceSessionHostIds()).toEqual(['local'])
|
||||
|
||||
store.getWorkspaceSession('ssh:not-persisted')
|
||||
store.setWorkspaceSession(getDefaultWorkspaceSession(), 'ssh:ssh-a')
|
||||
store.setWorkspaceSession(getDefaultWorkspaceSession(), 'runtime:environment-a')
|
||||
|
||||
expect(store.getWorkspaceSessionHostIds()).toEqual([
|
||||
'local',
|
||||
'ssh:ssh-a',
|
||||
'runtime:environment-a'
|
||||
])
|
||||
})
|
||||
|
||||
it('removeProject removes the derived project host setup compatibility record', async () => {
|
||||
const store = await createStore()
|
||||
store.addRepo(makeRepo({ id: 'r1' }))
|
||||
|
||||
@@ -5769,6 +5769,17 @@ export class Store {
|
||||
return this.state.workspaceSessionsByHostId?.[resolved] ?? getDefaultWorkspaceSession()
|
||||
}
|
||||
|
||||
getWorkspaceSessionHostIds(): ExecutionHostId[] {
|
||||
const hostIds = new Set<ExecutionHostId>([LOCAL_EXECUTION_HOST_ID])
|
||||
for (const key of Object.keys(this.state.workspaceSessionsByHostId ?? {})) {
|
||||
const hostId = normalizeExecutionHostId(key)
|
||||
if (hostId) {
|
||||
hostIds.add(hostId)
|
||||
}
|
||||
}
|
||||
return [...hostIds]
|
||||
}
|
||||
|
||||
readTerminalScrollbackSnapshot(ref: string): string | null {
|
||||
return readTerminalScrollbackSnapshotSync(ref, this.terminalScrollbackSnapshotStorage)
|
||||
}
|
||||
|
||||
@@ -17033,6 +17033,210 @@ describe('OrcaRuntimeService', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('adopts an SSH folder legacy worker through its SSH workspace-session partition', async () => {
|
||||
const connectionId = 'ssh-folder'
|
||||
const ptyId = `ssh:${connectionId}@@pty-folder-legacy`
|
||||
const workerPaneKey = `legacy-ssh-folder-worker:${HEADLESS_LEAF_ID}`
|
||||
const incarnationId = '67676767-6767-4767-8767-676767676767'
|
||||
const folderPath = '/srv/platform'
|
||||
const sshInitialSession: WorkspaceSessionState = {
|
||||
...getDefaultWorkspaceSession(),
|
||||
activeWorktreeId: TEST_FOLDER_WORKSPACE_KEY,
|
||||
tabsByWorktree: { [TEST_FOLDER_WORKSPACE_KEY]: [] },
|
||||
sleepingAgentSessionsByPaneKey: {
|
||||
[workerPaneKey]: {
|
||||
paneKey: workerPaneKey,
|
||||
tabId: 'legacy-ssh-folder-worker',
|
||||
worktreeId: TEST_FOLDER_WORKSPACE_KEY,
|
||||
agent: 'codex',
|
||||
providerSession: { key: 'session_id', id: 'legacy-ssh-folder-codex-session' },
|
||||
prompt: 'continue',
|
||||
state: 'working',
|
||||
capturedAt: 1,
|
||||
updatedAt: 1,
|
||||
origin: 'live',
|
||||
connectionId
|
||||
}
|
||||
}
|
||||
}
|
||||
const { runtimeStore } = makeRuntimeStoreWithWorkspaceSession(sshInitialSession)
|
||||
const localSession = getDefaultWorkspaceSession()
|
||||
let sshSession = sshInitialSession
|
||||
const getWorkspaceSession = vi.fn((hostId?: string | null) =>
|
||||
hostId === `ssh:${connectionId}` ? sshSession : localSession
|
||||
)
|
||||
const setWorkspaceSession = vi.fn((next: WorkspaceSessionState, hostId?: string | null) => {
|
||||
if (hostId !== `ssh:${connectionId}`) {
|
||||
throw new Error(`unexpected workspace-session host ${hostId ?? 'default'}`)
|
||||
}
|
||||
sshSession = next
|
||||
})
|
||||
const folderWorkspace = makeFolderWorkspace({ folderPath, connectionId })
|
||||
const projectGroup = makeFolderProjectGroup({ parentPath: folderPath })
|
||||
const runtime = new OrcaRuntimeService(
|
||||
{
|
||||
...runtimeStore,
|
||||
getFolderWorkspaces: () => [folderWorkspace],
|
||||
getProjectGroups: () => [projectGroup],
|
||||
getWorkspaceSession,
|
||||
setWorkspaceSession,
|
||||
flushOrThrow: vi.fn()
|
||||
} as never,
|
||||
undefined,
|
||||
{ canRecoverPersistentLocalPtys: () => true }
|
||||
)
|
||||
runtime.setOrchestrationDb({
|
||||
listLegacyWorkerTerminalRecoveryRows: () => [
|
||||
{
|
||||
dispatch_id: 'dispatch-ssh-folder',
|
||||
task_id: 'task-ssh-folder',
|
||||
dispatch_status: 'completed',
|
||||
contract_version: 0,
|
||||
assignee_handle: 'term_ssh_folder',
|
||||
assignee_pane_key: workerPaneKey,
|
||||
process_incarnation: `${ptyId}:${incarnationId}`,
|
||||
worker_state: 'ready',
|
||||
worktree_id: TEST_FOLDER_WORKSPACE_KEY,
|
||||
agent_terminal_handle: 'term_ssh_folder'
|
||||
}
|
||||
]
|
||||
} as unknown as OrchestrationDb)
|
||||
runtime.setPtyController({
|
||||
write: vi.fn(() => true),
|
||||
kill: vi.fn(() => true),
|
||||
getForegroundProcess: async () => null,
|
||||
listProcesses: async () => [
|
||||
{
|
||||
id: ptyId,
|
||||
incarnationId,
|
||||
terminalHandle: 'term_ssh_folder',
|
||||
title: 'SSH folder worker',
|
||||
cwd: folderPath,
|
||||
worktreeId: TEST_FOLDER_WORKSPACE_KEY,
|
||||
wslDistro: null
|
||||
}
|
||||
]
|
||||
})
|
||||
const revealTerminalSession = vi.fn().mockResolvedValue({ tabId: 'legacy-ssh-folder-worker' })
|
||||
runtime.setNotifier({ revealTerminalSession } as never)
|
||||
registerSshFilesystemProvider(connectionId, {
|
||||
stat: vi.fn(async () => ({ size: 0, type: 'directory', mtime: 1 }))
|
||||
} as never)
|
||||
|
||||
try {
|
||||
expect(runtime.prepareLegacyWorkerTerminalRecovery()).toMatchObject({
|
||||
blockedPanes: [expect.objectContaining({ paneKey: workerPaneKey })]
|
||||
})
|
||||
expect(
|
||||
sshSession.sleepingAgentSessionsByPaneKey?.[workerPaneKey]?.automaticResumeBlockedBy
|
||||
).toBe('legacy-orchestration-worker')
|
||||
expect(localSession.sleepingAgentSessionsByPaneKey?.[workerPaneKey]).toBeUndefined()
|
||||
await expect(
|
||||
runtime.reconcileLegacyWorkerTerminals({
|
||||
connectionId,
|
||||
materializeRenderer: true
|
||||
})
|
||||
).resolves.toMatchObject({
|
||||
adoptedDispatchIds: ['dispatch-ssh-folder'],
|
||||
exitedDispatchIds: [],
|
||||
deferredDispatchIds: []
|
||||
})
|
||||
} finally {
|
||||
unregisterSshFilesystemProvider(connectionId)
|
||||
}
|
||||
|
||||
expect(sshSession.sleepingAgentSessionsByPaneKey?.[workerPaneKey]).toBeUndefined()
|
||||
expect(localSession.sleepingAgentSessionsByPaneKey?.[workerPaneKey]).toBeUndefined()
|
||||
expect(getWorkspaceSession).toHaveBeenCalledWith(`ssh:${connectionId}`)
|
||||
expect(setWorkspaceSession).toHaveBeenCalledWith(expect.any(Object), `ssh:${connectionId}`)
|
||||
expect(sshSession.tabsByWorktree[TEST_FOLDER_WORKSPACE_KEY]).toContainEqual(
|
||||
expect.objectContaining({
|
||||
id: 'legacy-ssh-folder-worker',
|
||||
ptyId,
|
||||
worktreeId: TEST_FOLDER_WORKSPACE_KEY
|
||||
})
|
||||
)
|
||||
expect(revealTerminalSession).toHaveBeenCalledWith(TEST_FOLDER_WORKSPACE_KEY, {
|
||||
ptyId,
|
||||
title: 'SSH folder worker',
|
||||
activate: false,
|
||||
presentation: 'background',
|
||||
tabId: 'legacy-ssh-folder-worker',
|
||||
leafId: HEADLESS_LEAF_ID,
|
||||
focus: false
|
||||
})
|
||||
})
|
||||
|
||||
it('fences an unresolved folder legacy worker in its exact retained session partition', () => {
|
||||
const connectionId = 'ssh-unresolved-folder'
|
||||
const worktreeId = 'folder:missing-folder'
|
||||
const workerPaneKey = `legacy-unresolved-folder-worker:${HEADLESS_LEAF_ID}`
|
||||
const remoteInitialSession: WorkspaceSessionState = {
|
||||
...getDefaultWorkspaceSession(),
|
||||
tabsByWorktree: { [worktreeId]: [] },
|
||||
sleepingAgentSessionsByPaneKey: {
|
||||
[workerPaneKey]: {
|
||||
paneKey: workerPaneKey,
|
||||
tabId: 'legacy-unresolved-folder-worker',
|
||||
worktreeId,
|
||||
agent: 'codex',
|
||||
providerSession: { key: 'session_id', id: 'legacy-unresolved-folder-session' },
|
||||
prompt: 'continue',
|
||||
state: 'working',
|
||||
capturedAt: 1,
|
||||
updatedAt: 1,
|
||||
origin: 'live',
|
||||
connectionId
|
||||
}
|
||||
}
|
||||
}
|
||||
const localSession = getDefaultWorkspaceSession()
|
||||
let remoteSession = remoteInitialSession
|
||||
const getWorkspaceSession = vi.fn((hostId?: string | null) =>
|
||||
hostId === `ssh:${connectionId}` ? remoteSession : localSession
|
||||
)
|
||||
const setWorkspaceSession = vi.fn((next: WorkspaceSessionState, hostId?: string | null) => {
|
||||
if (hostId !== `ssh:${connectionId}`) {
|
||||
throw new Error(`unexpected workspace-session host ${hostId ?? 'default'}`)
|
||||
}
|
||||
remoteSession = next
|
||||
})
|
||||
const runtime = new OrcaRuntimeService({
|
||||
...store,
|
||||
getFolderWorkspaces: () => [],
|
||||
getWorkspaceSession,
|
||||
getWorkspaceSessionHostIds: () => ['local', `ssh:${connectionId}`],
|
||||
setWorkspaceSession,
|
||||
flushOrThrow: vi.fn()
|
||||
} as never)
|
||||
runtime.setOrchestrationDb({
|
||||
listLegacyWorkerTerminalRecoveryRows: () => [
|
||||
{
|
||||
dispatch_id: 'dispatch-unresolved-folder',
|
||||
task_id: 'task-unresolved-folder',
|
||||
dispatch_status: 'completed',
|
||||
contract_version: 0,
|
||||
assignee_handle: 'term_unresolved_folder',
|
||||
assignee_pane_key: workerPaneKey,
|
||||
process_incarnation: 'pty-unresolved-folder:68686868-6868-4868-8868-686868686868',
|
||||
worker_state: 'ready',
|
||||
worktree_id: worktreeId,
|
||||
agent_terminal_handle: 'term_unresolved_folder'
|
||||
}
|
||||
]
|
||||
} as unknown as OrchestrationDb)
|
||||
|
||||
expect(runtime.prepareLegacyWorkerTerminalRecovery()).toMatchObject({
|
||||
blockedPanes: [expect.objectContaining({ paneKey: workerPaneKey, worktreeId })]
|
||||
})
|
||||
expect(
|
||||
remoteSession.sleepingAgentSessionsByPaneKey?.[workerPaneKey]?.automaticResumeBlockedBy
|
||||
).toBe('legacy-orchestration-worker')
|
||||
expect(localSession.sleepingAgentSessionsByPaneKey?.[workerPaneKey]).toBeUndefined()
|
||||
expect(setWorkspaceSession).toHaveBeenCalledOnce()
|
||||
expect(setWorkspaceSession).toHaveBeenCalledWith(expect.any(Object), `ssh:${connectionId}`)
|
||||
})
|
||||
|
||||
it('adopts an SSH legacy worker only after its matching relay is ready', async () => {
|
||||
const connectionId = 'ssh-legacy-worker'
|
||||
const ptyId = `ssh:${connectionId}@@pty-legacy-worker`
|
||||
|
||||
@@ -1027,6 +1027,7 @@ type RuntimeStore = {
|
||||
removeWorkspaceLineage?: Store['removeWorkspaceLineage']
|
||||
getGitHubCache: Store['getGitHubCache']
|
||||
getWorkspaceSession?: Store['getWorkspaceSession']
|
||||
getWorkspaceSessionHostIds?: Store['getWorkspaceSessionHostIds']
|
||||
setWorkspaceSession?: Store['setWorkspaceSession']
|
||||
flushOrThrow?: Store['flushOrThrow']
|
||||
persistPtyBinding?: Store['persistPtyBinding']
|
||||
@@ -3585,32 +3586,43 @@ export class OrcaRuntimeService {
|
||||
>()
|
||||
const changedHostIds = new Set<ExecutionHostId>()
|
||||
for (const blocked of plan.blockedPanes) {
|
||||
const hostId = this.getWorkspaceSessionHostIdForWorktree(blocked.worktreeId)
|
||||
let state = sessions.get(hostId)
|
||||
if (!state) {
|
||||
const current = store.getWorkspaceSession(hostId)
|
||||
if (!current) {
|
||||
let hostIds: ExecutionHostId[]
|
||||
try {
|
||||
hostIds = [this.getWorkspaceSessionHostIdForWorktree(blocked.worktreeId)]
|
||||
} catch (error) {
|
||||
console.warn('[orchestration] legacy worker resume fence owner is unavailable', {
|
||||
worktreeId: blocked.worktreeId,
|
||||
error
|
||||
})
|
||||
hostIds = store.getWorkspaceSessionHostIds?.() ?? [LOCAL_EXECUTION_HOST_ID]
|
||||
}
|
||||
for (const hostId of hostIds) {
|
||||
let state = sessions.get(hostId)
|
||||
if (!state) {
|
||||
const current = store.getWorkspaceSession(hostId)
|
||||
if (!current) {
|
||||
continue
|
||||
}
|
||||
state = { current, next: structuredClone(current) }
|
||||
sessions.set(hostId, state)
|
||||
}
|
||||
const record = state.next.sleepingAgentSessionsByPaneKey?.[blocked.paneKey]
|
||||
if (
|
||||
!record ||
|
||||
!runtimeWorktreeIdsEqual(record.worktreeId, blocked.worktreeId) ||
|
||||
record.automaticResumeBlockedBy === 'legacy-orchestration-worker'
|
||||
) {
|
||||
continue
|
||||
}
|
||||
state = { current, next: structuredClone(current) }
|
||||
sessions.set(hostId, state)
|
||||
}
|
||||
const record = state.next.sleepingAgentSessionsByPaneKey?.[blocked.paneKey]
|
||||
if (
|
||||
!record ||
|
||||
!runtimeWorktreeIdsEqual(record.worktreeId, blocked.worktreeId) ||
|
||||
record.automaticResumeBlockedBy === 'legacy-orchestration-worker'
|
||||
) {
|
||||
continue
|
||||
}
|
||||
state.next.sleepingAgentSessionsByPaneKey = {
|
||||
...state.next.sleepingAgentSessionsByPaneKey,
|
||||
[blocked.paneKey]: {
|
||||
...record,
|
||||
automaticResumeBlockedBy: 'legacy-orchestration-worker'
|
||||
state.next.sleepingAgentSessionsByPaneKey = {
|
||||
...state.next.sleepingAgentSessionsByPaneKey,
|
||||
[blocked.paneKey]: {
|
||||
...record,
|
||||
automaticResumeBlockedBy: 'legacy-orchestration-worker'
|
||||
}
|
||||
}
|
||||
changedHostIds.add(hostId)
|
||||
}
|
||||
changedHostIds.add(hostId)
|
||||
}
|
||||
const changed = [...sessions].filter(([hostId]) => changedHostIds.has(hostId))
|
||||
if (changed.length === 0) {
|
||||
@@ -3981,8 +3993,20 @@ export class OrcaRuntimeService {
|
||||
}
|
||||
|
||||
private getWorkspaceSessionHostIdForWorktree(worktreeId: string): ExecutionHostId {
|
||||
const repo = this.store?.getRepo?.(getRepoIdFromWorktreeId(worktreeId))
|
||||
return repo ? getRepoExecutionHostId(repo) : 'local'
|
||||
const scope = parseWorkspaceKey(worktreeId)
|
||||
if (scope?.type === 'folder') {
|
||||
const workspace = this.store
|
||||
?.getFolderWorkspaces?.()
|
||||
.find((entry) => entry.id === scope.folderWorkspaceId)
|
||||
if (!workspace) {
|
||||
throw new Error('folder_workspace_not_found')
|
||||
}
|
||||
const connectionId = this.resolveFolderWorkspaceConnectionId(workspace)
|
||||
return connectionId ? toSshExecutionHostId(connectionId) : LOCAL_EXECUTION_HOST_ID
|
||||
}
|
||||
const resolvedWorktreeId = scope?.type === 'worktree' ? scope.worktreeId : worktreeId
|
||||
const repo = this.store?.getRepo?.(getRepoIdFromWorktreeId(resolvedWorktreeId))
|
||||
return repo ? getRepoExecutionHostId(repo) : LOCAL_EXECUTION_HOST_ID
|
||||
}
|
||||
|
||||
private getWorkspaceSessionForWorktree(worktreeId: string): WorkspaceSessionState | null {
|
||||
|
||||
Reference in New Issue
Block a user