Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions src/main/codex/codex-structured-dispatch-admission.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -319,7 +319,7 @@ describe('codex dispatch admission', () => {
})
})

it('revises a completed turn when its exact echo arrives late', async () => {
it('attributes an accepted echo before completing its turn', async () => {
const codex = fakeCodexAppServer({
'turn/start': () => ({ turn: { id: 'turn-1', status: 'inProgress' } })
})
Expand All @@ -328,25 +328,25 @@ describe('codex dispatch admission', () => {
const adapter = await acquiredCodexAdapter({ codex, settlements, sink: recorded.sink })
const connection = codex.connections[0]!

const sending = send(adapter, 'client-late-echo', 1_700_000_000_100)
const sending = send(adapter, 'client-echo', 1_700_000_000_100)
await openAfterTurnStarts(connection, 1, () => startTurn(connection, 'turn-1'))
await sending
echoUserMessage(connection, {
turnId: 'turn-1',
itemId: 'item-echo',
clientId: 'client-echo'
})
connection.handlers.onNotification?.('turn/completed', {
threadId: CODEX_TEST_THREAD_ID,
turn: { id: 'turn-1' }
})
echoUserMessage(connection, {
turnId: 'turn-1',
itemId: 'item-late',
clientId: 'client-late-echo'
})

expect(recorded.bodies.findLast((body) => body.kind === 'turn')).toMatchObject({
state: 'completed',
requestedAt: 1_700_000_000_100,
userItemId: agentJournalSubmissionKey('client-late-echo')
userItemId: agentJournalSubmissionKey('client-echo')
})
expect(settlements.map(({ clientMessageId }) => clientMessageId)).toEqual(['client-late-echo'])
expect(settlements.map(({ clientMessageId }) => clientMessageId)).toEqual(['client-echo'])
})

it('refuses overflow without discarding an older accepted send', async () => {
Expand Down
5 changes: 2 additions & 3 deletions src/main/codex/codex-structured-dispatch-echo.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -61,12 +61,11 @@ describe('the turn Codex answered a send into but has not opened', () => {
expect(echoes.answeredUnopenedTurn('thread-1', new Set(['turn-1']))).toBeNull()
})

it('is none once that turn ended, even with its send still armed for an echo', () => {
it('is none once that turn ended', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('client-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start')
// A completed end leaves its unechoed send armed.
expect(echoes.endTurn('thread-1', 'turn-1', { status: 'completed' })).toEqual([])
echoes.endTurn('thread-1', 'turn-1', { status: 'completed' })

expect(echoes.answeredUnopenedTurn('thread-1', NONE_OPEN)).toBeNull()
})
Expand Down
34 changes: 26 additions & 8 deletions src/main/codex/codex-structured-dispatch-echo.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ export const MAX_CODEX_RECORDED_TURN_ENDS = 64

/** How a primary-thread turn ended, as Codex reported it. */
export type CodexTurnEnd =
| { status: 'completed' }
| { status: 'completed'; detail?: ProviderDiagnostic }
| { status: 'interrupted' }
| { status: 'failed'; detail?: ProviderDiagnostic; notSignedIn?: true }

Expand Down Expand Up @@ -56,12 +56,14 @@ export type CodexDispatchEchoes = {
opened: (threadId: string, turnId: string) => void
/** Whether Codex opened this turn: an `error` naming one it never opened is that turn's end. */
hasOpened: (threadId: string, turnId: string) => boolean
/** A blocked prompt hook's reason, held only until that turn ends; the first one stands. */
notePromptHookReason: (threadId: string, turnId: string, detail: ProviderDiagnostic) => void
takePromptHookReason: (threadId: string, turnId: string) => ProviderDiagnostic | undefined
/** The latest armed send's answered turn a wait left unopened, neither open nor ended: Codex may
* still open it, though no wait is spent on it again. */
answeredTurnLeftUnopened: (threadId: string, openTurnIds: ReadonlySet<string>) => string | null
/**
* Records a turn's end and returns the sends bound to it that it settles: all of them unless it
* completed, which echoes its pending input first, so one it never echoed waits for recovery. An
* Records a turn's end and returns its unechoed sends: a completed turn refused them. An
* interrupt withdraws an un-echoed send, steered or the turn's own input: neither reached history.
*/
endTurn: (
Expand All @@ -87,11 +89,11 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes {
}
>()
const endedTurns = new Map<string, CodexTurnEnd>()
const promptHookReasons = new Map<string, ProviderDiagnostic>()
const unopenedTurns = new Set<string>()
const openedTurns = new Set<string>()
let nextSequence = 0
const turnKey = (threadId: string, turnId: string): string => JSON.stringify([threadId, turnId])
const settles = (end: CodexTurnEnd): boolean => end.status !== 'completed'
const leftUnopened = (threadId: string, turnId: string): void => {
unopenedTurns.add(turnKey(threadId, turnId))
for (const oldest of unopenedTurns) {
Expand Down Expand Up @@ -138,7 +140,7 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes {
}
entry.turn = { threadId, turnId, via }
const end = endedTurns.get(turnKey(threadId, turnId)) ?? null
if (end && settles(end)) {
if (end) {
armed.delete(clientMessageId)
}
return end
Expand All @@ -158,6 +160,24 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes {
}
},
hasOpened: (threadId, turnId) => openedTurns.has(turnKey(threadId, turnId)),
notePromptHookReason: (threadId, turnId, detail) => {
const key = turnKey(threadId, turnId)
if (!endedTurns.has(key) && !promptHookReasons.has(key)) {
promptHookReasons.set(key, detail)
}
for (const oldest of promptHookReasons.keys()) {
if (promptHookReasons.size <= MAX_CODEX_RECORDED_TURN_ENDS) {
break
}
promptHookReasons.delete(oldest)
}
},
takePromptHookReason: (threadId, turnId) => {
const key = turnKey(threadId, turnId)
const detail = promptHookReasons.get(key)
promptHookReasons.delete(key)
return detail
},
endTurn: (threadId, turnId, end) => {
const turn = turnKey(threadId, turnId)
endedTurns.delete(turn)
Expand All @@ -168,9 +188,6 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes {
}
endedTurns.delete(oldest)
}
if (!settles(end)) {
return []
}
const settled = [...armed].flatMap(([clientMessageId, entry]) =>
entry.turn && turnKey(entry.turn.threadId, entry.turn.turnId) === turn
? [{ clientMessageId, via: entry.turn.via }]
Expand All @@ -191,6 +208,7 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes {
clear: () => {
armed.clear()
endedTurns.clear()
promptHookReasons.clear()
unopenedTurns.clear()
openedTurns.clear()
nextSequence = 0
Expand Down
49 changes: 21 additions & 28 deletions src/main/codex/codex-structured-turn-end-settlement.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -136,20 +136,26 @@ describe('a Codex send its turn ended without echoing', () => {
])
})

it('stays pending, still armed, when its turn completes without echoing it', async () => {
it("names no hook's reason when the blocked turn refuses more than one send", async () => {
const rig = await turnEndRig()
await rig.sendAndOpen('client-1')
await rig.send('client-2')

rig.codex.connections[0]!.handlers.onNotification?.('hook/completed', {
threadId: CODEX_TEST_THREAD_ID,
turnId: 'turn-1',
run: {
eventName: 'userPromptSubmit',
status: 'blocked',
entries: [{ kind: 'feedback', text: 'Prompt blocked by policy.' }]
}
})
rig.turns.end('completed')
expect(rig.settlements).toEqual([])

// Codex echoes before a completed end; this late one only proves the send is still armed.
rig.turns.echo('client-1')
const generic = { rejection: { kind: 'providerRejected' } }
expect(rig.settlements).toEqual([
expect.objectContaining({
clientMessageId: 'client-1',
providerIdentity: expect.objectContaining({ turnId: 'turn-1' })
})
expect.objectContaining({ clientMessageId: 'client-1', ...generic }),
expect.objectContaining({ clientMessageId: 'client-2', ...generic })
])
})

Expand Down Expand Up @@ -253,22 +259,6 @@ describe('a Codex send its turn ended without echoing', () => {
expect(rig.settlements).toEqual([])
})

it('leaves a send pending, still armed, when its answer is read after its turn completed', async () => {
const rig = await turnEndRig()
const release = rig.turns.holdNextAnswer()
const sending = rig.send('client-1')
await vi.waitFor(() => expect(rig.turns.turnId).toBe('turn-1'))
rig.turns.start()
rig.turns.end('completed')
release()

await expect(sending).resolves.toEqual({ state: 'admitted' })
expect(rig.settlements).toEqual([])
rig.turns.echo('client-1')
expect(rig.settledIds()).toEqual(['client-1'])
expect(rig.settlements[0]).toHaveProperty('providerIdentity')
})

it('leaves a send whose answer timed out to its echo', async () => {
const rig = await turnEndRig()
rig.codex.routes['turn/start'] = () => {
Expand Down Expand Up @@ -358,7 +348,7 @@ describe('the turn a withdrawn Codex send names', () => {
})
})

it('is not named when its turn completes without echoing it: the send stays pending', async () => {
it('is refused in its completed turn even when the answer arrives after that end', async () => {
const rig = await turnEndRig()
const release = rig.turns.holdNextAnswer()
const sending = rig.send('client-1')
Expand All @@ -367,9 +357,12 @@ describe('the turn a withdrawn Codex send names', () => {
rig.turns.end('completed')
release()

await expect(sending).resolves.toEqual({ state: 'admitted' })
await rig.sendAndOpen('client-2')
rig.turns.end('completed')
await expect(sending).resolves.toMatchObject({
state: 'rejected',
rejection: { kind: 'providerRejected' },
answeredInTurn: { turn: rig.turnRecords[0], via: 'start' }
})
rig.turns.echo('client-1')
expect(rig.settlements).toEqual([])
})
})
Expand Down
98 changes: 67 additions & 31 deletions src/main/codex/codex-structured-turn-end-settlement.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,10 @@
// steered send waits in Codex's pending input, which the interrupt clears; a send that opened the
// turn is saved as the turn cancels, but Codex hard-aborts that save after 100 ms, so it is lost
// or reaches only the model's context, and the thread history shows no user message either way.
// So it is withdrawn, as a Stop's host-side withdrawal is. Any other end records pending
// input before `turn/completed`, a failed turn after its `error` frame, so only that
// frame settles: a failed turn that never echoed the send refused it, in Codex's words,
// and a completed one leaves it pending for the journal's recovery on exit. A turn Codex never
// opened has no `turn/completed`: before 0.148 its final `error` is its end (`unopenedTurnFailure`).
// So it is withdrawn, as a Stop's host-side withdrawal is. Codex records accepted input and emits
// its receipt before `turn/completed`, a failed turn's after its `error` (0.136 through current),
// so any other end refuses a send it never echoed, including one a prompt hook blocked.
// A turn Codex never opened has no `turn/completed`: before 0.148 its final `error` is its end.

import {
agentSessionFailureFact,
Expand Down Expand Up @@ -106,11 +105,11 @@ export function readCodexTurnEnd(method: string, params: unknown): CodexTurnEnd
return { status: 'completed' }
}

/** How an ended turn settles a send it never echoed; null leaves the send to its echo. */
/** How an ended turn settles a send it never echoed. */
export function codexTurnEndRejection(
end: CodexTurnEnd,
account?: CodexSession['account']
): AgentJournalDispatchRejection | null {
): AgentJournalDispatchRejection {
if (end.status === 'interrupted') {
return agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' })
}
Expand All @@ -122,7 +121,40 @@ export function codexTurnEndRejection(
})
)
}
return null
return codexDispatchRejection(
agentSessionFailureFact('providerRejected', end.detail ? { detail: end.detail } : {}),
account
)
}

/** A blocked prompt hook's reason for its turn: its verdict names no message. */
function notePromptHookReason(
session: Pick<CodexSession, 'threadId' | 'dispatchEchoes'>,
params: unknown,
turnId: string
): void {
const run = readCodexJournalRecord(readCodexJournalRecord(params).run)
if (
run.eventName !== 'userPromptSubmit' ||
(run.status !== 'blocked' && run.status !== 'stopped')
) {
return
}
const entries = Array.isArray(run.entries) ? run.entries.map(readCodexJournalRecord) : []
const entryText = (...kinds: string[]): unknown =>
entries.find(
(entry) =>
typeof entry.kind === 'string' &&
kinds.includes(entry.kind) &&
typeof entry.text === 'string' &&
entry.text.trim()
)?.text
const text =
entryText('feedback', 'stop') ?? entryText('error') ?? entryText('warning') ?? run.statusMessage
const detail = typeof text === 'string' ? providerDiagnostic(text, 'person') : undefined
if (detail) {
session.dispatchEchoes.notePromptHookReason(session.threadId, turnId, detail)
}
}

/** Records a turn Codex opened, ahead of the frames that may report the thread idle before it ends. */
Expand All @@ -144,31 +176,35 @@ export function settleCodexSendsInEndedTurn(
settle: (settlement: CodexTurnEndSettlement) => void
): void {
const turnId = readCodexTurnId(frame.params)
const end = turnId
? (readCodexTurnEnd(frame.method, frame.params) ??
unopenedTurnFailure(session, frame.method, frame.params, turnId))
: null
if (
!turnId ||
!end ||
(readCodexThreadId(frame.params) ?? session.threadId) !== session.threadId
) {
if (!turnId || (readCodexThreadId(frame.params) ?? session.threadId) !== session.threadId) {
return
}
const rejection = codexTurnEndRejection(end, session.account)
for (const { clientMessageId, via } of session.dispatchEchoes.endTurn(
session.threadId,
turnId,
end
)) {
if (rejection) {
settle({
clientMessageId,
state: 'rejected',
...codexAnsweredTurn(session, frame.sessionId, turnId, via),
...rejection
})
}
if (frame.method === 'hook/completed') {
notePromptHookReason(session, frame.params, turnId)
return
}
const end =
readCodexTurnEnd(frame.method, frame.params) ??
unopenedTurnFailure(session, frame.method, frame.params, turnId)
if (!end) {
return
}
const hookReason = session.dispatchEchoes.takePromptHookReason(session.threadId, turnId)
const refused = session.dispatchEchoes.endTurn(session.threadId, turnId, end)
// The hook's reason is that send's only when the turn refuses exactly one.
const rejection = codexTurnEndRejection(
end.status === 'completed' && hookReason && refused.length === 1
? { ...end, detail: hookReason }
: end,
session.account
)
for (const { clientMessageId, via } of refused) {
settle({
clientMessageId,
state: 'rejected',
...codexAnsweredTurn(session, frame.sessionId, turnId, via),
...rejection
})
}
}

Expand Down
Loading
Loading