Files
orca/src/main/codex/codex-structured-session-close.ts
T
Merge Sim 698eb3a74d fix(codex): settle a structured send on admission, and stop minting a colliding identity
Two sends could be written into the journal under one durable identity.

Codex coalesces a mid-turn `turn/start` into the running turn rather than
refusing it -- measured against real `codex app-server` builds 0.147.0,
0.150.1 and 0.153.4, none of which refuse and none of which fire a second
`turn/started`. The dispatch path read the turn id from the turn/start
response and stamped every accepted send `ordinal: 0`. Since a coalesced
send gets the running turn's id back, two submissions persisted the same
`providerItemId`. That string is durable, and it is the key a restore uses
to match a submission against provider history, so the second message's real
history row matched nothing and rendered as an extra bubble on replay.

On 0.147.0 it is worse than a collision: the coalesced response returns a
turn id that never starts and never completes, so the persisted key named a
turn absent from history and NEITHER message could match.

Identity is now minted from the echoed user message at `identityFor` -- the
single point that mints the journal row's own identity -- so the settled key
is by construction the one replay computes, rather than a parallel
calculation that can drift.

Dispatch returns `admitted` when the transport write completes; identity
settles on the echo through a channel that did not previously exist for
Codex. Waiters are keyed by client message id instead of being shifted off
the front of an array by arrival order, and they are cleared on session
close and child exit -- previously a timeout was the only thing that ever
ended one.

`TURN_ID_WAIT_MS` is deleted. It was never reachable on any build measured:
`readCodexTurnId` returns non-null on all three, so the 10s wait never
fired. The comment justifying it claimed older builds acknowledge before the
id exists, which no tested build does.

Three comments asserting Codex answers a mid-turn send with `turn already
running` are corrected. Their only backing was a test fixture inventing that
error string. The correction is factual only -- every changed line in
`src/main/runtime/orchestration/` is a comment, and mid-turn delivery is
still refused for both providers. Whether that policy is right is a separate
question; it was resting on a false premise.

Known gap, stated rather than implied: this prevents new collisions and does
not repair journals already written with a colliding or phantom key. Those
conversations keep duplicating on restore. Repairing them means re-matching
persisted submissions against provider history and rewriting
`providerItemId` -- which is what `journal-submission-reconciler.ts` is
written for, and it still has no production caller.
2026-09-11 11:31:17 -07:00

144 lines
5.2 KiB
TypeScript

import type { CodexAppServerConnection } from './codex-app-server-connection-types'
import { closeProcessRegistry } from '../../shared/child-process/close-process-registry'
import {
cancelCodexAcquisitionAttempt,
type CodexAcquisitionRegistry,
type CodexSession,
type CodexStructuredSessionAdapterDeps,
type CodexStructuredSessionEvent
} from './codex-structured-session-state'
import type { StructuredAgentSessionLifecycleEvent } from '../native-chat/agent-session-wire/structured-agent-session-adapter'
export function handleCodexSessionExit(input: {
sessions: Map<string, CodexSession>
sessionId: string
connection: CodexAppServerConnection | null
error: Error
prompts?: CodexSession['prompts']
allowFailedSettlement?: boolean
onEvent?: (event: CodexStructuredSessionEvent) => void
onBackgroundTasksChanged?: CodexStructuredSessionAdapterDeps['onBackgroundTasksChanged']
}): boolean {
const session = input.sessions.get(input.sessionId)
if (!session || session.connection !== input.connection || session.ended) {
input.prompts?.clear()
return false
}
session.exitObservedAt ??= Date.now()
const event: StructuredAgentSessionLifecycleEvent = {
type: 'ended',
sessionId: input.sessionId,
reason: input.error.message,
cause: session.requestedClose ? 'requested-close' : 'unexpected-exit',
fence: session.fence,
acquisitionGeneration: session.acquisitionGeneration,
observedAt: session.exitObservedAt
} as const
// A synchronous sink rejection (usually backpressure) is handed to host
// recovery, which appends the bounded fallback before reacquisition.
const admission = session.translator?.handle(event) ?? { accepted: true }
if (!admission.accepted) {
// The connection invokes onExit exactly once. Forward a flagged event so
// host recovery can append its no-new-blob fallback even when admission is
// backpressured; waiting for a second callback would strand the lease.
if (event.cause !== 'unexpected-exit' && !input.allowFailedSettlement) {
return false
}
event.settlementRetryRequired = true
}
session.ended = true
// Nothing can echo for this child any more; the journal's pending-submission
// recovery is what settles the sends these were armed for.
session.dispatchEchoes.clear()
session.backgroundTasks.clear()
input.onBackgroundTasksChanged?.(input.sessionId, null)
session.unbindReadingControl?.()
input.onEvent?.(event)
session.prompts.clear()
session.translator?.dispose()
return true
}
export async function closeCodexPublishedSession(
sessions: Map<string, CodexSession>,
sessionId: string,
onEvent?: (event: CodexStructuredSessionEvent) => void,
options?: {
allowFailedSettlement?: boolean
requestedClose?: boolean
expectedFence?: number
expectedAcquisitionGeneration?: string
unexpectedReason?: Error
}
): Promise<boolean> {
const session = sessions.get(sessionId)
if (!session) {
return true
}
if (
(options?.expectedFence !== undefined && session.fence !== options.expectedFence) ||
(options?.expectedAcquisitionGeneration !== undefined &&
session.acquisitionGeneration !== options.expectedAcquisitionGeneration)
) {
return false
}
// Sink-failure recovery force-closes the child but must preserve the
// observed-exit cause so host lease settlement runs as an unexpected death.
session.requestedClose = options?.requestedClose ?? true
// Keep the session indexed until the child exit is observed. A timeout or
// failed kill must leave the live connection available for a safe retry.
const exited = await session.connection.close()
if (exited !== true) {
return false
}
if (!session.ended) {
const handled = handleCodexSessionExit({
sessions,
sessionId,
connection: session.connection,
error: options?.unexpectedReason ?? new Error('codex session closed'),
prompts: session.prompts,
...(options?.allowFailedSettlement ? { allowFailedSettlement: true } : {}),
...(onEvent ? { onEvent } : {})
})
// Keep the closed session indexed when terminal settlement admission was
// rejected; a later close attempt retries the same stable lifecycle event.
if (!handled) {
return false
}
}
sessions.delete(sessionId)
return true
}
export async function closeCodexSession(
sessionId: string,
sessions: Map<string, CodexSession>,
acquisitions: CodexAcquisitionRegistry,
onEvent?: (event: CodexStructuredSessionEvent) => void
): Promise<boolean> {
const attempt = acquisitions.get(sessionId)
if (!(await cancelCodexAcquisitionAttempt(attempt))) {
return false
}
if (attempt) {
acquisitions.deleteIfCurrent(sessionId, attempt)
}
return closeCodexPublishedSession(sessions, sessionId, onEvent)
}
export async function closeAllCodexSessions(
sessions: Map<string, CodexSession>,
acquisitions: CodexAcquisitionRegistry,
close: (sessionId: string) => Promise<boolean>
): Promise<void> {
acquisitions.close()
await closeProcessRegistry({
attempts: 3,
hasEntries: () => sessions.size > 0 || acquisitions.size > 0,
entryIds: () => new Set([...sessions.keys(), ...acquisitions.sessionIds()]),
closeEntry: close,
failureMessage: 'codex structured session shutdown could not prove every child stopped'
})
}