mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 16:03:47 +00:00
fix(orchestration): apply the same honesty to the federation stop path
The federation host runs its own copy of the worker observation and stop logic, with the same two defects: `inspectRemoteAttachment` read a dropped relay's `connected: false` as `exited`, and `federationStop` settled the dispatch as stopped from a close it never confirmed — relaying a fabricated success all the way home to the coordinator. Two guards also had to move so the honest verdict does not become a new refusal. `federationRead` gated on `status !== 'running'`, which would have rejected a connected terminal the moment a stop lost contact with it; it now gates on `status === 'exited'`, which is equivalent for every pre-existing status given the two guards beside it. Local `workerStop` likewise still attempts the close when the verdict is `unverifiable` — losing contact is a reason to report the outcome honestly, never a reason to stop trying. The show observations now carry the reason alongside the status, so a bare `unverifiable` is actionable. Both are new optional fields.
This commit is contained in:
@@ -6,6 +6,7 @@ import type { RemoteDispatchAttachmentRow } from '../../orchestration/types'
|
||||
import { defineMethod, type RpcMethod } from '../core'
|
||||
import { OptionalFiniteNumber, requiredString } from '../schemas'
|
||||
import { readExactWorkerOutput } from './orchestration-worker-output'
|
||||
import { describeUnconfirmedAgentStop } from '../../../../shared/pty-liveness-verdict'
|
||||
|
||||
const FederationDispatchParams = z.object({
|
||||
dispatchId: requiredString('Missing Dispatch ID')
|
||||
@@ -36,7 +37,11 @@ export const ORCHESTRATION_FEDERATION_CONTROL_METHODS: RpcMethod[] = [
|
||||
runtimeEpoch: runtime.getRuntimeId(),
|
||||
attachment: exposeRemoteAttachment(attachment),
|
||||
terminal: observation.exact ? observation.terminal : null,
|
||||
observation: { status: observation.status, exactWorker: observation.exact }
|
||||
observation: {
|
||||
status: observation.status,
|
||||
exactWorker: observation.exact,
|
||||
...(observation.reason ? { reason: observation.reason } : {})
|
||||
}
|
||||
}
|
||||
}
|
||||
}),
|
||||
@@ -46,7 +51,10 @@ export const ORCHESTRATION_FEDERATION_CONTROL_METHODS: RpcMethod[] = [
|
||||
handler: async (params, { runtime, authenticatedCallerFingerprint }) => {
|
||||
requireHomeAttachment(runtime, params.dispatchId, authenticatedCallerFingerprint)
|
||||
const observation = await inspectRemoteAttachment(runtime, params.dispatchId)
|
||||
if (!observation.exact || !observation.terminal || observation.status !== 'running') {
|
||||
// Why `=== 'exited'` rather than `!== 'running'`: the other non-running
|
||||
// statuses are already covered by the two guards, and an unverifiable
|
||||
// terminal is still readable — losing stop-contact is not an exit.
|
||||
if (!observation.exact || !observation.terminal || observation.status === 'exited') {
|
||||
throw new OrchestrationError(
|
||||
'worker_identity_changed',
|
||||
`Remote Dispatch ${params.dispatchId} no longer resolves to its exact process.`
|
||||
@@ -134,6 +142,22 @@ export const ORCHESTRATION_FEDERATION_CONTROL_METHODS: RpcMethod[] = [
|
||||
}
|
||||
try {
|
||||
const close = await runtime.closeTerminal(observation.terminal.handle)
|
||||
if (close.ptyStopVerdict) {
|
||||
// The tab is retired but the process was never confirmed stopped, so
|
||||
// the coordinator must not be told this dispatch reached 'stopped'.
|
||||
const attachment = db.markRemoteAttachmentStopUnknown(
|
||||
params.dispatchId,
|
||||
describeUnconfirmedAgentStop(close)
|
||||
)
|
||||
return {
|
||||
dispatchId: params.dispatchId,
|
||||
state: attachment.state,
|
||||
alreadySettled: false,
|
||||
processAction: 'closed_agent_terminal',
|
||||
lastError: attachment.last_error,
|
||||
close
|
||||
}
|
||||
}
|
||||
const attachment = db.settleRemoteAttachmentStop(params.dispatchId)
|
||||
return {
|
||||
dispatchId: params.dispatchId,
|
||||
@@ -172,29 +196,44 @@ function requireHomeAttachment(
|
||||
return attachment
|
||||
}
|
||||
|
||||
async function inspectRemoteAttachment(runtime: OrcaRuntimeService, dispatchId: string) {
|
||||
async function inspectRemoteAttachment(
|
||||
runtime: OrcaRuntimeService,
|
||||
dispatchId: string
|
||||
): Promise<{
|
||||
terminal: Awaited<ReturnType<OrcaRuntimeService['showTerminal']>> | null
|
||||
exact: boolean
|
||||
status: 'unattached' | 'missing' | 'identity_changed' | 'running' | 'exited' | 'unverifiable'
|
||||
/** Set with `unverifiable`; names what we lost contact with. */
|
||||
reason?: string
|
||||
}> {
|
||||
const db = runtime.getOrchestrationDb()
|
||||
const attachment = db.getRemoteDispatchAttachment(dispatchId)
|
||||
if (!attachment?.terminal_handle) {
|
||||
return { terminal: null, exact: false, status: 'unattached' as const }
|
||||
return { terminal: null, exact: false, status: 'unattached' }
|
||||
}
|
||||
const terminal = await runtime.showTerminal(attachment.terminal_handle).catch(() => null)
|
||||
if (!terminal) {
|
||||
return { terminal: null, exact: false, status: 'missing' as const }
|
||||
return { terminal: null, exact: false, status: 'missing' }
|
||||
}
|
||||
const exact = db.isRemoteAttachmentProcessCurrent({
|
||||
dispatchId,
|
||||
paneKey: runtime.getTerminalPaneKey(attachment.terminal_handle),
|
||||
processIncarnation: runtime.getTerminalProcessIncarnation(attachment.terminal_handle)
|
||||
})
|
||||
if (!exact) {
|
||||
return { terminal, exact, status: 'identity_changed' }
|
||||
}
|
||||
// Why: the same rule as the local worker observation — the inventory only
|
||||
// iterates registered providers, so a dropped relay clears `connected` for
|
||||
// every remote PTY at once. Lost contact is not a death certificate.
|
||||
const verdict = runtime.getTerminalLivenessVerdict?.(attachment.terminal_handle) ?? null
|
||||
if (verdict?.status === 'unverifiable') {
|
||||
return { terminal, exact, status: 'unverifiable', reason: verdict.reason }
|
||||
}
|
||||
return {
|
||||
terminal,
|
||||
exact,
|
||||
status: exact
|
||||
? terminal.connected === false
|
||||
? ('exited' as const)
|
||||
: ('running' as const)
|
||||
: ('identity_changed' as const)
|
||||
status: verdict?.status !== 'live' && terminal.connected === false ? 'exited' : 'running'
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { ORCHESTRATION_CONTRACT_VERSION } from '../../../../shared/protocol-version'
|
||||
import { OrcaRuntimeService } from '../../orca-runtime'
|
||||
import { OrchestrationDb } from '../../orchestration/db'
|
||||
import { ORCHESTRATION_METHODS } from './orchestration'
|
||||
|
||||
// The federation host runs its own copy of the observation and stop logic, so
|
||||
// it needs the same rule: lost contact with a worker's host is not an exit, and
|
||||
// a close it could not confirm must not be relayed home as a settled stop.
|
||||
|
||||
const HOME_FINGERPRINT = 'home-peer-fingerprint'
|
||||
const DISPATCH_ID = 'ctx_federation_verdict'
|
||||
const HANDLE = 'term_remote_worker'
|
||||
const PANE_KEY = 'tab_remote:bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb'
|
||||
const INCARNATION = 'runtime:pty:7'
|
||||
const SSH_PROVIDER_GONE = 'its SSH provider is no longer registered'
|
||||
|
||||
describe('federation host liveness verdicts', () => {
|
||||
let db: OrchestrationDb
|
||||
let runtime: OrcaRuntimeService
|
||||
|
||||
beforeEach(() => {
|
||||
db = new OrchestrationDb(':memory:')
|
||||
runtime = new OrcaRuntimeService()
|
||||
runtime.setOrchestrationDb(db)
|
||||
vi.spyOn(runtime, 'getTerminalPaneKey').mockReturnValue(PANE_KEY)
|
||||
vi.spyOn(runtime, 'getTerminalProcessIncarnation').mockReturnValue(INCARNATION)
|
||||
vi.spyOn(runtime, 'showTerminal').mockResolvedValue({
|
||||
handle: HANDLE,
|
||||
worktreeId: 'repo::remote-worktree',
|
||||
connected: false,
|
||||
status: 'exited'
|
||||
} as never)
|
||||
db.createRemoteDispatchAttachment({
|
||||
dispatchId: DISPATCH_ID,
|
||||
taskId: 'task_remote',
|
||||
homePeerFingerprint: HOME_FINGERPRINT,
|
||||
protocolVersion: ORCHESTRATION_CONTRACT_VERSION,
|
||||
runtimeEpoch: runtime.getRuntimeId(),
|
||||
mutationReceipt: {
|
||||
callerFingerprint: HOME_FINGERPRINT,
|
||||
requestId: 'rpc_attach',
|
||||
method: 'orchestration.federationStart',
|
||||
payloadHash: 'hash'
|
||||
}
|
||||
})
|
||||
db.prepareRemoteAttachmentAuthority({
|
||||
dispatchId: DISPATCH_ID,
|
||||
paneKey: PANE_KEY,
|
||||
processIncarnation: INCARNATION,
|
||||
worktreeId: 'repo::remote-worktree',
|
||||
terminalHandle: HANDLE,
|
||||
setupState: 'not_applicable',
|
||||
effects: [{ kind: 'terminal', action: 'created', id: HANDLE }]
|
||||
})
|
||||
db.markRemoteAttachmentReady(DISPATCH_ID)
|
||||
})
|
||||
|
||||
afterEach(() => db.close())
|
||||
|
||||
async function call(name: string, params: Record<string, unknown>) {
|
||||
const method = ORCHESTRATION_METHODS.find((candidate) => candidate.name === name)
|
||||
if (!method) {
|
||||
throw new Error(`Method not found: ${name}`)
|
||||
}
|
||||
return method.handler(method.params!.parse(params), {
|
||||
runtime,
|
||||
authenticatedCallerFingerprint: HOME_FINGERPRINT
|
||||
} as never)
|
||||
}
|
||||
|
||||
it('reports lost contact as unverifiable rather than an observed exit', async () => {
|
||||
vi.spyOn(runtime, 'getTerminalLivenessVerdict').mockReturnValue({
|
||||
status: 'unverifiable',
|
||||
reason: SSH_PROVIDER_GONE
|
||||
})
|
||||
|
||||
await expect(
|
||||
call('orchestration.federationShow', { dispatchId: DISPATCH_ID })
|
||||
).resolves.toMatchObject({
|
||||
observation: { status: 'unverifiable', exactWorker: true, reason: SSH_PROVIDER_GONE }
|
||||
})
|
||||
})
|
||||
|
||||
it('still reports a locally observed exit as exited', async () => {
|
||||
await expect(
|
||||
call('orchestration.federationShow', { dispatchId: DISPATCH_ID })
|
||||
).resolves.toMatchObject({ observation: { status: 'exited', exactWorker: true } })
|
||||
})
|
||||
|
||||
it('still serves output for a terminal we merely lost stop-contact with', async () => {
|
||||
// Why this matters: the read gate used to be `status !== 'running'`, which
|
||||
// would refuse a connected terminal the moment a stop lost contact with it.
|
||||
vi.spyOn(runtime, 'showTerminal').mockResolvedValue({
|
||||
handle: HANDLE,
|
||||
worktreeId: 'repo::remote-worktree',
|
||||
connected: true
|
||||
} as never)
|
||||
vi.spyOn(runtime, 'getTerminalLivenessVerdict').mockReturnValue({
|
||||
status: 'unverifiable',
|
||||
reason: SSH_PROVIDER_GONE
|
||||
})
|
||||
|
||||
const outcome = await call('orchestration.federationRead', {
|
||||
dispatchId: DISPATCH_ID
|
||||
}).catch((error: unknown) => error)
|
||||
|
||||
expect(outcome).not.toMatchObject({ code: 'worker_identity_changed' })
|
||||
})
|
||||
|
||||
it('does not relay an unconfirmed close home as a settled stop', async () => {
|
||||
vi.spyOn(runtime, 'getTerminalLivenessVerdict').mockReturnValue({
|
||||
status: 'unverifiable',
|
||||
reason: SSH_PROVIDER_GONE
|
||||
})
|
||||
const closeTerminal = vi.spyOn(runtime, 'closeTerminal').mockResolvedValue({
|
||||
handle: HANDLE,
|
||||
tabId: 'tab_remote',
|
||||
ptyKilled: false,
|
||||
ptyStopVerdict: 'unverifiable',
|
||||
ptyStopReason: SSH_PROVIDER_GONE
|
||||
})
|
||||
|
||||
const stopped = (await call('orchestration.federationStop', { dispatchId: DISPATCH_ID })) as {
|
||||
state: string
|
||||
lastError?: string
|
||||
}
|
||||
|
||||
// Losing contact is a reason to report honestly, never to stop trying.
|
||||
expect(closeTerminal).toHaveBeenCalledWith(HANDLE)
|
||||
expect(stopped.state).not.toBe('stopped')
|
||||
expect(stopped.lastError).toContain('could not be confirmed stopped')
|
||||
})
|
||||
|
||||
it('still settles a confirmed close as a stop', async () => {
|
||||
vi.spyOn(runtime, 'closeTerminal').mockResolvedValue({
|
||||
handle: HANDLE,
|
||||
tabId: 'tab_remote',
|
||||
ptyKilled: true
|
||||
})
|
||||
|
||||
const stopped = (await call('orchestration.federationStop', { dispatchId: DISPATCH_ID })) as {
|
||||
state: string
|
||||
processAction: string
|
||||
}
|
||||
|
||||
expect(stopped.state).toBe('stopped')
|
||||
expect(stopped.processAction).toBe('closed_agent_terminal')
|
||||
})
|
||||
})
|
||||
@@ -122,7 +122,12 @@ export const ORCHESTRATION_WORKER_CONTROL_METHODS: RpcMethod[] = [
|
||||
dispatch,
|
||||
worker: exposeWorker(worker),
|
||||
terminal: observation.exact ? observation.terminal : null,
|
||||
observation: { status: observation.status, exactWorker: observation.exact },
|
||||
observation: {
|
||||
status: observation.status,
|
||||
exactWorker: observation.exact,
|
||||
// Why: a bare `unverifiable` is not actionable without naming what we lost.
|
||||
...(observation.reason ? { reason: observation.reason } : {})
|
||||
},
|
||||
terminalResource: resource ? exposeWorkerTerminalResource(resource) : null
|
||||
}
|
||||
}
|
||||
|
||||
@@ -86,7 +86,7 @@ export async function callFederatedWorkerShow(
|
||||
residualResources: unknown[]
|
||||
}
|
||||
terminal: unknown
|
||||
observation: { status: string; exactWorker: boolean }
|
||||
observation: { status: string; exactWorker: boolean; reason?: string }
|
||||
}> {
|
||||
return (await runtime.callOrchestrationWorkerServer(
|
||||
federated.environment_id,
|
||||
|
||||
@@ -66,17 +66,34 @@ describe('worker-stop against a terminal we lost contact with', () => {
|
||||
status: 'unverifiable',
|
||||
reason: 'its SSH provider is no longer registered'
|
||||
})
|
||||
const closeTerminal = vi.spyOn(runtime, 'closeTerminal').mockResolvedValue({
|
||||
handle: 'term_worker',
|
||||
tabId: 'tab_worker',
|
||||
ptyKilled: false,
|
||||
ptyStopVerdict: 'unverifiable',
|
||||
ptyStopReason: 'its SSH provider is no longer registered'
|
||||
})
|
||||
const dispatch = createWorker()
|
||||
|
||||
await expect(
|
||||
call('orchestration.workerShow', { dispatch: dispatch.id })
|
||||
).resolves.toMatchObject({
|
||||
observation: { status: 'unverifiable', exactWorker: true }
|
||||
observation: {
|
||||
status: 'unverifiable',
|
||||
exactWorker: true,
|
||||
reason: 'its SSH provider is no longer registered'
|
||||
}
|
||||
})
|
||||
|
||||
const stopped = (await call('orchestration.workerStop', { dispatch: dispatch.id })) as {
|
||||
state: string
|
||||
processAction: string
|
||||
lastError: string
|
||||
}
|
||||
// Losing contact is a reason to report honestly, never to stop trying.
|
||||
expect(closeTerminal).toHaveBeenCalledWith('term_worker')
|
||||
expect(stopped.processAction).toBe('closed_agent_terminal')
|
||||
expect(stopped.state).toBe('stop_unknown')
|
||||
expect(stopped.lastError).toContain('could not be confirmed stopped')
|
||||
expect(stopped.lastError).not.toContain('exited')
|
||||
})
|
||||
|
||||
@@ -2,6 +2,7 @@ import { z } from 'zod'
|
||||
import { OrchestrationError } from '../../orchestration/orchestration-error'
|
||||
import { defineMethod, type RpcMethod } from '../core'
|
||||
import { requiredString } from '../schemas'
|
||||
import { describeUnconfirmedAgentStop } from '../../../../shared/pty-liveness-verdict'
|
||||
import {
|
||||
inspectWorkerTerminal,
|
||||
resolvePinnedFederatedServer
|
||||
@@ -101,18 +102,17 @@ export const ORCHESTRATION_WORKER_STOP_METHODS: RpcMethod[] = [
|
||||
)
|
||||
}
|
||||
const observation = await inspectWorkerTerminal(runtime, db, params.dispatch)
|
||||
if (!observation.exact || observation.status !== 'running') {
|
||||
// Why `unverifiable` still proceeds: losing contact is a reason to report
|
||||
// the outcome honestly, never a reason to stop trying to stop the worker.
|
||||
if (
|
||||
!observation.exact ||
|
||||
(observation.status !== 'running' && observation.status !== 'unverifiable')
|
||||
) {
|
||||
return unknownReceipt(
|
||||
params.dispatch,
|
||||
db.markWorkerStopUnknown(
|
||||
params.dispatch,
|
||||
observation.status === 'unverifiable'
|
||||
? // Why: losing the link to the worker's host is not a death certificate,
|
||||
// so this must not settle as a stop or read as an observed exit.
|
||||
`The recorded worker process could not be confirmed stopped: ${
|
||||
observation.reason ?? 'its host could not be reached'
|
||||
}; no terminal was closed.`
|
||||
: `The recorded worker process is ${observation.status}; no terminal was closed.`
|
||||
`The recorded worker process is ${observation.status}; no terminal was closed.`
|
||||
),
|
||||
'none'
|
||||
)
|
||||
@@ -124,14 +124,7 @@ export const ORCHESTRATION_WORKER_STOP_METHODS: RpcMethod[] = [
|
||||
// settling here is the false success this receipt exists to prevent.
|
||||
return unknownReceipt(
|
||||
params.dispatch,
|
||||
db.markWorkerStopUnknown(
|
||||
params.dispatch,
|
||||
`The agent terminal was closed but its process could not be confirmed stopped: ${
|
||||
close.ptyStopVerdict === 'live'
|
||||
? 'it is still running'
|
||||
: (close.ptyStopReason ?? 'its host could not be reached')
|
||||
}.`
|
||||
),
|
||||
db.markWorkerStopUnknown(params.dispatch, describeUnconfirmedAgentStop(close)),
|
||||
'closed_agent_terminal'
|
||||
)
|
||||
}
|
||||
|
||||
@@ -13,10 +13,21 @@ export type PtyLivenessVerdict =
|
||||
|
||||
export const SSH_PROVIDER_UNREGISTERED_REASON = 'its SSH provider is no longer registered'
|
||||
export const NO_OBSERVING_PROVIDER_REASON = 'no registered provider can observe its host'
|
||||
export const PTY_STILL_RUNNING_NOTE = 'The PTY is still running.'
|
||||
|
||||
/** The one sentence every surface uses to admit a stop was not confirmed. */
|
||||
export function describeUnconfirmedStop(reason: string): string {
|
||||
return `The PTY was not confirmed stopped: ${reason}.`
|
||||
}
|
||||
|
||||
export const PTY_STILL_RUNNING_NOTE = 'The PTY is still running.'
|
||||
/** Words a close whose PTY teardown was never confirmed, for a stop receipt. */
|
||||
export function describeUnconfirmedAgentStop(close: {
|
||||
ptyStopVerdict?: 'live' | 'unverifiable'
|
||||
ptyStopReason?: string
|
||||
}): string {
|
||||
const detail =
|
||||
close.ptyStopVerdict === 'live'
|
||||
? 'it is still running'
|
||||
: (close.ptyStopReason ?? 'its host could not be reached')
|
||||
return `The agent terminal was closed but its process could not be confirmed stopped: ${detail}.`
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user