diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-admission.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-admission.ts new file mode 100644 index 00000000000..d43470a67a6 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-admission.ts @@ -0,0 +1,126 @@ +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentSessionOperationOutcome } from '../../../shared/agent-session-operation-ledger' +import type { + AgentSessionHandoffRequest, + AgentSessionHandoffResult, + AgentSessionHandoffStatus, + AgentSessionMutationResult, + AgentSessionWireRefusal +} from '../../../shared/agent-session-wire' +import { AGENT_SESSION_WIRE_REFUSAL_CODES } from '../../../shared/agent-session-wire' +import { + agentSessionFingerprintConflict, + computeAgentSessionPayloadFingerprint +} from '../../../shared/agent-session-mutation-envelope' +import type { StructuredAgentSessionHandoffOperationGuard } from './structured-agent-session-handoff-operation-guard' +import type { StructuredAgentSessionHandoffDeps } from './structured-agent-session-handoff-types' + +export type StructuredHandoffAdmission = + | { decision: 'continue'; record: AgentSessionRecord; fingerprint: string } + | { decision: 'replay'; outcome: AgentSessionOperationOutcome } + | { decision: 'refused'; refusal: AgentSessionWireRefusal } + +export async function admitStructuredHandoffRequest(input: { + deps: StructuredAgentSessionHandoffDeps + operationGuard: StructuredAgentSessionHandoffOperationGuard + callerKey: string + params: AgentSessionHandoffRequest + record: AgentSessionRecord + status?: AgentSessionHandoffStatus +}): Promise { + const action = input.params.action ?? 'start' + const requestFingerprint = computeAgentSessionPayloadFingerprint({ + method: 'agentSession.requestHandoff', + sessionId: input.record.sessionId, + fields: { direction: input.params.direction, mode: input.params.mode, action } + }) + const conflict = agentSessionFingerprintConflict(input.params.envelope, requestFingerprint) + if (conflict) { + return { decision: 'refused', refusal: conflict } + } + const fingerprint = computeAgentSessionPayloadFingerprint({ + method: 'agentSession.requestHandoff.operation', + sessionId: input.record.sessionId, + fields: { direction: input.params.direction } + }) + const operation = await input.operationGuard.check({ + callerKey: input.callerKey, + sessionId: input.record.sessionId, + operationId: input.params.envelope.clientOperationId, + fingerprint, + action, + ...(input.status ? { status: input.status } : {}), + now: input.deps.now() + }) + if (operation.decision === 'replay') { + return { decision: 'replay', outcome: operation.outcome } + } + if (operation.decision === 'refused') { + return { + decision: 'refused', + refusal: { + code: operation.code as 'agent_session_operation_conflict', + message: 'This handoff operation could not be admitted.' + } + } + } + if (input.params.envelope.expectedRuntimeFence !== input.record.lease.runtimeFence) { + await input.deps.store.recordOperationOutcome({ + callerKey: input.callerKey, + operationId: input.params.envelope.clientOperationId, + outcome: { status: 'failed', code: 'agent_session_checkpoint_stale' } + }) + return { + decision: 'refused', + refusal: { + code: 'agent_session_checkpoint_stale', + message: 'The session owner changed before the handoff request arrived.', + currentFence: input.record.lease.runtimeFence + } + } + } + return { decision: 'continue', record: input.record, fingerprint } +} + +export function replayedStructuredHandoffRefusal( + outcome: AgentSessionOperationOutcome +): AgentSessionWireRefusal | null { + if ( + outcome.status !== 'failed' || + !AGENT_SESSION_WIRE_REFUSAL_CODES.includes( + outcome.code as (typeof AGENT_SESSION_WIRE_REFUSAL_CODES)[number] + ) + ) { + return null + } + return { + code: outcome.code as (typeof AGENT_SESSION_WIRE_REFUSAL_CODES)[number], + message: 'This handoff request was previously refused.' + } +} + +export async function refuseAdmittedStructuredHandoff(input: { + deps: StructuredAgentSessionHandoffDeps + callerKey: string + params: AgentSessionHandoffRequest + refusal: AgentSessionWireRefusal +}): Promise> { + await input.deps.store.recordOperationOutcome({ + callerKey: input.callerKey, + operationId: input.params.envelope.clientOperationId, + outcome: { status: 'failed', code: input.refusal.code } + }) + return { ok: false, refusal: input.refusal } +} + +export function structuredHandoffRetryIsAdmissible( + status: AgentSessionHandoffStatus, + params: AgentSessionHandoffRequest +): boolean { + return ( + status.phase === 'failed' && + status.direction === params.direction && + status.operationId === params.envelope.clientOperationId && + status.error?.recoverableOwner !== 'none' + ) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-flow-context.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-flow-context.ts new file mode 100644 index 00000000000..2aa2eb440ae --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-flow-context.ts @@ -0,0 +1,33 @@ +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentSessionHandoffStatus } from '../../../shared/agent-session-wire' +import type { + StructuredAgentSessionHandoffDeps, + StructuredAgentSessionHandoffFlowContext, + StructuredTuiOwner +} from './structured-agent-session-handoff-types' + +export function createStructuredHandoffFlowContext(input: { + deps: StructuredAgentSessionHandoffDeps + owner: (sessionId: string) => StructuredTuiOwner | undefined + retainOwner: (sessionId: string, owner: StructuredTuiOwner) => void + releaseOwner: (sessionId: string) => void + setStatus: (sessionId: string, status: AgentSessionHandoffStatus) => void + enterPreparing: StructuredAgentSessionHandoffFlowContext['enterPreparing'] + publishStage: StructuredAgentSessionHandoffFlowContext['publishStage'] + requireRecord: (sessionId: string) => AgentSessionRecord +}): StructuredAgentSessionHandoffFlowContext { + return input +} + +export async function stopStructuredNativeTurn( + deps: StructuredAgentSessionHandoffDeps, + sessionId: string, + turnId: string +): Promise { + const record = deps.store.getRecord(sessionId) + if (!record || record.lease.runtimeKind !== 'native') { + return false + } + const session = deps.session(sessionId) + return (await deps.acquireNativeStop?.(sessionId, turnId, session.fence)) ?? false +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-flow-runner.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-flow-runner.ts new file mode 100644 index 00000000000..875d7c92204 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-flow-runner.ts @@ -0,0 +1,79 @@ +import type { AgentSessionHandoffRequest } from '../../../shared/agent-session-wire' +import { stopStructuredNativeTurn } from './structured-agent-session-handoff-flow-context' +import { handoffStructuredSessionToTui } from './structured-agent-session-handoff-forward' +import type { StructuredAgentSessionHandoffOperationGuard } from './structured-agent-session-handoff-operation-guard' +import { handoffStructuredSessionToNative } from './structured-agent-session-handoff-reverse' +import type { + StructuredAgentSessionHandoffDeps, + StructuredAgentSessionHandoffFlowContext +} from './structured-agent-session-handoff-types' + +export class StructuredAgentSessionHandoffFlowRunner { + private readonly active = new Set>() + + constructor( + private readonly input: { + deps: StructuredAgentSessionHandoffDeps + operationGuard: StructuredAgentSessionHandoffOperationGuard + flowContext: () => StructuredAgentSessionHandoffFlowContext + fail: (params: AgentSessionHandoffRequest, error: unknown) => void + } + ) {} + + async drain(): Promise { + await Promise.allSettled(this.active) + } + + begin(input: { + callerKey: string + params: AgentSessionHandoffRequest + turnId: string | null + fingerprint: string + }): void { + const { callerKey, params, turnId, fingerprint } = input + const sessionId = params.envelope.sessionId + this.input.operationGuard.start(sessionId, { + callerKey, + operationId: params.envelope.clientOperationId, + fingerprint + }) + const flow = this.run(params, turnId) + .then(() => + this.input.deps.store.recordOperationOutcome({ + callerKey, + operationId: params.envelope.clientOperationId, + outcome: { status: 'succeeded', sessionId } + }) + ) + .catch(async (error) => { + await this.input.deps.store.recordOperationOutcome({ + callerKey, + operationId: params.envelope.clientOperationId, + outcome: { status: 'failed', code: 'agent_session_handoff_failed' } + }) + this.input.fail(params, error) + }) + .finally(() => this.input.operationGuard.finish(sessionId, params.envelope.clientOperationId)) + this.active.add(flow) + void flow.finally(() => this.active.delete(flow)) + } + + private run(params: AgentSessionHandoffRequest, turnId: string | null): Promise { + const sessionId = params.envelope.sessionId + return this.input.deps.schedule(sessionId, async () => { + if (turnId && params.mode === 'stop-turn') { + const stopped = await stopStructuredNativeTurn(this.input.deps, sessionId, turnId) + if (!stopped) { + throw new Error('The current turn did not acknowledge cancellation.') + } + } + await (params.direction === 'to-tui' + ? handoffStructuredSessionToTui(this.input.flowContext(), params, params.action === 'retry') + : handoffStructuredSessionToNative( + this.input.flowContext(), + params, + params.action === 'retry' + )) + }) + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-operation-guard.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-operation-guard.ts index c24bca96070..00896472a00 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-operation-guard.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-operation-guard.ts @@ -1,27 +1,64 @@ import type { AgentSessionHandoffStatus } from '../../../shared/agent-session-wire' +import type { AgentSessionOperationOutcome } from '../../../shared/agent-session-operation-ledger' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' -type ActiveOperation = { operationId: string; fingerprint: string } +type ActiveOperation = { callerKey: string; operationId: string; fingerprint: string } + +export type HandoffOperationDecision = + | { decision: 'new' } + | { decision: 'replay'; outcome: AgentSessionOperationOutcome } + | { decision: 'retry' } + | { decision: 'refused'; code: 'agent_session_operation_conflict' | string } export class StructuredAgentSessionHandoffOperationGuard { private readonly activeBySession = new Map() - check(input: { + constructor(private readonly store: AgentSessionRecordStore) {} + + async check(input: { + callerKey: string sessionId: string operationId: string fingerprint: string action: 'start' | 'cancel-queued' | 'retry' status?: AgentSessionHandoffStatus - }): 'new' | 'replay' | 'conflict' { + now: number + }): Promise { const active = this.activeBySession.get(input.sessionId) if ( - (input.status?.operationId === input.operationId && - input.status.phase !== 'idle' && - input.action === 'start') || - (active?.operationId === input.operationId && active.fingerprint === input.fingerprint) + active?.operationId === input.operationId && + (active.fingerprint !== input.fingerprint || active.callerKey !== input.callerKey) ) { - return 'replay' + return { decision: 'refused', code: 'agent_session_operation_conflict' } } - return active?.operationId === input.operationId ? 'conflict' : 'new' + const ledger = await this.store.admitOperation({ + callerKey: input.callerKey, + operationId: input.operationId, + fingerprint: input.fingerprint, + now: input.now + }) + if (ledger.decision === 'refused') { + return { decision: 'refused', code: ledger.code } + } + if (ledger.decision === 'admit') { + return { decision: 'new' } + } + if (input.action === 'retry' && ledger.row.outcome.status === 'failed') { + await this.store.recordOperationOutcome({ + callerKey: input.callerKey, + operationId: input.operationId, + outcome: { status: 'pending' } + }) + return { decision: 'retry' } + } + if ( + ledger.row.outcome.status === 'pending' && + !active && + input.status?.operationId !== input.operationId + ) { + return { decision: 'new' } + } + return { decision: 'replay', outcome: ledger.row.outcome } } start(sessionId: string, operation: ActiveOperation): void { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-restart.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-restart.ts new file mode 100644 index 00000000000..85dde4a990b --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-restart.ts @@ -0,0 +1,211 @@ +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { + AgentSessionHandoffRequest, + AgentSessionHandoffStatus +} from '../../../shared/agent-session-wire' +import { + abandonStoredAgentSessionHandoffAttempt, + setStoredAgentSessionHandoffStage, + stopStoredAgentSessionOwnerForHandoff +} from '../../runtime/agent-session-handoff-record-transitions' +import { handoffStructuredSessionToTui } from './structured-agent-session-handoff-forward' +import { handoffStructuredSessionToNative } from './structured-agent-session-handoff-reverse' +import { idleStructuredHandoffStatus } from './structured-agent-session-handoff-status' +import type { + StructuredAgentSessionHandoffDeps, + StructuredAgentSessionHandoffFlowContext, + StructuredTuiOwner +} from './structured-agent-session-handoff-types' + +type RestartAccess = { + deps: StructuredAgentSessionHandoffDeps + requireRecord: (sessionId: string) => AgentSessionRecord + flowContext: () => StructuredAgentSessionHandoffFlowContext + retainOwner: (sessionId: string, owner: StructuredTuiOwner) => void + setStatus: (sessionId: string, status: AgentSessionHandoffStatus) => void +} + +export async function restoreStructuredAgentSessionHandoff( + input: RestartAccess, + sessionId: string +): Promise { + const initial = input.requireRecord(sessionId) + const operationId = initial.lease.handoffOperationId + const initialStage = initial.lease.handoffStage + if (initialStage === 'recovering' || initialStage === 'manual-recovery') { + if (operationId) { + await input.deps.store.recordOperationOutcome({ + operationId, + outcome: { status: 'failed', code: 'agent_session_ownership_unknown' } + }) + } + input.setStatus(sessionId, idleStructuredHandoffStatus(initial)) + return + } + let lastError: unknown + for (let attempt = 0; attempt < 3; attempt += 1) { + try { + await restoreOnce(input, input.requireRecord(sessionId)) + const settled = input.requireRecord(sessionId) + if (settled.lease.handoffStage !== null || settled.lease.handoffOperationId !== null) { + throw new Error('Restart handoff reconciliation did not settle the transfer.') + } + if (operationId) { + await input.deps.store.recordOperationOutcome({ + operationId, + outcome: { status: 'succeeded', sessionId } + }) + } + return + } catch (error) { + lastError = error + if (attempt < 2) { + await new Promise((resolve) => setTimeout(resolve, 100 * 2 ** attempt)) + } + } + } + const current = input.requireRecord(sessionId) + const failed = await setStoredAgentSessionHandoffStage(input.deps.store, { + sessionId, + fence: current.lease.runtimeFence, + stage: 'manual-recovery', + handoffOperationId: current.lease.handoffOperationId, + now: input.deps.now() + }) + const status = idleStructuredHandoffStatus(failed) + if (operationId) { + await input.deps.store.recordOperationOutcome({ + operationId, + outcome: { status: 'failed', code: 'agent_session_handoff_failed' } + }) + } + input.setStatus(sessionId, { + ...status, + ...(status.error + ? { + error: { + ...status.error, + details: lastError instanceof Error ? lastError.message : String(lastError) + } + } + : {}) + }) +} + +async function restoreOnce(input: RestartAccess, record: AgentSessionRecord): Promise { + if (!input.deps.transport) { + if (record.lease.handoffStage !== null || record.lease.runtimeKind === 'tui') { + throw new Error('Agent TUI handoff recovery is unavailable on this host.') + } + return + } + if (record.lease.handoffStage === null && record.lease.runtimeKind === 'tui') { + await restoreLiveTui(input, record) + return + } + if (!record.lease.handoffOperationId) { + return + } + if (record.lease.handoffStage === 'preparing') { + await restorePreparing(input, record) + return + } + if (record.lease.handoffStage === 'new-owner-proving') { + await restoreProving(input, record) + return + } + if (record.lease.handoffStage === 'old-owner-stopped') { + await continueHandoff(input, record) + } +} + +async function restoreLiveTui(input: RestartAccess, record: AgentSessionRecord): Promise { + const owner = await input.deps.transport!.recoverTuiOwner(record) + input.retainOwner(record.sessionId, owner) + input.setStatus(record.sessionId, { + owner: 'tui', + direction: null, + phase: 'idle', + stage: null, + operationId: null, + terminal: owner.terminal, + hostLabel: input.deps.transport?.hostLabel + }) +} + +async function restorePreparing(input: RestartAccess, record: AgentSessionRecord): Promise { + if (record.lease.runtimeKind === 'tui') { + const owner = await input.deps.transport!.recoverTuiOwner(record) + await input.deps.transport!.reproveTuiOwner({ record, owner }) + const settled = await setStoredAgentSessionHandoffStage(input.deps.store, { + sessionId: record.sessionId, + fence: record.lease.runtimeFence, + stage: null, + handoffOperationId: null, + now: input.deps.now() + }) + await restoreLiveTui(input, settled) + return + } + await input.deps.transport!.stopRecoveredOwner(record) + const stopped = await stopStoredAgentSessionOwnerForHandoff(input.deps.store, { + sessionId: record.sessionId, + expectedFence: record.lease.runtimeFence, + operationId: record.lease.handoffOperationId!, + now: input.deps.now() + }) + await continueHandoff(input, stopped) +} + +async function restoreProving(input: RestartAccess, record: AgentSessionRecord): Promise { + const operationId = record.lease.handoffOperationId! + if (record.lease.runtimeKind === 'tui') { + const owner = await input.deps.transport!.recoverTuiOwner(record) + const reproved = await input.deps.transport!.reproveTuiOwner({ record, owner }) + await input.deps.store.proveOwner({ + sessionId: record.sessionId, + fence: record.lease.runtimeFence, + link: reproved.link, + now: input.deps.now() + }) + input.retainOwner(record.sessionId, reproved) + input.setStatus(record.sessionId, { + owner: 'tui', + direction: null, + phase: 'idle', + stage: null, + operationId: null, + terminal: reproved.terminal, + hostLabel: input.deps.transport?.hostLabel + }) + return + } + await input.deps.transport!.stopRecoveredOwner(record) + const stopped = await abandonStoredAgentSessionHandoffAttempt(input.deps.store, { + sessionId: record.sessionId, + expectedFence: record.lease.runtimeFence, + operationId, + recoverableRuntimeKind: 'tui', + now: input.deps.now() + }) + await continueHandoff(input, stopped) +} + +async function continueHandoff(input: RestartAccess, record: AgentSessionRecord): Promise { + const direction = record.lease.runtimeKind === 'native' ? 'to-tui' : 'to-native' + const operationId = record.lease.handoffOperationId! + const params: AgentSessionHandoffRequest = { + envelope: { + sessionId: record.sessionId, + clientOperationId: operationId, + expectedRuntimeFence: record.lease.runtimeFence, + payloadFingerprint: 'restart-reconciliation' + }, + direction, + mode: 'now', + action: 'retry' + } + await (direction === 'to-tui' + ? handoffStructuredSessionToTui(input.flowContext(), params, true) + : handoffStructuredSessionToNative(input.flowContext(), params, true)) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-reverse.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-reverse.ts index 0f10cfe2019..4e86bfcd6b9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-reverse.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-reverse.ts @@ -16,7 +16,7 @@ export async function handoffStructuredSessionToNative( const sessionId = params.envelope.sessionId const operationId = params.envelope.clientOperationId let record = context.requireRecord(sessionId) - const owner = context.owner(sessionId) + let owner = context.owner(sessionId) let transcriptPath = owner?.transcriptPath if (!retry || record.lease.handoffStage === 'preparing') { if (record.lease.handoffStage === null) { @@ -25,6 +25,9 @@ export async function handoffStructuredSessionToNative( if (!owner) { throw new Error('The owning agent terminal could not be identified.') } + owner = await deps.transport!.reproveTuiOwner({ record, owner }) + context.retainOwner(sessionId, owner) + transcriptPath = owner.transcriptPath ?? transcriptPath context.setStatus(sessionId, { owner: 'tui', direction: 'to-native', diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-types.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-types.ts index acb3d92bfe0..32979dc6460 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-types.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-types.ts @@ -21,6 +21,12 @@ export type StructuredAgentSessionHandoffTransport = { fence: number spawnToken: string }): Promise + reproveTuiOwner(input: { + record: AgentSessionRecord + owner: StructuredTuiOwner + }): Promise + recoverTuiOwner(record: AgentSessionRecord): Promise + stopRecoveredOwner(record: AgentSessionRecord): Promise waitForTuiExit(owner: StructuredTuiOwner): Promise<{ transcriptPath?: string }> tuiStatus(owner: StructuredTuiOwner): 'idle' | 'busy' stopFailedTuiLaunch?(owner: StructuredTuiOwner): Promise diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts index 155a064ae00..05b6ad2b316 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts @@ -11,6 +11,11 @@ import type { AgentSessionHandoffStatus } from '../../../shared/agent-session-wire' import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { + reserveStoredAgentSessionHandoffOwner, + setStoredAgentSessionHandoffStage, + stopStoredAgentSessionOwnerForHandoff +} from '../../runtime/agent-session-handoff-record-transitions' import { openAgentSessionJournal } from '../agent-session-journal/journal-store' import { StructuredAgentSessionHandoffCoordinator } from './structured-agent-session-handoff' import type { @@ -34,11 +39,17 @@ let launchTui: ReturnType > +let reproveTuiOwner: ReturnType< + typeof vi.fn +> let stopFailedTuiLaunch: ReturnType< typeof vi.fn> > let acquireNativeStop: ReturnType Promise>> let acquireNativeCalls: number +let stopRecoveredOwner: ReturnType< + typeof vi.fn +> let operations: number function operationId(): string { @@ -68,6 +79,10 @@ function request( } } +function submit(params: AgentSessionHandoffRequest) { + return coordinator.request('client-1', params) +} + function process(spawnToken: string, pid: number) { return { hostId: 'local', @@ -163,6 +178,15 @@ function createCoordinator(): StructuredAgentSessionHandoffCoordinator { transport: { hostLabel: 'Test host', launchTui, + reproveTuiOwner, + recoverTuiOwner: async (record) => { + const owner = makeTuiOwner( + record.lease.runtimeFence, + record.lease.ownerProcess?.spawnToken ?? record.lease.reservedSpawnToken ?? 'recovered' + ) + return { ...owner, process: record.lease.ownerProcess ?? owner.process } + }, + stopRecoveredOwner, waitForTuiExit, tuiStatus: () => (tuiIdle ? 'idle' : 'busy'), stopFailedTuiLaunch @@ -205,7 +229,7 @@ async function waitForPhase(phase: AgentSessionHandoffStatus['phase']): Promise< } async function moveToTui(): Promise { - expect(coordinator.request(request('to-tui', 'now'))).toMatchObject({ ok: true }) + expect(await submit(request('to-tui', 'now'))).toMatchObject({ ok: true }) await vi.waitFor(() => expect(coordinator.status(SESSION)).toMatchObject({ owner: 'tui' })) } @@ -217,6 +241,7 @@ beforeEach(async () => { importFailure = null nativeAcquireFailure = null acquireNativeCalls = 0 + stopRecoveredOwner = vi.fn(async () => undefined) store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' }) await establishNativeOwner() journal = await openAgentSessionJournal({ @@ -231,12 +256,14 @@ beforeEach(async () => { }) launchTui = vi.fn(async ({ fence, spawnToken }) => makeTuiOwner(fence, spawnToken)) waitForTuiExit = vi.fn(async (owner) => ({ transcriptPath: owner.transcriptPath })) + reproveTuiOwner = vi.fn(async ({ owner }) => owner) stopFailedTuiLaunch = vi.fn(async () => undefined) acquireNativeStop = vi.fn(async () => true) coordinator = createCoordinator() }) afterEach(async () => { + await coordinator.drain() await rm(root, { recursive: true, force: true }) }) @@ -254,7 +281,7 @@ describe('structured session ownership handoff', () => { handoffStage: null }) - expect(coordinator.request(request('to-native', 'after-turn'))).toMatchObject({ ok: true }) + expect(await submit(request('to-native', 'after-turn'))).toMatchObject({ ok: true }) await vi.waitFor(() => expect(coordinator.status(SESSION)).toMatchObject({ owner: 'native' })) const messages = journal.snapshot().items.filter((item) => item.body.kind === 'message') @@ -269,21 +296,38 @@ describe('structured session ownership handoff', () => { it('queues a busy native turn, supports cancel, and starts only after idle', async () => { await appendStatus('running') - expect(coordinator.request(request('to-tui', 'after-turn'))).toMatchObject({ ok: true }) + expect(await submit(request('to-tui', 'after-turn'))).toMatchObject({ ok: true }) expect(coordinator.status(SESSION).phase).toBe('queued') expect( - coordinator.request(request('to-tui', 'after-turn', { action: 'cancel-queued' })) + await submit(request('to-tui', 'after-turn', { action: 'cancel-queued' })) ).toMatchObject({ ok: true }) await appendStatus('completed') await new Promise((resolve) => setTimeout(resolve, 200)) expect(launchTui).not.toHaveBeenCalled() await appendStatus('running') - expect(coordinator.request(request('to-tui', 'after-turn'))).toMatchObject({ ok: true }) + expect(await submit(request('to-tui', 'after-turn'))).toMatchObject({ ok: true }) await appendStatus('completed') await vi.waitFor(() => expect(launchTui).toHaveBeenCalledOnce()) }) + it('durably replays a completed queued cancellation after restart', async () => { + await appendStatus('running') + expect(await submit(request('to-tui', 'after-turn'))).toMatchObject({ ok: true }) + const cancellation = request('to-tui', 'after-turn', { action: 'cancel-queued' }) + expect(await submit(cancellation)).toMatchObject({ ok: true, replayed: false }) + await coordinator.drain() + + store = await AgentSessionRecordStore.open({ + directory: join(root, 'store'), + hostId: 'local' + }) + coordinator = createCoordinator() + + expect(await submit(cancellation)).toMatchObject({ ok: true, replayed: true }) + expect(launchTui).not.toHaveBeenCalled() + }) + it('replays a duplicate active handoff without launching a second owner', async () => { let finishLaunch!: () => void launchTui.mockImplementationOnce( @@ -294,7 +338,7 @@ describe('structured session ownership handoff', () => { ) const operation = operationId() const first = request('to-tui', 'now', { operationId: operation }) - expect(coordinator.request(first)).toMatchObject({ ok: true, replayed: false }) + expect(await submit(first)).toMatchObject({ ok: true, replayed: false }) await vi.waitFor(() => expect(launchTui).toHaveBeenCalledOnce()) expect(store.getRecord(SESSION)?.lease).toMatchObject({ runtimeKind: 'tui', @@ -302,22 +346,135 @@ describe('structured session ownership handoff', () => { handoffStage: 'new-owner-proving' }) expect(coordinator.status(SESSION).owner).not.toBe('tui') - expect(coordinator.request(first)).toMatchObject({ ok: true, replayed: true }) + expect(await submit(first)).toMatchObject({ ok: true, replayed: true }) finishLaunch() await vi.waitFor(() => expect(coordinator.status(SESSION).owner).toBe('tui')) expect(launchTui).toHaveBeenCalledOnce() }) + it('replays the same completed handoff operation after the coordinator restarts', async () => { + const operation = operationId() + const first = request('to-tui', 'now', { operationId: operation }) + expect(await submit(first)).toMatchObject({ ok: true, replayed: false }) + await vi.waitFor(() => expect(coordinator.status(SESSION).owner).toBe('tui')) + await coordinator.drain() + + store = await AgentSessionRecordStore.open({ + directory: join(root, 'store'), + hostId: 'local' + }) + coordinator = createCoordinator() + expect(await submit(first)).toMatchObject({ ok: true, replayed: true }) + expect(launchTui).toHaveBeenCalledOnce() + }) + + it('continues a persisted preparing stage after restart instead of stranding it', async () => { + const operation = operationId() + const first = request('to-tui', 'now', { operationId: operation }) + await store.admitOperation({ + callerKey: 'client-1', + operationId: operation, + fingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.requestHandoff.operation', + sessionId: SESSION, + fields: { direction: 'to-tui' } + }), + now: NOW + }) + await setStoredAgentSessionHandoffStage(store, { + sessionId: SESSION, + fence: 1, + stage: 'preparing', + handoffOperationId: operation, + now: NOW + }) + coordinator = createCoordinator() + + await coordinator.restore(SESSION) + + expect(stopRecoveredOwner).toHaveBeenCalledOnce() + expect(store.getRecord(SESSION)?.lease).toMatchObject({ + runtimeKind: 'tui', + claimStatus: 'live', + handoffStage: null + }) + await coordinator.drain() + store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' }) + coordinator = createCoordinator() + expect(await submit(first)).toMatchObject({ ok: true, replayed: true }) + expect(launchTui).toHaveBeenCalledOnce() + }) + + it('finishes a live new-owner-proving stage after restart', async () => { + const operation = operationId() + let record = await setStoredAgentSessionHandoffStage(store, { + sessionId: SESSION, + fence: 1, + stage: 'preparing', + handoffOperationId: operation, + now: NOW + }) + record = await stopStoredAgentSessionOwnerForHandoff(store, { + sessionId: SESSION, + expectedFence: record.lease.runtimeFence, + operationId: operation, + now: NOW + }) + const spawnToken = 'restarted-tui' + record = await reserveStoredAgentSessionHandoffOwner(store, { + sessionId: SESSION, + expectedFence: record.lease.runtimeFence, + runtimeKind: 'tui', + spawnToken, + operationId: operation, + claimKeyId: 'key-1', + now: NOW + }) + await store.commitProcessIdentity({ + sessionId: SESSION, + fence: record.lease.runtimeFence, + process: process(spawnToken, 4400), + now: NOW + }) + coordinator = createCoordinator() + + await coordinator.restore(SESSION) + + expect(store.getRecord(SESSION)?.lease).toMatchObject({ + runtimeKind: 'tui', + claimStatus: 'live', + handoffStage: null + }) + expect(coordinator.status(SESSION)).toMatchObject({ owner: 'tui', phase: 'idle' }) + }) + it('rejects cancellation once a queued handoff is no longer pending', async () => { expect( - coordinator.request(request('to-tui', 'after-turn', { action: 'cancel-queued' })) + await submit(request('to-tui', 'after-turn', { action: 'cancel-queued' })) ).toMatchObject({ ok: false, refusal: { code: 'agent_session_operation_conflict' } }) expect(coordinator.status(SESSION)).toMatchObject({ owner: 'native', phase: 'idle' }) }) + it('durably replays a queued-cancellation refusal as a refusal', async () => { + const cancellation = request('to-tui', 'after-turn', { action: 'cancel-queued' }) + expect(await submit(cancellation)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_conflict' } + }) + + store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' }) + coordinator = createCoordinator() + + expect(await submit(cancellation)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_conflict' } + }) + expect(launchTui).not.toHaveBeenCalled() + }) + it('requires acknowledged cancellation before stopping a native turn', async () => { await appendStatus('running') - expect(coordinator.request(request('to-tui', 'stop-turn'))).toMatchObject({ ok: true }) + expect(await submit(request('to-tui', 'stop-turn'))).toMatchObject({ ok: true }) await vi.waitFor(() => expect(coordinator.status(SESSION).owner).toBe('tui')) expect(acquireNativeStop).toHaveBeenCalledWith('turn-1') expect(acquireNativeStop.mock.invocationCallOrder[0]).toBeLessThan( @@ -328,16 +485,42 @@ describe('structured session ownership handoff', () => { it('queues a busy TUI from host status instead of trusting a stale journal', async () => { await moveToTui() tuiIdle = false - expect(coordinator.request(request('to-native', 'after-turn'))).toMatchObject({ ok: true }) + expect(await submit(request('to-native', 'after-turn'))).toMatchObject({ ok: true }) expect(coordinator.status(SESSION).phase).toBe('queued') expect(waitForTuiExit).not.toHaveBeenCalled() tuiIdle = true await vi.waitFor(() => expect(waitForTuiExit).toHaveBeenCalledOnce()) }) + it('re-proves the current TUI handle before accepting its exit and importing history', async () => { + await moveToTui() + const reverse = request('to-native', 'now') + expect(await submit(reverse)).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(coordinator.status(SESSION).owner).toBe('native')) + + expect(reproveTuiOwner).toHaveBeenCalledOnce() + expect(reproveTuiOwner.mock.invocationCallOrder[0]).toBeLessThan( + waitForTuiExit.mock.invocationCallOrder[0] ?? Infinity + ) + }) + + it('does not close or import when the current TUI handle cannot be re-proved', async () => { + await moveToTui() + reproveTuiOwner.mockRejectedValueOnce(new Error('provider handle changed')) + expect(await submit(request('to-native', 'now'))).toMatchObject({ ok: true }) + await waitForPhase('failed') + + expect(waitForTuiExit).not.toHaveBeenCalled() + expect(store.getRecord(SESSION)?.lease).toMatchObject({ + runtimeKind: 'tui', + claimStatus: 'live', + handoffStage: 'preparing' + }) + }) + it('recovers native ownership when TUI launch or proof fails', async () => { launchTui.mockRejectedValueOnce(new Error('resume proof failed')) - expect(coordinator.request(request('to-tui', 'now'))).toMatchObject({ ok: true }) + expect(await submit(request('to-tui', 'now'))).toMatchObject({ ok: true }) await waitForPhase('failed') expect(coordinator.status(SESSION)).toMatchObject({ owner: 'native', @@ -355,7 +538,7 @@ describe('structured session ownership handoff', () => { ...makeTuiOwner(fence, spawnToken), link: link(fence + 1, 'bad-proof') })) - expect(coordinator.request(request('to-tui', 'now'))).toMatchObject({ ok: true }) + expect(await submit(request('to-tui', 'now'))).toMatchObject({ ok: true }) await waitForPhase('failed') expect(stopFailedTuiLaunch).toHaveBeenCalledOnce() expect(store.getRecord(SESSION)?.lease.runtimeKind).toBe('native') @@ -368,9 +551,9 @@ describe('structured session ownership handoff', () => { })) stopFailedTuiLaunch.mockRejectedValueOnce(new Error('exit unproved')) const failedOperation = operationId() - expect( - coordinator.request(request('to-tui', 'now', { operationId: failedOperation })) - ).toMatchObject({ ok: true }) + expect(await submit(request('to-tui', 'now', { operationId: failedOperation }))).toMatchObject({ + ok: true + }) await waitForPhase('failed') expect(coordinator.status(SESSION)).toMatchObject({ @@ -380,9 +563,7 @@ describe('structured session ownership handoff', () => { }) expect(store.getRecord(SESSION)?.lease.handoffStage).toBe('manual-recovery') expect( - coordinator.request( - request('to-tui', 'now', { action: 'retry', operationId: failedOperation }) - ) + await submit(request('to-tui', 'now', { action: 'retry', operationId: failedOperation })) ).toMatchObject({ ok: false, refusal: { code: 'agent_session_operation_conflict' } }) }) @@ -391,7 +572,7 @@ describe('structured session ownership handoff', () => { importFailure = new Error('history unavailable') const retryOperation = operationId() expect( - coordinator.request(request('to-native', 'after-turn', { operationId: retryOperation })) + await submit(request('to-native', 'after-turn', { operationId: retryOperation })) ).toMatchObject({ ok: true }) await waitForPhase('failed') expect(coordinator.status(SESSION)).toMatchObject({ @@ -406,9 +587,7 @@ describe('structured session ownership handoff', () => { }) expect( - coordinator.request( - request('to-native', 'now', { action: 'retry', operationId: retryOperation }) - ) + await submit(request('to-native', 'now', { action: 'retry', operationId: retryOperation })) ).toMatchObject({ ok: true }) await vi.waitFor(() => expect(coordinator.status(SESSION).owner).toBe('native')) expect(waitForTuiExit).toHaveBeenCalledOnce() @@ -419,7 +598,7 @@ describe('structured session ownership handoff', () => { nativeAcquireFailure = new Error('native proof unavailable') const retryOperation = operationId() expect( - coordinator.request(request('to-native', 'after-turn', { operationId: retryOperation })) + await submit(request('to-native', 'after-turn', { operationId: retryOperation })) ).toMatchObject({ ok: true }) await waitForPhase('failed') @@ -442,9 +621,7 @@ describe('structured session ownership handoff', () => { operationId: retryOperation }) expect( - coordinator.request( - request('to-native', 'now', { action: 'retry', operationId: retryOperation }) - ) + await submit(request('to-native', 'now', { action: 'retry', operationId: retryOperation })) ).toMatchObject({ ok: true }) await vi.waitFor(() => expect(coordinator.status(SESSION).owner).toBe('native')) expect(waitForTuiExit).toHaveBeenCalledOnce() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.ts index e394a4f37d0..55542d61d38 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.ts @@ -3,18 +3,22 @@ import type { AgentSessionHandoffRequest, AgentSessionHandoffResult, AgentSessionHandoffStatus, - AgentSessionMutationResult + AgentSessionMutationResult, + AgentSessionWireRefusal } from '../../../shared/agent-session-wire' -import { - agentSessionFingerprintConflict, - computeAgentSessionPayloadFingerprint -} from '../../../shared/agent-session-mutation-envelope' import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection' import { setStoredAgentSessionHandoffStage } from '../../runtime/agent-session-handoff-record-transitions' -import { handoffStructuredSessionToTui } from './structured-agent-session-handoff-forward' -import { handoffStructuredSessionToNative } from './structured-agent-session-handoff-reverse' +import { + admitStructuredHandoffRequest, + refuseAdmittedStructuredHandoff, + replayedStructuredHandoffRefusal, + structuredHandoffRetryIsAdmissible +} from './structured-agent-session-handoff-admission' +import { createStructuredHandoffFlowContext } from './structured-agent-session-handoff-flow-context' +import { StructuredAgentSessionHandoffFlowRunner } from './structured-agent-session-handoff-flow-runner' import { StructuredAgentSessionHandoffQueue } from './structured-agent-session-handoff-queue' import { StructuredAgentSessionHandoffOperationGuard } from './structured-agent-session-handoff-operation-guard' +import { restoreStructuredAgentSessionHandoff } from './structured-agent-session-handoff-restart' import { structuredHandoffRefusal as refusal, structuredHandoffSuccess @@ -35,9 +39,18 @@ export class StructuredAgentSessionHandoffCoordinator { private readonly statuses = new Map() private readonly queue = new StructuredAgentSessionHandoffQueue() private readonly tuiOwners = new Map() - private readonly operationGuard = new StructuredAgentSessionHandoffOperationGuard() + private readonly operationGuard: StructuredAgentSessionHandoffOperationGuard + private readonly flowRunner: StructuredAgentSessionHandoffFlowRunner - constructor(private readonly deps: StructuredAgentSessionHandoffDeps) {} + constructor(private readonly deps: StructuredAgentSessionHandoffDeps) { + this.operationGuard = new StructuredAgentSessionHandoffOperationGuard(deps.store) + this.flowRunner = new StructuredAgentSessionHandoffFlowRunner({ + deps, + operationGuard: this.operationGuard, + flowContext: () => this.flowContext(), + fail: (params, error) => this.fail(params, error) + }) + } status(sessionId: string): AgentSessionHandoffStatus { return ( @@ -45,95 +58,103 @@ export class StructuredAgentSessionHandoffCoordinator { ) } - request( + async drain(): Promise { + await this.flowRunner.drain() + } + + async restore(sessionId: string): Promise { + await restoreStructuredAgentSessionHandoff( + { + deps: this.deps, + requireRecord: (id) => this.requireRecord(id), + flowContext: () => this.flowContext(), + retainOwner: (id, owner) => this.tuiOwners.set(id, owner), + setStatus: (id, status) => this.setStatus(id, status) + }, + sessionId + ) + } + + async request( + callerKey: string, params: AgentSessionHandoffRequest - ): AgentSessionMutationResult { + ): Promise> { const record = this.requireRecord(params.envelope.sessionId) - const fields = { - direction: params.direction, - mode: params.mode, - action: params.action ?? 'start' - } - const fingerprint = computeAgentSessionPayloadFingerprint({ - method: 'agentSession.requestHandoff', - sessionId: record.sessionId, - fields - }) - const conflict = agentSessionFingerprintConflict(params.envelope, fingerprint) - if (conflict) { - return { ok: false, refusal: conflict } - } const currentStatus = this.statuses.get(record.sessionId) - const operationDecision = this.operationGuard.check({ - sessionId: record.sessionId, - operationId: params.envelope.clientOperationId, - fingerprint, - action: fields.action, + const admission = await admitStructuredHandoffRequest({ + deps: this.deps, + operationGuard: this.operationGuard, + callerKey, + params, + record, ...(currentStatus ? { status: currentStatus } : {}) }) - if (operationDecision === 'replay') { + if (admission.decision === 'replay') { + const replayedRefusal = replayedStructuredHandoffRefusal(admission.outcome) + if (replayedRefusal) { + return { ok: false, refusal: replayedRefusal } + } return this.success(record.sessionId, true) } - if (operationDecision === 'conflict') { - return { - ok: false, - refusal: refusal( - 'agent_session_operation_conflict', - 'This handoff operation is already running with different parameters.' - ) - } + if (admission.decision === 'refused') { + return { ok: false, refusal: admission.refusal } } - if (params.envelope.expectedRuntimeFence !== record.lease.runtimeFence) { - return { - ok: false, - refusal: { - code: 'agent_session_checkpoint_stale', - message: 'The session owner changed before the handoff request arrived.', - currentFence: record.lease.runtimeFence - } - } - } - if (fields.action === 'cancel-queued') { + const { fingerprint } = admission + const action = params.action ?? 'start' + if (action === 'cancel-queued') { if (currentStatus?.phase !== 'queued' || currentStatus.direction !== params.direction) { - return { - ok: false, - refusal: refusal('agent_session_operation_conflict', 'No matching queued handoff exists.') - } + return this.refuseAdmitted( + callerKey, + params, + 'agent_session_operation_conflict', + 'No matching queued handoff exists.' + ) } this.queue.cancel(record.sessionId) this.setStatus(record.sessionId, idleStructuredHandoffStatus(record)) + await this.deps.store.recordOperationOutcome({ + callerKey, + operationId: params.envelope.clientOperationId, + outcome: { status: 'succeeded', sessionId: record.sessionId } + }) return this.success(record.sessionId, false) } if (!this.deps.transport) { - return { - ok: false, - refusal: refusal( - 'structured_agent_session_unsupported', - 'Agent TUI handoff is unavailable on this host.' + return this.refuseAdmitted( + callerKey, + params, + 'structured_agent_session_unsupported', + 'Agent TUI handoff is unavailable on this host.' + ) + } + if (action === 'retry') { + if (!structuredHandoffRetryIsAdmissible(this.status(record.sessionId), params)) { + return this.refuseAdmitted( + callerKey, + params, + 'agent_session_operation_conflict', + 'This handoff is no longer retryable.' ) } - } - if (fields.action === 'retry') { - return this.retry(params) + this.begin(callerKey, params, null, fingerprint) + return this.success(record.sessionId, false) } const expectedOwner = params.direction === 'to-tui' ? 'native' : 'tui' if (record.lease.runtimeKind !== expectedOwner || record.lease.claimStatus !== 'live') { - return { - ok: false, - refusal: refusal( - 'agent_session_conflict', - `The ${expectedOwner} runtime does not own this session.` - ) - } + return this.refuseAdmitted( + callerKey, + params, + 'agent_session_conflict', + `The ${expectedOwner} runtime does not own this session.` + ) } if (structuredSessionHasPendingPrompt(this.deps.session(record.sessionId).journal)) { - return { - ok: false, - refusal: refusal( - 'agent_session_conflict', - 'Resolve the pending question or approval before switching.' - ) - } + return this.refuseAdmitted( + callerKey, + params, + 'agent_session_conflict', + 'Resolve the pending question or approval before switching.' + ) } const turnId = activeStructuredAgentSessionTurnId( this.deps.session(record.sessionId).journal.snapshot().items @@ -141,45 +162,41 @@ export class StructuredAgentSessionHandoffCoordinator { const busy = expectedOwner === 'native' ? turnId !== null : this.tuiStatus(record.sessionId) !== 'idle' if (busy && params.mode === 'now') { - return { - ok: false, - refusal: refusal('agent_session_conflict', 'The current turn must finish before switching.') - } + return this.refuseAdmitted( + callerKey, + params, + 'agent_session_conflict', + 'The current turn must finish before switching.' + ) } if (busy && params.mode === 'after-turn') { - this.queueAfterTurn(params) + this.queueAfterTurn(callerKey, params, fingerprint) return this.success(record.sessionId, false) } if (busy && expectedOwner === 'tui' && params.mode === 'stop-turn') { - return { - ok: false, - refusal: refusal( - 'structured_agent_session_unsupported', - 'Exit the agent terminal after this turn to continue in chat.' - ) - } + return this.refuseAdmitted( + callerKey, + params, + 'structured_agent_session_unsupported', + 'Exit the agent terminal after this turn to continue in chat.' + ) } - this.begin(params, turnId, fingerprint) + this.begin(callerKey, params, turnId, fingerprint) return this.success(record.sessionId, false) } - private retry( - params: AgentSessionHandoffRequest - ): AgentSessionMutationResult { - const status = this.status(params.envelope.sessionId) - if ( - status?.phase !== 'failed' || - status.direction !== params.direction || - status.operationId !== params.envelope.clientOperationId || - status.error?.recoverableOwner === 'none' - ) { - return { - ok: false, - refusal: refusal('agent_session_operation_conflict', 'This handoff is no longer retryable.') - } - } - this.begin(params, null, params.envelope.payloadFingerprint) - return this.success(params.envelope.sessionId, false) + private async refuseAdmitted( + callerKey: string, + params: AgentSessionHandoffRequest, + code: AgentSessionWireRefusal['code'], + message: string + ): Promise> { + return refuseAdmittedStructuredHandoff({ + deps: this.deps, + callerKey, + params, + refusal: refusal(code, message) + }) } private success( @@ -189,7 +206,11 @@ export class StructuredAgentSessionHandoffCoordinator { return structuredHandoffSuccess(this.deps, sessionId, replayed, this.status(sessionId)) } - private queueAfterTurn(params: AgentSessionHandoffRequest): void { + private queueAfterTurn( + callerKey: string, + params: AgentSessionHandoffRequest, + fingerprint: string + ): void { const sessionId = params.envelope.sessionId this.setStatus(sessionId, { owner: params.direction === 'to-tui' ? 'native' : 'tui', @@ -209,35 +230,23 @@ export class StructuredAgentSessionHandoffCoordinator { : this.tuiStatus(sessionId) === 'idle', () => { const next = { ...params, mode: 'now' as const } - this.begin(next, null, params.envelope.payloadFingerprint) + this.begin(callerKey, next, null, fingerprint) } ) } private begin( + callerKey: string, params: AgentSessionHandoffRequest, turnId: string | null, fingerprint: string ): void { - const sessionId = params.envelope.sessionId - this.operationGuard.start(sessionId, { - operationId: params.envelope.clientOperationId, + this.flowRunner.begin({ + callerKey, + params, + turnId, fingerprint }) - void this.deps - .schedule(sessionId, async () => { - if (turnId && params.mode === 'stop-turn') { - const stopped = await this.stopNativeTurn(sessionId, turnId) - if (!stopped) { - throw new Error('The current turn did not acknowledge cancellation.') - } - } - await (params.direction === 'to-tui' - ? handoffStructuredSessionToTui(this.flowContext(), params, params.action === 'retry') - : handoffStructuredSessionToNative(this.flowContext(), params, params.action === 'retry')) - }) - .catch((error) => this.fail(params, error)) - .finally(() => this.operationGuard.finish(sessionId, params.envelope.clientOperationId)) } private tuiStatus(sessionId: string): 'idle' | 'busy' { @@ -246,7 +255,7 @@ export class StructuredAgentSessionHandoffCoordinator { } private flowContext(): StructuredAgentSessionHandoffFlowContext { - return { + return createStructuredHandoffFlowContext({ deps: this.deps, owner: (sessionId) => this.tuiOwners.get(sessionId), retainOwner: (sessionId, owner) => this.tuiOwners.set(sessionId, owner), @@ -258,17 +267,7 @@ export class StructuredAgentSessionHandoffCoordinator { this.enterPreparing(record, operationId, direction), publishStage: (record, direction) => this.publishStage(record, direction), requireRecord: (sessionId) => this.requireRecord(sessionId) - } - } - - private async stopNativeTurn(sessionId: string, turnId: string): Promise { - const record = this.requireRecord(sessionId) - if (record.lease.runtimeKind !== 'native') { - return false - } - const session = this.deps.session(sessionId) - const result = await this.deps.acquireNativeStop?.(sessionId, turnId, session.fence) - return result ?? false + }) } private async enterPreparing( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts index 70569ffa37d..c1df075905f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts @@ -5,11 +5,28 @@ import { type DeferredStructuredAgentSessionEventSink } from './structured-agent-session-event-sink' import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host' +import { StructuredAgentSessionLeaseRenewer } from './structured-agent-session-lease-renewer' export class StructuredAgentSessionHostRuntimeState { private readonly eventSinks = new Map() + private readonly leaseRenewer: StructuredAgentSessionLeaseRenewer - constructor(private readonly deps: StructuredAgentSessionHostDeps) {} + constructor(private readonly deps: StructuredAgentSessionHostDeps) { + this.leaseRenewer = new StructuredAgentSessionLeaseRenewer({ + store: deps.store, + probe: (record) => this.probeRecord(record), + now: () => deps.now?.() ?? Date.now(), + onError: ({ sessionId, error }) => deps.onEventSinkError?.({ sessionId, error }) + }) + } + + startLeaseRenewal(): void { + this.leaseRenewer.start() + } + + stopLeaseRenewal(): void { + this.leaseRenewer.stop() + } eventSinkFor(sessionId: string): DeferredStructuredAgentSessionEventSink { const existing = this.eventSinks.get(sessionId) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts index edbc0114db3..59321c2ecf5 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts @@ -221,6 +221,7 @@ beforeEach(async () => { }) afterEach(async () => { + await host.flushAllStreamedEvents() await rm(root, { recursive: true, force: true }) }) @@ -228,7 +229,6 @@ describe('attach', () => { it('reserves the lease, spawns through the adapter, and opens the journal', async () => { const result = await host.attach(CALLER, attachParams()) expect(result).toMatchObject({ ok: true, replayed: false }) - expect(host.hasSession(SESSION)).toBe(true) const record = store.getRecord(SESSION) expect(record?.lease.ownerProcess?.pid).toBe(4242) expect(record?.lease.handoffStage).toBeNull() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 80bd58109af..bd8223bf8f5 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -113,6 +113,7 @@ export class StructuredAgentSessionHost { subscribers: this.subscribers, now: this.now }) + this.runtimeState.startLeaseRenewal() } private now = (): number => this.deps.now?.() ?? Date.now() @@ -151,7 +152,8 @@ export class StructuredAgentSessionHost { ), serialize: (sessionId, task) => this.serialize(sessionId, task), hasSession: (sessionId) => this.sessions.has(sessionId), - onReadable: (sessionId, restored) => this.sessions.set(sessionId, restored) + onReadable: (sessionId, restored) => this.sessions.set(sessionId, restored), + restoreHandoff: (sessionId) => this.handoffs.restore(sessionId) }) } @@ -233,6 +235,7 @@ export class StructuredAgentSessionHost { /** Settles final provider rows after every child has stopped producing them. */ async flushAllStreamedEvents(): Promise { + this.runtimeState.stopLeaseRenewal() // An acquired provider can still be opening its journal; bind before draining. await this.tasks.drainAttaches() await this.runtimeState.flushAllEventSinks() @@ -307,10 +310,11 @@ export class StructuredAgentSessionHost { } requestHandoff( + caller: StructuredAgentSessionCaller, params: AgentSessionHandoffRequest - ): AgentSessionMutationResult { + ): Promise> { this.requireSession(params.envelope.sessionId) - return this.handoffs.request(params) + return this.handoffs.request(caller.callerKey, params) } handoffStatus(sessionId: string): AgentSessionHandoffStatus { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.test.ts new file mode 100644 index 00000000000..e0dd07c344c --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.test.ts @@ -0,0 +1,130 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { StructuredAgentSessionLeaseRenewer } from './structured-agent-session-lease-renewer' + +const NOW = 1_800_000_000_000 +const roots: string[] = [] + +async function liveStore(): Promise { + const root = await mkdtemp(join(tmpdir(), 'orca-lease-renewer-')) + roots.push(root) + const store = await AgentSessionRecordStore.open({ directory: root, hostId: 'local' }) + const reserved = await store.reserveOwner({ + sessionId: 'session-renewal', + location: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'folder' + }, + provider: 'codex', + accountHome: { variable: 'CODEX_HOME', path: root }, + runtimeKind: 'native', + expectedFence: null, + spawnToken: 'spawn-renewal', + claimKeyId: 'key-1', + handoffOperationId: null, + probe: { outcome: 'reservation-unused' }, + operation: { + callerKey: 'test', + operationId: `${NOW}-00000000000000000000000000000001`, + fingerprint: 'create' + }, + now: NOW + }) + await store.commitProcessIdentity({ + sessionId: 'session-renewal', + fence: reserved.record.lease.runtimeFence, + process: { + hostId: 'local', + pid: 4242, + processStartTimeMs: NOW - 1_000, + spawnToken: 'spawn-renewal' + }, + now: NOW + }) + await store.proveOwner({ + sessionId: 'session-renewal', + fence: reserved.record.lease.runtimeFence, + link: { + linkId: 'link-renewal', + handle: { provider: 'codex', threadId: 'thread-renewal' }, + origin: 'created', + mintedAtFence: reserved.record.lease.runtimeFence, + observedAt: NOW + }, + now: NOW + }) + return store +} + +afterEach(async () => { + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))) +}) + +describe('structured agent-session lease renewal', () => { + it('drives renewal on the production interval', async () => { + vi.useFakeTimers() + const store = await liveStore() + let now = NOW + const renewer = new StructuredAgentSessionLeaseRenewer({ + store, + probe: async () => ({ + outcome: 'identity-matched', + matchedOn: ['process-start-time'] + }), + now: () => now + }) + try { + renewer.start() + now += 10_000 + await vi.advanceTimersByTimeAsync(10_000) + await vi.waitFor(() => + expect(store.getRecord('session-renewal')?.lease.lastRenewedAt).toBe(now) + ) + } finally { + renewer.stop() + vi.useRealTimers() + } + }) + + it('renews every live owner only after re-proving its child identity', async () => { + const store = await liveStore() + const probe = vi.fn(async () => ({ + outcome: 'identity-matched' as const, + matchedOn: ['process-start-time' as const] + })) + const renewer = new StructuredAgentSessionLeaseRenewer({ + store, + probe, + now: () => NOW + 10_000 + }) + + await renewer.renewNow() + + expect(probe).toHaveBeenCalledOnce() + expect(store.getRecord('session-renewal')?.lease.lastRenewedAt).toBe(NOW + 10_000) + }) + + it('stops extending the lease when child proof is no longer sufficient', async () => { + const store = await liveStore() + const onError = vi.fn() + const renewer = new StructuredAgentSessionLeaseRenewer({ + store, + probe: async () => ({ outcome: 'indeterminate', reason: 'probe unavailable' }), + now: () => NOW + 10_000, + onError + }) + + await renewer.renewNow() + + expect(store.getRecord('session-renewal')?.lease.lastRenewedAt).toBe(NOW) + expect(onError).toHaveBeenCalledWith({ + sessionId: 'session-renewal', + error: expect.any(Error) + }) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.ts new file mode 100644 index 00000000000..f9f9b7b6773 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-lease-renewer.ts @@ -0,0 +1,73 @@ +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { + AGENT_SESSION_LEASE_TTL_MS, + type AgentSessionRecordStore +} from '../../runtime/agent-session-record-store' + +const RENEW_INTERVAL_MS = Math.floor(AGENT_SESSION_LEASE_TTL_MS / 3) + +export class StructuredAgentSessionLeaseRenewer { + private timer: ReturnType | null = null + private running = false + + constructor( + private readonly input: { + store: AgentSessionRecordStore + probe: (record: AgentSessionRecord) => Promise + now: () => number + onError?: (input: { sessionId: string; error: unknown }) => void + intervalMs?: number + } + ) {} + + start(): void { + if (this.timer) { + return + } + this.timer = setInterval(() => void this.renewNow(), this.input.intervalMs ?? RENEW_INTERVAL_MS) + this.timer.unref?.() + } + + stop(): void { + if (this.timer) { + clearInterval(this.timer) + this.timer = null + } + } + + async renewNow(): Promise { + if (this.running) { + return + } + this.running = true + try { + await Promise.all( + this.input.store + .listRecords() + .filter( + (record) => + !record.lease.unreconciled && + record.lease.claimStatus === 'live' && + record.lease.ownerProcess !== null + ) + .map((record) => this.renewRecord(record)) + ) + } finally { + this.running = false + } + } + + private async renewRecord(record: AgentSessionRecord): Promise { + try { + await this.input.store.renewLease({ + sessionId: record.sessionId, + fence: record.lease.runtimeFence, + childProbe: await this.input.probe(record), + now: this.input.now() + }) + } catch (error) { + this.input.onError?.({ sessionId: record.sessionId, error }) + } + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts index c2d2657e7e5..082f79ca194 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts @@ -18,6 +18,7 @@ export async function restoreStructuredAgentSessionsOnRestart(input: { serialize: (sessionId: string, task: () => Promise) => Promise hasSession: (sessionId: string) => boolean onReadable: (sessionId: string, restored: RestoredStructuredAgentSessionRead) => void + restoreHandoff: (sessionId: string) => Promise }): Promise { await Promise.all( input.records.map(async ({ sessionId }) => { @@ -49,6 +50,9 @@ export async function restoreStructuredAgentSessionsOnRestart(input: { input.onReadable(sessionId, restored) } }) + if (input.hasSession(sessionId)) { + await input.restoreHandoff(sessionId) + } }) ) } diff --git a/src/main/runtime/agent-session-lease-transitions.ts b/src/main/runtime/agent-session-lease-transitions.ts index ff3eb1f69e6..3cc134ba9e0 100644 --- a/src/main/runtime/agent-session-lease-transitions.ts +++ b/src/main/runtime/agent-session-lease-transitions.ts @@ -6,7 +6,6 @@ * observed process identity, and a proved provider handle — in that order, at one fence. */ -import { isDeepStrictEqual } from 'node:util' import { adjudicateAgentSessionRestart, evaluateAgentSessionAcquisition, @@ -24,6 +23,7 @@ import type { AgentSessionProcessIdentity, AgentSessionRecord } from '../../shared/agent-session-record' +import { adjudicateRestartedAgentSessionHandoff } from './agent-session-restart-handoff-adjudication' export type AgentSessionReservation = { runtimeKind: AgentSessionOwnerRuntimeKind @@ -50,13 +50,6 @@ export function assertFence(lease: AgentSessionLease, fence: number): void { } } -export function agentSessionReconciliationTargetMatches( - current: AgentSessionRecord, - probed: AgentSessionRecord -): boolean { - return isDeepStrictEqual(current, probed) -} - /** * Compare-and-swap reservation. Writes the intent at fence + 1 before any process exists, so the * loser of a concurrent swap is refused and never spawns. @@ -256,6 +249,12 @@ export function applyAgentSessionRestartAdjudication(args: { lastRenewedAt: args.now }) } + if ( + record.lease.handoffStage === 'preparing' || + record.lease.handoffStage === 'new-owner-proving' + ) { + return adjudicateRestartedAgentSessionHandoff(record, args.probe, args.now) + } const adjudication = adjudicateAgentSessionRestart({ lease: record.lease, probe: args.probe, diff --git a/src/main/runtime/agent-session-process-identity-probe.test.ts b/src/main/runtime/agent-session-process-identity-probe.test.ts index 57e0361baa7..5f310a2a6ee 100644 --- a/src/main/runtime/agent-session-process-identity-probe.test.ts +++ b/src/main/runtime/agent-session-process-identity-probe.test.ts @@ -154,13 +154,18 @@ describe('owner identity probe', () => { ).resolves.toEqual({ outcome: 'identity-matched', matchedOn: ['spawn-token'] }) }) - it('returns no start time on Windows rather than guessing one', async () => { - await expect(readProcessStartTimeMs(process.pid, 'win32')).resolves.toBeNull() + it('reads a CIM-backed start time on Windows when running there', async () => { + const observed = await readProcessStartTimeMs(process.pid, 'win32') + expect(observed === null).toBe(process.platform !== 'win32') }) it('reads a start time for the current process on this platform', async () => { const observed = await readProcessStartTimeMs(process.pid) - if (process.platform === 'linux' || process.platform === 'darwin') { + if ( + process.platform === 'linux' || + process.platform === 'darwin' || + process.platform === 'win32' + ) { expect(observed).not.toBeNull() expect(Math.abs((observed as number) - (Date.now() - process.uptime() * 1000))).toBeLessThan( 60_000 diff --git a/src/main/runtime/agent-session-process-identity-probe.ts b/src/main/runtime/agent-session-process-identity-probe.ts index 8d63fd485ac..93d5427f3aa 100644 --- a/src/main/runtime/agent-session-process-identity-probe.ts +++ b/src/main/runtime/agent-session-process-identity-probe.ts @@ -74,9 +74,26 @@ async function readDarwinProcessStartTimeMs(pid: number): Promise } } +async function readWindowsProcessStartTimeMs(pid: number): Promise { + try { + const script = + `$p = Get-CimInstance Win32_Process -Filter "ProcessId = ${pid}"; ` + + 'if ($p) { $p.CreationDate.ToUniversalTime().ToString("o") }' + const { stdout } = await execFileAsync( + 'powershell.exe', + ['-NoProfile', '-NonInteractive', '-Command', script], + { timeout: PROCESS_START_TIME_TIMEOUT_MS, windowsHide: true } + ) + const parsed = Date.parse(stdout.trim()) + return Number.isFinite(parsed) ? parsed : null + } catch { + return null + } +} + /** - * Windows returns null: reading a start time needs a CIM query Orca does not make here, so a - * Windows child without a token echo qualifies for manual recovery only. + * Process start time is the cross-platform PID-reuse guard when no provider hook can echo the + * spawn token back to the owner probe. */ export async function readProcessStartTimeMs( pid: number, @@ -88,6 +105,9 @@ export async function readProcessStartTimeMs( if (platform === 'darwin') { return readDarwinProcessStartTimeMs(pid) } + if (platform === 'win32') { + return readWindowsProcessStartTimeMs(pid) + } return null } diff --git a/src/main/runtime/agent-session-reconciliation-target.ts b/src/main/runtime/agent-session-reconciliation-target.ts new file mode 100644 index 00000000000..03fd77be7dd --- /dev/null +++ b/src/main/runtime/agent-session-reconciliation-target.ts @@ -0,0 +1,9 @@ +import { isDeepStrictEqual } from 'node:util' +import type { AgentSessionRecord } from '../../shared/agent-session-record' + +export function agentSessionReconciliationTargetMatches( + current: AgentSessionRecord, + probed: AgentSessionRecord +): boolean { + return isDeepStrictEqual(current, probed) +} diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index de8cb9c771f..1f3922b5732 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -10,8 +10,9 @@ */ import { - pruneAgentSessionOperationRows, agentSessionOperationKey, + pruneAgentSessionOperationRows, + settleAgentSessionOperation, type AgentSessionOperationDecision, type AgentSessionOperationOutcome, type AgentSessionOperationRow @@ -31,7 +32,6 @@ import { } from '../../shared/agent-session-record' import { applyAgentSessionRestartAdjudication, - agentSessionReconciliationTargetMatches, commitAgentSessionProcessIdentity, evictAgentSessionOwner, proveAgentSessionOwner, @@ -39,6 +39,7 @@ import { setAgentSessionJournalCheckpoint, type AgentSessionProcessIdentityCommit } from './agent-session-lease-transitions' +import { agentSessionReconciliationTargetMatches } from './agent-session-reconciliation-target' import { setAgentSessionReservationProcesslessProof, type AgentSessionReservationProcesslessProof @@ -291,16 +292,12 @@ export class AgentSessionRecordStore { } async recordOperationOutcome(args: { - callerKey: string + callerKey?: string operationId: string outcome: AgentSessionOperationOutcome }): Promise { await this.transact(() => { - const key = agentSessionOperationKey(args.callerKey, args.operationId) - const row = this.state.operations.get(key) - if (row) { - this.state.operations.set(key, { ...row, outcome: args.outcome }) - } + this.state.operations = settleAgentSessionOperation(this.state.operations, args) }) } diff --git a/src/main/runtime/agent-session-restart-handoff-adjudication.test.ts b/src/main/runtime/agent-session-restart-handoff-adjudication.test.ts new file mode 100644 index 00000000000..a18f55f333f --- /dev/null +++ b/src/main/runtime/agent-session-restart-handoff-adjudication.test.ts @@ -0,0 +1,79 @@ +import { describe, expect, it } from 'vitest' +import type { AgentSessionRecord } from '../../shared/agent-session-record' +import { adjudicateRestartedAgentSessionHandoff } from './agent-session-restart-handoff-adjudication' + +const NOW = 1_800_000_000_000 + +function record(stage: 'preparing' | 'new-owner-proving'): AgentSessionRecord { + return { + schemaVersion: 1, + sessionId: 'session-restart', + location: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'folder' + }, + provider: 'codex', + providerHandleChain: [], + accountHome: { variable: 'CODEX_HOME', path: '/tmp/codex' }, + lease: { + sessionId: 'session-restart', + runtimeKind: stage === 'preparing' ? 'native' : 'tui', + runtimeFence: 4, + handoffStage: stage, + provenHandleLinkId: null, + ownerProcess: { + hostId: 'local', + pid: 4242, + processStartTimeMs: NOW - 1_000, + spawnToken: 'spawn-restart' + }, + reservedSpawnToken: 'spawn-restart', + leaseDeadlineAt: NOW + 30_000, + lastRenewedAt: NOW, + handoffOperationId: 'handoff-op-1', + journalCheckpoint: null, + claimKeyId: 'key-1', + claimStatus: stage === 'preparing' ? 'live' : 'reserved', + unreconciled: true, + deathEvidence: null + }, + createdAt: NOW, + updatedAt: NOW + } +} + +describe('restarted handoff adjudication', () => { + it('continues a dead preparing owner at old-owner-stopped under the same operation', () => { + expect( + adjudicateRestartedAgentSessionHandoff( + record('preparing'), + { outcome: 'pid-absent' }, + NOW + 1_000 + ).lease + ).toMatchObject({ + runtimeKind: 'native', + runtimeFence: 5, + handoffStage: 'old-owner-stopped', + handoffOperationId: 'handoff-op-1', + claimStatus: 'released', + ownerProcess: null + }) + }) + + it('rolls a dead proving target back to the source kind before retry', () => { + expect( + adjudicateRestartedAgentSessionHandoff( + record('new-owner-proving'), + { outcome: 'pid-absent' }, + NOW + 1_000 + ).lease + ).toMatchObject({ + runtimeKind: 'native', + runtimeFence: 5, + handoffStage: 'old-owner-stopped', + handoffOperationId: 'handoff-op-1' + }) + }) +}) diff --git a/src/main/runtime/agent-session-restart-handoff-adjudication.ts b/src/main/runtime/agent-session-restart-handoff-adjudication.ts new file mode 100644 index 00000000000..86a42d7dbc2 --- /dev/null +++ b/src/main/runtime/agent-session-restart-handoff-adjudication.ts @@ -0,0 +1,52 @@ +import { + adjudicateAgentSessionRestart, + type AgentSessionOwnerProbe +} from '../../shared/agent-session-lease-adjudication' +import type { AgentSessionRecord } from '../../shared/agent-session-record' + +export function adjudicateRestartedAgentSessionHandoff( + record: AgentSessionRecord, + probe: AgentSessionOwnerProbe, + now: number +): AgentSessionRecord { + const adjudication = adjudicateAgentSessionRestart({ + lease: record.lease, + probe, + observedAt: now + }) + if (adjudication.disposition === 'readopt') { + return updateLease(record, { ...record.lease, unreconciled: false, lastRenewedAt: now }) + } + if (adjudication.disposition !== 'evicted') { + return updateLease(record, { + ...record.lease, + handoffStage: 'recovering', + claimStatus: + adjudication.disposition === 'conflicted' ? 'conflicted' : record.lease.claimStatus, + unreconciled: false, + lastRenewedAt: now + }) + } + const provingTarget = record.lease.handoffStage === 'new-owner-proving' + return updateLease(record, { + ...record.lease, + runtimeKind: provingTarget + ? record.lease.runtimeKind === 'native' + ? 'tui' + : 'native' + : record.lease.runtimeKind, + runtimeFence: adjudication.nextFence, + handoffStage: 'old-owner-stopped', + ownerProcess: null, + reservedSpawnToken: null, + processlessAt: null, + claimStatus: 'released', + unreconciled: false, + lastRenewedAt: now, + deathEvidence: adjudication.evidence + }) +} + +function updateLease(record: AgentSessionRecord, lease: AgentSessionRecord['lease']) { + return { ...record, lease, updatedAt: lease.lastRenewedAt } +} diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 5227e5f7885..28a2d68b10f 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -75,10 +75,13 @@ import { ensureStructuredAgentSessionHost as installStructuredAgentSessionHost } import { getStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry' import type { AgentSessionAttachParams } from '../native-chat/agent-session-wire/structured-agent-session-attach' import type { StructuredAgentSessionHandoffTransport } from '../native-chat/agent-session-wire/structured-agent-session-handoff-types' +import type { AgentSessionRecord } from '../../shared/agent-session-record' import { codexProviderHandleLink } from '../codex/codex-structured-owner-identity' import { proveCodexTuiRollout } from '../codex/codex-tui-rollout-proof' -import { readProcessStartTimeMs } from './agent-session-process-identity-probe' +import { probeAgentSessionProcessIdentity } from './agent-session-process-identity-probe' import { waitForStructuredTuiExitProof } from './structured-tui-exit-proof' +import { readStructuredTuiProcessIdentity } from './structured-tui-process-identity' +import { hasStructuredTuiIdleEvidence } from './structured-tui-idle-evidence' import { getProfileUserDataPath } from '../orca-profiles/profile-storage-paths' import { getSystemCodexHomePath } from '../codex/codex-home-paths' import { @@ -9120,10 +9123,6 @@ export class OrcaRuntimeService { if (!terminal.processId || !terminal.paneKey || !terminal.tabId) { throw new Error('The resumed terminal did not publish a process identity.') } - await this.waitForTerminal(terminal.handle, { - condition: 'tui-idle', - timeoutMs: 30_000 - }) const proof = await this.waitForStructuredTuiProof({ handle: terminal.handle, paneKey: terminal.paneKey, @@ -9139,12 +9138,12 @@ export class OrcaRuntimeService { tabId: terminal.tabId, paneKey: terminal.paneKey }, - process: { + process: await readStructuredTuiProcessIdentity({ hostId: record.location.executionHostId, - pid: terminal.processId, - processStartTimeMs: await readProcessStartTimeMs(terminal.processId), - spawnToken - }, + rootPid: terminal.processId, + spawnToken, + agent: 'codex' + }), link: codexProviderHandleLink({ threadId: head.handle.threadId, resumed: true, @@ -9169,6 +9168,64 @@ export class OrcaRuntimeService { }) return owner.transcriptPath ? { transcriptPath: owner.transcriptPath } : {} }, + reproveTuiOwner: async ({ record, owner }) => { + const proof = await probeAgentSessionProcessIdentity({ identity: owner.process }) + if (proof.outcome !== 'identity-matched' || proof.matchedOn.length === 0) { + throw new Error('The owning Codex child process could not be re-proved.') + } + const head = record.providerHandleChain.at(-1) + if (head?.handle.provider !== 'codex') { + throw new Error('agent_session_identity_required') + } + const tuiProof = await this.waitForStructuredTuiProof({ + handle: owner.terminal.handle, + paneKey: owner.terminal.paneKey, + threadId: head.handle.threadId, + spawnToken: owner.process.spawnToken, + codexHome: record.accountHome.path, + sessionId: record.sessionId + }) + return { ...owner, transcriptPath: tuiProof.transcriptPath } + }, + recoverTuiOwner: async (record) => { + const identity = record.lease.ownerProcess + const head = record.providerHandleChain.at(-1) + if (!identity || head?.handle.provider !== 'codex') { + throw new Error('agent_session_identity_required') + } + const candidate = [...this.ptysById.values()].find( + (pty) => + pty.connected && + pty.launchToken === identity.spawnToken && + pty.launchAgent === 'codex' && + pty.tabId && + pty.paneKey + ) + const handle = candidate ? this.handleByPtyId.get(candidate.ptyId) : null + if (!candidate?.tabId || !candidate.paneKey || !handle) { + throw new Error('The owning agent terminal could not be recovered.') + } + const proof = await this.waitForStructuredTuiProof({ + handle, + paneKey: candidate.paneKey, + threadId: head.handle.threadId, + spawnToken: identity.spawnToken, + codexHome: record.accountHome.path, + sessionId: record.sessionId + }) + return { + terminal: { handle, tabId: candidate.tabId, paneKey: candidate.paneKey }, + process: identity, + link: codexProviderHandleLink({ + threadId: head.handle.threadId, + resumed: true, + fence: record.lease.runtimeFence, + observedAt: Date.now() + }), + transcriptPath: proof.transcriptPath + } + }, + stopRecoveredOwner: (record) => this.stopStructuredSessionProcess(record), tuiStatus: (owner) => this.structuredTuiStatus(owner.terminal.handle), stopFailedTuiLaunch: async (owner) => { await this.closeTerminal(owner.terminal.handle) @@ -9180,6 +9237,37 @@ export class OrcaRuntimeService { } } + private async stopStructuredSessionProcess(record: AgentSessionRecord): Promise { + const identity = record.lease.ownerProcess + if (!identity) { + return + } + const proof = await probeAgentSessionProcessIdentity({ identity }) + if (proof.outcome === 'pid-absent' || proof.outcome === 'identity-mismatch') { + return + } + if (proof.outcome !== 'identity-matched' || proof.matchedOn.length === 0) { + throw new Error('The recovered owner process could not be stopped safely.') + } + try { + process.kill(identity.pid, 'SIGTERM') + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ESRCH') { + throw error + } + return + } + const deadline = Date.now() + 15_000 + while (Date.now() < deadline) { + const current = await probeAgentSessionProcessIdentity({ identity }) + if (current.outcome === 'pid-absent' || current.outcome === 'identity-mismatch') { + return + } + await new Promise((resolve) => setTimeout(resolve, 100)) + } + throw new Error('The recovered owner process did not exit.') + } + private structuredTuiStatus(handle: string): 'idle' | 'busy' { const pty = this.getLivePtyForHandle(handle) if (pty) { @@ -9188,21 +9276,22 @@ export class OrcaRuntimeService { pty.pty.tailPartialLine, pty.pty.preview ) - return !detectTerminalWaitBlockedReason(text) && - (pty.pty.lastAgentStatus === 'idle' || - this.getAdoptedPtyExplicitIdleStatus(pty.pty) === 'idle' || - isKnownReadyPromptPreview(text)) + return hasStructuredTuiIdleEvidence({ + blocked: detectTerminalWaitBlockedReason(text) !== null, + status: pty.pty.lastAgentStatus, + statusObservedLive: pty.pty.lastAgentStatusObservedLive + }) ? 'idle' : 'busy' } try { const { leaf } = this.getLiveLeafForHandle(handle) const text = buildTerminalWaitText(leaf.tailBuffer, leaf.tailPartialLine, leaf.preview) - const title = leaf.paneTitle ?? this.tabs.get(leaf.tabId)?.title - return !detectTerminalWaitBlockedReason(text) && - (leaf.lastAgentStatus === 'idle' || - (title && detectExplicitIdleStatusFromTitle(title) === 'idle') || - isKnownReadyPromptPreview(text)) + return hasStructuredTuiIdleEvidence({ + blocked: detectTerminalWaitBlockedReason(text) !== null, + status: leaf.lastAgentStatus, + statusObservedLive: leaf.lastAgentStatusObservedLive + }) ? 'idle' : 'busy' } catch { diff --git a/src/main/runtime/rpc/methods/structured-agent-session.ts b/src/main/runtime/rpc/methods/structured-agent-session.ts index 7bb796a20be..3501f9a0a65 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.ts @@ -214,7 +214,7 @@ export const STRUCTURED_AGENT_SESSION_METHODS: RpcAnyMethod[] = [ defineMethod({ name: 'agentSession.requestHandoff', params: HandoffParams, - handler: async (params, ctx) => requireHost(ctx).requestHandoff(params) + handler: async (params, ctx) => requireHost(ctx).requestHandoff(callerFor(ctx), params) }), defineMethod({ name: 'agentSession.handoffStatus', diff --git a/src/main/runtime/structured-tui-idle-evidence.test.ts b/src/main/runtime/structured-tui-idle-evidence.test.ts new file mode 100644 index 00000000000..7177cb9bfc3 --- /dev/null +++ b/src/main/runtime/structured-tui-idle-evidence.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, it } from 'vitest' +import { hasStructuredTuiIdleEvidence } from './structured-tui-idle-evidence' + +describe('structured TUI idle evidence', () => { + it('does not treat a ready prompt preview as proof that a turn is idle', () => { + const readyPreview = ' >_ OpenAI Codex\n model: gpt-5.5\n directory: /workspace' + expect(readyPreview).toContain('OpenAI Codex') + expect( + hasStructuredTuiIdleEvidence({ blocked: false, status: null, statusObservedLive: false }) + ).toBe(false) + }) + + it('requires an explicit idle state and still rejects blocked prompts', () => { + expect( + hasStructuredTuiIdleEvidence({ blocked: false, status: 'idle', statusObservedLive: true }) + ).toBe(true) + expect( + hasStructuredTuiIdleEvidence({ blocked: true, status: 'idle', statusObservedLive: true }) + ).toBe(false) + }) + + it('does not authorize a restored idle status before live observation', () => { + expect( + hasStructuredTuiIdleEvidence({ blocked: false, status: 'idle', statusObservedLive: false }) + ).toBe(false) + }) +}) diff --git a/src/main/runtime/structured-tui-idle-evidence.ts b/src/main/runtime/structured-tui-idle-evidence.ts new file mode 100644 index 00000000000..d91518cdd4c --- /dev/null +++ b/src/main/runtime/structured-tui-idle-evidence.ts @@ -0,0 +1,9 @@ +import type { AgentStatus } from '../../shared/agent-title-core' + +export function hasStructuredTuiIdleEvidence(input: { + blocked: boolean + status: AgentStatus | null + statusObservedLive: boolean +}): boolean { + return !input.blocked && input.status === 'idle' && input.statusObservedLive +} diff --git a/src/main/runtime/structured-tui-process-identity.test.ts b/src/main/runtime/structured-tui-process-identity.test.ts new file mode 100644 index 00000000000..59a06833fa6 --- /dev/null +++ b/src/main/runtime/structured-tui-process-identity.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it, vi } from 'vitest' +import { readStructuredTuiProcessIdentity } from './structured-tui-process-identity' + +describe('structured TUI process identity', () => { + it('binds the direct Codex child instead of the PTY shell pid', async () => { + const readStartTime = vi.fn(async () => 1_700_000_000_000) + await expect( + readStructuredTuiProcessIdentity({ + hostId: 'local', + rootPid: 100, + spawnToken: 'spawn-1', + agent: 'codex', + platform: 'darwin', + readPosixRows: async () => [ + { pid: 100, ppid: 1, stat: 'Ss', command: '/bin/zsh' }, + { pid: 101, ppid: 100, stat: 'S+', command: 'node /opt/codex/bin/codex resume abc' }, + { pid: 102, ppid: 101, stat: 'S+', command: '/opt/codex/vendor/codex' } + ], + readStartTime + }) + ).resolves.toEqual({ + hostId: 'local', + pid: 101, + processStartTimeMs: 1_700_000_000_000, + spawnToken: 'spawn-1' + }) + expect(readStartTime).toHaveBeenCalledWith(101, 'darwin') + }) + + it('fails closed when sibling Codex children make the owner ambiguous', async () => { + await expect( + readStructuredTuiProcessIdentity({ + hostId: 'local', + rootPid: 100, + spawnToken: 'spawn-1', + agent: 'codex', + platform: 'win32', + readWindowsRows: async () => [ + { pid: 100, ppid: 1, name: 'pwsh.exe', command: 'pwsh.exe', executablePath: '' }, + { pid: 101, ppid: 100, name: 'codex.exe', command: 'codex resume a', executablePath: '' }, + { pid: 102, ppid: 100, name: 'codex.exe', command: 'codex resume b', executablePath: '' } + ] + }) + ).rejects.toThrow('one exact Codex child process') + }) + + it('fails closed when the process snapshot omitted the PTY root', async () => { + await expect( + readStructuredTuiProcessIdentity({ + hostId: 'local', + rootPid: 100, + spawnToken: 'spawn-1', + agent: 'codex', + platform: 'darwin', + readPosixRows: async () => [ + { pid: 101, ppid: 100, stat: 'S+', command: 'codex resume abc' } + ] + }) + ).rejects.toThrow('root process was not present') + }) +}) diff --git a/src/main/runtime/structured-tui-process-identity.ts b/src/main/runtime/structured-tui-process-identity.ts new file mode 100644 index 00000000000..0566ec20eba --- /dev/null +++ b/src/main/runtime/structured-tui-process-identity.ts @@ -0,0 +1,96 @@ +import { recognizeAgentProcessFromCommandLine } from '../../shared/agent-process-recognition' +import { + getFreshProcessTableSnapshot, + type ProcessTableRow +} from '../../shared/process-table-snapshot' +import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record' +import { queryWindowsProcessRowsFresh } from '../providers/windows-foreground-process-rows' +import { readProcessStartTimeMs } from './agent-session-process-identity-probe' + +type ProcessRow = { pid: number; ppid: number; command: string; foreground: boolean } + +function descendants(rows: ProcessRow[], rootPid: number): (ProcessRow & { depth: number })[] { + const children = new Map() + for (const row of rows) { + children.set(row.ppid, [...(children.get(row.ppid) ?? []), row]) + } + const found: (ProcessRow & { depth: number })[] = [] + const pending = [{ pid: rootPid, depth: 0 }] + const seen = new Set() + while (pending.length > 0) { + const current = pending.pop()! + if (seen.has(current.pid)) { + continue + } + seen.add(current.pid) + const row = rows.find((candidate) => candidate.pid === current.pid) + if (row) { + found.push({ ...row, depth: current.depth }) + } + for (const child of children.get(current.pid) ?? []) { + pending.push({ pid: child.pid, depth: current.depth + 1 }) + } + } + return found +} + +export function resolveStructuredTuiChildPid( + rows: ProcessRow[], + rootPid: number, + agent: 'codex' +): number | null { + const candidates = descendants(rows, rootPid).filter( + (row) => recognizeAgentProcessFromCommandLine(row.command)?.agent === agent + ) + const foreground = candidates.filter((row) => row.foreground) + const eligible = foreground.length > 0 ? foreground : candidates + eligible.sort((left, right) => left.depth - right.depth || left.pid - right.pid) + if (eligible.length === 0 || eligible[1]?.depth === eligible[0]?.depth) { + return null + } + return eligible[0]!.pid +} + +function posixRows(rows: ProcessTableRow[]): ProcessRow[] { + return rows.map((row) => ({ + pid: row.pid, + ppid: row.ppid, + command: row.command, + foreground: row.stat.includes('+') + })) +} + +export async function readStructuredTuiProcessIdentity(input: { + hostId: string + rootPid: number + spawnToken: string + agent: 'codex' + platform?: NodeJS.Platform + readPosixRows?: () => Promise + readWindowsRows?: typeof queryWindowsProcessRowsFresh + readStartTime?: (pid: number, platform?: NodeJS.Platform) => Promise +}): Promise { + const platform = input.platform ?? process.platform + const rows: ProcessRow[] = + platform === 'win32' + ? (await (input.readWindowsRows ?? queryWindowsProcessRowsFresh)()).map((row) => ({ + pid: row.pid, + ppid: row.ppid, + command: row.command, + foreground: false + })) + : posixRows(await (input.readPosixRows ?? getFreshProcessTableSnapshot)()) + if (!rows.some((row) => row.pid === input.rootPid)) { + throw new Error('The terminal root process was not present in the process snapshot.') + } + const pid = resolveStructuredTuiChildPid(rows, input.rootPid, input.agent) + if (pid === null) { + throw new Error('The resumed terminal did not expose one exact Codex child process.') + } + return { + hostId: input.hostId, + pid, + processStartTimeMs: await (input.readStartTime ?? readProcessStartTimeMs)(pid, platform), + spawnToken: input.spawnToken + } +} diff --git a/src/shared/agent-session-operation-ledger.ts b/src/shared/agent-session-operation-ledger.ts index 864c2c66485..ceddfd1d14b 100644 --- a/src/shared/agent-session-operation-ledger.ts +++ b/src/shared/agent-session-operation-ledger.ts @@ -52,6 +52,28 @@ export function agentSessionOperationKey(callerKey: string, operationId: string) return `${callerKey}${OPERATION_KEY_SEPARATOR}${operationId}` } +export function settleAgentSessionOperation( + rows: ReadonlyMap, + args: { + /** Restart reconciliation omits this because the lease persists no client identity. */ + callerKey?: string + operationId: string + outcome: AgentSessionOperationOutcome + } +): Map { + const targetKey = args.callerKey + ? agentSessionOperationKey(args.callerKey, args.operationId) + : null + return new Map( + [...rows].map(([key, row]) => [ + key, + (targetKey ? key === targetKey : row.operationId === args.operationId) + ? { ...row, outcome: args.outcome } + : row + ]) + ) +} + /** * Retention floor. The tombstone must outlive the window in which its id could still be admitted * as new, plus the accepted future skew — otherwise a retry arriving in the gap becomes a second