mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 00:03:47 +00:00
fix(native-chat): serialize structured handoff recovery
This commit is contained in:
@@ -70,6 +70,7 @@ export async function admitStructuredHandoffRequest(input: {
|
||||
operationId: input.params.envelope.clientOperationId,
|
||||
outcome: { status: 'failed', code: 'agent_session_checkpoint_stale' }
|
||||
})
|
||||
input.operationGuard.finish(input.record.sessionId, input.params.envelope.clientOperationId)
|
||||
return {
|
||||
decision: 'refused',
|
||||
refusal: {
|
||||
|
||||
+25
-3
@@ -1,5 +1,7 @@
|
||||
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
|
||||
import type { AgentSessionHandoffStatus } from '../../../shared/agent-session-wire'
|
||||
import { setStoredAgentSessionHandoffStage } from '../../runtime/agent-session-handoff-record-transitions'
|
||||
import { switchingStructuredHandoffStatus } from './structured-agent-session-handoff-status'
|
||||
import type {
|
||||
StructuredAgentSessionHandoffDeps,
|
||||
StructuredAgentSessionHandoffFlowContext,
|
||||
@@ -12,11 +14,31 @@ export function createStructuredHandoffFlowContext(input: {
|
||||
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
|
||||
const publishStage: StructuredAgentSessionHandoffFlowContext['publishStage'] = (
|
||||
record,
|
||||
direction
|
||||
) => {
|
||||
input.setStatus(
|
||||
record.sessionId,
|
||||
switchingStructuredHandoffStatus(record, direction, input.deps.transport?.hostLabel)
|
||||
)
|
||||
}
|
||||
return {
|
||||
...input,
|
||||
publishStage,
|
||||
enterPreparing: async (record, operationId, direction) => {
|
||||
const prepared = await setStoredAgentSessionHandoffStage(input.deps.store, {
|
||||
sessionId: record.sessionId,
|
||||
fence: record.lease.runtimeFence,
|
||||
stage: 'preparing',
|
||||
handoffOperationId: operationId,
|
||||
now: input.deps.now()
|
||||
})
|
||||
publishStage(prepared, direction)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function stopStructuredNativeTurn(
|
||||
|
||||
+5
-3
@@ -44,19 +44,21 @@ export class StructuredAgentSessionHandoffFlowRunner {
|
||||
fingerprint
|
||||
})
|
||||
const flow = this.run(params, turnId, tuiAlreadyExited)
|
||||
.then(() =>
|
||||
this.input.deps.store.recordOperationOutcome({
|
||||
.then(() => {
|
||||
this.input.operationGuard.finish(sessionId, params.envelope.clientOperationId)
|
||||
return 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.operationGuard.finish(sessionId, params.envelope.clientOperationId)
|
||||
this.input.fail(params, error)
|
||||
})
|
||||
.finally(() => this.input.operationGuard.finish(sessionId, params.envelope.clientOperationId))
|
||||
|
||||
+161
@@ -0,0 +1,161 @@
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import {
|
||||
agentSessionLeaseFixture,
|
||||
agentSessionRecordFixture
|
||||
} from '../../../shared/agent-session-record.test-fixture'
|
||||
import type {
|
||||
AgentSessionHandoffRequest,
|
||||
AgentSessionHandoffStatus
|
||||
} from '../../../shared/agent-session-wire'
|
||||
import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
|
||||
import { queuedStructuredHandoffCanBegin } from './structured-agent-session-handoff-queue'
|
||||
import { StructuredAgentSessionHandoffOperationGuard } from './structured-agent-session-handoff-operation-guard'
|
||||
|
||||
const NOW = 1_800_000_000_000
|
||||
const SESSION = 'session-alpha-1'
|
||||
const OPERATION_A = `${NOW}-00000000000000000000000000000001`
|
||||
const OPERATION_B = `${NOW}-00000000000000000000000000000002`
|
||||
|
||||
let root: string | null = null
|
||||
|
||||
afterEach(async () => {
|
||||
if (root) {
|
||||
await rm(root, { recursive: true, force: true })
|
||||
root = null
|
||||
}
|
||||
})
|
||||
|
||||
async function createGuard() {
|
||||
root = await mkdtemp(join(tmpdir(), 'orca-handoff-operation-guard-'))
|
||||
const store = await AgentSessionRecordStore.open({ directory: root, hostId: 'local' })
|
||||
return { guard: new StructuredAgentSessionHandoffOperationGuard(store), store }
|
||||
}
|
||||
|
||||
function status(phase: 'switching' | 'queued' | 'idle'): AgentSessionHandoffStatus {
|
||||
return {
|
||||
owner: phase === 'idle' ? 'native' : 'none',
|
||||
direction: phase === 'idle' ? null : 'to-tui',
|
||||
phase,
|
||||
stage: phase === 'switching' ? 'preparing' : null,
|
||||
operationId: phase === 'idle' ? null : OPERATION_A
|
||||
}
|
||||
}
|
||||
|
||||
describe('structured handoff operation ownership', () => {
|
||||
it('reserves one winner across concurrent admissions', async () => {
|
||||
const { guard } = await createGuard()
|
||||
const check = (operationId: string) =>
|
||||
guard.check({
|
||||
callerKey: operationId,
|
||||
sessionId: SESSION,
|
||||
operationId,
|
||||
fingerprint: operationId,
|
||||
action: 'start',
|
||||
now: NOW
|
||||
})
|
||||
|
||||
const decisions = await Promise.all([check(OPERATION_A), check(OPERATION_B)])
|
||||
|
||||
expect(decisions.map(({ decision }) => decision).sort()).toEqual(['new', 'refused'])
|
||||
})
|
||||
|
||||
it.each(['switching', 'queued'] as const)(
|
||||
'durably refuses a distinct operation while the %s operation owns the session',
|
||||
async (phase) => {
|
||||
const { guard } = await createGuard()
|
||||
guard.start(SESSION, {
|
||||
callerKey: 'client-a',
|
||||
operationId: OPERATION_A,
|
||||
fingerprint: 'fingerprint-a'
|
||||
})
|
||||
|
||||
expect(
|
||||
await guard.check({
|
||||
callerKey: 'client-b',
|
||||
sessionId: SESSION,
|
||||
operationId: OPERATION_B,
|
||||
fingerprint: 'fingerprint-b',
|
||||
action: 'start',
|
||||
status: status(phase),
|
||||
now: NOW
|
||||
})
|
||||
).toEqual({ decision: 'refused', code: 'agent_session_operation_conflict' })
|
||||
|
||||
guard.finish(SESSION, OPERATION_A)
|
||||
expect(
|
||||
await guard.check({
|
||||
callerKey: 'client-b',
|
||||
sessionId: SESSION,
|
||||
operationId: OPERATION_B,
|
||||
fingerprint: 'fingerprint-b',
|
||||
action: 'start',
|
||||
status: status('idle'),
|
||||
now: NOW
|
||||
})
|
||||
).toMatchObject({
|
||||
decision: 'replay',
|
||||
outcome: { status: 'failed', code: 'agent_session_operation_conflict' }
|
||||
})
|
||||
}
|
||||
)
|
||||
|
||||
it('admits only cancellation beside a queued operation', async () => {
|
||||
const { guard } = await createGuard()
|
||||
guard.start(SESSION, {
|
||||
callerKey: 'client-a',
|
||||
operationId: OPERATION_A,
|
||||
fingerprint: 'fingerprint-a'
|
||||
})
|
||||
|
||||
await expect(
|
||||
guard.check({
|
||||
callerKey: 'client-b',
|
||||
sessionId: SESSION,
|
||||
operationId: OPERATION_B,
|
||||
fingerprint: 'fingerprint-b',
|
||||
action: 'cancel-queued',
|
||||
status: status('queued'),
|
||||
now: NOW
|
||||
})
|
||||
).resolves.toEqual({ decision: 'new' })
|
||||
})
|
||||
})
|
||||
|
||||
describe('queued handoff fence revalidation', () => {
|
||||
const params: AgentSessionHandoffRequest = {
|
||||
envelope: {
|
||||
sessionId: SESSION,
|
||||
clientOperationId: OPERATION_A,
|
||||
expectedRuntimeFence: 7,
|
||||
payloadFingerprint: 'fingerprint'
|
||||
},
|
||||
direction: 'to-tui',
|
||||
mode: 'after-turn',
|
||||
action: 'start'
|
||||
}
|
||||
const queued = status('queued')
|
||||
|
||||
it('accepts the same live owner and fence', () => {
|
||||
const record = agentSessionRecordFixture(
|
||||
agentSessionLeaseFixture({ runtimeKind: 'native', ownerProcess: null })
|
||||
)
|
||||
expect(queuedStructuredHandoffCanBegin(record, queued, params)).toBe(true)
|
||||
})
|
||||
|
||||
it.each([
|
||||
agentSessionLeaseFixture({ runtimeKind: 'native', runtimeFence: 8, ownerProcess: null }),
|
||||
agentSessionLeaseFixture({ runtimeKind: 'tui' }),
|
||||
agentSessionLeaseFixture({
|
||||
runtimeKind: 'native',
|
||||
ownerProcess: null,
|
||||
handoffStage: 'preparing'
|
||||
})
|
||||
])('refuses a changed durable owner or fence', (lease) => {
|
||||
expect(queuedStructuredHandoffCanBegin(agentSessionRecordFixture(lease), queued, params)).toBe(
|
||||
false
|
||||
)
|
||||
})
|
||||
})
|
||||
+56
-7
@@ -24,13 +24,6 @@ export class StructuredAgentSessionHandoffOperationGuard {
|
||||
status?: AgentSessionHandoffStatus
|
||||
now: number
|
||||
}): Promise<HandoffOperationDecision> {
|
||||
const active = this.activeBySession.get(input.sessionId)
|
||||
if (
|
||||
active?.operationId === input.operationId &&
|
||||
(active.fingerprint !== input.fingerprint || active.callerKey !== input.callerKey)
|
||||
) {
|
||||
return { decision: 'refused', code: 'agent_session_operation_conflict' }
|
||||
}
|
||||
const ledger = await this.store.admitOperation({
|
||||
callerKey: input.callerKey,
|
||||
operationId: input.operationId,
|
||||
@@ -40,7 +33,35 @@ export class StructuredAgentSessionHandoffOperationGuard {
|
||||
if (ledger.decision === 'refused') {
|
||||
return { decision: 'refused', code: ledger.code }
|
||||
}
|
||||
const active = this.activeBySession.get(input.sessionId)
|
||||
const queuedCancellation =
|
||||
input.action === 'cancel-queued' &&
|
||||
input.status?.phase === 'queued' &&
|
||||
input.status.operationId === active?.operationId
|
||||
const activeConflict = Boolean(
|
||||
active &&
|
||||
((active.operationId === input.operationId &&
|
||||
(active.fingerprint !== input.fingerprint || active.callerKey !== input.callerKey)) ||
|
||||
(active.operationId !== input.operationId && !queuedCancellation))
|
||||
)
|
||||
const queuedConflict = Boolean(
|
||||
!active &&
|
||||
input.status?.phase === 'queued' &&
|
||||
input.status.operationId !== input.operationId &&
|
||||
input.action !== 'cancel-queued'
|
||||
)
|
||||
if (activeConflict || queuedConflict) {
|
||||
if (ledger.decision === 'admit') {
|
||||
await this.store.recordOperationOutcome({
|
||||
callerKey: input.callerKey,
|
||||
operationId: input.operationId,
|
||||
outcome: { status: 'failed', code: 'agent_session_operation_conflict' }
|
||||
})
|
||||
}
|
||||
return { decision: 'refused', code: 'agent_session_operation_conflict' }
|
||||
}
|
||||
if (ledger.decision === 'admit') {
|
||||
this.reserve(input)
|
||||
return { decision: 'new' }
|
||||
}
|
||||
if (input.action === 'retry' && ledger.row.outcome.status === 'failed') {
|
||||
@@ -49,6 +70,7 @@ export class StructuredAgentSessionHandoffOperationGuard {
|
||||
operationId: input.operationId,
|
||||
outcome: { status: 'pending' }
|
||||
})
|
||||
this.reserve(input)
|
||||
return { decision: 'retry' }
|
||||
}
|
||||
if (
|
||||
@@ -56,6 +78,7 @@ export class StructuredAgentSessionHandoffOperationGuard {
|
||||
!active &&
|
||||
input.status?.operationId !== input.operationId
|
||||
) {
|
||||
this.reserve(input)
|
||||
return { decision: 'new' }
|
||||
}
|
||||
return { decision: 'replay', outcome: ledger.row.outcome }
|
||||
@@ -65,9 +88,35 @@ export class StructuredAgentSessionHandoffOperationGuard {
|
||||
this.activeBySession.set(sessionId, operation)
|
||||
}
|
||||
|
||||
private reserve(input: {
|
||||
action: 'start' | 'cancel-queued' | 'retry' | 'recover'
|
||||
callerKey: string
|
||||
sessionId: string
|
||||
operationId: string
|
||||
fingerprint: string
|
||||
}): void {
|
||||
if (input.action !== 'cancel-queued') {
|
||||
this.start(input.sessionId, input)
|
||||
}
|
||||
}
|
||||
|
||||
finish(sessionId: string, operationId: string): void {
|
||||
if (this.activeBySession.get(sessionId)?.operationId === operationId) {
|
||||
this.activeBySession.delete(sessionId)
|
||||
}
|
||||
}
|
||||
|
||||
async settle(
|
||||
sessionId: string,
|
||||
operationId: string,
|
||||
outcome: AgentSessionOperationOutcome
|
||||
): Promise<void> {
|
||||
const active = this.activeBySession.get(sessionId)
|
||||
await this.store.recordOperationOutcome({
|
||||
...(active?.operationId === operationId ? { callerKey: active.callerKey } : {}),
|
||||
operationId,
|
||||
outcome
|
||||
})
|
||||
this.finish(sessionId, operationId)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ import type {
|
||||
AgentSessionHandoffRequest,
|
||||
AgentSessionHandoffStatus
|
||||
} from '../../../shared/agent-session-wire'
|
||||
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
|
||||
import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection'
|
||||
import type {
|
||||
StructuredAgentSessionHandoffDeps,
|
||||
@@ -53,13 +54,35 @@ export class StructuredAgentSessionHandoffQueue {
|
||||
}
|
||||
}
|
||||
|
||||
export function queuedStructuredHandoffCanBegin(
|
||||
record: AgentSessionRecord,
|
||||
status: AgentSessionHandoffStatus,
|
||||
params: AgentSessionHandoffRequest
|
||||
): boolean {
|
||||
const expectedOwner = params.direction === 'to-tui' ? 'native' : 'tui'
|
||||
return (
|
||||
record.sessionId === params.envelope.sessionId &&
|
||||
status.phase === 'queued' &&
|
||||
status.direction === params.direction &&
|
||||
status.operationId === params.envelope.clientOperationId &&
|
||||
record.lease.runtimeFence === params.envelope.expectedRuntimeFence &&
|
||||
record.lease.runtimeKind === expectedOwner &&
|
||||
record.lease.claimStatus === 'live' &&
|
||||
record.lease.handoffStage === null &&
|
||||
!record.lease.unreconciled
|
||||
)
|
||||
}
|
||||
|
||||
export function enqueueStructuredHandoffAfterTurn(input: {
|
||||
deps: StructuredAgentSessionHandoffDeps
|
||||
queue: StructuredAgentSessionHandoffQueue
|
||||
params: AgentSessionHandoffRequest
|
||||
tuiOwner: StructuredTuiOwner | undefined
|
||||
status: () => AgentSessionHandoffStatus
|
||||
requireRecord: () => AgentSessionRecord
|
||||
setStatus: (status: AgentSessionHandoffStatus) => void
|
||||
begin: (params: AgentSessionHandoffRequest, tuiAlreadyExited: boolean) => void
|
||||
refuse: (record: AgentSessionRecord) => void
|
||||
}): void {
|
||||
const { deps, params, queue, tuiOwner } = input
|
||||
const sessionId = params.envelope.sessionId
|
||||
@@ -83,6 +106,14 @@ export function enqueueStructuredHandoffAfterTurn(input: {
|
||||
: null
|
||||
return tuiReadiness !== null
|
||||
},
|
||||
() => input.begin({ ...params, mode: 'now' }, tuiReadiness === 'exited')
|
||||
() => {
|
||||
const record = input.requireRecord()
|
||||
const status = input.status()
|
||||
if (!queuedStructuredHandoffCanBegin(record, status, params)) {
|
||||
input.refuse(record)
|
||||
return
|
||||
}
|
||||
input.begin({ ...params, mode: 'now' }, tuiReadiness === 'exited')
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
@@ -371,10 +371,10 @@ describe.each(STRUCTURED_HANDOFF_PROVIDER_CASES)(
|
||||
|
||||
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 queued = request('to-tui', 'after-turn')
|
||||
expect(await submit(queued)).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'),
|
||||
@@ -383,7 +383,7 @@ describe.each(STRUCTURED_HANDOFF_PROVIDER_CASES)(
|
||||
coordinator = createCoordinator()
|
||||
|
||||
expect(await submit(cancellation)).toMatchObject({ ok: true, replayed: true })
|
||||
expect(launchTui).not.toHaveBeenCalled()
|
||||
expect((await submit(queued)).ok).toBe(false)
|
||||
})
|
||||
|
||||
it('replays a duplicate active handoff without launching a second owner', async () => {
|
||||
|
||||
@@ -7,7 +7,6 @@ import type {
|
||||
AgentSessionWireRefusal
|
||||
} from '../../../shared/agent-session-wire'
|
||||
import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection'
|
||||
import { setStoredAgentSessionHandoffStage } from '../../runtime/agent-session-handoff-record-transitions'
|
||||
import {
|
||||
admitStructuredHandoffRequest,
|
||||
refuseAdmittedStructuredHandoff,
|
||||
@@ -34,8 +33,7 @@ import {
|
||||
failedStructuredHandoffStatus,
|
||||
idleStructuredHandoffStatus,
|
||||
structuredSessionHasPendingPrompt,
|
||||
structuredTuiStatus,
|
||||
switchingStructuredHandoffStatus
|
||||
structuredTuiStatus
|
||||
} from './structured-agent-session-handoff-status'
|
||||
import type {
|
||||
StructuredAgentSessionHandoffDeps,
|
||||
@@ -61,9 +59,8 @@ export class StructuredAgentSessionHandoffCoordinator {
|
||||
}
|
||||
|
||||
status(sessionId: string): AgentSessionHandoffStatus {
|
||||
return (
|
||||
this.statuses.get(sessionId) ?? idleStructuredHandoffStatus(this.requireRecord(sessionId))
|
||||
)
|
||||
const current = this.statuses.get(sessionId)
|
||||
return current ?? idleStructuredHandoffStatus(this.requireRecord(sessionId))
|
||||
}
|
||||
|
||||
drain = (): Promise<void> => this.flowRunner.drain()
|
||||
@@ -108,7 +105,11 @@ export class StructuredAgentSessionHandoffCoordinator {
|
||||
const { fingerprint } = admission
|
||||
const action = params.action ?? 'start'
|
||||
if (action === 'cancel-queued') {
|
||||
if (currentStatus?.phase !== 'queued' || currentStatus.direction !== params.direction) {
|
||||
if (
|
||||
currentStatus?.phase !== 'queued' ||
|
||||
currentStatus.operationId === null ||
|
||||
currentStatus.direction !== params.direction
|
||||
) {
|
||||
return this.refuseAdmitted(
|
||||
callerKey,
|
||||
params,
|
||||
@@ -117,6 +118,10 @@ export class StructuredAgentSessionHandoffCoordinator {
|
||||
)
|
||||
}
|
||||
this.queue.cancel(record.sessionId)
|
||||
await this.operationGuard.settle(record.sessionId, currentStatus.operationId, {
|
||||
status: 'failed',
|
||||
code: 'agent_session_operation_conflict'
|
||||
})
|
||||
this.setStatus(record.sessionId, idleStructuredHandoffStatus(record))
|
||||
await this.deps.store.recordOperationOutcome({
|
||||
callerKey,
|
||||
@@ -206,8 +211,11 @@ export class StructuredAgentSessionHandoffCoordinator {
|
||||
queue: this.queue,
|
||||
params,
|
||||
tuiOwner,
|
||||
status: () => this.status(record.sessionId),
|
||||
requireRecord: () => this.requireRecord(record.sessionId),
|
||||
setStatus: (status) => this.setStatus(record.sessionId, status),
|
||||
begin: (next, exited) => this.begin(callerKey, next, null, fingerprint, exited)
|
||||
begin: (next, exited) => this.begin(callerKey, next, null, fingerprint, exited),
|
||||
refuse: (latest) => this.refuseQueued(params, latest)
|
||||
})
|
||||
return this.success(record.sessionId, false)
|
||||
}
|
||||
@@ -229,17 +237,30 @@ export class StructuredAgentSessionHandoffCoordinator {
|
||||
code: AgentSessionWireRefusal['code'],
|
||||
message: string
|
||||
): Promise<AgentSessionMutationResult<AgentSessionHandoffResult>> {
|
||||
return refuseAdmittedStructuredHandoff({
|
||||
const result = await refuseAdmittedStructuredHandoff({
|
||||
deps: this.deps,
|
||||
callerKey,
|
||||
params,
|
||||
refusal: refusal(code, message)
|
||||
})
|
||||
this.operationGuard.finish(params.envelope.sessionId, params.envelope.clientOperationId)
|
||||
return result
|
||||
}
|
||||
|
||||
private success = (sessionId: string, replayed: boolean) =>
|
||||
structuredHandoffSuccess(this.deps, sessionId, replayed, this.status(sessionId))
|
||||
|
||||
private refuseQueued(params: AgentSessionHandoffRequest, record: AgentSessionRecord): void {
|
||||
const settlement = this.operationGuard
|
||||
.settle(record.sessionId, params.envelope.clientOperationId, {
|
||||
status: 'failed',
|
||||
code: 'agent_session_checkpoint_stale'
|
||||
})
|
||||
.then(() => this.setStatus(record.sessionId, idleStructuredHandoffStatus(record)))
|
||||
.catch((error) => this.fail(params, error))
|
||||
this.flowRunner.track(settlement)
|
||||
}
|
||||
|
||||
private begin(
|
||||
callerKey: string,
|
||||
params: AgentSessionHandoffRequest,
|
||||
@@ -263,35 +284,10 @@ export class StructuredAgentSessionHandoffCoordinator {
|
||||
retainOwner: (sessionId, owner) => this.tuiOwners.set(sessionId, owner),
|
||||
releaseOwner: (sessionId) => void this.tuiOwners.delete(sessionId),
|
||||
setStatus: (sessionId, status) => this.setStatus(sessionId, status),
|
||||
enterPreparing: (record, operationId, direction) =>
|
||||
this.enterPreparing(record, operationId, direction),
|
||||
publishStage: (record, direction) => this.publishStage(record, direction),
|
||||
requireRecord: (sessionId) => this.requireRecord(sessionId)
|
||||
})
|
||||
}
|
||||
|
||||
private async enterPreparing(
|
||||
record: AgentSessionRecord,
|
||||
operationId: string,
|
||||
direction: 'to-tui' | 'to-native'
|
||||
): Promise<void> {
|
||||
const prepared = await setStoredAgentSessionHandoffStage(this.deps.store, {
|
||||
sessionId: record.sessionId,
|
||||
fence: record.lease.runtimeFence,
|
||||
stage: 'preparing',
|
||||
handoffOperationId: operationId,
|
||||
now: this.deps.now()
|
||||
})
|
||||
this.publishStage(prepared, direction)
|
||||
}
|
||||
|
||||
private publishStage(record: AgentSessionRecord, direction: 'to-tui' | 'to-native'): void {
|
||||
this.setStatus(
|
||||
record.sessionId,
|
||||
switchingStructuredHandoffStatus(record, direction, this.deps.transport?.hostLabel)
|
||||
)
|
||||
}
|
||||
|
||||
private fail(params: AgentSessionHandoffRequest, error: unknown): void {
|
||||
const record = this.requireRecord(params.envelope.sessionId)
|
||||
this.setStatus(
|
||||
|
||||
@@ -71,19 +71,21 @@ export function beginStructuredManualRecovery(input: {
|
||||
throw new Error('The TUI owner proof is still unavailable.')
|
||||
}
|
||||
})
|
||||
.then(() =>
|
||||
deps.store.recordOperationOutcome({
|
||||
.then(() => {
|
||||
operationGuard.finish(sessionId, params.envelope.clientOperationId)
|
||||
return deps.store.recordOperationOutcome({
|
||||
callerKey,
|
||||
operationId: params.envelope.clientOperationId,
|
||||
outcome: { status: 'succeeded', sessionId }
|
||||
})
|
||||
)
|
||||
})
|
||||
.catch(async (error) => {
|
||||
await deps.store.recordOperationOutcome({
|
||||
callerKey,
|
||||
operationId: params.envelope.clientOperationId,
|
||||
outcome: { status: 'failed', code: 'agent_session_handoff_failed' }
|
||||
})
|
||||
operationGuard.finish(sessionId, params.envelope.clientOperationId)
|
||||
const status = idleStructuredHandoffStatus(requireRecord(sessionId))
|
||||
setStatus(sessionId, {
|
||||
...status,
|
||||
|
||||
@@ -107,13 +107,18 @@ describe('admission through the store', () => {
|
||||
}
|
||||
)
|
||||
|
||||
it('admits only the reserved TUI proof token before ownership', () => {
|
||||
it('admits only the reserved TUI proof token while the new process is proving', () => {
|
||||
publish(
|
||||
agentSessionLeaseFixture({
|
||||
runtimeKind: 'tui',
|
||||
claimStatus: 'reserved',
|
||||
handoffStage: 'new-owner-proving',
|
||||
ownerProcess: null,
|
||||
ownerProcess: {
|
||||
hostId: 'local',
|
||||
pid: 4200,
|
||||
processStartTimeMs: 10,
|
||||
spawnToken: 'proof-token'
|
||||
},
|
||||
reservedSpawnToken: 'proof-token'
|
||||
})
|
||||
)
|
||||
@@ -127,6 +132,26 @@ describe('admission through the store', () => {
|
||||
).toBe(false)
|
||||
})
|
||||
|
||||
it('refuses proof input that does not match the committed process identity', () => {
|
||||
publish(
|
||||
agentSessionLeaseFixture({
|
||||
runtimeKind: 'tui',
|
||||
claimStatus: 'reserved',
|
||||
handoffStage: 'new-owner-proving',
|
||||
ownerProcess: {
|
||||
hostId: 'local',
|
||||
pid: 4200,
|
||||
processStartTimeMs: 10,
|
||||
spawnToken: 'other-token'
|
||||
},
|
||||
reservedSpawnToken: 'proof-token'
|
||||
})
|
||||
)
|
||||
expect(gate.admitProof(PTY_ID, { sessionId: SESSION_ID, spawnToken: 'proof-token' })).toBe(
|
||||
false
|
||||
)
|
||||
})
|
||||
|
||||
it('refuses proof input after ownership is live', () => {
|
||||
publish(agentSessionLeaseFixture({ reservedSpawnToken: 'proof-token' }))
|
||||
expect(gate.admitProof(PTY_ID, { sessionId: SESSION_ID, spawnToken: 'proof-token' })).toBe(
|
||||
|
||||
@@ -79,8 +79,8 @@ export class AgentSessionPtyWriteGate {
|
||||
lease?.runtimeKind === 'tui' &&
|
||||
lease.claimStatus === 'reserved' &&
|
||||
lease.handoffStage === 'new-owner-proving' &&
|
||||
lease.ownerProcess === null &&
|
||||
lease.reservedSpawnToken === authority.spawnToken &&
|
||||
(lease.ownerProcess === null || lease.ownerProcess.spawnToken === authority.spawnToken) &&
|
||||
!lease.unreconciled
|
||||
)
|
||||
}
|
||||
|
||||
@@ -34,7 +34,7 @@ function notifier(revealTerminalSession: ReturnType<typeof vi.fn>) {
|
||||
}
|
||||
|
||||
describe('structured TUI launch tab binding', () => {
|
||||
it('rebuilds a Claude proving link only from launch-token-bound hook evidence', async () => {
|
||||
it('rebuilds a Claude proving link from current launch-token-bound hook evidence', async () => {
|
||||
const paneKey = 'tab-claude:leaf-claude'
|
||||
const spawnToken = 'claude-restart-token'
|
||||
const sessionId = '019fd532-7c11-7a90-b6de-4e1a2c3d5f61'
|
||||
@@ -113,12 +113,85 @@ describe('structured TUI launch tab binding', () => {
|
||||
}
|
||||
})
|
||||
expect(reproved.link.linkId).not.toBe('claude-old')
|
||||
expect(attestAgentHookCompatibilityAuthority).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('requires restored hook attestation after the runtime restarts', async () => {
|
||||
const paneKey = 'tab-restored:leaf-restored'
|
||||
const spawnToken = 'restored-token'
|
||||
const sessionId = '019fd532-7c11-7a90-b6de-4e1a2c3d5f62'
|
||||
const transcriptPath = '/tmp/claude-home/projects/worktree/restored.jsonl'
|
||||
const attestAgentHookCompatibilityAuthority = vi.fn(() => ({
|
||||
paneKey,
|
||||
source: 'hydrated_commitment' as const
|
||||
}))
|
||||
const runtime = new OrcaRuntimeService(null, undefined, {
|
||||
attestAgentHookCompatibilityAuthority,
|
||||
getAgentProviderSessionRowsForPane: () => [
|
||||
{
|
||||
paneKey,
|
||||
connectionId: null,
|
||||
state: 'done',
|
||||
prompt: '',
|
||||
agentType: 'claude',
|
||||
receivedAt: 10,
|
||||
stateStartedAt: 10,
|
||||
providerSession: { key: 'session_id', id: sessionId, transcriptPath }
|
||||
}
|
||||
]
|
||||
})
|
||||
const internal = runtime as unknown as {
|
||||
createStructuredAgentSessionHandoffTransport(): StructuredAgentSessionHandoffTransport
|
||||
ptysById: Map<string, unknown>
|
||||
restoredOrchestrationAuthorityByPtyId: Map<string, unknown>
|
||||
}
|
||||
internal.ptysById.set('pty-restored', {
|
||||
ptyId: 'pty-restored',
|
||||
worktreeId: WORKTREE_ID,
|
||||
connectionId: null,
|
||||
tabId: 'tab-restored',
|
||||
paneKey,
|
||||
launchToken: spawnToken,
|
||||
launchAgent: 'claude',
|
||||
connected: true
|
||||
})
|
||||
internal.restoredOrchestrationAuthorityByPtyId.set('pty-restored', {})
|
||||
const record = {
|
||||
sessionId: 'session-restored',
|
||||
accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/tmp/claude-home' },
|
||||
providerHandleChain: [
|
||||
{
|
||||
linkId: 'claude-restored',
|
||||
handle: { provider: 'claude', sessionId, leafUuid: 'leaf-restored' },
|
||||
origin: 'created',
|
||||
mintedAtFence: 1,
|
||||
observedAt: 1
|
||||
}
|
||||
],
|
||||
lease: {
|
||||
runtimeFence: 4,
|
||||
ownerProcess: {
|
||||
hostId: 'local',
|
||||
pid: 4545,
|
||||
processStartTimeMs: 30,
|
||||
spawnToken
|
||||
},
|
||||
provenHandleLinkId: null
|
||||
}
|
||||
} as never
|
||||
|
||||
await internal.createStructuredAgentSessionHandoffTransport().recoverTuiOwner(record)
|
||||
|
||||
expect(attestAgentHookCompatibilityAuthority).toHaveBeenCalledWith({
|
||||
paneKey,
|
||||
launchTokenHash: createHash('sha256').update(spawnToken).digest('hex'),
|
||||
connectionId: null,
|
||||
terminalProvenance: 'restored'
|
||||
})
|
||||
attestAgentHookCompatibilityAuthority.mockReturnValueOnce(null as never)
|
||||
await expect(
|
||||
internal.createStructuredAgentSessionHandoffTransport().recoverTuiOwner(record)
|
||||
).rejects.toThrow('launch-token authority')
|
||||
})
|
||||
|
||||
it('proves the published launch tab before returning its revealed renderer binding', async () => {
|
||||
|
||||
@@ -9417,14 +9417,16 @@ export class OrcaRuntimeService {
|
||||
if (!paneKey) {
|
||||
throw new Error('The owning Claude terminal did not retain its pane identity.')
|
||||
}
|
||||
const attestation = this.attestAgentHookCompatibilityAuthorityFn?.({
|
||||
paneKey,
|
||||
launchTokenHash: createHash('sha256').update(input.spawnToken).digest('hex'),
|
||||
connectionId: input.candidate.connectionId,
|
||||
terminalProvenance: 'restored'
|
||||
})
|
||||
if (attestation?.paneKey !== paneKey) {
|
||||
throw new Error('The owning Claude terminal did not retain launch-token authority.')
|
||||
if (this.restoredOrchestrationAuthorityByPtyId.has(input.candidate.ptyId)) {
|
||||
const attestation = this.attestAgentHookCompatibilityAuthorityFn?.({
|
||||
paneKey,
|
||||
launchTokenHash: createHash('sha256').update(input.spawnToken).digest('hex'),
|
||||
connectionId: input.candidate.connectionId,
|
||||
terminalProvenance: 'restored'
|
||||
})
|
||||
if (attestation?.paneKey !== paneKey) {
|
||||
throw new Error('The owning Claude terminal did not retain launch-token authority.')
|
||||
}
|
||||
}
|
||||
const matches = (this.getAgentProviderSessionRowsForPaneFn?.(paneKey) ?? []).filter(
|
||||
(row) =>
|
||||
|
||||
@@ -197,7 +197,7 @@ export const HandoffParams = z
|
||||
envelope: MutationEnvelope,
|
||||
direction: z.enum(['to-tui', 'to-native']),
|
||||
mode: z.enum(['now', 'after-turn', 'stop-turn']),
|
||||
action: z.enum(['start', 'cancel-queued', 'retry']).optional()
|
||||
action: z.enum(['start', 'cancel-queued', 'retry', 'recover']).optional()
|
||||
})
|
||||
.strict()
|
||||
|
||||
|
||||
@@ -331,6 +331,21 @@ describe('method routing', () => {
|
||||
expect(response).toMatchObject({ ok: true })
|
||||
expect(hostCalls.requestHandoff).toHaveBeenCalledWith({ callerKey: 'mobile-device-a' }, params)
|
||||
})
|
||||
|
||||
it('accepts manual proof recovery at the RPC boundary', async () => {
|
||||
const params = {
|
||||
envelope: envelope(),
|
||||
direction: 'to-tui',
|
||||
mode: 'now',
|
||||
action: 'recover'
|
||||
}
|
||||
|
||||
await expect(call('agentSession.requestHandoff', params)).resolves.toMatchObject({ ok: true })
|
||||
expect(hostCalls.requestHandoff).toHaveBeenCalledWith(
|
||||
{ callerKey: 'trusted-local:runtime' },
|
||||
params
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
describe('mobile image provenance', () => {
|
||||
|
||||
Reference in New Issue
Block a user