fix structured TUI adoption cleanup races

This commit is contained in:
Brennan Benson
2026-08-23 12:02:26 -07:00
parent c69922bc40
commit 24a9e9862a
7 changed files with 354 additions and 74 deletions
@@ -64,7 +64,7 @@ export function isStructuredTuiAdoptionReservation(
lease.runtimeKind === 'tui' &&
lease.claimStatus === 'reserved' &&
lease.handoffStage === 'new-owner-proving' &&
lease.ownerProcess === null &&
(lease.ownerProcess === null || lease.ownerProcess.spawnToken === spawnToken) &&
lease.reservedSpawnToken === spawnToken &&
!lease.unreconciled
)
@@ -117,10 +117,9 @@ export async function reserveStructuredTuiAdoption(
}
/**
* Undo a reservation whose proof never landed. `processlessAt` is the durable "nothing ever
* spawned under this token" proof, which is what lets the next attempt supersede the record
* instead of being refused by it forever — and what stops a restart from latching it into
* recovery.
* Undo a reservation whose proof never landed. Adoption only attributes its synthetic token to an
* existing process, so clearing a partially committed identity truthfully restores the durable
* "nothing spawned under this token" proof needed by the next attempt.
*/
export async function releaseStructuredTuiAdoptionReservation(
input: StructuredTuiAdoptionReservationRelease & {
@@ -138,12 +137,18 @@ export async function releaseStructuredTuiAdoptionReservation(
return
}
const now = input.now()
await input.deps.store.setReservationProcesslessProof({
sessionId: input.sessionId,
fence: input.fence,
spawnToken: input.spawnToken,
processlessAt: now,
now
await input.deps.store.transitionHandoff(input.sessionId, (current) => {
if (
current.lease.runtimeFence !== input.fence ||
!isStructuredTuiAdoptionReservation(current, input.spawnToken)
) {
return current
}
return {
...current,
lease: { ...current.lease, ownerProcess: null, processlessAt: now },
updatedAt: now
}
})
}
@@ -18,6 +18,7 @@ export type StructuredTuiAdoptionRequest = {
caller: StructuredAgentSessionCaller
params: AgentSessionAttachParams
owner: StructuredTuiOwner
onOwnerProven?: () => void
}
export function adoptStructuredTuiOwner(
@@ -56,6 +57,7 @@ async function adopt(
} else {
assertMatchingTuiOwner(record, owner)
}
input.onOwnerProven?.()
const attached = await attachJournal({
record,
params,
@@ -92,12 +94,16 @@ async function proveAdoptedTuiOwner(
): Promise<AgentSessionRecord> {
const { sessionId } = record
const fence = record.lease.runtimeFence
await input.deps.store.commitProcessIdentity({
sessionId,
fence,
process: input.owner.process,
now: input.now()
})
if (record.lease.ownerProcess === null) {
await input.deps.store.commitProcessIdentity({
sessionId,
fence,
process: input.owner.process,
now: input.now()
})
} else if (!isDeepStrictEqual(record.lease.ownerProcess, input.owner.process)) {
throw new Error('agent_session_conflict')
}
return input.deps.store.proveOwner({
sessionId,
fence,
@@ -40,6 +40,7 @@ type ProofRig = {
proofWrites: string[]
refusedWrites: string[]
codexHome: string
historyFilePath: ReturnType<typeof vi.fn>
}
let root: string
@@ -90,6 +91,7 @@ async function buildRig(): Promise<ProofRig> {
hostId: 'local'
})
agentSessionPtyWriteGate.attachRecordLookup((sessionId) => store.getRecord(sessionId))
const historyFilePath = vi.fn(async () => null)
const host = new StructuredAgentSessionHost({
store,
adapter: {
@@ -97,7 +99,8 @@ async function buildRig(): Promise<ProofRig> {
dispatch: vi.fn(),
cancelTurn: vi.fn(),
answerPrompt: vi.fn(),
setOption: vi.fn()
setOption: vi.fn(),
historyFilePath
},
journalRoot: join(root, 'journals'),
claimKeyId: 'key-1',
@@ -110,7 +113,15 @@ async function buildRig(): Promise<ProofRig> {
setStructuredAgentSessionHost(host)
const runtime = new OrcaRuntimeService({ getSettings: () => ({ agentDefaultEnv: {} }) } as never)
const state: ProofRig = { runtime, store, host, proofWrites: [], refusedWrites: [], codexHome }
const state: ProofRig = {
runtime,
store,
host,
proofWrites: [],
refusedWrites: [],
codexHome,
historyFilePath
}
const ptyRecord = {
ptyId: PTY_ID,
connected: true,
@@ -255,8 +266,7 @@ describe('structured Codex adoption against the real PTY write gate', () => {
})
}, 30_000)
it('lets the attempt holding the reservation win when two adoptions overlap', async () => {
// Both attempts park here: reservation taken and pane bound, proof not yet written.
it('deduplicates a replay while a third fresh operation supersedes it', async () => {
const arrived: string[] = []
let releaseBoth: () => void = () => {}
const bothArrived = new Promise<void>((resolve) => {
@@ -280,13 +290,17 @@ describe('structured Codex adoption against the real PTY write gate', () => {
}
)
const superseded = rig.runtime.adoptStructuredAgentSessionTerminal(
{ ...adoptInput(), threadId: THREAD_ID },
{ callerKey: 'renderer-1' }
)
const replayedInput = { ...adoptInput(), threadId: THREAD_ID }
const superseded = rig.runtime.adoptStructuredAgentSessionTerminal(replayedInput, {
callerKey: 'renderer-1'
})
superseded.catch(() => undefined)
const replay = rig.runtime.adoptStructuredAgentSessionTerminal(replayedInput, {
callerKey: 'renderer-1'
})
expect(replay).toBe(superseded)
await vi.waitFor(() => expect(arrived).toHaveLength(1))
// The second attempt carries the same sessionId and supersedes the first one's reservation.
// A fresh operation is independent and supersedes the shared delivery's reservation.
const winner = rig.runtime.adoptStructuredAgentSessionTerminal(
{ ...adoptInput(), threadId: THREAD_ID },
{ callerKey: 'renderer-1' }
@@ -294,6 +308,7 @@ describe('structured Codex adoption against the real PTY write gate', () => {
await expect(winner).resolves.toMatchObject({ ok: true })
await expect(superseded).rejects.toThrow('could not verify its Codex session')
await expect(replay).rejects.toThrow('could not verify its Codex session')
// The loser's cleanup ran against a pane and a reservation that were no longer its own.
expect(agentSessionPtyWriteGate.boundSessionId(PTY_ID)).toBe(SESSION_ID)
expect(rig.store.getRecord(SESSION_ID)).toMatchObject({
@@ -301,6 +316,140 @@ describe('structured Codex adoption against the real PTY write gate', () => {
})
}, 30_000)
it('rolls back a process identity when proving the owner fails', async () => {
vi.spyOn(rig.store, 'proveOwner').mockRejectedValueOnce(new Error('fault after process commit'))
await expect(
rig.runtime.adoptStructuredAgentSessionTerminal(
{ ...adoptInput(), threadId: THREAD_ID },
{ callerKey: 'renderer-1' }
)
).rejects.toThrow('fault after process commit')
expect(rig.store.getRecord(SESSION_ID)?.lease).toMatchObject({
claimStatus: 'reserved',
ownerProcess: null,
processlessAt: expect.any(Number)
})
expect(agentSessionPtyWriteGate.boundSessionId(PTY_ID)).toBeNull()
await expect(
rig.runtime.adoptStructuredAgentSessionTerminal(
{ ...adoptInput(), threadId: THREAD_ID },
{ callerKey: 'renderer-1' }
)
).resolves.toMatchObject({ ok: true })
}, 30_000)
it('preserves a just-proved owner when the next runtime step fails', async () => {
const settlePtyAttempt =
agentSessionPtyWriteGate.settlePtyAttempt.bind(agentSessionPtyWriteGate)
const settle = vi.spyOn(agentSessionPtyWriteGate, 'settlePtyAttempt')
settle.mockImplementationOnce((ptyId, attemptToken) => {
const settled = settlePtyAttempt(ptyId, attemptToken)
expect(settled).toBe(true)
throw new Error('fault after owner proof')
})
await expect(
rig.runtime.adoptStructuredAgentSessionTerminal(
{ ...adoptInput(), threadId: THREAD_ID },
{ callerKey: 'renderer-1' }
)
).rejects.toThrow('fault after owner proof')
expect(rig.store.getRecord(SESSION_ID)?.lease.claimStatus).toBe('live')
expect(agentSessionPtyWriteGate.boundSessionId(PTY_ID)).toBe(SESSION_ID)
}, 30_000)
it('keeps the pane settled when journal attachment fails after owner proof', async () => {
rig.historyFilePath.mockRejectedValueOnce(new Error('journal attach fault'))
const input = { ...adoptInput(), threadId: THREAD_ID }
await expect(
rig.runtime.adoptStructuredAgentSessionTerminal(input, { callerKey: 'renderer-1' })
).rejects.toThrow('journal attach fault')
expect(rig.store.getRecord(SESSION_ID)?.lease).toMatchObject({
claimStatus: 'live',
ownerProcess: { pid: 4242 }
})
expect(agentSessionPtyWriteGate.boundSessionId(PTY_ID)).toBe(SESSION_ID)
expect(agentSessionPtyWriteGate.admit(PTY_ID)).toMatchObject({
admitted: true,
sessionId: SESSION_ID
})
await expect(
rig.runtime.adoptStructuredAgentSessionTerminal(input, { callerKey: 'renderer-1' })
).resolves.toMatchObject({ ok: true, replayed: true })
}, 30_000)
it('keeps the pane settled when outcome persistence fails', async () => {
vi.spyOn(rig.store, 'recordOperationOutcome').mockRejectedValueOnce(
new Error('outcome persistence fault')
)
const input = { ...adoptInput(), threadId: THREAD_ID }
await expect(
rig.runtime.adoptStructuredAgentSessionTerminal(input, { callerKey: 'renderer-1' })
).rejects.toThrow('outcome persistence fault')
expect(rig.store.getRecord(SESSION_ID)?.lease.claimStatus).toBe('live')
expect(agentSessionPtyWriteGate.boundSessionId(PTY_ID)).toBe(SESSION_ID)
await expect(
rig.runtime.adoptStructuredAgentSessionTerminal(input, { callerKey: 'renderer-1' })
).resolves.toMatchObject({ ok: true, replayed: true })
}, 30_000)
it('retries a transient real-store reservation release without masking the proof error', async () => {
await rm(join(rig.codexHome, 'sessions'), { recursive: true, force: true })
const release = vi
.spyOn(rig.store, 'transitionHandoff')
.mockRejectedValueOnce(new Error('transient store write fault'))
.mockRejectedValueOnce(new Error('transient store write fault'))
.mockRejectedValueOnce(new Error('transient store write fault'))
await expect(
rig.runtime.adoptStructuredAgentSessionTerminal(
{ ...adoptInput(), threadId: THREAD_ID },
{ callerKey: 'renderer-1' }
)
).rejects.toThrow('did not prove the expected Codex rollout')
expect(release).toHaveBeenCalledTimes(3)
expect(agentSessionPtyWriteGate.boundSessionId(PTY_ID)).toBe(SESSION_ID)
await vi.waitFor(
() => {
expect(release).toHaveBeenCalledTimes(4)
expect(rig.store.getRecord(SESSION_ID)?.lease).toMatchObject({
claimStatus: 'reserved',
ownerProcess: null,
processlessAt: expect.any(Number)
})
expect(agentSessionPtyWriteGate.boundSessionId(PTY_ID)).toBeNull()
},
{ timeout: 3_000 }
)
}, 30_000)
it('does not recreate a pane binding when the pane exits during settlement', async () => {
const settle = vi.spyOn(agentSessionPtyWriteGate, 'settlePtyAttempt')
settle.mockImplementationOnce((ptyId) => {
const internal = rig.runtime as unknown as {
ptysById: Map<string, { connected: boolean }>
}
internal.ptysById.get(ptyId)!.connected = false
agentSessionPtyWriteGate.unbindPty(ptyId)
return false
})
await expect(
rig.runtime.adoptStructuredAgentSessionTerminal(
{ ...adoptInput(), threadId: THREAD_ID },
{ callerKey: 'renderer-1' }
)
).resolves.toMatchObject({ ok: true })
expect(rig.store.getRecord(SESSION_ID)?.lease.claimStatus).toBe('live')
expect(agentSessionPtyWriteGate.boundSessionId(PTY_ID)).toBeNull()
}, 30_000)
it('refuses a pane another structured session already owns', async () => {
agentSessionPtyWriteGate.bindPty(PTY_ID, 'someone-elses-session')
+57 -29
View File
@@ -119,6 +119,8 @@ import {
import { waitForStructuredTuiExitProof } from './structured-tui-exit-proof'
import { stopAdoptedCodexTui } from './adopted-codex-tui-stop'
import { readStructuredTuiProcessIdentity } from './structured-tui-process-identity'
import { StructuredTuiAdoptionInflight } from './structured-tui-adoption-inflight'
import { releaseStructuredTuiAdoptionReservation } from './structured-tui-adoption-reservation-release'
import { hasStructuredTuiIdleEvidence } from './structured-tui-idle-evidence'
import { evaluateStructuredTuiRecoveryClaim } from './structured-tui-recovery-claim-match'
import { getProfileUserDataPath } from '../orca-profiles/profile-storage-paths'
@@ -3191,6 +3193,9 @@ export class OrcaRuntimeService {
private structuredAgentSessionTabRestorePromise: Promise<void> | null = null
private structuredAgentSessionStartupRestorePromise: Promise<void> | null = null
private readonly adoptedStructuredTuiOwners = new Map<string, StructuredTuiOwner>()
private readonly structuredTuiAdoptionInflight = new StructuredTuiAdoptionInflight<
AgentSessionMutationResult<AgentSessionAttachResult>
>()
private leaves = new Map<string, RuntimeLeafRecord>()
// Why: PTY output is a per-keystroke hot path. Looking up affected leaves by
// ptyId keeps active TUI redraws independent of the total open terminal count.
@@ -11426,7 +11431,7 @@ export class OrcaRuntimeService {
}
}
async adoptStructuredAgentSessionTerminal(
adoptStructuredAgentSessionTerminal(
input: {
envelope: {
sessionId: string
@@ -11441,6 +11446,20 @@ export class OrcaRuntimeService {
threadId?: string
},
caller: StructuredAgentSessionCaller
): Promise<AgentSessionMutationResult<AgentSessionAttachResult>> {
return this.structuredTuiAdoptionInflight.run(
{
callerKey: caller.callerKey,
operationId: input.envelope.clientOperationId,
fingerprint: input.envelope.payloadFingerprint
},
() => this.adoptStructuredAgentSessionTerminalOnce(input, caller)
)
}
private async adoptStructuredAgentSessionTerminalOnce(
input: Parameters<OrcaRuntimeService['adoptStructuredAgentSessionTerminal']>[0],
caller: StructuredAgentSessionCaller
): Promise<AgentSessionMutationResult<AgentSessionAttachResult>> {
const pty = this.ptysById.get(input.ptyId)
if (
@@ -11504,7 +11523,7 @@ export class OrcaRuntimeService {
}
const { fence, spawnToken } = reserved
agentSessionPtyWriteGate.bindPtyForAttempt(input.ptyId, sessionId, spawnToken)
let owned = false
let ownerProven = false
try {
return await this.attachAdoptedStructuredTui({
input,
@@ -11515,32 +11534,38 @@ export class OrcaRuntimeService {
spawnToken,
rootProcessId,
pty,
onOwned: () => {
owned = true
onOwnerProven: (owner) => {
if (ownerProven) {
return
}
ownerProven = true
this.adoptedStructuredTuiOwners.set(sessionId, owner)
const settled = agentSessionPtyWriteGate.settlePtyAttempt(input.ptyId, spawnToken)
const currentPty = this.ptysById.get(input.ptyId)
if (
!settled &&
currentPty?.connected &&
currentPty.paneKey === input.paneKey &&
agentSessionPtyWriteGate.boundSessionId(input.ptyId) === null
) {
agentSessionPtyWriteGate.bindPty(input.ptyId, sessionId)
}
}
})
} finally {
if (owned) {
agentSessionPtyWriteGate.settlePtyAttempt(input.ptyId, spawnToken)
} else {
// A failed proof must leave nothing behind: an unreleased reservation is exactly the
// latched lease that makes the next attempt refuse itself. Both releases are scoped to this
// attempt's own spawn token, because a second adoption of the same pane carries the same
// sessionId — clearing by pty alone unbinds the pane the attempt that superseded us is
// proving on, and then neither can finish.
agentSessionPtyWriteGate.releasePtyAttempt(input.ptyId, spawnToken)
await host
.releaseAdoptedTuiReservation({ sessionId, fence, spawnToken })
.catch((error: unknown) => {
// Swallowed on purpose: the proof error below is the one the user must see. But a
// release that never persisted leaves the lease latched, so it cannot also be silent.
console.warn('[native-chat] adopted TUI reservation release failed', {
sessionId,
fence,
spawnToken,
error
})
if (!ownerProven) {
const error = await releaseStructuredTuiAdoptionReservation(
() => host.releaseAdoptedTuiReservation({ sessionId, fence, spawnToken }),
() => agentSessionPtyWriteGate.releasePtyAttempt(input.ptyId, spawnToken)
)
if (error !== null) {
console.warn('[native-chat] adopted TUI reservation release failed', {
sessionId,
fence,
spawnToken,
error
})
}
}
}
}
@@ -11554,7 +11579,7 @@ export class OrcaRuntimeService {
spawnToken: string
rootProcessId: number
pty: RuntimePtyWorktreeRecord
onOwned: () => void
onOwnerProven: (owner: StructuredTuiOwner) => void
}): Promise<AgentSessionMutationResult<AgentSessionAttachResult>> {
const { input, caller, host, intent, fence, spawnToken, rootProcessId, pty } = args
const readPty = (): RuntimePtyWorktreeRecord => {
@@ -11649,11 +11674,14 @@ export class OrcaRuntimeService {
runtimeKind: 'tui',
providerHandle: { kind: 'codex', threadId }
}
const adopted = await host.adoptTuiOwner({ caller, params, owner })
const adopted = await host.adoptTuiOwner({
caller,
params,
owner,
onOwnerProven: () => args.onOwnerProven(owner)
})
if (adopted.ok) {
// The pane stays bound to the reservation this call already made; only the owner is new.
this.adoptedStructuredTuiOwners.set(input.envelope.sessionId, owner)
args.onOwned()
args.onOwnerProven(owner)
}
return adopted
}
@@ -0,0 +1,28 @@
const MAX_IN_FLIGHT_ADOPTIONS = 4_096
export class StructuredTuiAdoptionInflight<Result> {
private readonly operations = new Map<string, Promise<Result>>()
run(
input: { callerKey: string; operationId: string; fingerprint: string },
adopt: () => Promise<Result>
): Promise<Result> {
const key = `${input.callerKey}\0${input.operationId}\0${input.fingerprint}`
const existing = this.operations.get(key)
if (existing) {
return existing
}
if (this.operations.size >= MAX_IN_FLIGHT_ADOPTIONS) {
return Promise.reject(new Error('Too many structured TUI adoptions are still pending.'))
}
const operation = adopt()
this.operations.set(key, operation)
const drop = (): void => {
if (this.operations.get(key) === operation) {
this.operations.delete(key)
}
}
void operation.then(drop, drop)
return operation
}
}
@@ -0,0 +1,27 @@
const RELEASE_ATTEMPTS = 3
const BACKGROUND_RETRY_MS = 1_000
function retryInBackground(release: () => Promise<void>, onReleased: () => void): void {
const timer = setTimeout(() => {
void release().then(onReleased, () => retryInBackground(release, onReleased))
}, BACKGROUND_RETRY_MS)
timer.unref?.()
}
export async function releaseStructuredTuiAdoptionReservation(
release: () => Promise<void>,
onReleased: () => void
): Promise<unknown | null> {
let failure: unknown = null
for (let attempt = 0; attempt < RELEASE_ATTEMPTS; attempt += 1) {
try {
await release()
onReleased()
return null
} catch (error) {
failure = error
}
}
retryInBackground(release, onReleased)
return failure
}
@@ -10,22 +10,15 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
import type { AppState } from '../types'
import { createTestStore, makeUnifiedTab, makeTabGroup } from './store-test-helpers'
import type * as StructuredNativeChatToggle from './structured-native-chat-toggle'
vi.mock('sonner', () => ({ toast: { info: vi.fn(), success: vi.fn(), error: vi.fn() } }))
const { mockSetTerminalNativeChatMode } = vi.hoisted(() => ({
mockSetTerminalNativeChatMode: vi.fn()
}))
vi.mock('./structured-native-chat-toggle', async (importOriginal) => ({
...(await importOriginal<typeof StructuredNativeChatToggle>()),
setTerminalNativeChatMode: mockSetTerminalNativeChatMode
}))
const runtimeCall = vi.fn()
const mockApi = {
ui: { set: vi.fn().mockResolvedValue(undefined) },
settings: { get: vi.fn().mockResolvedValue({}), set: vi.fn().mockResolvedValue(undefined) }
settings: { get: vi.fn().mockResolvedValue({}), set: vi.fn().mockResolvedValue(undefined) },
runtime: { call: runtimeCall }
}
// @ts-expect-error -- partial window stub is sufficient for these store-only tests
@@ -37,13 +30,53 @@ describe('structured chat route out of setTabViewMode', () => {
let store: ReturnType<typeof createTestStore>
beforeEach(() => {
mockSetTerminalNativeChatMode.mockReset()
mockSetTerminalNativeChatMode.mockResolvedValue('structured')
let handoffStarted = false
runtimeCall.mockReset()
runtimeCall.mockImplementation(async ({ method }: { method: string }) => {
if (method === 'agentSession.adoptTerminal') {
return { ok: true, result: { ok: true, fence: 1 } }
}
if (method === 'agentSession.handoff') {
handoffStarted = true
return { ok: true, result: { ok: true } }
}
if (method === 'agentSession.handoffStatus') {
return {
ok: true,
result: handoffStarted
? { owner: 'native', direction: null, phase: 'idle' }
: { owner: 'tui', direction: null, phase: 'idle' }
}
}
throw new Error(`Unexpected runtime method: ${method}`)
})
store = createTestStore()
store.setState({
activeRepoId: 'repo1',
activeWorktreeId: WT,
projects: [{ id: 'repo1' }],
repos: [{ id: 'repo1', path: '/tmp/repo', connectionId: null }],
worktreesByRepo: {
repo1: [{ id: WT, repoId: 'repo1', path: '/tmp/feature', hostId: 'local' }]
},
detectedWorktreesByRepo: {},
settings: {},
unifiedTabsByWorktree: {
[WT]: [makeUnifiedTab({ id: 'codex-tab', worktreeId: WT, groupId: 'g-1' })]
},
tabsByWorktree: { [WT]: [{ id: 'codex-tab', launchAgent: 'codex' }] },
terminalLayoutsByTabId: {
'codex-tab': {
activeLeafId: 'leaf-adopt',
ptyIdsByLeafId: { 'leaf-adopt': 'pty-adopt' }
}
},
agentStatusByPaneKey: {
'codex-tab:leaf-adopt': {
agentType: 'codex',
providerSession: { id: 'thread-adopt' }
}
},
groupsByWorktree: {
[WT]: [
makeTabGroup({
@@ -54,22 +87,26 @@ describe('structured chat route out of setTabViewMode', () => {
})
]
}
} as Partial<AppState>)
} as unknown as Partial<AppState>)
})
it('hands an explicit chat mode to the structured adoption path', () => {
it('hands an explicit chat mode to the structured adoption path', async () => {
store.getState().setTabViewMode('codex-tab', 'chat')
expect(mockSetTerminalNativeChatMode).toHaveBeenCalledWith(
expect.objectContaining({ tabId: 'codex-tab', mode: 'chat' })
await vi.waitFor(() =>
expect(runtimeCall).toHaveBeenCalledWith(
expect.objectContaining({ method: 'agentSession.adoptTerminal' })
)
)
})
it('hands a toggled chat mode to the same path', () => {
it('hands a toggled chat mode to the same path', async () => {
store.getState().toggleTabViewMode('codex-tab')
expect(mockSetTerminalNativeChatMode).toHaveBeenCalledWith(
expect.objectContaining({ tabId: 'codex-tab', mode: 'chat' })
await vi.waitFor(() =>
expect(runtimeCall).toHaveBeenCalledWith(
expect.objectContaining({ method: 'agentSession.adoptTerminal' })
)
)
})
})