Skip to content
Merged
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
14 changes: 7 additions & 7 deletions src/main/codex/codex-structured-dispatch-echo.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -133,25 +133,25 @@ describe('the turn Codex answered a send into but has not opened', () => {
it('is the turn the latest armed send was answered into', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('client-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start')
echoes.arm('client-2')
echoes.bindTurn('client-2', 'thread-1', 'turn-2')
echoes.bindTurn('client-2', 'thread-1', 'turn-2', 'start')

expect(echoes.answeredUnopenedTurn('thread-1', NONE_OPEN)).toBe('turn-2')
})

it('is none once Codex opened that turn', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('client-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start')

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', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('client-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-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([])

Expand All @@ -161,9 +161,9 @@ describe('the turn Codex answered a send into but has not opened', () => {
it('skips a turn a wait left unopened, and still names an earlier one', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('client-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start')
echoes.arm('client-2')
echoes.bindTurn('client-2', 'thread-1', 'turn-2')
echoes.bindTurn('client-2', 'thread-1', 'turn-2', 'start')

echoes.leftUnopened('thread-1', 'turn-2')

Expand All @@ -174,7 +174,7 @@ describe('the turn Codex answered a send into but has not opened', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('client-1')
echoes.arm('client-2')
echoes.bindTurn('client-2', 'thread-2', 'turn-2')
echoes.bindTurn('client-2', 'thread-2', 'turn-2', 'start')

expect(echoes.answeredUnopenedTurn('thread-1', NONE_OPEN)).toBeNull()
})
Expand Down
36 changes: 26 additions & 10 deletions src/main/codex/codex-structured-dispatch-echo.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
import type { ProviderDiagnostic } from '../../shared/agent-session-failure'
import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types'
import type {
AgentJournalItemIdentity,
AgentJournalTurnJoin
} from '../../shared/agent-session-journal-types'

/** Sends awaiting their echo. One bound to a turn that ended without taking it settles from that
* end; any other whose echo never arrives is retired by the journal's recovery on exit. */
Expand Down Expand Up @@ -35,10 +38,15 @@ export type CodexDispatchEchoes = {
/** Drops an armed send whose write never reached the provider. */
disarm: (clientMessageId: string) => void
/**
* Binds a send to the turn Codex answered it into. Returns that turn's end when the answer is
* read after it; a send that end settles is no longer armed.
* Binds a send to the turn Codex answered it into, and how it joined that turn. Returns that
* turn's end when the answer is read after it; a send that end settles is no longer armed.
*/
bindTurn: (clientMessageId: string, threadId: string, turnId: string) => CodexTurnEnd | null
bindTurn: (
clientMessageId: string,
threadId: string,
turnId: string,
via: AgentJournalTurnJoin
) => CodexTurnEnd | null
/** The turn the latest armed send was answered into that is neither in `openTurnIds`, ended, nor
* left unopened through a wait: one Codex has picked for the send but not opened. */
answeredUnopenedTurn: (threadId: string, openTurnIds: ReadonlySet<string>) => string | null
Expand All @@ -52,7 +60,11 @@ export type CodexDispatchEchoes = {
* completed, which echoes its pending input first, so one it never echoed waits for recovery. An
* interrupt withdraws an un-echoed send, steered or the turn's own input: neither reached history.
*/
endTurn: (threadId: string, turnId: string, end: CodexTurnEnd) => string[]
endTurn: (
threadId: string,
turnId: string,
end: CodexTurnEnd
) => { clientMessageId: string; via: AgentJournalTurnJoin }[]
/** Submission origin for this exact send, retained until its echo settles it. */
requestOrigin: (clientMessageId: string) => CodexDispatchRequestOrigin | null
/** Highest causal sequence assigned to a dispatch in this session. */
Expand All @@ -64,7 +76,11 @@ export type CodexDispatchEchoes = {
export function createCodexDispatchEchoes(): CodexDispatchEchoes {
const armed = new Map<
string,
{ requestedAt: number | null; sequence: number; turn?: { threadId: string; turnId: string } }
{
requestedAt: number | null
sequence: number
turn?: { threadId: string; turnId: string; via: AgentJournalTurnJoin }
}
>()
const endedTurns = new Map<string, CodexTurnEnd>()
const unopenedTurns = new Set<string>()
Expand Down Expand Up @@ -101,12 +117,12 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes {
},
settle: (clientMessageId) => armed.delete(clientMessageId),
disarm: (clientMessageId) => void armed.delete(clientMessageId),
bindTurn: (clientMessageId, threadId, turnId) => {
bindTurn: (clientMessageId, threadId, turnId, via) => {
const entry = armed.get(clientMessageId)
if (!entry) {
return null
}
entry.turn = { threadId, turnId }
entry.turn = { threadId, turnId, via }
const end = endedTurns.get(turnKey(threadId, turnId)) ?? null
if (end && settles(end)) {
armed.delete(clientMessageId)
Expand Down Expand Up @@ -141,10 +157,10 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes {
}
const settled = [...armed].flatMap(([clientMessageId, entry]) =>
entry.turn && turnKey(entry.turn.threadId, entry.turn.turnId) === turn
? [clientMessageId]
? [{ clientMessageId, via: entry.turn.via }]
: []
)
for (const clientMessageId of settled) {
for (const { clientMessageId } of settled) {
armed.delete(clientMessageId)
}
return settled
Expand Down
2 changes: 1 addition & 1 deletion src/main/codex/codex-structured-session-adapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap
}
if (event.type === 'notification') {
// Only an admitted turn end settles; a refused one settles on the retry that lands.
settleCodexSendsInEndedTurn(session, event.method, event.params, (settlement) =>
settleCodexSendsInEndedTurn(session, event, (settlement) =>
this.deps.onDispatchSettledLate?.({ sessionId: event.sessionId, ...settlement })
)
// After the admission check, so a refused frame is observed by the child records
Expand Down
6 changes: 5 additions & 1 deletion src/main/codex/codex-structured-session-state.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger'
import type {
AgentJournalAnsweredTurnIdentity,
AgentJournalItemIdentity,
AgentSessionJournalIdentity
} from '../../shared/agent-session-journal-types'
Expand Down Expand Up @@ -87,7 +88,10 @@ export type CodexStructuredSessionAdapterDeps = {
onDispatchSettledLate?: (
input: { sessionId: string; clientMessageId: string } & (
| { providerIdentity: AgentJournalItemIdentity }
| ({ state: 'rejected' } & AgentJournalDispatchRejection)
| ({
state: 'rejected'
answeredInTurn: AgentJournalAnsweredTurnIdentity
} & AgentJournalDispatchRejection)
)
) => void
/** Codex reported its thread not running with no turn open: a send whose
Expand Down
106 changes: 98 additions & 8 deletions src/main/codex/codex-structured-turn-end-settlement.test.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
import { describe, expect, it, vi } from 'vitest'
import type { AgentJournalItemBody } from '../../shared/agent-session-journal-types'
import type {
AgentJournalItemBody,
AgentJournalItemIdentity
} from '../../shared/agent-session-journal-types'
import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key'
import { classifyDispatchRejection } from '../../shared/structured-agent-session-dispatch-rejection'
import { createCodexDispatchEchoes } from './codex-structured-dispatch-echo'
import {
Expand All @@ -20,11 +24,17 @@ async function turnEndRig() {
Object.assign(codex.routes, turns.routes)
const settlements: LateSettlement[] = []
const bodies: AgentJournalItemBody[] = []
const turnRecords: AgentJournalItemIdentity[] = []
const adapter = await acquiredCodexAdapter({
codex,
settlements,
sink: {
appendItem: (_identity, body) => bodies.push(body),
appendItem: (identity, body) => {
bodies.push(body)
if (body.kind === 'turn') {
turnRecords.push(identity)
}
},
appendTombstone: () => {},
publish: () => {}
}
Expand All @@ -46,7 +56,18 @@ async function turnEndRig() {
const settledIds = () => settlements.map(({ clientMessageId }) => clientMessageId)
const categoryOf = (settlement: LateSettlement | undefined) =>
settlement && 'state' in settlement ? classifyDispatchRejection(settlement).category : null
return { codex, turns, adapter, send, sendAndOpen, settlements, settledIds, categoryOf, bodies }
return {
codex,
turns,
adapter,
send,
sendAndOpen,
settlements,
settledIds,
categoryOf,
bodies,
turnRecords
}
}

describe('a Codex send its turn ended without echoing', () => {
Expand Down Expand Up @@ -286,34 +307,103 @@ describe('a Codex send its turn ended without echoing', () => {
})
})

describe('the turn a withdrawn Codex send names', () => {
it('is the record of the turn it opened, when that turn is interrupted', async () => {
const rig = await turnEndRig()
await rig.sendAndOpen('client-1')

rig.turns.end('interrupted')

expect(rig.turnRecords.length).toBeGreaterThan(0)
expect(new Set(rig.turnRecords.map(agentJournalItemKey)).size).toBe(1)
expect(rig.settlements).toEqual([
expect.objectContaining({
clientMessageId: 'client-1',
answeredInTurn: { turn: rig.turnRecords[0], via: 'start' }
})
])
})

it('is the running turn for a send steered into it', async () => {
const rig = await turnEndRig()
await rig.sendAndOpen('client-1')
await rig.send('client-2')

rig.turns.end('interrupted')

expect(rig.settlements.map((settlement) => [settlement.clientMessageId, settlement])).toEqual([
[
'client-1',
expect.objectContaining({ answeredInTurn: { turn: rig.turnRecords[0], via: 'start' } })
],
[
'client-2',
expect.objectContaining({ answeredInTurn: { turn: rig.turnRecords[0], via: 'steer' } })
]
])
})

it('is the ended turn whose end was read before the answer', 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('interrupted')
release()

await expect(sending).resolves.toMatchObject({
state: 'rejected',
answeredInTurn: { turn: rig.turnRecords[0], via: 'start' }
})
})

it('is not named when its turn completes without echoing it: the send stays pending', 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' })
await rig.sendAndOpen('client-2')
rig.turns.end('completed')
expect(rig.settlements).toEqual([])
})
})

describe('a send bound to a turn', () => {
it('dies with the settlement its turn end makes', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('client-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start')

expect(echoes.endTurn('thread-1', 'turn-1', { status: 'interrupted' })).toEqual(['client-1'])
expect(echoes.endTurn('thread-1', 'turn-1', { status: 'interrupted' })).toEqual([
{ clientMessageId: 'client-1', via: 'start' }
])
expect(echoes.size).toBe(0)
expect(echoes.settle('client-1')).toBe(false)
})

it('dies with its child, which forgets recorded turn ends too', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('client-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start')
echoes.endTurn('thread-2', 'turn-2', { status: 'interrupted' })

echoes.clear()

expect(echoes.size).toBe(0)
echoes.arm('client-2')
expect(echoes.bindTurn('client-2', 'thread-2', 'turn-2')).toBeNull()
expect(echoes.bindTurn('client-2', 'thread-2', 'turn-2', 'start')).toBeNull()
})

it('is matched by thread as well as turn id', () => {
const echoes = createCodexDispatchEchoes()
echoes.arm('client-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1')
echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start')

expect(echoes.endTurn('thread-2', 'turn-1', { status: 'interrupted' })).toEqual([])
expect(echoes.size).toBe(1)
Expand Down
26 changes: 19 additions & 7 deletions src/main/codex/codex-structured-turn-end-settlement.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,9 @@ import {
agentSessionFailureWords,
type AgentJournalDispatchRejection
} from '../../shared/agent-session-failure-words'
import type { AgentJournalAnsweredTurnIdentity } from '../../shared/agent-session-journal-types'
import type { CodexTurnEnd } from './codex-structured-dispatch-echo'
import { codexTurnLifecycleIdentity } from './codex-structured-journal-translation-turns'
import type { CodexSession } from './codex-structured-session-state'
import {
readCodexThreadId,
Expand All @@ -41,6 +43,8 @@ export function codexDispatchRejection(
export type CodexTurnEndSettlement = {
clientMessageId: string
state: 'rejected'
/** The turn Codex answered the send into, whose end settled it, and how the send joined it. */
answeredInTurn: AgentJournalAnsweredTurnIdentity
} & AgentJournalDispatchRejection

function errorDetail(params: unknown): ProviderDiagnostic | undefined {
Expand Down Expand Up @@ -80,19 +84,27 @@ export function codexTurnEndRejection(end: CodexTurnEnd): AgentJournalDispatchRe
/** Settles the sends bound to the turn this admitted notification ended. */
export function settleCodexSendsInEndedTurn(
session: Pick<CodexSession, 'threadId' | 'dispatchEchoes'>,
method: string,
params: unknown,
frame: { sessionId: string; method: string; params: unknown },
settle: (settlement: CodexTurnEndSettlement) => void
): void {
const turnId = readCodexTurnId(params)
const end = readCodexTurnEnd(method, params)
if (!turnId || !end || (readCodexThreadId(params) ?? session.threadId) !== session.threadId) {
const turnId = readCodexTurnId(frame.params)
const end = readCodexTurnEnd(frame.method, frame.params)
if (
!turnId ||
!end ||
(readCodexThreadId(frame.params) ?? session.threadId) !== session.threadId
) {
return
}
const rejection = codexTurnEndRejection(end)
for (const clientMessageId of session.dispatchEchoes.endTurn(session.threadId, turnId, end)) {
const turn = codexTurnLifecycleIdentity(frame.sessionId, turnId)
for (const { clientMessageId, via } of session.dispatchEchoes.endTurn(
session.threadId,
turnId,
end
)) {
if (rejection) {
settle({ clientMessageId, state: 'rejected', ...rejection })
settle({ clientMessageId, state: 'rejected', answeredInTurn: { turn, via }, ...rejection })
}
}
}
Loading
Loading