diff --git a/src/main/claude/claude-message-journaling.ts b/src/main/claude/claude-message-journaling.ts index 0ade48d58125..9d75db1fca9f 100644 --- a/src/main/claude/claude-message-journaling.ts +++ b/src/main/claude/claude-message-journaling.ts @@ -63,9 +63,7 @@ export function journalClaudeMessage( startsTurn: boolean, observedAt: number, /** Host clock on the submission that produced this send, when known. */ - requestedAt?: number, - /** The submission this send echo acknowledged. */ - openedBy?: string + requestedAt?: number ): boolean { const envelope = readClaudeMessageEnvelope(message) if (!envelope) { @@ -178,7 +176,6 @@ export function journalClaudeMessage( startsTurn, observedAt, ...(requestedAt === undefined ? {} : { requestedAt }), - ...(openedBy === undefined ? {} : { openedBy }), userItemId: agentJournalItemKey(identity) }) if (sendEchoTurn) { diff --git a/src/main/claude/claude-open-turn.ts b/src/main/claude/claude-open-turn.ts index ddc1bd553907..1cf0a8d048cc 100644 --- a/src/main/claude/claude-open-turn.ts +++ b/src/main/claude/claude-open-turn.ts @@ -60,11 +60,6 @@ export class ClaudeOpenTurn { } /** The open turn's row, where a fact about the running turn lands. */ - /** The submission that opened the open turn, when known (`ClaudeCurrentTurn.openedBy`). */ - get openedBy(): string | null { - return this.current?.openedBy ?? null - } - get identity(): AgentJournalItemIdentity | null { return this.current ? claudeCurrentTurnIdentity(this.current) : null } diff --git a/src/main/claude/claude-result-journaling.ts b/src/main/claude/claude-result-journaling.ts index a4226067b98e..1473886b9eac 100644 --- a/src/main/claude/claude-result-journaling.ts +++ b/src/main/claude/claude-result-journaling.ts @@ -59,9 +59,7 @@ export function journalClaudeResult( // Read before the settle below closes the turn: an error end the journal's Stop rule makes a // person's cancellation is theirs to decide as the end is written (`turnEndAfterStop`). const turnId = settlesTurn ? turn.id : null - const leftToStop = - turnId !== null && - sink.journalStopDecidesTurn?.(turnId, observedAt, turn.openedBy ?? undefined) === true + const leftToStop = turnId !== null && sink.journalStopDecidesTurn?.(turnId, observedAt) === true if (settlesTurn) { prompts.retryPendingCancellations() turn.suppressReopenOnFailure(message.is_error === true) diff --git a/src/main/claude/claude-structured-journal-translation.ts b/src/main/claude/claude-structured-journal-translation.ts index 3442dcc956b7..070798bc5216 100644 --- a/src/main/claude/claude-structured-journal-translation.ts +++ b/src/main/claude/claude-structured-journal-translation.ts @@ -179,10 +179,8 @@ export function createClaudeJournalTranslator( message: Record, startsTurn: boolean, observedAt: number, - requestedAt?: number, - openedBy?: string - ): boolean => - journalClaudeMessage(messageContext, message, startsTurn, observedAt, requestedAt, openedBy) + requestedAt?: number + ): boolean => journalClaudeMessage(messageContext, message, startsTurn, observedAt, requestedAt) return { handle: (event) => { @@ -250,8 +248,7 @@ export function createClaudeJournalTranslator( event.message, event.startsTurn === true, event.observedAt ?? Date.now(), - event.requestedAt, - event.clientMessageId + event.requestedAt ) ) { providerFallback.append( diff --git a/src/main/claude/claude-turn-lifecycle-item.ts b/src/main/claude/claude-turn-lifecycle-item.ts index 1d0db5fa21f4..e816be7f2b29 100644 --- a/src/main/claude/claude-turn-lifecycle-item.ts +++ b/src/main/claude/claude-turn-lifecycle-item.ts @@ -19,8 +19,6 @@ export type ClaudeCurrentTurn = { /** Provider key of the user echo, or the lifecycle row itself when provider * output opened a turn with no user row to receive its timing. */ userItemId: string - /** The submission whose send opened the turn, while its echo has yet to land in the journal. */ - openedBy?: string /** Present when the turn is the host's record of a conversation command. */ command?: ClaudeCommandTurn } diff --git a/src/main/claude/claude-turn-opening.ts b/src/main/claude/claude-turn-opening.ts index b5e13a39c71c..1406928b37ed 100644 --- a/src/main/claude/claude-turn-opening.ts +++ b/src/main/claude/claude-turn-opening.ts @@ -30,8 +30,6 @@ export type ClaudeSendEchoTurnInput = { requestedAt?: number /** Provider key of the user row this turn is anchored to. */ userItemId: string - /** The submission this echo acknowledged. */ - openedBy?: string } /** The turn a replayed send echo opens, or null when this frame is not one. */ @@ -48,8 +46,7 @@ export function claudeTurnOpenedBySendEcho( turnId: envelope.uuid, startedAt: input.observedAt, ...(input.requestedAt === undefined ? {} : { requestedAt: input.requestedAt }), - userItemId: input.userItemId, - ...(input.openedBy === undefined ? {} : { openedBy: input.openedBy }) + userItemId: input.userItemId } : null } diff --git a/src/main/codex/codex-structured-conversation-stop.test.ts b/src/main/codex/codex-structured-conversation-stop.test.ts index 62ecaf8f2ea1..1e548dd0909f 100644 --- a/src/main/codex/codex-structured-conversation-stop.test.ts +++ b/src/main/codex/codex-structured-conversation-stop.test.ts @@ -1,5 +1,6 @@ // A Codex Stop that names no turn. The fake keeps Codex 0.157's turn bookkeeping: it answers -// `turn/start` before it opens the turn, and refuses an interrupt until then. +// `turn/start` before it opens the turn, and refuses an interrupt with no turn active until the +// thread runs. import { describe, expect, it, vi } from 'vitest' import { structuredAgentSessionCommandTurn } from '../native-chat/agent-session-wire/structured-agent-session-command-turn' @@ -29,7 +30,7 @@ describe('a Codex Stop that names no turn', () => { const rig = await codexTurnLifecycleRig() await openedTurn(rig) - await expect(stop(rig, () => null)).resolves.toEqual({ cancelled: true }) + await expect(stop(rig, () => null)).resolves.toEqual({ cancelled: true, turnId: 'turn-1' }) expect(rig.interrupts().map((call) => call.params?.turnId)).toEqual(['turn-1']) }) diff --git a/src/main/codex/codex-structured-dispatch-echo.ts b/src/main/codex/codex-structured-dispatch-echo.ts index 775cac5bca4f..1eee3bc9a0a7 100644 --- a/src/main/codex/codex-structured-dispatch-echo.ts +++ b/src/main/codex/codex-structured-dispatch-echo.ts @@ -44,6 +44,9 @@ export type CodexDispatchEchoes = { answeredUnopenedTurn: (threadId: string, openTurnIds: ReadonlySet) => string | null /** Codex did not open this answered turn within a wait, so no later wait is spent on it. */ leftUnopened: (threadId: string, turnId: string) => void + /** 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 | 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 @@ -68,6 +71,19 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes { let nextSequence = 0 const turnKey = (threadId: string, turnId: string): string => JSON.stringify([threadId, turnId]) const settles = (end: CodexTurnEnd): boolean => end.status !== 'completed' + const answeredTurn = ( + threadId: string, + openTurnIds: ReadonlySet, + admits: (key: string) => boolean + ): string | null => { + const answered = [...armed.values()].flatMap(({ turn }) => { + const key = turn?.threadId === threadId ? turnKey(threadId, turn.turnId) : null + return turn && key && !openTurnIds.has(turn.turnId) && !endedTurns.has(key) && admits(key) + ? [turn.turnId] + : [] + }) + return answered.at(-1) ?? null + } return { arm(clientMessageId, requestedAt) { const existing = armed.get(clientMessageId) @@ -97,17 +113,10 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes { } return end }, - answeredUnopenedTurn: (threadId, openTurnIds) => { - const answered = [...armed.values()].flatMap(({ turn }) => - turn?.threadId === threadId && - !openTurnIds.has(turn.turnId) && - !endedTurns.has(turnKey(threadId, turn.turnId)) && - !unopenedTurns.has(turnKey(threadId, turn.turnId)) - ? [turn.turnId] - : [] - ) - return answered.at(-1) ?? null - }, + answeredUnopenedTurn: (threadId, openTurnIds) => + answeredTurn(threadId, openTurnIds, (key) => !unopenedTurns.has(key)), + answeredTurnLeftUnopened: (threadId, openTurnIds) => + answeredTurn(threadId, openTurnIds, (key) => unopenedTurns.has(key)), leftUnopened: (threadId, turnId) => { unopenedTurns.add(turnKey(threadId, turnId)) for (const oldest of unopenedTurns) { diff --git a/src/main/codex/codex-structured-prompt-ownership.test.ts b/src/main/codex/codex-structured-prompt-ownership.test.ts index ef1f8cc426f8..a0b2dc4146ed 100644 --- a/src/main/codex/codex-structured-prompt-ownership.test.ts +++ b/src/main/codex/codex-structured-prompt-ownership.test.ts @@ -509,7 +509,7 @@ describe('Codex live prompt ownership', () => { await expect( adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) - ).resolves.toEqual({ cancelled: true }) + ).resolves.toEqual({ cancelled: true, turnId: 'turn-1' }) expect(completionThreads(events)).toEqual([]) }) diff --git a/src/main/codex/codex-structured-prompt-ownership.ts b/src/main/codex/codex-structured-prompt-ownership.ts index effab9662bf5..fdef6e750f48 100644 --- a/src/main/codex/codex-structured-prompt-ownership.ts +++ b/src/main/codex/codex-structured-prompt-ownership.ts @@ -12,7 +12,7 @@ import { } from './codex-structured-prompt-replies' import { requireLiveCodexSession, type CodexSession } from './codex-structured-session-state' import { interruptCodexTurn } from './codex-structured-turn-cancellation' -import { codexRunningOrOpeningTurn } from './codex-structured-turn-open-wait' +import { codexStopTarget, interruptOpeningCodexTurn } from './codex-structured-turn-open-wait' type CancelInput = Parameters[0] type AnswerInput = Parameters[0] @@ -22,10 +22,25 @@ function providerTurnId(session: CodexSession, turnId: string): string | undefin return session.translator ? session.translator.commandProviderTurnId(turnId) : turnId } +/** A send whose answer this child lost: Codex may still open a turn for it that no Stop can name. + * A pending one Codex answered is not: its turn is the one the Stop interrupts. */ +function sendInDoubt(dispatchStatus: CancelInput['dispatchStatus']): boolean { + return dispatchStatus?.state === 'unknown' && !dispatchStatus.recovered +} + +/** The Stop's own answer, naming the journal turn it stopped when it took. */ +function stoppedTurn( + outcome: AgentSessionCancelOutcome, + turnId: string +): AgentSessionCancelOutcome { + return outcome.cancelled ? { ...outcome, turnId } : outcome +} + /** * A Stop that names no turn: interrupt the turn the journal shows, or else the one Codex reported * started and not yet ended, which the journal can trail by a publish, or else the one Codex - * answered a send into, once it opens. + * answered a send into, at once (`interruptOpeningCodexTurn`). A turn that may still open but + * cannot be reached, or a send whose answer was lost, answers refused, and the host ends the child. */ async function cancelCodexConversation( input: Parameters[0], @@ -34,20 +49,32 @@ async function cancelCodexConversation( const { request, sessions, requestTimeoutMs } = input const liveTurnId = request.resolveLiveTurnId?.() ?? null // A turn the journal shows that Codex has not started yet (a compaction's) has nothing to stop. - const turnId = - liveTurnId === null - ? await codexRunningOrOpeningTurn(session) - : providerTurnId(session, liveTurnId) - // The wait for the turn to open can outlive the session it began on. - if ( - !turnId || - sessions.get(request.sessionId) !== session || - session.ended || - session.fence !== request.fence - ) { + const target = liveTurnId === null ? codexStopTarget(session) : { turnId: liveTurnId } + // Read again after any wait for the turn to open, which can outlive the session it began on. + const current = (): boolean => + sessions.get(request.sessionId) === session && !session.ended && session.fence === request.fence + if (!current()) { + return { cancelled: false } + } + if (target === null) { + return sendInDoubt(request.dispatchStatus) + ? { cancelled: false, refusal: { turnMayOpen: true } } + : { cancelled: false } + } + const journalTurnId = 'opening' in target ? target.opening : target.turnId + const turnId = liveTurnId === null ? journalTurnId : providerTurnId(session, liveTurnId) + if (!turnId) { return { cancelled: false } } - return interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs }) + const interrupt = () => + interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs }) + if (!('opening' in target)) { + return stoppedTurn(await interrupt(), journalTurnId) + } + return stoppedTurn( + await interruptOpeningCodexTurn(session, turnId, interrupt, current), + journalTurnId + ) } export async function cancelCodexStructuredTurn(input: { @@ -67,7 +94,10 @@ export async function cancelCodexStructuredTurn(input: { return { cancelled: false } } if (!prompt) { - return interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs }) + return stoppedTurn( + await interruptCodexTurn({ session, threadId: session.threadId, turnId, requestTimeoutMs }), + requestedTurnId + ) } if (session.fence !== request.fence) { return { cancelled: false } diff --git a/src/main/codex/codex-structured-session-cancel.test.ts b/src/main/codex/codex-structured-session-cancel.test.ts index cd30bfa2e6d1..939d5faf0cdb 100644 --- a/src/main/codex/codex-structured-session-cancel.test.ts +++ b/src/main/codex/codex-structured-session-cancel.test.ts @@ -143,7 +143,7 @@ describe('CodexStructuredSessionAdapter.cancelTurn', () => { await expect( adapter.cancelTurn({ sessionId: 'session-1', turnId: 'turn-1', fence: 7 }) - ).resolves.toEqual({ cancelled: true }) + ).resolves.toEqual({ cancelled: true, turnId: 'turn-1' }) expect(codex.connections[0].calls.at(-1)).toEqual({ method: 'turn/interrupt', params: { threadId: THREAD_ID, turnId: 'turn-1' } @@ -288,7 +288,7 @@ describe('a Codex Stop is the interrupt alone', () => { params: expect.objectContaining({ turn: { id: 'turn-1', status: 'interrupted' } }) }) ) - await expect(stopped).resolves.toEqual({ cancelled: true }) + await expect(stopped).resolves.toEqual({ cancelled: true, turnId: 'turn-1' }) }) it('reads and kills no processes, from the send through the Stop', async () => { @@ -299,7 +299,7 @@ describe('a Codex Stop is the interrupt alone', () => { for (const work of Object.values(processWork)) { expect(work).not.toHaveBeenCalled() } - await expect(stopped).resolves.toEqual({ cancelled: true }) + await expect(stopped).resolves.toEqual({ cancelled: true, turnId: 'turn-1' }) for (const work of Object.values(processWork)) { expect(work).not.toHaveBeenCalled() } diff --git a/src/main/codex/codex-structured-turn-open-wait.test.ts b/src/main/codex/codex-structured-turn-open-wait.test.ts index c86a39fb59ef..d669c088d080 100644 --- a/src/main/codex/codex-structured-turn-open-wait.test.ts +++ b/src/main/codex/codex-structured-turn-open-wait.test.ts @@ -1,7 +1,8 @@ // A Codex Stop that names no turn, sent after Codex answered a cold send and before it opened that -// turn, waits for the turn to open, or provably not to, and never the send itself. The fake keeps -// Codex 0.157's turn bookkeeping: it answers before it opens the turn, and refuses an interrupt -// until then. +// turn, interrupts that turn at once. Refused as finding no turn while the turn has yet to run, it +// waits for the turn to open, or provably not to, and interrupts once more; it never waits for the +// send itself. The fake keeps Codex 0.157's turn bookkeeping: it answers before it opens the turn, +// and refuses an interrupt with no turn active until the thread runs. import { afterEach, describe, expect, it, vi } from 'vitest' import { @@ -15,6 +16,7 @@ type Rig = Awaited> const ADMITTED = { state: 'admitted' } const REFUSED = { cancelled: false } +const MAY_OPEN = { cancelled: false, refusal: { turnMayOpen: true } } const stop = (rig: Rig) => rig.adapter.cancelTurn({ sessionId: 'session-1', fence: 7 }) @@ -26,12 +28,23 @@ async function answeredColdSend(rig: Rig): Promise { expect(await settledWithin(sending)).toEqual(ADMITTED) } -/** A Stop sent in the window, which Codex would refuse if it reached Codex now. */ +const interruptedTurns = (rig: Rig) => rig.interrupts().map((call) => call.params?.turnId) + +/** A Stop under fake timers, read without awaiting it. */ +function pressed(rig: Rig): () => unknown { + let outcome: unknown = 'held' + void stop(rig).then((value) => { + outcome = value + }) + return () => outcome +} + +/** A Stop sent in the window before Codex's thread runs the turn: Codex refused its interrupt. */ async function waitingStop(rig: Rig) { await answeredColdSend(rig) const stopping = stop(rig) expect(await settledWithin(stopping)).toBe('held') - expect(rig.interrupts()).toEqual([]) + expect(interruptedTurns(rig)).toEqual(['turn-1']) // Wrapped: an async function returning the promise itself would wait for it. return { stopping } } @@ -51,15 +64,26 @@ describe('a Codex send answered before its turn opens', () => { }) describe("a no-turn Stop in the window between Codex's answer and its turn opening", () => { - it('waits for the turn to open, then stops it', async () => { + it("is taken at once once Codex's thread runs the turn", async () => { + const rig = await codexTurnLifecycleRig() + await answeredColdSend(rig) + rig.turns.run() + + expect(await settledWithin(stop(rig))).toEqual({ cancelled: true, turnId: 'turn-1' }) + expect(interruptedTurns(rig)).toEqual(['turn-1']) + await vi.waitFor(() => expect(rig.turns.turnId).toBeNull()) + }) + + it('refused before the thread runs, waits for the turn to open and interrupts once more', async () => { const rig = await codexTurnLifecycleRig() const { stopping } = await waitingStop(rig) rig.turns.start() - expect(await settledWithin(stopping)).toEqual({ cancelled: true }) - expect(rig.interrupts().map((call) => call.params?.turnId)).toEqual(['turn-1']) - expect(rig.turns.turnId).toBeNull() + expect(await settledWithin(stopping)).toEqual({ cancelled: true, turnId: 'turn-1' }) + expect(interruptedTurns(rig)).toEqual(['turn-1', 'turn-1']) + // Codex answers the interrupt, then ends the turn. + await vi.waitFor(() => expect(rig.turns.turnId).toBeNull()) }) it('does not wait when Codex opened the turn before its answer was read', async () => { @@ -71,7 +95,7 @@ describe("a no-turn Stop in the window between Codex's answer and its turn openi release() await sending - expect(await settledWithin(stop(rig))).toEqual({ cancelled: true }) + expect(await settledWithin(stop(rig))).toEqual({ cancelled: true, turnId: 'turn-1' }) }) it('does not wait for a send Codex steered into the running turn', async () => { @@ -80,7 +104,7 @@ describe("a no-turn Stop in the window between Codex's answer and its turn openi rig.turns.start() expect(await settledWithin(rig.send('client-2'))).toEqual(ADMITTED) - expect(await settledWithin(stop(rig))).toEqual({ cancelled: true }) + expect(await settledWithin(stop(rig))).toEqual({ cancelled: true, turnId: 'turn-1' }) expect(rig.interrupts().map((call) => call.params?.turnId)).toEqual(['turn-1']) }) @@ -91,19 +115,41 @@ describe("a no-turn Stop in the window between Codex's answer and its turn openi rig.turns.end('interrupted') expect(await settledWithin(stopping)).toEqual(REFUSED) - expect(rig.interrupts()).toEqual([]) + expect(interruptedTurns(rig)).toEqual(['turn-1']) + }) + + // Codex ended the turn as the interrupt reached it: it refuses the turn as its last ended one. + it('stops nothing, and waits for nothing, when Codex already ended the turn', async () => { + const rig = await codexTurnLifecycleRig() + await answeredColdSend(rig) + const interrupt = rig.codex.routes['turn/interrupt']! + rig.codex.routes['turn/interrupt'] = (params) => { + rig.turns.end('completed') + return interrupt(params) + } + + expect(await settledWithin(stop(rig))).toEqual(REFUSED) + expect(interruptedTurns(rig)).toEqual(['turn-1']) }) + // The turn neither opened nor ended, and its send is still owed: no press calls that nothing. it.each(['idle', 'systemError'])( - 'stops nothing when Codex reports the thread %s', + 'answers refused, on the first press and on a second alike, when Codex reports the thread %s', async (type) => { const rig = await codexTurnLifecycleRig() const { stopping } = await waitingStop(rig) rig.notify('thread/status/changed', { threadId: CODEX_TEST_THREAD_ID, status: { type } }) - expect(await settledWithin(stopping)).toEqual(REFUSED) - expect(rig.interrupts()).toEqual([]) + expect(await settledWithin(stopping)).toEqual(MAY_OPEN) + // The second press's interrupt is refused too; it waits its own bound, then agrees. + vi.useFakeTimers() + const second = pressed(rig) + await vi.advanceTimersByTimeAsync(CODEX_TURN_OPEN_WAIT_MS - 1) + expect(second()).toBe('held') + await vi.advanceTimersByTimeAsync(1) + expect(second()).toEqual(MAY_OPEN) + expect(interruptedTurns(rig)).toEqual(['turn-1', 'turn-1']) } ) @@ -115,7 +161,7 @@ describe("a no-turn Stop in the window between Codex's answer and its turn openi expect(await settledWithin(stopping)).toBe('held') rig.turns.start() - expect(await settledWithin(stopping)).toEqual({ cancelled: true }) + expect(await settledWithin(stopping)).toEqual({ cancelled: true, turnId: 'turn-1' }) }) it('stops nothing when the child exits', async () => { @@ -125,23 +171,68 @@ describe("a no-turn Stop in the window between Codex's answer and its turn openi rig.codex.connections[0]!.handlers.onExit?.(new Error('codex app-server exited')) expect(await settledWithin(stopping)).toEqual(REFUSED) - expect(rig.interrupts()).toEqual([]) + expect(interruptedTurns(rig)).toEqual(['turn-1']) }) - it('stops nothing once its bound runs out', async () => { + // The turn may still open and run: the host ends the child (`performCancel`). + it('answers refused once its bound runs out', async () => { const rig = await codexTurnLifecycleRig() await answeredColdSend(rig) vi.useFakeTimers() - let outcome: unknown = 'held' - void stop(rig).then((value) => { - outcome = value - }) + const first = pressed(rig) await vi.advanceTimersByTimeAsync(CODEX_TURN_OPEN_WAIT_MS - 1) - expect(outcome).toBe('held') + expect(first()).toBe('held') await vi.advanceTimersByTimeAsync(1) - expect(outcome).toEqual(REFUSED) + expect(first()).toEqual(MAY_OPEN) + expect(interruptedTurns(rig)).toEqual(['turn-1']) + }) + + // A press after an earlier wait gave up waits again, so a turn that opens then is still stopped. + it('waits again on a second press, and stops the turn if it opens then', async () => { + const rig = await codexTurnLifecycleRig() + await answeredColdSend(rig) + vi.useFakeTimers() + const first = pressed(rig) + await vi.advanceTimersByTimeAsync(CODEX_TURN_OPEN_WAIT_MS) + expect(first()).toEqual(MAY_OPEN) + + const second = pressed(rig) + await vi.advanceTimersByTimeAsync(CODEX_TURN_OPEN_WAIT_MS - 1) + expect(second()).toBe('held') + rig.turns.start() + await vi.advanceTimersByTimeAsync(0) + + expect(second()).toEqual({ cancelled: true, turnId: 'turn-1' }) + expect(interruptedTurns(rig)).toEqual(['turn-1', 'turn-1', 'turn-1']) + }) +}) + +describe('a no-turn Stop with no turn Codex answered', () => { + it('answers refused when the answer to the latest send was lost: Codex may still open its turn', async () => { + const rig = await codexTurnLifecycleRig() + + const outcome = await rig.adapter.cancelTurn({ + sessionId: 'session-1', + fence: 7, + dispatchStatus: { state: 'unknown', recovered: false } + }) + + expect(outcome).toEqual({ cancelled: false, refusal: { turnMayOpen: true } }) expect(rig.interrupts()).toEqual([]) }) + + it('stops nothing when nothing was sent, or the lost answer outlived its child', async () => { + const rig = await codexTurnLifecycleRig() + + expect(await stop(rig)).toEqual(REFUSED) + expect( + await rig.adapter.cancelTurn({ + sessionId: 'session-1', + fence: 7, + dispatchStatus: { state: 'unknown', recovered: true } + }) + ).toEqual(REFUSED) + }) }) diff --git a/src/main/codex/codex-structured-turn-open-wait.ts b/src/main/codex/codex-structured-turn-open-wait.ts index 9f30b292ff0c..9dd071656f35 100644 --- a/src/main/codex/codex-structured-turn-open-wait.ts +++ b/src/main/codex/codex-structured-turn-open-wait.ts @@ -1,6 +1,7 @@ // A Stop's or a send's wait for the turn Codex answered a send into to open, or provably not // to: it ended, the thread stopped running, or the child is gone. Held in memory only. +import type { AgentSessionCancelOutcome } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { CodexDispatchEchoes } from './codex-structured-dispatch-echo' import { codexThreadStoppedRunning, @@ -61,18 +62,88 @@ export function createCodexTurnOpenWaits(): CodexTurnOpenWaits { } } -/** - * The turn a Stop or a send names: the latest Codex reported started and not ended, or else the - * one Codex answered a send into, once it opens. Null when none is running and that one ends, the - * thread stops running or the child exits first, or the wait runs out; each such turn is waited - * for once. - */ -export async function codexRunningOrOpeningTurn(session: { +/** What a send finds to steer, or a Stop to interrupt: a turn running, or one Codex answered a send + * into that has not opened and has not ended, so may still open while its send is owed; null when + * neither. */ +export type CodexStopTarget = { turnId: string } | { opening: string } | null + +type CodexOpeningTurnSession = { threadId: string activeTurnIds?: ReadonlySet - dispatchEchoes: Pick + dispatchEchoes: Pick< + CodexDispatchEchoes, + 'answeredUnopenedTurn' | 'leftUnopened' | 'answeredTurnLeftUnopened' + > turnOpenWaits: Pick -}): Promise { +} + +/** The latest turn Codex reported started and not ended, or else the one it answered a send into + * that has neither opened nor ended, whether or not a wait already gave up on it. */ +export function codexStopTarget(session: CodexOpeningTurnSession): CodexStopTarget { + const running = [...(session.activeTurnIds ?? [])].at(-1) + if (running) { + return { turnId: running } + } + const open = session.activeTurnIds ?? new Set() + const opening = + session.dispatchEchoes.answeredUnopenedTurn(session.threadId, open) ?? + session.dispatchEchoes.answeredTurnLeftUnopened(session.threadId, open) + return opening ? { opening } : null +} + +/** Codex's -32600 for an interrupt that finds no turn: before the turn has started running, or + * once it has ended. */ +const CODEX_NO_ACTIVE_TURN_TO_INTERRUPT = 'no active turn to interrupt' + +/** + * A Stop of a turn Codex answered a send into that has not opened. Codex takes an interrupt as + * soon as the turn starts, which can be before its `turn/started` is read, so it goes out at once. + * Refused as finding no turn while that turn has yet to start, the Stop waits for it to open, + * bounded, and sends it once more. A turn that ended meanwhile was nothing to stop. One that never + * opens fails the Stop (`turnMayOpen`): it may still open and run, so the host ends the child. + */ +export async function interruptOpeningCodexTurn( + session: CodexOpeningTurnSession, + turnId: string, + interrupt: () => Promise, + /** Whether the session the Stop began on still runs it, read after the wait. */ + stillCurrent: () => boolean +): Promise { + const open = (): ReadonlySet => session.activeTurnIds ?? new Set() + const first = await interrupt() + if (first.cancelled || first.refusal?.detail?.text !== CODEX_NO_ACTIVE_TURN_TO_INTERRUPT) { + return first + } + const opened = (): boolean => open().has(turnId) + const stillOpening = (): boolean => { + const target = codexStopTarget(session) + return target !== null && 'opening' in target && target.opening === turnId + } + if (!opened() && stillOpening()) { + await session.turnOpenWaits.wait(turnId, CODEX_TURN_OPEN_WAIT_MS) + if (!stillCurrent()) { + return { cancelled: false } + } + } + if (opened()) { + return interrupt() + } + if (!stillOpening()) { + // It ended: the Stop found no turn running. + return { cancelled: false } + } + return { cancelled: false, refusal: { turnMayOpen: true } } +} + +/** + * The turn a send steers into: the running one, or else the one Codex answered a send into, once + * it opens. Before 0.148 Codex refuses a steer until it opens the turn, so the send waits for that, + * once per turn: a turn a wait left unopened is not waited for again. Null when no turn is running + * by then. + */ +export async function codexRunningOrOpeningTurn( + session: CodexOpeningTurnSession +): Promise { const running = [...(session.activeTurnIds ?? [])].at(-1) if (running) { return running @@ -84,7 +155,6 @@ export async function codexRunningOrOpeningTurn(session: { if (!answered) { return null } - // Codex refuses an interrupt, and before 0.148 a steer, until it opens the turn. await session.turnOpenWaits.wait(answered, CODEX_TURN_OPEN_WAIT_MS) if (session.activeTurnIds?.has(answered)) { return answered diff --git a/src/main/codex/codex-structured-turn-steer.test.ts b/src/main/codex/codex-structured-turn-steer.test.ts index c1b982709b30..15f95aa29275 100644 --- a/src/main/codex/codex-structured-turn-steer.test.ts +++ b/src/main/codex/codex-structured-turn-steer.test.ts @@ -223,12 +223,16 @@ describe('a Codex send made after Codex answered an earlier one, before it opene expect(await rig.sending).toEqual({ state: 'admitted' }) expect(rig.methods()).toEqual(['thread/start', 'turn/start', 'turn/steer']) expect(await rig.adapter.cancelTurn({ sessionId: 'session-1', fence: 7 })).toEqual({ - cancelled: true + cancelled: true, + turnId: 'turn-1' }) - expect(rig.settlements.map(({ clientMessageId }) => clientMessageId).sort()).toEqual([ - 'client-1', - 'client-2' - ]) + // Codex answers the interrupt, then ends the turn, which settles both. + await vi.waitFor(() => + expect(rig.settlements.map(({ clientMessageId }) => clientMessageId).sort()).toEqual([ + 'client-1', + 'client-2' + ]) + ) }) it('starts its own turn when that turn ends without opening', async () => { diff --git a/src/main/codex/codex-turn-lifecycle-fake.ts b/src/main/codex/codex-turn-lifecycle-fake.ts index a958faca1eaa..e0747f38b0b0 100644 --- a/src/main/codex/codex-turn-lifecycle-fake.ts +++ b/src/main/codex/codex-turn-lifecycle-fake.ts @@ -2,9 +2,13 @@ // it, for tests. From app-server `turn_processor.rs`: `turn/start` picks the turn before it // answers, a send while a turn is open is steered into it under the same id with no // second `turn/started`, `turn/steer` refuses with -32600 unless its `expectedTurnId` is -// the running turn, and `turn_interrupt_inner` refuses with -32600 until the -// turn has started. The answer can be held, so a test can deliver it after the -// turn's own frames, as the wire allows. `legacyStartAnswers` is a Codex before 0.148, +// the running turn, and `turn_interrupt_inner`, with no turn active, refuses with -32600 "no +// active turn to interrupt" until the thread runs (`run`, or `start`) and once the turn has ended. +// An interrupt Codex takes is answered before its turn ends: on `TurnAborted`, +// `bespoke_event_handling.rs` answers pending interrupts, then sends `turn/completed` +// (interrupted) on the same channel; here that end comes on a later read. The `turn/start` +// answer can be held, so a test can deliver it after the turn's own frames, as the wire +// allows. `legacyStartAnswers` is a Codex before 0.148, // whose `turn/start` answers a steered send with its submission id, a turn that never opens, // and whose `turn/steer` finds no turn to steer until the picked one has started. @@ -22,7 +26,13 @@ export type CodexTurnLifecycleFake = { holdNextAnswer: () => () => void /** Codex emits `turn/started` for the turn it picked last. */ start: () => void - /** Codex ends the picked or running turn on its own. */ + /** The thread runs the picked turn, whose `turn/started` has yet to be read: an interrupt of it + * is taken, and that `turn/started` comes ahead of the answer. */ + run: () => void + /** Codex takes an interrupt of the running turn: it drops it as its active turn before it + * answers, and that turn's `turn/completed` comes later (`end`). */ + takeInterrupt: () => string + /** Codex ends the picked or running turn on its own, or the one whose interrupt it took. */ end: (status: 'completed' | 'interrupted' | 'failed', errorMessage?: string) => void /** Codex fails the picked turn before starting it, which reports an `error` and no turn end. */ failUnopened: (message: string) => void @@ -51,14 +61,41 @@ export function codexTurnLifecycleFake( let active: string | null = null let lastTurn: string | null = null let held: Promise | null = null + let pickedRuns = false + // The turn whose interrupt Codex took and whose `turn/completed` is still to come. + let aborting: string | null = null const finish = (turnId: string, status: string, errorMessage?: string): void => { - picked = null - active = null + // A taken interrupt's turn is no longer active: a turn picked since stays picked. + if (aborting === turnId) { + aborting = null + } else { + picked = null + active = null + pickedRuns = false + } notify()('turn/completed', { threadId, turn: { id: turnId, status, ...(errorMessage ? { error: { message: errorMessage } } : {}) } }) } + + const takeInterrupt = (): string => { + if (!active) { + throw new Error('no running turn to interrupt') + } + aborting = active + active = null + picked = null + return aborting + } + const open = (): void => { + if (!picked) { + throw new Error('no picked turn to start') + } + active = picked + pickedRuns = false + notify()('turn/started', { threadId, turn: { id: active, status: 'inProgress' } }) + } return { routes: { 'turn/start': () => { @@ -88,6 +125,9 @@ export function codexTurnLifecycleFake( }, 'turn/interrupt': (params) => { const turnId = params?.turnId + if (!active && pickedRuns && picked === turnId) { + open() + } if (!active) { throw refusal('turn/interrupt', 'no active turn to interrupt') } @@ -97,8 +137,13 @@ export function codexTurnLifecycleFake( `expected active turn id ${String(turnId)} but found ${active}` ) } - // Codex answers the interrupt once the turn has aborted. - finish(active, 'interrupted') + const aborted = takeInterrupt() + // Answered, then the turn's end on the next read; a turn that ends meanwhile is not ended twice. + setTimeout(() => { + if (aborting === aborted) { + finish(aborted, 'interrupted') + } + }, 0) return {} } }, @@ -109,15 +154,17 @@ export function codexTurnLifecycleFake( }) return () => release() }, - start: () => { - if (!picked) { - throw new Error('no picked turn to start') + start: () => open(), + takeInterrupt, + run: () => { + if (!picked || active) { + throw new Error('no picked turn to run') } - active = picked - notify()('turn/started', { threadId, turn: { id: active, status: 'inProgress' } }) + pickedRuns = true }, end: (status, errorMessage) => { - const turnId = active ?? picked + // A taken interrupt's turn/completed comes ahead of anything a later turn sends. + const turnId = aborting ?? active ?? picked if (!turnId) { throw new Error('no turn to end') } diff --git a/src/main/native-chat/agent-session-journal/journal-stop-marks.ts b/src/main/native-chat/agent-session-journal/journal-stop-marks.ts index e79958a952c7..baba17c0058d 100644 --- a/src/main/native-chat/agent-session-journal/journal-stop-marks.ts +++ b/src/main/native-chat/agent-session-journal/journal-stop-marks.ts @@ -1,12 +1,15 @@ // What the journal answers about its Stops beyond the queue's pause: the latest Stop event, which -// the turn-end rule reads (`journal-stop-turn-end.ts`), and whether a person's still decides. +// the turn-end rule reads (`journal-stop-turn-end.ts`), whether a person's still decides, and what +// a person's Stop that named no turn binds while and after it settles. import type { JournalReducerState } from './journal-reducer' import { + beginJournalStopSettle, latestAcceptedSendUnopened, personStopDecidesTurn, type JournalLatestStop } from './journal-stop-turn-end' +import type { JournalStopSettle } from './queued-message-pause' export class JournalStopMarks { constructor(private readonly deps: { state: () => JournalReducerState }) {} @@ -21,7 +24,23 @@ export class JournalStopMarks { } /** `personStopDecidesTurn`: a person's Stop decides how turn `turnId` ends. */ - personStopDecides(turnId: string | null, endedAt?: number, openedBy?: string): boolean { - return personStopDecidesTurn(this.deps.state(), turnId, endedAt, openedBy) + personStopDecides(turnId: string | null, endedAt?: number): boolean { + return personStopDecidesTurn(this.deps.state(), turnId, endedAt) + } + + /** `beginJournalStopSettle`: a turn that ends from here until `settled` is the Stop's. */ + beginSettle(): JournalStopSettle | null { + return beginJournalStopSettle(this.deps.state()) + } + + /** Closes a settle `beginSettle` opened, binding `turnId` when the Stop stopped one. */ + settled(settle: JournalStopSettle | null, turnId?: string): void { + if (!settle) { + return + } + settle.settling = false + if (turnId !== undefined) { + settle.turnId = turnId + } } } diff --git a/src/main/native-chat/agent-session-journal/journal-stop-turn-end.test.ts b/src/main/native-chat/agent-session-journal/journal-stop-turn-end.test.ts new file mode 100644 index 000000000000..59ab1bbb990f --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-stop-turn-end.test.ts @@ -0,0 +1,90 @@ +// The turn-end rule as a Stop that named no turn binds it: every turn that ends while the Stop +// settles, then only the turn it stopped. + +import { describe, expect, it } from 'vitest' +import type { AgentJournalItemBody } from '../../../shared/agent-session-journal-types' +import { createJournalReducerState, type JournalReducerState } from './journal-reducer' +import type { JournalStopEvent } from './journal-row-schema' +import { + beginJournalStopSettle, + personStopDecidesTurn, + turnEndAfterStop +} from './journal-stop-turn-end' +import { JournalStopMarks } from './journal-stop-marks' + +const STOPPED_AT = 1_000 + +function stateWith(event?: JournalStopEvent): JournalReducerState { + const state = createJournalReducerState('session-1', 'epoch-1') + state.queuePauseMarks.latestStop = event ? { sequence: 5, event } : null + return state +} + +type TurnBody = Extract + +function ended(turnId: string): TurnBody { + return { kind: 'turn', turnId, state: 'interrupted', completedAt: STOPPED_AT + 10 } +} + +const turnless: JournalStopEvent = { reason: 'user-stop', at: STOPPED_AT } + +describe("a person's Stop that named no turn", () => { + it('makes any turn that ends while it settles theirs', () => { + const state = stateWith(turnless) + const marks = new JournalStopMarks({ state: () => state }) + marks.beginSettle() + + expect(turnEndAfterStop(state, 'turn-item', ended('turn-a'))).toMatchObject({ + outcome: 'cancellation' + }) + expect(personStopDecidesTurn(state, null)).toBe(true) + }) + + it('once settled, binds only the turn it stopped', () => { + const state = stateWith(turnless) + const marks = new JournalStopMarks({ state: () => state }) + marks.settled(marks.beginSettle(), 'turn-a') + + expect(turnEndAfterStop(state, 'turn-item', ended('turn-a'))).toMatchObject({ + outcome: 'cancellation' + }) + expect(turnEndAfterStop(state, 'other-item', ended('turn-b'))).not.toHaveProperty('outcome') + expect(personStopDecidesTurn(state, null)).toBe(false) + expect(personStopDecidesTurn(state, 'turn-a')).toBe(true) + }) + + it('binds nothing once it settled having stopped nothing', () => { + const state = stateWith(turnless) + const marks = new JournalStopMarks({ state: () => state }) + marks.settled(marks.beginSettle()) + + expect(turnEndAfterStop(state, 'turn-item', ended('turn-a'))).not.toHaveProperty('outcome') + }) + + it('binds nothing before it settles: a relaunch folds the event alone', () => { + const state = stateWith(turnless) + + expect(turnEndAfterStop(state, 'turn-item', ended('turn-a'))).not.toHaveProperty('outcome') + expect(personStopDecidesTurn(state, null)).toBe(false) + }) + + it('keeps the turn an earlier press bound when pressed again', () => { + const state = stateWith(turnless) + const marks = new JournalStopMarks({ state: () => state }) + marks.settled(marks.beginSettle(), 'turn-a') + marks.settled(marks.beginSettle()) + + expect(turnEndAfterStop(state, 'turn-item', ended('turn-a'))).toMatchObject({ + outcome: 'cancellation' + }) + }) +}) + +describe('a Stop with nothing in memory to bind', () => { + it.each([ + ['one that named its turn', { reason: 'user-stop', turnId: 'turn-a', at: STOPPED_AT }], + ['a host stop', { reason: 'host-stop', at: STOPPED_AT }] + ] satisfies [string, JournalStopEvent][])('opens no settle: %s', (_label, event) => { + expect(beginJournalStopSettle(stateWith(event))).toBeNull() + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-stop-turn-end.ts b/src/main/native-chat/agent-session-journal/journal-stop-turn-end.ts index 4c28a0422a6d..6b71477f74e4 100644 --- a/src/main/native-chat/agent-session-journal/journal-stop-turn-end.ts +++ b/src/main/native-chat/agent-session-journal/journal-stop-turn-end.ts @@ -1,11 +1,11 @@ -// What a Stop decides about the turn it named: the one rule every turn-end write passes through. +// What a Stop decides about the turn it stopped: the one rule every turn-end write passes through. // -// A turn a person's Stop or close of this chat named, or, when it named none, a turn opened by a -// send it stopped, ending with no verdict of its own after that Stop's event, ends as their -// cancellation. A host stop, an eviction and no Stop at all leave the end as written. It runs -// where each row is built, inside the journal's serialized write, so it reads every Stop folded -// before the end: the adapter's settle, the host's fallback and a relaunch's settle all write -// through it, and every client folds the row it wrote. +// A turn a person's Stop or close of this chat named, or, when it named none, one that ended while +// that Stop settled or that its settle bound, ending with no verdict of its own after that Stop's +// event, ends as their cancellation. A host stop, an eviction and no Stop at all leave the end as +// written. It runs where each row is built, inside the journal's serialized write: the adapter's +// end, the host's fallback and a relaunch's settle all write through it, and every client folds +// the row it wrote. What an unnamed Stop binds is held in memory, so after a relaunch it binds none. import { agentJournalSubmissionKey, @@ -16,10 +16,9 @@ import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' -import { isUnansweredStructuredAgentSessionDispatch } from '../../../shared/structured-agent-session-unanswered-dispatch' import type { JournalReducerState } from './journal-reducer' import type { JournalStopEvent } from './journal-row-schema' -import type { JournalQueuePauseMarks } from './queued-message-pause' +import type { JournalQueuePauseMarks, JournalStopSettle } from './queued-message-pause' export type JournalLatestStop = NonNullable @@ -40,7 +39,8 @@ type TurnEndState = Pick< 'items' | 'queuePauseMarks' | 'submissions' | 'aliases' > -/** The send whose journal item `userItemId` (a turn's opener, by its own or its provider key) is. */ +/** The send whose journal item `userItemId` (a turn's opener, by its own or its provider key) is. + * Only says whether a send's turn has opened; it never decides whose a turn's end is. */ function openingSubmission( state: TurnEndState, userItemId: string | undefined @@ -83,72 +83,36 @@ export function latestAcceptedSendUnopened(state: TurnEndState): boolean { return true } -/** A send a Stop that named no turn stopped: one already handed to the agent at the Stop's - * position, whose turn had not opened. A card the Stop held, or anything sent after it, is not. */ -function isStopTarget( - state: TurnEndState, - stop: JournalLatestStop, - submission: AgentJournalSubmission -): boolean { - if ( - submission.dispatchState === 'rejected' || - (submission.handoverRecorded === true && submission.handedOverAt === undefined) - ) { - return false - } - // A handed-over send's item sits at its handover (`placeHandedOverMessage`). - const handedOver = state.items.get(agentJournalSubmissionKey(submission.clientMessageId)) - return handedOver !== undefined && handedOver.sequence < stop.sequence -} - /** Whether `stop`, a person's, makes the end of turn `turnId` theirs: it named that turn, or named - * none and stopped the send that opened it (`userItemId`). */ -function stopIsTurnCancellation( - state: TurnEndState, - stop: JournalLatestStop, - turnId: string, - userItemId: string | undefined -): boolean { + * none and the turn ends while it settles, or its settle bound that turn. */ +function stopIsTurnCancellation(stop: JournalLatestStop, turnId: string): boolean { if (!stopIsAPersons(stop.event.reason)) { return false } if (stop.event.turnId !== undefined) { return stop.event.turnId === turnId } - const opener = openingSubmission(state, userItemId) - return opener !== undefined && isStopTarget(state, stop, opener) + return stop.settle?.settling === true || stop.settle?.turnId === turnId } -/** THE rule: whether the latest Stop makes turn `turnId`, opened by `userItemId` and ending at - * `endedAt` with no verdict of its own, a person's cancellation. An exit the provider saw before - * the Stop was news, whenever its end is written. */ +/** THE rule: whether the latest Stop makes turn `turnId`, ending at `endedAt` with no verdict of its + * own, a person's cancellation. An exit the provider saw before the Stop was news, whenever its + * end is written. */ function stopEndsTurnAsCancellation( state: TurnEndState, turnId: string, - userItemId: string | undefined, endedAt: number | undefined ): boolean { const stop = state.queuePauseMarks.latestStop return ( stop !== null && - stopIsTurnCancellation(state, stop, turnId, userItemId) && + stopIsTurnCancellation(stop, turnId) && (endedAt === undefined || endedAt >= stop.event.at) ) } -/** With no turn running, the work in flight is the person's Stop's: every send still unanswered is - * one it stopped. */ -function unansweredSendsAreStopTargets(state: TurnEndState, stop: JournalLatestStop): boolean { - const unanswered = [...state.submissions.values()].filter((submission) => - isUnansweredStructuredAgentSessionDispatch(submission) - ) - return ( - unanswered.length > 0 && unanswered.every((submission) => isStopTarget(state, stop, submission)) - ) -} - /** - * Whether a person's Stop decides the end of turn `turnId` (null: the sends in flight with no turn + * Whether a person's Stop decides the end of turn `turnId` (null: the work in flight with no turn * running), by `turnEndAfterStop`'s rule: ending at `endedAt` it is their cancellation, and still * running it is theirs to end. For a writer that must choose before the end is written: a host * stop must not supersede it, and a Claude error result naming no reason leaves its verdict to it. @@ -156,29 +120,36 @@ function unansweredSendsAreStopTargets(state: TurnEndState, stop: JournalLatestS export function personStopDecidesTurn( state: TurnEndState, turnId: string | null, - endedAt?: number, - /** The submission that opened the turn, for one whose rows have yet to land. */ - openedBy?: string + endedAt?: number ): boolean { const stop = state.queuePauseMarks.latestStop if (stop === null || !stopIsAPersons(stop.event.reason)) { return false } if (turnId === null) { - return stop.event.turnId === undefined && unansweredSendsAreStopTargets(state, stop) + return stop.event.turnId === undefined && stop.settle?.settling === true + } + return stopEndsTurnAsCancellation(state, turnId, endedAt) +} + +/** Opens the settle of the latest Stop, a person's that named no turn: until it closes, every turn + * that ends is that Stop's. Null when there is none, or it named its turn. */ +export function beginJournalStopSettle( + state: Pick +): JournalStopSettle | null { + const stop = state.queuePauseMarks.latestStop + if (stop === null || stop.event.turnId !== undefined || !stopIsAPersons(stop.event.reason)) { + return null } - const turn = [...state.items.values()] - .map((item) => readAgentJournalTurn(item.body)) - .find((candidate) => candidate?.turnId === turnId) - const userItemId = - turn?.userItemId ?? (openedBy === undefined ? undefined : agentJournalSubmissionKey(openedBy)) - return stopEndsTurnAsCancellation(state, turnId, userItemId, endedAt) + // A repeat press keeps the turn an earlier one bound. + stop.settle = { settling: true, ...(stop.settle?.turnId ? { turnId: stop.settle.turnId } : {}) } + return stop.settle } /** * The body to write for item `itemId`: unchanged unless it ends, with no verdict of its own and no - * earlier than the latest Stop event, a person's, which named it, or stopped the send that opened - * it, while it was still open (running, or unproven). A provider's own verdict always stands. + * earlier than the latest Stop event, a person's that named it or bound it, while it was still open + * (running, or unproven). A provider's own verdict always stands. */ export function turnEndAfterStop( state: TurnEndState, @@ -193,8 +164,7 @@ export function turnEndAfterStop( if (previous && previous.state !== 'running' && previous.state !== 'unverifiable') { return body } - const userItemId = body.userItemId ?? previous?.userItemId - return stopEndsTurnAsCancellation(state, body.turnId, userItemId, body.completedAt) + return stopEndsTurnAsCancellation(state, body.turnId, body.completedAt) ? { ...body, outcome: 'cancellation' } : body } diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index 2349524ccfe9..db63aea8870f 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -155,6 +155,10 @@ export class AgentSessionJournal { return this.state.epoch } + get agent(): AgentSessionJournalIdentity['agent'] { + return this.identity.agent + } + /** Whether a row at this sequence was on disk when this handle opened, so an earlier handle * wrote it. Sequences restart with each epoch, so a row of a later epoch never was. */ wroteBeforeOpen(sequence: number | undefined): boolean { diff --git a/src/main/native-chat/agent-session-journal/queued-message-pause.ts b/src/main/native-chat/agent-session-journal/queued-message-pause.ts index 9039760ba43c..63dc6823df28 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-pause.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-pause.ts @@ -15,9 +15,13 @@ import type { JournalStopEvent, JournalTombstoneRow } from './journal-row-schema export type QueuePauseReason = 'stopped' | 'cleared' | 'restarted' +/** What a person's Stop that named no turn binds, held in memory and never read from a row, so a + * reopen binds nothing: while it settles, every turn that ends; once settled, the turn it stopped. */ +export type JournalStopSettle = { settling: boolean; turnId?: string } + /** The latest Stop event, whatever its reason, and the latest Resume row, folded by the reducer. */ export type JournalQueuePauseMarks = { - latestStop: { sequence: number; event: JournalStopEvent } | null + latestStop: { sequence: number; event: JournalStopEvent; settle?: JournalStopSettle } | null /** 0 when none. */ resumedSequence: number } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-stop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-stop.ts index 4709234ad5c6..a6b12b4f4fd1 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-stop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-stop.ts @@ -10,10 +10,13 @@ import type { ProviderDiagnostic } from '../../../shared/agent-session-failure' /** `refusal`: the provider answered the Stop and declined it, in its own words when it gave any. * `turnNotRunning`: its refusal is the kind it gives for a turn not running there, so the Stop keeps - * the child; absent, it could not interrupt that turn, which may run on. */ + * the child; absent, it could not interrupt that turn, which may run on. `turnMayOpen`: no turn + * was there to interrupt, but one a send is owed may still open. `turnId`: the journal turn a Stop + * that took interrupted, which the Stop then binds. */ export type AgentSessionCancelOutcome = { cancelled: boolean - refusal?: { detail?: ProviderDiagnostic; turnNotRunning?: true } + refusal?: { detail?: ProviderDiagnostic; turnNotRunning?: true; turnMayOpen?: true } + turnId?: string } /** Where a card's Cancel goes: a dismissal (`dismissPrompt`), or the chat's Stop. */ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts index c89c0fc3d6bd..b55ebbda2d79 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts @@ -109,9 +109,17 @@ export function mutateWithChatStop( } return { ok: true, value: { ...named, cancelled: withdrewAny } } } + const reach = hadQueued ? 'unrecorded' : stopReachesUnrecordedWork(ctx, turnId) + const priorStop = ctx.journal.stopMarks.latest() // Issued, not awaited, before the interrupt or any child end; the `finally` awaits it. - const effect = - hadQueued || stopReachesUnrecordedWork(ctx, turnId) ? tookEffect() : Promise.resolve() + const effect = reach === 'unrecorded' ? tookEffect() : Promise.resolve() + // Its settle binds the latest Stop only when that Stop is this press's own, or the one in + // force this press repeats; a late Stop or an event not yet written binds nothing. A write + // is in the fold by its call's return, so a newer latest Stop is this press's event. + const ownsLatestStop = + reach === 'unrecorded' + ? ctx.journal.stopMarks.latest()?.sequence !== priorStop?.sequence + : reach === 'repeat' try { return await performCancel( { ...ctx, failureTextContext: structuredAgentSessionFailureWordsContext(record) }, @@ -130,7 +138,8 @@ export function mutateWithChatStop( endSession: (owed) => { windDown = owed }, - withdrewQueued: withdrew + withdrewQueued: withdrew, + ...(ownsLatestStop ? { opensSettle: true as const } : {}) } ) } finally { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts index aab219c779d6..6f03cc1345cc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-stop.test.ts @@ -240,7 +240,9 @@ describe('a Stop that names no turn', () => { expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } }) expect(closeSession).toHaveBeenCalledExactlyOnceWith(SESSION) - expect(await statusRows()).toEqual(['Cancellation requested.']) + // Its turn never opened, so the child's end took the send back: no row stays to report on it. + expect((await submission(id))?.dispatchState).toBe('rejected') + expect(await statusRows()).toEqual([]) }) it('says the agent did not stop, in its words, when it refused because its turn is not running', async () => { @@ -271,7 +273,9 @@ describe('a Stop that names no turn', () => { expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } }) expect(closeSession).toHaveBeenCalledExactlyOnceWith(SESSION) - expect(await statusRows()).toEqual(['Cancellation requested.']) + // Its turn never opened, so the child's end took the send back: no row stays to report on it. + expect((await submission(id))?.dispatchState).toBe('rejected') + expect(await statusRows()).toEqual([]) }) it('says the agent did not stop, in its words, when it could not interrupt and the child end is unproven', async () => { @@ -295,7 +299,7 @@ describe('a Stop that names no turn', () => { expect(await statusRows()).toEqual(["Codex didn't stop: failed to interrupt turn."]) }) - it('reads as requested when the child exit was proven and only a later cleanup step failed', async () => { + it('counts as stopped when the child exit was proven and only a later cleanup step failed', async () => { const { id, result } = send('hello') await result await eventually(async () => expect((await submission(id))?.handedOverAt).toBeDefined()) @@ -312,7 +316,8 @@ describe('a Stop that names no turn', () => { fields: expect.objectContaining({ scope: 'stop-child', sessionId: SESSION }) }) ) - expect(await statusRows()).toEqual(['Cancellation requested.']) + expect((await submission(id))?.dispatchState).toBe('rejected') + expect(await statusRows()).toEqual([]) }) it('says the Stop is unconfirmed, not that nothing ran, when neither the interrupt nor the child end is proven', async () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts index f0372c8e58de..639b34cb7955 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts @@ -38,23 +38,20 @@ import { exitedRootTurnScope, runningRootTurnScope } from './structured-agent-session-exit-turn-scope' +import { + withdrawCodexSendsNoTurnOpenedFor, + type UnopenedSendJournal +} from './structured-agent-session-unopened-send-withdrawal' /** Bounds the exit reason the lease keeps as log evidence; a provider diagnostic is held to the * same cap. */ export const MAX_UNEXPECTED_EXIT_REASON_CHARS = MAX_PROVIDER_DIAGNOSTIC_CHARS -type DeadGenerationSubmission = Pick< - ReturnType[number], - 'clientMessageId' | 'dispatchState' | 'recovered' | 'handoverRecorded' | 'handedOverAt' -> - -export type DeadGenerationJournal = { +export type DeadGenerationJournal = UnopenedSendJournal & { appendLifecycleBatch: AgentSessionJournal['appendLifecycleBatch'] markPendingSubmissionsUnknown: AgentSessionJournal['markPendingSubmissionsUnknown'] rejectPendingSubmissions: AgentSessionJournal['rejectPendingSubmissions'] - snapshot: () => Pick, 'items'> pendingSubmissions?: AgentSessionJournal['pendingSubmissions'] - submissions?: () => DeadGenerationSubmission[] } export type StructuredAgentSessionUnfinishedWork = { @@ -139,6 +136,9 @@ export async function settleStructuredAgentSessionDeadGeneration(input: { const startupFailure = input.exitedDuringStartup ? structuredAgentSessionStartFailure({ exit: input.exitFailure }, input.failureTextContext) : null + if (!startupFailure) { + await withdrawCodexSendsNoTurnOpenedFor(input.journal, input.fence) + } await (startupFailure ? input.journal.rejectPendingSubmissions(input.fence, startupFailure) : input.journal.markPendingSubmissionsUnknown(input.fence, input.pendingSubmissionReason)) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts index bdc07a3b3d54..afcecadc7445 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts @@ -85,8 +85,8 @@ export class StructuredAgentSessionSinkQueue { journalLinkage = (): StructuredAgentSessionLinkageJournal | null => this.target?.journal ?? null - journalStopDecidesTurn = (turnId: string, endedAt: number, openedBy?: string): boolean => - this.target?.journal.stopMarks.personStopDecides(turnId, endedAt, openedBy) ?? false + journalStopDecidesTurn = (turnId: string, endedAt: number): boolean => + this.target?.journal.stopMarks.personStopDecides(turnId, endedAt) ?? false bindReadingControl(control: StructuredAgentSessionReadingControl): () => void { this.readingControl = control diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts index 49572abdcc00..12966b73a97b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts @@ -142,9 +142,8 @@ export type StructuredAgentSessionEventSink = { /** The bound journal's producer linkage; null until bound. */ journalLinkage?(): StructuredAgentSessionLinkageJournal | null /** Whether the bound journal's Stop rule makes turn `turnId`, ending at `endedAt` with no verdict - * of its own, a person's cancellation (`personStopDecidesTurn`); false until bound. `openedBy`: - * the submission that opened it, for a turn whose rows have yet to land. */ - journalStopDecidesTurn?(turnId: string, endedAt: number, openedBy?: string): boolean + * of its own, a person's cancellation (`personStopDecidesTurn`); false until bound. */ + journalStopDecidesTurn?(turnId: string, endedAt: number): boolean appendLifecycleBatch?( settlementId: string, mutations: readonly JournalLifecycleMutationInput[], diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts index 60c20cb768d9..2c312ef17e84 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-lifetime.ts @@ -140,7 +140,7 @@ export async function stopStructuredAgentSessionAgentUnderSerialize( // Judged before the kill: a stop that ends nothing writes nothing. const recorded = (await stopEndsWork(context, sessionId, session, ending)) ? recordStopEvent(context, sessionId, session, ending) - : Promise.resolve() + : Promise.resolve(null) // The obligation OUTLIVES the child. `child` is ended the instant the adapter proves the exit, // so a step that aborts after that point would otherwise leave the retry reading "no child // here" and skipping the settlement and the lease release it still owes. @@ -233,6 +233,9 @@ export async function stopStructuredAgentSessionAgentUnderSerialize( session.owesProviderChildWindDown = { ...owed, failedAt: session.journal.cursor() } } throw error + } finally { + // A person's close binds what its child's end cut; done, proven or not, it binds no more. + void recorded.then((settle) => session.journal.stopMarks.settled(settle)) } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-event.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-event.ts index 757a76a90590..70cfd07bb7a6 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-event.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-event.ts @@ -7,6 +7,8 @@ import type { StructuredAgentSessionStopCause } from './structured-agent-session import type { StructuredAgentSessionLifetimeContext } from './structured-agent-session-host-lifetime' import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' +import type { JournalStopSettle } from '../agent-session-journal/queued-message-pause' +import { sentSinceStop } from './structured-agent-session-queued-stop' /** How a stop ends the child, and why (`lastEndedChild`). A person's Stop wrote its event in its * own step (`recorded` names its reason); any other stop names the reason its event records, with @@ -63,23 +65,37 @@ export async function stopEndsWork( journal.submissions(), child.fence ) - // A host stop of work a person's Stop is already ending must not supersede that Stop's reason. + return working && (ending.cause === 'user-close' || !defersToPersonsStop(session)) +} + +/** A host stop of work a person's Stop is already ending must not supersede that Stop's reason: + * the turn it decides, or, with no turn running, the work under the queue pause it still holds, + * with nothing sent since, which a host's event would lift. */ +function defersToPersonsStop(session: StructuredAgentSessionHostSession): boolean { + const { journal } = session + const live = journal.activeTurnId() + if (live !== null) { + return journal.stopMarks.personStopDecides(live) + } + const inForce = journal.queuedMessages.userStopInForce() return ( - working && - (ending.cause === 'user-close' || !journal.stopMarks.personStopDecides(journal.activeTurnId())) + (inForce !== null && !sentSinceStop(journal, inForce)) || + journal.stopMarks.personStopDecides(null) ) } /** Writes this stop's event (`JournalStopEvent`). Issued before the kill and never awaited by it: - * bookkeeping, reported on failure. */ + * bookkeeping, reported on failure. A person's close that named no turn opens its settle once its + * event lands, as a person's Stop does, so a turn the child's end cuts is theirs; the caller + * closes it once that end is done. */ export function recordStopEvent( context: StructuredAgentSessionLifetimeContext, sessionId: string, session: StructuredAgentSessionHostSession, ending: StructuredAgentSessionStopEnding -): Promise { +): Promise { if ('recorded' in ending) { - return Promise.resolve() + return Promise.resolve(null) } const turnId = session.journal.activeTurnId() return session.journal @@ -88,12 +104,14 @@ export function recordStopEvent( structuredAgentSessionConversationFence(context.deps.store, sessionId) ) .then( - () => undefined, - (error: unknown) => + () => (ending.cause === 'user-close' ? session.journal.stopMarks.beginSettle() : null), + (error: unknown) => { context.deps.logger.warn("a host stop's Stop event row skipped", { scope: 'stop-event', sessionId, error }) + return null + } ) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts index ccdc429e34e2..03c1042309fa 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts @@ -58,30 +58,42 @@ export async function runRecordedStop( } /** - * Whether a Stop reaching a running agent stops anything no Stop event records yet. Not when it - * names a turn already over (a late Stop from a phone), nor when it repeats the Stop still in force - * with nothing sent since, on the same turn or one that opened after a Stop pressed before any - * turn showed: a card queued between the presses then sends normally, as after one Stop. + * What a Stop reaching a running agent stops: work no Stop event records yet (`unrecorded`), or + * only what the Stop still in force already records, which it repeats with nothing sent since, on + * the same turn or one that opened after a Stop pressed before any turn showed (`repeat`): a card + * queued between the presses then sends normally, as after one Stop. `late`: it names a turn + * already over, as a late Stop from a phone can. */ export function stopReachesUnrecordedWork( ctx: Pick, namedTurnId: string | undefined -): boolean { +): 'unrecorded' | 'repeat' | 'late' { const live = ctx.journal.activeTurnId() // No turn published yet while the agent works: the named one may still be opening. if ( structuredAgentSessionStopNamesTurnNotLive(namedTurnId, live) && (live !== null || !isMainAgentWorking(ctx)) ) { - return false + return 'late' } const inForce = ctx.journal.queuedMessages.userStopInForce() if (inForce === null) { - return true + return 'unrecorded' } - // Sent after that Stop and not refused, even if its fate is unknown: this interrupt may send it - // back to waiting, so this Stop must hold it. A send with no sequence is an older host's. - const sentSince = ctx.journal + // This interrupt may send a later send back to waiting, so this Stop must hold it. + return sentSinceStop(ctx.journal, inForce) || + structuredAgentSessionStopNamesTurnNotLive(inForce.event.turnId, live) + ? 'unrecorded' + : 'repeat' +} + +/** Whether anything was sent after the Stop in force and not refused, even if its fate is + * unknown. A send with no sequence is an older host's. */ +export function sentSinceStop( + journal: Pick, + inForce: { sequence: number } +): boolean { + return journal .submissions() .some( (entry) => @@ -89,7 +101,6 @@ export function stopReachesUnrecordedWork( entry.acceptedSequence !== undefined && entry.acceptedSequence > inForce.sequence ) - return sentSince || structuredAgentSessionStopNamesTurnNotLive(inForce.event.turnId, live) } /** The Stop's withdrawal of every queued send, issued at once and never awaited ahead of the diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts index 2654b20ac730..a91ed7c8cbd4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts @@ -1,9 +1,9 @@ -// Which turn a person's Stop that named no turn binds: only a turn a send it stopped opens. A Stop -// of a start that never landed stopped a send that opens no turn; a card it held, which Resume -// releases, and anything sent after it open their own; a rewind keeps the binding. Turn rows name -// the send that opened them, as Codex writes them. +// Which turn a person's Stop that named no turn binds: one that ends while the Stop settles, and +// the turn its interrupt took. A Stop that stopped nothing binds no turn that opens after it, nor +// does a Stop of a start that never landed; a card it held, which Resume releases, and anything +// sent after it end as their own. Turn rows name the send that opened them, as Codex writes them. -import { afterEach, describe, expect, it } from 'vitest' +import { afterEach, describe, expect, it, vi } from 'vitest' import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' import { AGENT_JOURNAL_THREAD_SCOPE, @@ -12,10 +12,11 @@ import { import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import type { JournalStopEvent } from '../agent-session-journal/journal-row-schema' import { settleStaleStructuredAgentSessionState } from './structured-agent-session-dead-generation-settlement' -import { HOST_TEST_SESSION } from './structured-agent-session-host-test-data' +import { HOST_TEST_SESSION, hostTestOperationId } from './structured-agent-session-host-test-data' import { createQueuedMessageTestRig, eventually, + QUEUED_RIG_CALLER, type QueuedMessageTestRig } from './structured-agent-session-queued-message-rig.test-fixture' @@ -63,6 +64,18 @@ async function laterTurn() { .find((turn) => turn?.turnId === 'turn-later') } +/** Every end row written for the later turn, in order: a relabel would show as two. */ +function laterTurnEndRows() { + const since = journal().readSince({ epoch: journal().epoch, sequence: 0 }) + if (!since.ok) { + throw new Error(`expected rows, got reset ${since.reset}`) + } + return since.rows.flatMap((row) => { + const turn = row.kind === 'item' ? readAgentJournalTurn(row.body) : undefined + return turn?.turnId === 'turn-later' && turn.state !== 'running' ? [turn] : [] + }) +} + /** The turn send `clientMessageId` opens, running, named by its row as Codex writes it. */ async function turnOpenedBy(clientMessageId: string, state: 'running' | 'interrupted' = 'running') { await journal().appendItem( @@ -185,19 +198,22 @@ describe('a Stop of a start that never landed binds no later turn', () => { }) }) -describe('a Stop pressed before its send opened a turn binds only that turn', () => { +describe('a Stop pressed before its send opened a turn binds only the turn it stopped', () => { /** The send the Stop stopped is handed over and unopened; a card waits behind it. */ - async function stopBeforeTheTurnShowed(): Promise<{ stopped: string; held: string }> { + async function stopBeforeTheTurnShowed( + answer: Awaited> = { cancelled: true } + ): Promise<{ stopped: string; held: string }> { rig = await createQueuedMessageTestRig() const stopped = await rig.workingSend() const held = await queuedDraft('queued behind the turn') + rig.cancelTurn.mockResolvedValueOnce(answer) expect(await rig.stop()).toMatchObject({ ok: true }) expect(journal().stopMarks.latest()?.event).not.toHaveProperty('turnId') return { stopped, held } } - it("binds the stopped send's own turn", async () => { - const { stopped } = await stopBeforeTheTurnShowed() + it('binds the turn its interrupt took, whose end lands after the Stop answered', async () => { + const { stopped } = await stopBeforeTheTurnShowed({ cancelled: true, turnId: 'turn-later' }) await rig.settleAccepted(stopped, 'stopped') await turnOpenedBy(stopped, 'interrupted') @@ -205,6 +221,34 @@ describe('a Stop pressed before its send opened a turn binds only that turn', () expect(await laterTurn()).toMatchObject({ state: 'interrupted', outcome: 'cancellation' }) }) + // The interrupt reaches the turn as it opens; its end is written before the Stop answers. + it('reads a turn that ends while the Stop settles as interrupted from its first end row', async () => { + rig = await createQueuedMessageTestRig() + const stopped = await rig.workingSend() + rig.cancelTurn.mockImplementationOnce(async () => { + await turnOpenedBy(stopped) + await turnOpenedBy(stopped, 'interrupted') + return { cancelled: true, turnId: 'turn-later' } + }) + + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + + expect(laterTurnEndRows()).toEqual([ + expect.objectContaining({ state: 'interrupted', outcome: 'cancellation' }) + ]) + expect(journal().activeTurnId()).toBeNull() + }) + + it('binds no turn that opens after a Stop that stopped nothing', async () => { + const { stopped } = await stopBeforeTheTurnShowed({ cancelled: false }) + await rig.settleAccepted(stopped, 'stopped') + + await turnOpenedBy(stopped) + await turnOpenedBy(stopped, 'interrupted') + + await expectNews() + }) + it("writes the host's event when it evicts the turn of the card Resume sent", async () => { const { stopped, held } = await stopBeforeTheTurnShowed() await rig.settleAccepted(stopped, 'stopped') @@ -229,14 +273,116 @@ describe('a Stop pressed before its send opened a turn binds only that turn', () }) }) -describe('a host stop with no turn running after a Stop that named none', () => { - it('writes nothing while every unanswered send is one the Stop stopped', async () => { +describe('a press that writes no Stop event of its own', () => { + // A turnless Stop long since settled, then a later turn the person's own send opened. + async function laterTurnAfterAnEarlierStop(): Promise { rig = await createQueuedMessageTestRig() await rig.workingSend() expect(await rig.stop()).toMatchObject({ ok: true }) expect(journal().stopMarks.latest()?.event).not.toHaveProperty('turnId') + const sent = await rig.workingSend() + await rig.settleAccepted(sent, 'sent') + await turnOpenedBy(sent) + return sent + } + + it('reopens no earlier Stop: a late Stop naming a turn already over binds no turn', async () => { + const sent = await laterTurnAfterAnEarlierStop() + const earlier = journal().stopMarks.latest() + rig.cancelTurn.mockImplementationOnce(async () => { + await turnOpenedBy(sent, 'interrupted') + return { cancelled: false } + }) + + const late = await rig.host.cancel(QUEUED_RIG_CALLER, { + envelope: rig.envelope({ turnId: 'turn-gone' }, 'agentSession.cancel', hostTestOperationId()), + turnId: 'turn-gone' + }) + + expect(late).toMatchObject({ ok: true }) + expect(journal().stopMarks.latest()).toBe(earlier) + await expectNews() + }) + + // The event is in the fold by its write's return, so one that failed leaves the earlier Stop latest. + it('reopens no earlier Stop: a press whose event row failed binds no turn', async () => { + const sent = await laterTurnAfterAnEarlierStop() + const earlier = journal().stopMarks.latest() + vi.spyOn(journal(), 'appendStopEvent').mockRejectedValueOnce(new Error('disk full')) + rig.cancelTurn.mockImplementationOnce(async () => { + await turnOpenedBy(sent, 'interrupted') + return { cancelled: false } + }) + + expect(await rig.stop()).toMatchObject({ ok: true }) + + expect(journal().stopMarks.latest()).toBe(earlier) + await expectNews() + }) +}) + +describe("a Stop's settle that ends the turn its interrupt took", () => { + it("ends a turn still running at the Stop's settle, as the Stop's, once", async () => { + rig = await createQueuedMessageTestRig() + const stopped = await rig.workingSend() + await rig.settleAccepted(stopped, 'stopped') + await turnOpenedBy(stopped) + rig.cancelTurn.mockResolvedValueOnce({ cancelled: true, turnId: 'turn-later' }) + + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + + expect(await laterTurn()).toMatchObject({ state: 'interrupted', outcome: 'cancellation' }) + expect(journal().activeTurnId()).toBeNull() + }) +}) + +describe('a host stop with no turn running after a Stop that named none', () => { + it('defers to that Stop while it settles, and not after', async () => { + rig = await createQueuedMessageTestRig() + await rig.workingSend() + let whileSettling: boolean | undefined + rig.cancelTurn.mockImplementationOnce(async () => { + whileSettling = journal().stopMarks.personStopDecides(null) + return { cancelled: true } + }) + + expect(await rig.stop()).toMatchObject({ ok: true }) + + expect(whileSettling).toBe(true) + expect(journal().stopMarks.personStopDecides(null)).toBe(false) + }) + + it.each([ + ['stopped nothing', { cancelled: false }], + ['took', { cancelled: true }] + ] as const)( + 'writes nothing while the Stop, which %s, still pauses the queue, and the card stays held', + async (_, answer) => { + rig = await createQueuedMessageTestRig() + await rig.workingSend() + const card = await queuedDraft('queued behind the send') + rig.cancelTurn.mockResolvedValueOnce(answer) + expect(await rig.stop()).toMatchObject({ ok: true }) + expect(journal().stopMarks.latest()?.event).not.toHaveProperty('turnId') + + expect(await evictedAt()).toEqual(['user-stop']) + expect(await rig.queuePause()).toMatchObject({ reason: 'stopped' }) + expect(await rig.handoff(card)).toBeUndefined() + } + ) + + // Codex could not reach a turn still able to open, and its process could not be ended. + it('writes nothing after a Stop whose kill failed, and the card stays held', async () => { + rig = await createQueuedMessageTestRig() + await rig.workingSend() + const card = await queuedDraft('queued behind the send') + rig.cancelTurn.mockResolvedValueOnce({ cancelled: false, refusal: { turnMayOpen: true } }) + rig.closeSession.mockRejectedValueOnce(new Error('the kill timed out')) + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: false } }) expect(await evictedAt()).toEqual(['user-stop']) + expect(await rig.queuePause()).toMatchObject({ reason: 'stopped' }) + expect(await rig.handoff(card)).toBeUndefined() }) it("writes the host's event when a send after the Stop is unanswered beside the stopped one", async () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-restart.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-restart.test.ts index 50056eedd031..916654cac648 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-restart.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-restart.test.ts @@ -1,7 +1,7 @@ -// A Stop that took effect still decides its turn after Orca restarts before the turn's end was -// written: the relaunch's settle reads the Stop's event, so the turn reads "Interrupted after N" -// with the muted mark, not "Failed". A turn nobody stopped, and one a Stop never named, still read -// as the news they are. +// A Stop that took effect still decides the turn it named after Orca restarts before the turn's end +// was written: the relaunch's settle reads the Stop's event, so the turn reads "Interrupted after N" +// with the muted mark, not "Failed". A turn nobody stopped, one a Stop never named, and one a +// turnless Stop was still settling when Orca died, still read as the news they are. import { afterEach, describe, expect, it } from 'vitest' import type { AgentSessionDeathEvidence } from '../../../shared/agent-session-record' @@ -138,6 +138,26 @@ describe('a restart between a Stop and its turn end', () => { expect(errorRows).toEqual([]) }) + // A Stop naming its turn, as the phone sends one: its event names the turn, which is what a + // relaunch reads, so it still counts though Orca died with its interrupt out. + it.each([ + ['Claude', CLAUDE_TURN], + ['Codex', CODEX_TURN] + ])('reads Interrupted after N for a %s Stop that named its turn', async (_label, identity) => { + await runningTurn(identity) + rig.cancelTurn.mockImplementationOnce(() => new Promise(() => undefined)) + void rig.host.cancel(QUEUED_RIG_CALLER, { + envelope: rig.envelope({ turnId: TURN }, 'agentSession.cancel', hostTestOperationId()), + turnId: TURN + }) + await expect.poll(() => journal().stopMarks.latest()?.event.turnId).toBe(TURN) + + await restartAndSettle() + + expect(settled().turn).toMatchObject({ state: 'interrupted', outcome: 'cancellation' }) + expect(settled().label).toMatch(/^Interrupted after /) + }) + it('reads Interrupted after N, marked interrupted: a close of the chat that died midway', async () => { await runningTurn(CODEX_TURN) // The host dies inside the close: the provider's close never answers, and nothing settles. @@ -273,10 +293,11 @@ describe('a restart between a Stop and its turn end', () => { expect(settled().turn).not.toHaveProperty('outcome') }) - // A Stop pressed before any turn showed stopped the turn its send was about to open, and no other. + // A Stop pressed before any turn showed stopped the turn its interrupt took, and no other. it('reads a turn a host send opened after the turnless Stop ended its own turn as no Stop of its', async () => { rig = await createQueuedMessageTestRig() const stopped = await rig.workingSend() + rig.cancelTurn.mockResolvedValueOnce({ cancelled: true, turnId: TURN }) expect(await rig.stop()).toMatchObject({ ok: true }) expect(journal().stopMarks.latest()?.event.turnId).toBeUndefined() await rig.settleAccepted(stopped, 'stopped') @@ -323,6 +344,35 @@ describe('a restart between a Stop and its turn end', () => { expect(settled(NEXT_TURN).turn).not.toHaveProperty('outcome') }) + // What a turnless Stop binds is held in memory: a relaunch finds it unsettled, and it binds none. + it('reads the turn a turnless Stop was settling when Orca died as no Stop of its', async () => { + rig = await createQueuedMessageTestRig() + const stopped = await rig.workingSend() + // Orca dies while the interrupt is out: the Stop never settles. + rig.cancelTurn.mockImplementationOnce(async () => { + await journal().appendItem( + CODEX_TURN, + { + kind: 'turn', + turnId: TURN, + state: 'running', + startedAt: Date.now(), + userItemId: agentJournalSubmissionKey(stopped) + }, + { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + return new Promise(() => undefined) + }) + void rig.stop() + await expect.poll(() => journal().activeTurnId()).toBe(TURN) + expect(journal().stopMarks.latest()?.event.turnId).toBeUndefined() + + await restartAndSettle('exit-observed') + + expect(settled(TURN).turn).toMatchObject({ state: 'interrupted' }) + expect(settled(TURN).turn).not.toHaveProperty('outcome') + }) + // A refusal is no record: the first Stop stays the one in force, so pressing again repeats it. it('writes nothing for a Stop pressed again after the provider refused one', async () => { await runningTurn(CODEX_TURN) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-note-withdrawn-send.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-note-withdrawn-send.test.ts new file mode 100644 index 000000000000..a3c8853ef67a --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-note-withdrawn-send.test.ts @@ -0,0 +1,148 @@ +// A Stop that ends the child after its interrupt failed: its row reports on the run it stopped, so +// one whose child end took back the send it found, with no turn running, leaves none. Its message +// is back in the composer, and a row would sit under the turn before as if that turn were stopped. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' +import { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import { createTrackedJournalOpener } from '../agent-session-journal/journal-host-database-test-support' +import { createStructuredAgentSessionLogger } from './structured-agent-session-logger' +import { performCancel, type AgentSessionTurnContext } from './structured-agent-session-turns' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: 'session-1', + workspaceId: 'workspace-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } +} + +let root: string | null = null +const journals = createTrackedJournalOpener() + +afterEach(async () => { + await journals.closeAll() + if (root) { + await rm(root, { recursive: true, force: true }) + root = null + } +}) + +async function stopEndingTheChild(options: { + turnRunning: boolean + takesSendBack: boolean + earlierProcessSendInDoubt?: boolean +}) { + root = await mkdtemp(join(tmpdir(), 'orca-stop-note-withdrawn-')) + const journal = await journals.open({ identity: IDENTITY, stateDirectory: root }) + if (options.turnRunning) { + await journal.appendItem( + { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal: 0 }, + { kind: 'turn', turnId: 'turn-1', state: 'running' }, + { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + } + if (options.earlierProcessSendInDoubt) { + await journal.appendSubmission({ + clientMessageId: 'send-0', + payloadFingerprint: 'send-0', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'before the crash' }] }, + fence: 1 + }) + await journal.resolveDispatch({ + clientMessageId: 'send-0', + state: 'unknown', + reason: 'host_restarted', + fence: 1, + recovered: true + }) + } + await journal.appendSubmission({ + clientMessageId: 'send-1', + payloadFingerprint: 'send-1', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'look around' }] }, + fence: 1 + }) + const ctx: AgentSessionTurnContext = { + logger: createStructuredAgentSessionLogger(), + sessionId: 'session-1', + journal, + fence: 1, + adapter: { + acquire: vi.fn(), + dispatch: vi.fn(), + closeSession: vi.fn(), + cancelTurn: vi.fn(async () => ({ + cancelled: false, + refusal: { detail: { text: 'failed to interrupt turn', audience: 'person' as const } } + })), + answerPrompt: vi.fn(), + setOption: vi.fn() + }, + persistOptions: async () => undefined, + resolvedBy: 'client-1', + publish: vi.fn(), + now: () => 1 + } + const result = await performCancel(ctx, { + clientOperationId: 'stop-1', + stopChild: async () => { + if (options.takesSendBack) { + await journal.resolveDispatch({ + clientMessageId: 'send-1', + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('cancelled'), { + surface: 'rejection' + }), + fence: 1, + recovered: true + }) + } + } + }) + const rows = journal + .snapshot() + .items.flatMap((item) => (item.body.kind === 'status' ? [item.body.text] : [])) + return { cancelled: result.ok && result.value.cancelled, rows } +} + +describe('a Stop that ends the child after its interrupt failed', () => { + it('leaves no row when the child end took back the send, with no turn running', async () => { + expect(await stopEndingTheChild({ turnRunning: false, takesSendBack: true })).toEqual({ + cancelled: true, + rows: [] + }) + }) + + it('says cancellation was requested when the send stays in doubt', async () => { + expect(await stopEndingTheChild({ turnRunning: false, takesSendBack: false })).toEqual({ + cancelled: true, + rows: ['Cancellation requested.'] + }) + }) + + // That send was never this child's to take back, so it does not keep the row. + it("leaves no row when the only other send in doubt was an earlier process's", async () => { + expect( + await stopEndingTheChild({ + turnRunning: false, + takesSendBack: true, + earlierProcessSendInDoubt: true + }) + ).toEqual({ cancelled: true, rows: [] }) + }) + + it('says cancellation was requested on the turn it stopped, whatever became of the send', async () => { + expect(await stopEndingTheChild({ turnRunning: true, takesSendBack: true })).toEqual({ + cancelled: true, + rows: ['Cancellation requested.'] + }) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-settle-binding.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-settle-binding.test.ts new file mode 100644 index 000000000000..14e62854bbcc --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-settle-binding.test.ts @@ -0,0 +1,197 @@ +// A Stop pressed before its send's turn showed binds the turn that send opens only by stopping it: +// a Codex turn that did not open in time ends with the child the Stop ends, and a Claude turn that +// opens before the Stop's child end is proven ends as the Stop's. Nothing reads as running after. + +import { afterEach, describe, expect, it } from 'vitest' +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalItemIdentity +} from '../../../shared/agent-session-journal-types' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' +import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/structured-agent-session-main-agent-working' +import { HOST_TEST_SESSION } from './structured-agent-session-host-test-data' +import { + createQueuedMessageTestRig, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' + +let rig: QueuedMessageTestRig + +afterEach(() => rig.dispose()) + +const LATER: AgentJournalItemIdentity = { + provider: 'codex', + threadId: 'thread-1', + turnId: 'turn-later', + ordinal: 1000 +} + +const OPENED: AgentJournalItemIdentity = { + provider: 'codex', + threadId: 'thread-1', + turnId: 'turn-opened', + ordinal: 999 +} + +function journal() { + const open = rig.host.collaboratorsForTests().sessions.get(HOST_TEST_SESSION)?.journal + if (!open) { + throw new Error('expected the conversation open') + } + return open +} + +/** The turn `clientMessageId` opens, running, as the provider writes it. */ +async function turnOpens(clientMessageId: string): Promise { + await journal().appendItem( + OPENED, + { + kind: 'turn', + turnId: 'turn-opened', + state: 'running', + startedAt: Date.now(), + userItemId: agentJournalSubmissionKey(clientMessageId) + }, + { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) +} + +function openedTurn() { + return journal() + .snapshot() + .items.map((item) => readAgentJournalTurn(item.body)) + .find((turn) => turn?.turnId === 'turn-opened') +} + +function nothingRuns(): boolean { + return !isStructuredAgentSessionMainAgentWorking( + journal().activeTurnId(), + journal().submissions() + ) +} + +/** Drains the Stop's next step on the session's lane: a wind-down it handed the child's end to. */ +async function windDown(): Promise { + await rig.host['tasks'].serialize(HOST_TEST_SESSION, async () => {}) +} + +describe('a Codex Stop whose answered turn did not open in time', () => { + // Codex admitted the send, answered into a turn it has not opened; the Stop's wait ran out. + async function stopOfAnsweredSend(): Promise { + rig = await createQueuedMessageTestRig() + const sent = await rig.workingSend() + rig.cancelTurn.mockResolvedValueOnce({ cancelled: false, refusal: { turnMayOpen: true } }) + return sent + } + + it('ends the child, so the turn can never run', async () => { + await stopOfAnsweredSend() + + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + + expect(rig.closeSession).toHaveBeenCalled() + expect(openedTurn()).toBeUndefined() + expect(nothingRuns()).toBe(true) + }) + + it('reads a turn that opened before the child end as interrupted by the Stop', async () => { + const sent = await stopOfAnsweredSend() + rig.closeSession.mockImplementationOnce(async () => { + await turnOpens(sent) + return true + }) + + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + + expect(openedTurn()).toMatchObject({ state: 'interrupted', outcome: 'cancellation' }) + expect(nothingRuns()).toBe(true) + }) +}) + +describe('a Codex Stop that could not reach a turn still able to open, whose kill failed', () => { + it('says the Stop is unconfirmed, never that no turn was running', async () => { + rig = await createQueuedMessageTestRig() + await rig.workingSend() + rig.cancelTurn.mockResolvedValueOnce({ cancelled: false, refusal: { turnMayOpen: true } }) + rig.closeSession.mockRejectedValueOnce(new Error('the kill timed out')) + + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: false } }) + + const rows = journal() + .snapshot() + .items.flatMap((item) => (item.body.kind === 'status' ? [item.body.failure?.kind] : [])) + expect(rows).toEqual(['cancelUnconfirmed']) + }) +}) + +describe('a Claude Stop pressed before its send echoed', () => { + it('reads the turn that opens before its child end is proven as interrupted by the Stop', async () => { + rig = await createQueuedMessageTestRig({ stopEndsSession: true }) + const sent = await rig.workingSend() + rig.closeSession.mockImplementationOnce(async () => { + await turnOpens(sent) + return true + }) + + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + await windDown() + + expect(rig.closeSession).toHaveBeenCalled() + expect(openedTurn()).toMatchObject({ state: 'interrupted', outcome: 'cancellation' }) + expect(nothingRuns()).toBe(true) + }) +}) + +describe("a person's close pressed before its send's turn showed", () => { + it('reads the turn its child end cut as interrupted by the person', async () => { + rig = await createQueuedMessageTestRig() + const sent = await rig.workingSend() + rig.closeSession.mockImplementationOnce(async () => { + await turnOpens(sent) + return true + }) + + await rig.host.close(HOST_TEST_SESSION, 'user-close') + + const { items } = await rig.host.journalSnapshot(HOST_TEST_SESSION) + const turn = items + .map((item) => readAgentJournalTurn(item.body)) + .find((entry) => entry?.turnId === 'turn-opened') + expect(turn).toMatchObject({ state: 'interrupted', outcome: 'cancellation' }) + }) + + // A /clear stops the agent as the person's close and keeps the conversation, so the close's + // settle must have closed by the time a later turn ends. + it('binds no turn that ends on its own after a close the conversation outlived', async () => { + rig = await createQueuedMessageTestRig() + await rig.workingSend() + await rig.host['lifetime'].stopAgent(HOST_TEST_SESSION, { cause: 'user-close' }) + expect(journal().stopMarks.latest()?.event).toMatchObject({ reason: 'user-close' }) + expect(journal().stopMarks.latest()?.event).not.toHaveProperty('turnId') + + await journal().appendItem( + LATER, + { kind: 'turn', turnId: 'turn-later', state: 'running', startedAt: Date.now() }, + { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + await journal().appendItem( + LATER, + { + kind: 'turn', + turnId: 'turn-later', + state: 'interrupted', + startedAt: Date.now(), + completedAt: Date.now() + 5 + }, + { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + + const later = journal() + .snapshot() + .items.map((item) => readAgentJournalTurn(item.body)) + .find((turn) => turn?.turnId === 'turn-later') + expect(later).toMatchObject({ state: 'interrupted' }) + expect(later).not.toHaveProperty('outcome') + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts index 5e26d1b620f8..6337d65f5537 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-cancel.ts @@ -2,7 +2,11 @@ import { agentChildWorkStopTargets } from '../../../shared/agent-child-work-stop import { agentSessionFailureFact } from '../../../shared/agent-session-failure' import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' -import type { AgentJournalStatusItem } from '../../../shared/agent-session-journal-types' +import type { + AgentJournalStatusItem, + AgentJournalSubmission +} from '../../../shared/agent-session-journal-types' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import type { AgentSessionCancelResult } from '../../../shared/agent-session-wire' import { latestJournalDispatchObservation } from '../agent-session-journal/journal-dispatch-observation' import type { AgentSessionCancelOutcome } from './structured-agent-session-adapter' @@ -22,6 +26,8 @@ import { structuredAgentSessionStoppedTurnId } from './structured-agent-session-turn-stop-notes' import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/structured-agent-session-main-agent-working' +import { runningTurnLifecycleRevisions } from './structured-agent-session-stale-turn-verdict' +import { sendStopCanTakeBack } from './structured-agent-session-unopened-send-withdrawal' import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' /** Whether the fold reads working. Every write has landed by its call's return, and the open paid @@ -52,25 +58,46 @@ function stillRunsStoppedTurn( return stoppedTurnId === null || ctx.journal.activeTurnId() === stoppedTurnId } -/** The row for a Stop the provider declined, in its words when it gave any. */ +/** Whether the child's end took back every send the Stop found in flight, none of them having run + * (`withdrawCodexSendsNoTurnOpenedFor`). */ +function tookBackEverySend( + ctx: Pick, + inFlight: readonly AgentJournalSubmission[] +): boolean { + const rejected = new Set( + ctx.journal + .submissions() + .flatMap((entry) => (entry.dispatchState === 'rejected' ? [entry.clientMessageId] : [])) + ) + return inFlight.length > 0 && inFlight.every((entry) => rejected.has(entry.clientMessageId)) +} + +/** The row for a Stop the provider declined, in its words when it gave any. One that could not + * reach a turn still able to open says the Stop is unconfirmed, never that nothing ran. */ function stopRefusedNote( ctx: Pick, refusal: AgentSessionCancelOutcome['refusal'] ): AgentJournalStatusItem { const detail = refusal?.detail + const fact = refusal?.turnMayOpen + ? agentSessionFailureFact('cancelUnconfirmed') + : agentSessionFailureFact('stopRefused', detail ? { detail } : {}) return { kind: 'status', - ...agentSessionFailureWords(agentSessionFailureFact('stopRefused', detail ? { detail } : {}), { - ...ctx.failureTextContext, - surface: 'row' - }) + ...agentSessionFailureWords(fact, { ...ctx.failureTextContext, surface: 'row' }) } } /** What a Stop that ends the provider's session leaves its next serialized step: whether the * provider took the interrupt, so its wind-down is worth waiting on, and when the interrupt went - * out. No turn id: that step runs right behind the Stop, so no later turn can slip in between. */ -export type StructuredAgentSessionStopWindDown = { waitsForProvider: boolean; stoppedAt: number } + * out. No turn id: that step runs right behind the Stop, so no later turn can slip in between. + * `settled` closes the Stop's settle once the wind-down finishes, whether or not the child's exit + * was proven: until then, what ends is the Stop's. */ +export type StructuredAgentSessionStopWindDown = { + waitsForProvider: boolean + stoppedAt: number + settled?: () => void +} /** * A session-ending Stop's second step, queued behind its first in the same tick so nothing sent @@ -91,35 +118,45 @@ export async function endStoppedStructuredAgentSession( await stopChild() } catch (error) { onError(error) + } finally { + windDown.settled?.() + } +} + +/** A turn the Stop's interrupt took that still reads running: the Stop ends it, once, while it + * still binds it. The provider's stream lands as it arrives, so the fold already holds any end it + * sent. Bookkeeping: a failure is reported, never the Stop's. */ +async function endStoppedTurnAtSettle(ctx: AgentSessionTurnContext, turnId: string): Promise { + try { + const running = ctx.journal + .snapshot() + .items.filter((item) => readAgentJournalTurn(item.body)?.turnId === turnId) + const mutations = runningTurnLifecycleRevisions(running, { + state: 'interrupted', + completedAt: ctx.now() + }) + if (mutations.length > 0) { + await ctx.journal.appendLifecycleBatch({ + settlementId: `stop-settled:${turnId}`, + mutations, + fence: ctx.fence + }) + } + } catch (error) { + ctx.logger.warn("ending a stopped turn at its Stop's settle failed", { + scope: 'stop-settle', + sessionId: ctx.sessionId, + error + }) } } +/** What the Stop's settle binds: the turn it stopped, and whether its wind-down closes it. */ +type StopSettleBinding = { turnId?: string; closedByWindDown?: true } + export async function performCancel( ctx: AgentSessionTurnContext, - input: { - clientOperationId: string - /** Absent: whatever the conversation has in flight; present: only while that turn is current. */ - turnId?: string - scope?: 'background-tasks' - taskId?: string - prompt?: { itemId: string; expectedRevision: number } - /** Ends the provider child, for a running command the provider did not take the Stop on, or a - * turn whose interrupt failed. */ - stopChild?: () => Promise - /** A child end after a failed interrupt that threw. */ - onStopChildError?: (error: unknown) => void - /** After that throw: whether the host let go of the child, its exit proven before a later - * cleanup step failed. */ - childReleased?: () => boolean - /** Hands the child's end to the Stop's next serialized step, for a provider whose Stop ends - * its session. */ - endSession?: (windDown: StructuredAgentSessionStopWindDown) => void - /** Whether the host's withdrawal of queued messages for this Stop withdrew any; awaited only - * after the interrupt. */ - withdrewQueued?: Promise - /** The session's child records: a background Stop reaches the tasks they offer a stop. */ - childWork?: () => readonly AgentChildWorkView[] | undefined - } + input: PerformCancelInput ): Promise> { if (input.prompt) { const validated = validatePendingPrompt(ctx, input.prompt) @@ -127,6 +164,52 @@ export async function performCancel( return answerCancelOfSettledPrompt(ctx, { ...input, prompt: input.prompt }, validated) } } + // A person's Stop that named no turn binds what ends while it settles (`beginJournalStopSettle`). + const settle = input.opensSettle ? ctx.journal.stopMarks.beginSettle() : null + const binding: StopSettleBinding = {} + const close = (): void => ctx.journal.stopMarks.settled(settle, binding.turnId) + try { + return await cancelAndNote(ctx, input, binding, close) + } finally { + if (!binding.closedByWindDown) { + close() + } + } +} + +type PerformCancelInput = { + clientOperationId: string + /** Absent: whatever the conversation has in flight; present: only while that turn is current. */ + turnId?: string + scope?: 'background-tasks' + taskId?: string + prompt?: { itemId: string; expectedRevision: number } + /** Ends the provider child, for a running command the provider did not take the Stop on, or a + * turn whose interrupt failed. */ + stopChild?: () => Promise + /** A child end after a failed interrupt that threw. */ + onStopChildError?: (error: unknown) => void + /** After that throw: whether the host let go of the child, its exit proven before a later + * cleanup step failed. */ + childReleased?: () => boolean + /** Hands the child's end to the Stop's next serialized step, for a provider whose Stop ends + * its session. */ + endSession?: (windDown: StructuredAgentSessionStopWindDown) => void + /** Whether the host's withdrawal of queued messages for this Stop withdrew any; awaited only + * after the interrupt. */ + withdrewQueued?: Promise + /** The session's child records: a background Stop reaches the tasks they offer a stop. */ + childWork?: () => readonly AgentChildWorkView[] | undefined + /** The latest Stop event is this press's own, or the in-force one it repeats: its settle binds. */ + opensSettle?: true +} + +async function cancelAndNote( + ctx: AgentSessionTurnContext, + input: PerformCancelInput, + binding: StopSettleBinding, + closeSettle: () => void +): Promise> { let cancelled = false let note: AgentJournalStatusItem | null = { kind: 'status', @@ -155,11 +238,15 @@ export async function performCancel( const endsSession = input.endSession !== undefined && ctx.adapter.stopEndsSession?.(ctx.sessionId) === true const stoppedAt = Date.now() + // Read before the cancel: the sends a child end may take back. + const sentBeforeStop = ctx.journal.submissions().filter(sendStopCanTakeBack) // The provider's own answer; unset when its cancel threw, leaving the effect unknown. let taken: boolean | undefined // The provider could not interrupt the turn, or its cancel threw: the turn may run on. let interruptFailed = false let refusal: AgentSessionCancelOutcome['refusal'] + // The turn the provider says this Stop's interrupt took. + let stoppedTurn: string | undefined try { const dispatchStatus = latestJournalDispatchObservation(ctx.journal, ctx.fence) const outcome: AgentSessionCancelOutcome = stoppedBefore @@ -187,6 +274,7 @@ export async function performCancel( taken = outcome.cancelled cancelled = outcome.cancelled refusal = outcome.refusal + stoppedTurn = outcome.turnId interruptFailed = refusal !== undefined && refusal.turnNotRunning !== true // Working is read first: a working session never takes this branch, so a child end below never // waits on the withdrawal. On an idle queue the withdrawal has already landed in the fold. @@ -217,7 +305,8 @@ export async function performCancel( if (endsSession && (!namesTurnNotLive || taken !== false)) { // An interrupt the provider took is worth waiting on, turn row or not: a Stop before the echo // has none, and the echo still opens the turn the Stop interrupted. - input.endSession?.({ waitsForProvider: taken === true, stoppedAt }) + binding.closedByWindDown = true + input.endSession?.({ waitsForProvider: taken === true, stoppedAt, settled: closeSettle }) cancelled = true // The child's end confirms the Stop, so a refused or unconfirmed interrupt says nothing more. if (note !== null) { @@ -246,7 +335,13 @@ export async function performCancel( } if (ended) { cancelled = true - note = { kind: 'status', text: STOP_NOTE_CANCELLATION_REQUESTED } + // A child end that took back every send it found, with no turn running, ended a run that + // never started: its message is back in the composer, and a row would sit under the turn + // before as if that turn were stopped. + note = + stoppedTurnId === null && tookBackEverySend(ctx, sentBeforeStop) + ? null + : { kind: 'status', text: STOP_NOTE_CANCELLATION_REQUESTED } } else if (taken !== undefined) { // The turn was just read running, so a named Stop says it was refused rather than nothing. note = stopRefusedNote(ctx, refusal) @@ -255,6 +350,12 @@ export async function performCancel( // Nothing was left of the turn it named and nothing else ended: a Stop that ends nothing writes no row. note = null } + if (cancelled) { + binding.turnId = stoppedTurn ?? stoppedTurnId ?? undefined + } + if (taken === true && stoppedTurn !== undefined && !binding.closedByWindDown && !input.scope) { + await endStoppedTurnAtSettle(ctx, stoppedTurn) + } const value = { ...(input.turnId !== undefined ? { turnId: input.turnId } : {}), cancelled } if (input.scope || note === null) { return { ok: true, value } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.test.ts new file mode 100644 index 000000000000..d4a78117a1e0 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.test.ts @@ -0,0 +1,107 @@ +// Which sends a Stop's child end takes back: the same rule decides the withdrawal and whether the +// Stop writes a row, so a send it must leave alone is left alone by both. + +import { describe, expect, it, vi } from 'vitest' +import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import { + sendStopCanTakeBack, + withdrawCodexSendsNoTurnOpenedFor, + type UnopenedSendJournal +} from './structured-agent-session-unopened-send-withdrawal' + +type Send = Pick< + AgentJournalSubmission, + | 'clientMessageId' + | 'dispatchState' + | 'recovered' + | 'handoverRecorded' + | 'handedOverAt' + | 'acceptedSequence' +> + +/** A send that started its own turn before a person's Stop at sequence 10, no turn having opened. */ +function sendBeforeStop(overrides: Partial = {}): Send { + return { + clientMessageId: 'send-1', + dispatchState: 'pending', + handoverRecorded: true, + handedOverAt: 5, + acceptedSequence: 4, + ...overrides + } +} + +function journalWith(send: Send) { + const resolveDispatch = vi.fn(async () => ({ epoch: 'epoch-1', sequence: 11 })) + const journal: UnopenedSendJournal = { + agent: 'codex', + queuedMessages: { + userStopInForce: () => ({ sequence: 10, event: { reason: 'user-stop', at: 1 } }) + }, + snapshot: () => ({ + items: [ + { + itemId: agentJournalSubmissionKey(send.clientMessageId), + revision: 0, + sequence: 5, + observedAt: 5, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'look around' }] }, + turnScope: { kind: 'thread' } + } + ] + }), + submissions: () => [send], + resolveDispatch + } + return { journal, resolveDispatch } +} + +describe('a send a Stop can take back', () => { + it('is one still being handed over, or one this process lost the answer to', () => { + expect(sendStopCanTakeBack(sendBeforeStop())).toBe(true) + expect(sendStopCanTakeBack(sendBeforeStop({ dispatchState: 'unknown' }))).toBe(true) + }) + + // An earlier process's send: its child may have started it before the crash. + it('is never one an earlier process left in doubt', () => { + expect(sendStopCanTakeBack(sendBeforeStop({ dispatchState: 'unknown', recovered: true }))).toBe( + false + ) + }) + + // The card holds its text and returns paused; nothing was handed to the agent. + it("is never a queued card's send that was not handed over", () => { + expect(sendStopCanTakeBack(sendBeforeStop({ handedOverAt: undefined }))).toBe(false) + }) +}) + +describe("a Codex child's end under a person's Stop", () => { + it('withdraws a send whose answer this process lost', async () => { + const { journal, resolveDispatch } = journalWith(sendBeforeStop({ dispatchState: 'unknown' })) + + await withdrawCodexSendsNoTurnOpenedFor(journal, 1) + + expect(resolveDispatch).toHaveBeenCalledWith( + expect.objectContaining({ clientMessageId: 'send-1', state: 'rejected' }) + ) + }) + + it('leaves a send an earlier process left in doubt as it is', async () => { + const { journal, resolveDispatch } = journalWith( + sendBeforeStop({ dispatchState: 'unknown', recovered: true }) + ) + + await withdrawCodexSendsNoTurnOpenedFor(journal, 1) + + expect(resolveDispatch).not.toHaveBeenCalled() + }) + + it("leaves a queued card's send as it is", async () => { + const { journal, resolveDispatch } = journalWith(sendBeforeStop({ handedOverAt: undefined })) + + await withdrawCodexSendsNoTurnOpenedFor(journal, 1) + + expect(resolveDispatch).not.toHaveBeenCalled() + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.ts new file mode 100644 index 000000000000..64a5296c0086 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.ts @@ -0,0 +1,94 @@ +// Which unanswered sends a dying Codex child takes with it as never sent, read from the journal at +// the settlement that lands, so a retried wind-down reads the same rows. + +import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' +import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import type { AgentType } from '../../../shared/agent-status-types' +import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' + +/** What the derivation reads; absent members (a narrow double) withdraw nothing. */ +export type UnopenedSendJournal = { + agent?: AgentType + queuedMessages?: Pick + snapshot: () => Pick, 'items'> + submissions?: () => Pick< + AgentJournalSubmission, + | 'clientMessageId' + | 'dispatchState' + | 'recovered' + | 'handoverRecorded' + | 'handedOverAt' + | 'acceptedSequence' + >[] + resolveDispatch?: AgentSessionJournal['resolveDispatch'] +} + +/** A send a dying child can take back: one still being handed over, or one this process left in + * doubt (its answer was lost). Never a queued card's, nor a send an earlier process left behind. */ +export function sendStopCanTakeBack( + entry: Pick< + AgentJournalSubmission, + 'dispatchState' | 'recovered' | 'handoverRecorded' | 'handedOverAt' + > +): boolean { + return ( + !isQueuedAgentJournalSubmission(entry) && + (entry.dispatchState === 'pending' || + (entry.dispatchState === 'unknown' && entry.recovered !== true)) + ) +} + +/** + * Withdraws the sends a Codex child left unanswered when a person's Stop, in force since they were + * sent, ends it: each one that started its own turn (handed over with no turn running, so its + * message belongs to no turn) when no turn has opened since. Codex records a prompt only once its + * turn starts (core tasks/regular.rs:50, session/turn.rs:886-902), so they never ran. A send that + * joined a running turn may be in it, so it, a send with no recorded place, and any other end stay + * in doubt. + */ +export async function withdrawCodexSendsNoTurnOpenedFor( + journal: UnopenedSendJournal, + fence: number +): Promise { + const stop = journal.agent === 'codex' ? journal.queuedMessages?.userStopInForce() : null + if (!stop || !journal.submissions || !journal.resolveDispatch) { + return + } + const { items } = journal.snapshot() + const turns = items.flatMap((item) => { + const turn = readAgentJournalTurn(item.body) + return turn ? [{ running: turn.state === 'running', sequence: item.sequence }] : [] + }) + // Its message's place, recorded at handover: the turn it joined, or the conversation. + const startedOwnTurn = (clientMessageId: string): boolean => + items.find((item) => item.itemId === agentJournalSubmissionKey(clientMessageId))?.turnScope + ?.kind === 'thread' + const opened = (acceptedSequence: number): boolean => + turns.some((turn) => turn.running || turn.sequence > acceptedSequence) + const unopened = journal + .submissions() + .filter( + (entry) => + sendStopCanTakeBack(entry) && + entry.acceptedSequence !== undefined && + entry.acceptedSequence < stop.sequence && + startedOwnTurn(entry.clientMessageId) && + !opened(entry.acceptedSequence) + ) + const withdrawn = agentSessionFailureWords(agentSessionFailureFact('cancelled'), { + surface: 'rejection' + }) + for (const entry of unopened) { + await journal.resolveDispatch({ + clientMessageId: entry.clientMessageId, + state: 'rejected', + ...withdrawn, + fence, + recovered: true + }) + } +} diff --git a/src/main/runtime/structured-agent-session-codex-interrupt-order.test.ts b/src/main/runtime/structured-agent-session-codex-interrupt-order.test.ts new file mode 100644 index 000000000000..de09e72b7e5c --- /dev/null +++ b/src/main/runtime/structured-agent-session-codex-interrupt-order.test.ts @@ -0,0 +1,238 @@ +// Codex answers an interrupt it took before it ends the turn: on `TurnAborted` the app-server +// answers pending interrupts, then sends `turn/completed` (interrupted) on the same channel. The +// Stop's settle, which runs between the two, must label the turn the Stop's: no end row for it +// ever reads as a failure. Driven through the shipped host, journal and Codex adapter; only the +// Codex child is fake, and the test delivers the turn's end after the Stop has answered. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { + CodexAppServerConnection, + CodexAppServerConnectionHandlers, + openCodexAppServerConnection +} from '../codex/codex-app-server-connection' +import { codexTurnLifecycleFake } from '../codex/codex-turn-lifecycle-fake' +import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-session-mutation-envelope' +import type { AgentJournalTurnItem } from '../../shared/agent-session-journal-types' +import { owesStructuredAgentSessionWork } from '../../shared/structured-agent-session-owed-work' +import { + HOST_TEST_SESSION as SESSION, + HOST_TEST_THREAD as THREAD, + hostTestAttachParams, + hostTestMessage +} from '../native-chat/agent-session-wire/structured-agent-session-host-test-data' +import type { StructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-host' +import { + liveTestJournalRows, + openTestJournalHostDatabase +} from '../native-chat/agent-session-journal/journal-host-database-test-support' +import { parseJournalRow } from '../native-chat/agent-session-journal/journal-row-schema' +import { createStructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' +import { + ensureStructuredAgentSessionHost, + stopStructuredAgentSessionRuntime +} from './structured-agent-session-runtime' + +const CALLER = { callerKey: 'codex-interrupt-order-test' } +const MODEL = { + model: 'gpt-test', + displayName: 'GPT Test', + hidden: false, + supportedReasoningEfforts: [], + defaultReasoningEffort: null, + isDefault: true +} + +let root: string +let host: StructuredAgentSessionHost +let fence: number +let handlers: CodexAppServerConnectionHandlers | undefined +let answers: number +let interrupts: number +let turns: ReturnType +let opened: Set +let operations = 0 + +/** The durable ledger stamps its own clock and refuses an id far from it. */ +const operationId = (): string => `${Date.now()}-${(++operations).toString(16).padStart(32, '0')}` + +function envelope(method: string, fields: Record) { + return { + sessionId: SESSION, + clientOperationId: operationId(), + expectedRuntimeFence: fence, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method, + sessionId: SESSION, + fields + }) + } +} + +async function send(text: string): Promise { + const body = hostTestMessage(text) + const sent = await host.send(CALLER, { envelope: envelope('agentSession.send', { body }), body }) + if (!sent.ok) { + throw new Error(JSON.stringify(sent.refusal)) + } + return sent.value.clientMessageId +} + +async function stop(turnId?: string): Promise { + const stopped = await host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', turnId === undefined ? {} : { turnId }), + ...(turnId === undefined ? {} : { turnId }) + }) + if (!stopped.ok) { + throw new Error(JSON.stringify(stopped.refusal)) + } +} + +/** Every end row written for turn-1, as a row or a lifecycle batch's mutation, in order. */ +function turnEndRows(): AgentJournalTurnItem[] { + return liveTestJournalRows(openTestJournalHostDatabase(root).db, SESSION).flatMap((stored) => { + const parsed = parseJournalRow(stored.rowJson) + if (!parsed.ok) { + return [] + } + const { row } = parsed + const bodies = row.kind === 'lifecycle-batch' ? row.mutations : [row] + return bodies.flatMap((entry) => + entry.kind === 'item' && + entry.body.kind === 'turn' && + entry.body.turnId === 'turn-1' && + entry.body.state !== 'running' + ? [entry.body] + : [] + ) + }) +} + +/** The Stop has answered with Codex's end still to come; then Codex ends the turn. */ +async function expectInterruptedThroughout(): Promise { + expect(interrupts).toBe(1) + expect(turnEndRows()).toEqual([ + expect.objectContaining({ state: 'interrupted', outcome: 'cancellation' }) + ]) + + turns.end('interrupted') + + await vi.waitFor(async () => expect(await owesWork()).toBe(false)) + const ends = turnEndRows() + expect(ends.length).toBeGreaterThan(0) + for (const end of ends) { + expect(end).toMatchObject({ state: 'interrupted', outcome: 'cancellation' }) + } +} + +async function owesWork(): Promise { + const snapshot = await host.journalSnapshot(SESSION) + return owesStructuredAgentSessionWork(snapshot.items, snapshot.submissions, fence) +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-codex-interrupt-order-')) + answers = 0 + interrupts = 0 + opened = new Set() + turns = codexTurnLifecycleFake(THREAD, () => (method, params) => { + if (method === 'turn/started') { + opened.add(turns.turnId ?? '') + } + handlers?.onNotification?.(method, params) + }) + const openConnection: typeof openCodexAppServerConnection = async ( + _launch, + connectionHandlers = {} + ) => { + handlers = connectionHandlers + const connection: CodexAppServerConnection = { + pid: 4321, + closed: false, + request: async (method) => { + if (method === 'thread/start' || method === 'thread/resume') { + return { thread: { id: THREAD } } + } + if (method === 'model/list') { + return { data: [MODEL], nextCursor: null } + } + if (method === 'turn/start') { + answers += 1 + return turns.routes['turn/start']() + } + if (method === 'turn/interrupt') { + // Taken: answered now; the test sends the turn's end once the Stop has answered. A turn + // Codex runs but whose turn/started was not yet read gets that frame ahead of the answer. + interrupts += 1 + if (turns.turnId !== null && !opened.has(turns.turnId)) { + turns.start() + } + return {} + } + return {} + }, + notify: () => {}, + respond: () => {}, + respondWithError: () => {}, + close: async () => true + } + return connection + } + host = await ensureStructuredAgentSessionHost({ + logger: createStructuredAgentSessionLogger(), + stateDirectory: root, + hostId: 'local', + claimKeyId: 'key-1', + resolveWorkspacePath: async () => root, + resolveClaudeAuthPolicy: () => ({ stripAuthEnv: true }), + resolveCodexCommand: () => 'codex', + resolveEnvironment: async () => ({ PATH: process.env.PATH }), + openCodexConnection: openConnection, + readProcessStartTime: async () => 1_700_000_000_000 + }) + const attachParams = hostTestAttachParams(null, { providerHandle: undefined }) + attachParams.envelope.clientOperationId = operationId() + const attached = await host.attach(CALLER, attachParams) + if (!attached.ok) { + throw new Error(JSON.stringify(attached.refusal)) + } + fence = attached.value.fence +}) + +afterEach(async () => { + await stopStructuredAgentSessionRuntime() + await rm(root, { recursive: true, force: true }) +}) + +describe("a Codex Stop answered before Codex's turn/completed (interrupted)", () => { + it.each([ + ['names the running turn', 'turn-1'], + ['names no turn', undefined] + ] as const)( + 'reads the turn as interrupted by the Stop, never failed, when it %s', + async (_, named) => { + await send('look around') + await vi.waitFor(() => expect(answers).toBe(1)) + turns.start() + + await stop(named) + + await expectInterruptedThroughout() + } + ) + + // No turn showed when the person pressed Stop, which interrupts the answered turn at once: the + // turn that opens is bound by the Stop's settle. + it('reads the turn that opens behind an in-flight turn/start as interrupted, never failed', async () => { + const release = turns.holdNextAnswer() + await send('look around') + await vi.waitFor(() => expect(answers).toBe(1)) + const stopping = stop() + release() + await stopping + + await expectInterruptedThroughout() + }) +}) diff --git a/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts b/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts index 57e632a6c85c..b7bbd283fd8c 100644 --- a/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts +++ b/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts @@ -1,7 +1,8 @@ // A Codex send the turn it went into never took: the Stop that interrupts that turn // withdraws it, and nothing reads as working after. A send made while that turn runs, a -// queued card's Send-now included, goes in as `turn/steer` naming it. A Stop or a send made -// before Codex opens that turn waits for it to open, since Codex refuses an interrupt until then. +// queued card's Send-now included, goes in as `turn/steer` naming it. A Stop made before Codex +// opens that turn interrupts it at once; refused before Codex's thread runs it, the Stop waits +// for it to open and interrupts once more. A send made then waits for it to open. // Driven through the shipped host, journal and Codex adapter; only the Codex child is fake, // keeping Codex 0.157's turn bookkeeping. @@ -9,10 +10,11 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import type { - CodexAppServerConnection, - CodexAppServerConnectionHandlers, - openCodexAppServerConnection +import { + CodexAppServerRequestError, + type CodexAppServerConnection, + type CodexAppServerConnectionHandlers, + type openCodexAppServerConnection } from '../codex/codex-app-server-connection' import { codexTurnLifecycleFake } from '../codex/codex-turn-lifecycle-fake' import { settledWithin } from '../codex/codex-structured-dispatch-test-support' @@ -22,6 +24,7 @@ import { computeAgentSessionPayloadFingerprint } from '../../shared/agent-sessio import type { AgentJournalSubmission } from '../../shared/agent-session-journal-types' import { classifyDispatchRejection } from '../../shared/structured-agent-session-dispatch-rejection' import { owesStructuredAgentSessionWork } from '../../shared/structured-agent-session-owed-work' +import { readAgentJournalTurn } from '../../shared/agent-session-turn-record' import { HOST_TEST_SESSION as SESSION, HOST_TEST_THREAD as THREAD, @@ -87,6 +90,13 @@ let answers: number let steers: number let interrupts: number let childCloses: number +let connections: number +/** Closes that fail before one goes through, as a kill that times out. */ +let failingCloses: number +/** Codex fails every interrupt it is sent, as one it could not submit (-32603). */ +let interruptsFail: boolean +/** Codex takes the interrupt and answers it; the test sends that turn's end later. */ +let interruptEndHeld: boolean let turns: ReturnType let operations = 0 @@ -171,6 +181,12 @@ async function settled(): Promise<{ } } +async function statusRows(): Promise { + return (await host.journalSnapshot(SESSION)).items.flatMap((item) => + item.body.kind === 'status' ? [item.body.text] : [] + ) +} + function verdictOf(submissions: readonly AgentJournalSubmission[], clientMessageId: string) { const submission = submissions.find((entry) => entry.clientMessageId === clientMessageId) return submission?.dispatchState === 'rejected' @@ -187,6 +203,10 @@ beforeEach(async () => { steers = 0 interrupts = 0 childCloses = 0 + connections = 0 + failingCloses = 0 + interruptsFail = false + interruptEndHeld = false turns = codexTurnLifecycleFake( THREAD, () => (method, params) => handlers?.onNotification?.(method, params) @@ -196,6 +216,7 @@ beforeEach(async () => { connectionHandlers = {} ) => { handlers = connectionHandlers + connections += 1 const connection: CodexAppServerConnection = { pid: 4321, closed: false, @@ -214,6 +235,20 @@ beforeEach(async () => { steers += 1 return turns.routes['turn/steer'](params) } + if (method === 'turn/interrupt' && interruptsFail) { + interrupts += 1 + throw new CodexAppServerRequestError( + 'turn/interrupt', + -32603, + 'codex app-server turn/interrupt failed: could not submit', + 'could not submit' + ) + } + if (method === 'turn/interrupt' && interruptEndHeld) { + interrupts += 1 + turns.takeInterrupt() + return {} + } if (method === 'turn/interrupt') { interrupts += 1 return turns.routes['turn/interrupt'](params) @@ -224,6 +259,10 @@ beforeEach(async () => { respond: () => {}, respondWithError: () => {}, close: async () => { + if (failingCloses > 0) { + failingCloses -= 1 + throw new Error('the kill timed out') + } childCloses += 1 return true } @@ -472,7 +511,7 @@ describe('a second send made after Codex answered the first, before it opened th }) describe('a Stop sent after Codex answered a cold send, before it opened the turn', () => { - it('waits for the turn to open, then stops it and withdraws the send', async () => { + it('refused before the thread runs it, waits for the turn to open, then stops it and withdraws the send', async () => { const sent = await send('look around') await vi.waitFor(() => expect(answers).toBe(1)) @@ -482,17 +521,16 @@ describe('a Stop sent after Codex answered a cold send, before it opened the tur vi.useFakeTimers({ toFake: ['Date', 'setTimeout', 'clearTimeout'] }) const stopping = stop() await began.promise - // Without the wait, it would reach Codex now, which would refuse it, and be done. const held = settledWithin(stopping, 1_000) await vi.advanceTimersByTimeAsync(1_000) expect(openWaits.ended).toEqual([]) expect(await held).toBe('held') - expect(interrupts).toBe(0) + expect(interrupts).toBe(1) turns.start() await stopping - expect(interrupts).toBe(1) - expect(turns.turnId).toBeNull() + expect(interrupts).toBe(2) + await vi.waitFor(() => expect(turns.turnId).toBeNull()) await vi.waitFor(async () => expect(verdictOf((await settled()).submissions, sent)).toBe('withdrawn') ) @@ -501,14 +539,8 @@ describe('a Stop sent after Codex answered a cold send, before it opened the tur }) describe('a Stop in that window that the turn never opens for', () => { - async function statusRows(): Promise { - return (await host.journalSnapshot(SESSION)).items.flatMap((item) => - item.body.kind === 'status' ? [item.body.text] : [] - ) - } - - async function waitingStop(): Promise<{ stopping: Promise }> { - await send('look around') + async function waitingStop(): Promise<{ stopping: Promise; sent: string }> { + const sent = await send('look around') await vi.waitFor(() => expect(answers).toBe(1)) await host.flushStreamedEvents(SESSION) const began = Promise.withResolvers() @@ -519,7 +551,7 @@ describe('a Stop in that window that the turn never opens for', () => { const held = settledWithin(stopping, 200) await vi.advanceTimersByTimeAsync(200) expect(await held).toBe('held') - return { stopping } + return { stopping, sent } } it('says Codex had no turn running when that turn ends first', async () => { @@ -528,12 +560,13 @@ describe('a Stop in that window that the turn never opens for', () => { turns.end('interrupted') await stopping - expect(interrupts).toBe(0) + expect(interrupts).toBe(1) + expect(childCloses).toBe(0) expect(await statusRows()).toContain('Codex had no turn running to stop.') }) it('lets a chat closed behind it close within its bound and one eviction', async () => { - const { stopping } = await waitingStop() + const { stopping, sent } = await waitingStop() const closing = host.close(SESSION, 'evict') const closedWithinBound = settledWithin( @@ -549,8 +582,10 @@ describe('a Stop in that window that the turn never opens for', () => { expect(await closedWithinBound).not.toBe('held') expect(await settledWithin(stopping, 0)).not.toBe('held') expect(childCloses).toBe(1) - expect(interrupts).toBe(0) - expect(await statusRows()).toContain('Codex had no turn running to stop.') + expect(interrupts).toBe(1) + // Its wait ran out with the turn still able to open, so the Stop ended the child and took the + // send back. + expect(verdictOf((await settled()).submissions, sent)).toBe('withdrawn') }) it('lets the app quit behind it within the eviction budget, and still close the child', async () => { @@ -570,6 +605,206 @@ describe('a Stop in that window that the turn never opens for', () => { }) }) +describe("a Stop pressed while Codex's turn/start is in flight", () => { + function turnRow( + items: Awaited>['items'] + ) { + return items + .map((item) => readAgentJournalTurn(item.body)) + .find((turn) => turn?.turnId === 'turn-1') + } + + it("interrupts the turn that opens, which reads as the Stop's, and nothing reads as running", async () => { + const release = turns.holdNextAnswer() + await send('look around') + await vi.waitFor(() => expect(answers).toBe(1)) + // Queued behind the send's handover on the session's lane, so it runs after Codex answers. + const stopping = stop() + release() + await vi.waitFor(() => expect(openWaits.turnIds).toContain('turn-1')) + turns.start() + await stopping + + // Refused before the thread ran the turn, then taken once it opened. + expect(interrupts).toBe(2) + expect(turnRow((await host.journalSnapshot(SESSION)).items)).toMatchObject({ + state: 'interrupted', + outcome: 'cancellation' + }) + // Codex's end, after its answer, settles the send. + await vi.waitFor(async () => expect((await settled()).owesWork).toBe(false)) + }) + + it('ends the child when its interrupt was refused and the turn has not opened by the end of its wait', async () => { + const release = turns.holdNextAnswer() + await send('look around') + await vi.waitFor(() => expect(answers).toBe(1)) + const stopping = stop() + release() + + expect(await settledWithin(stopping, CODEX_TURN_OPEN_WAIT_MS + 2_000)).not.toBe('held') + + expect(interrupts).toBe(1) + expect(childCloses).toBe(1) + expect((await settled()).owesWork).toBe(false) + expect(turnRow((await host.journalSnapshot(SESSION)).items)).toBeUndefined() + }) + + // Codex records a prompt only once its turn starts, so a send whose turn never opened never ran: + // the Stop takes it back, and no send in doubt holds the client's queue. + async function stoppedBeforeItsTurnOpened(): Promise { + const release = turns.holdNextAnswer() + const sent = await send('look around') + await vi.waitFor(() => expect(answers).toBe(1)) + const stopping = stop() + release() + expect(await settledWithin(stopping, CODEX_TURN_OPEN_WAIT_MS + 2_000)).not.toBe('held') + return sent + } + + async function nextSendStartsANewChild(): Promise { + turns = codexTurnLifecycleFake( + THREAD, + () => (method, params) => handlers?.onNotification?.(method, params) + ) + const next = await send('try again') + await vi.waitFor(() => expect(answers).toBe(2)) + expect(connections).toBe(2) + return next + } + + it('withdraws the send whose turn never opened, and the next send starts a new child', async () => { + const sent = await stoppedBeforeItsTurnOpened() + + expect(childCloses).toBe(1) + expect(verdictOf((await settled()).submissions, sent)).toBe('withdrawn') + await nextSendStartsANewChild() + }) + + // Nothing it stopped stays in the transcript, so a Stop row would read as stopping the turn before. + it('leaves no Stop row under the turn before once it takes the send back', async () => { + const warmUp = await send('warm up') + await vi.waitFor(() => expect(answers).toBe(1)) + turns.start() + turns.echo(warmUp) + turns.end('completed') + const release = turns.holdNextAnswer() + const sent = await send('look around') + await vi.waitFor(() => expect(answers).toBe(2)) + const stopping = stop() + release() + expect(await settledWithin(stopping, CODEX_TURN_OPEN_WAIT_MS + 2_000)).not.toBe('held') + + expect(verdictOf((await settled()).submissions, sent)).toBe('withdrawn') + expect(await statusRows()).toEqual([]) + expect(turnRow((await host.journalSnapshot(SESSION)).items)).toMatchObject({ + state: 'completed' + }) + }) + + // Its turn/start answer lost, the send is in doubt rather than pending; the child's end takes it + // back all the same, so it writes no row either. + it('leaves no Stop row under the turn before when it takes back a send whose answer was lost', async () => { + const warmUp = await send('warm up') + await vi.waitFor(() => expect(answers).toBe(1)) + turns.start() + turns.echo(warmUp) + turns.end('completed') + const answer = turns.routes['turn/start'] + turns.routes['turn/start'] = () => { + turns.routes['turn/start'] = answer + throw new Error('codex app-server request timed out: turn/start') + } + const sent = await send('look around') + await vi.waitFor(() => expect(answers).toBe(2)) + await vi.waitFor(async () => + expect(verdictOf((await settled()).submissions, sent)).toBe('unknown') + ) + + expect(await settledWithin(stop(), CODEX_TURN_OPEN_WAIT_MS + 2_000)).not.toBe('held') + + expect(childCloses).toBe(1) + expect(verdictOf((await settled()).submissions, sent)).toBe('withdrawn') + expect(await statusRows()).toEqual([]) + }) + + it('withdraws it too when the child end fails and a later retry lands it', async () => { + failingCloses = 1 + const sent = await stoppedBeforeItsTurnOpened() + expect(childCloses).toBe(0) + + await nextSendStartsANewChild() + + expect(childCloses).toBe(1) + expect(verdictOf((await settled()).submissions, sent)).toBe('withdrawn') + }) + + // A turn that opened may hold the send: the child's end leaves it in doubt, as before. + it('leaves a send in doubt when its turn opened before the child end', async () => { + const sent = await send('look around') + await vi.waitFor(() => expect(answers).toBe(1)) + turns.start() + interruptsFail = true + + await stop() + + expect(childCloses).toBe(1) + expect(verdictOf((await settled()).submissions, sent)).toBe('unknown') + expect(await statusRows()).toEqual(['Cancellation requested.']) + }) + + // Handed over once the stopped turn read ended, it started its own turn, however late that + // turn's own end then arrives; a second Stop finds its turn never opened. + it('withdraws a send made after a Stop that started its own turn, which never opened', async () => { + const opening = await send('look around') + await vi.waitFor(() => expect(answers).toBe(1)) + turns.start() + turns.echo(opening) + interruptEndHeld = true + await stop() + interruptEndHeld = false + const next = await send('run this after the stop') + await vi.waitFor(() => expect(answers).toBe(2)) + // The stopped turn's own end arrives after the next send was handed over. + turns.end('interrupted') + + expect(await settledWithin(stop(), CODEX_TURN_OPEN_WAIT_MS + 2_000)).not.toBe('held') + + expect(childCloses).toBe(1) + expect(verdictOf((await settled()).submissions, next)).toBe('withdrawn') + }) + + // Codex drains a steer into the running turn, so it may hold one it never echoed. + it('leaves a send steered into the open turn in doubt when the child end follows', async () => { + const opening = await send('look around') + await vi.waitFor(() => expect(answers).toBe(1)) + turns.start() + turns.echo(opening) + const steered = await send('and check the tests') + await vi.waitFor(() => expect(steers).toBe(1)) + interruptsFail = true + + await stop('turn-1') + + expect(childCloses).toBe(1) + expect(verdictOf((await settled()).submissions, steered)).toBe('unknown') + }) + + // Sent after the Stop, to a child that then dies before its turn opens: the Stop, still in force, + // was never this send's. + it('leaves a send made after the Stop in doubt when its child dies', async () => { + await stoppedBeforeItsTurnOpened() + const next = await nextSendStartsANewChild() + + handlers?.onExit?.(new Error('codex app-server exited')) + + await vi.waitFor(async () => + expect(verdictOf((await settled()).submissions, next)).not.toBe('pending') + ) + expect(verdictOf((await settled()).submissions, next)).toBe('unknown') + }) +}) + describe('a cold send with no Stop behind it', () => { /** Well inside one eviction step's budget, and far inside the bound a Stop waits. */ const PROMPTLY_MS = 1_000 @@ -587,6 +822,17 @@ describe('a cold send with no Stop behind it', () => { expect(childCloses).toBe(1) }) + // Only a person's Stop takes a send back: a host's close leaves it in doubt, as before. + it('leaves its send in doubt when the host closes the chat', async () => { + await answeredColdSend() + const sent = (await settled()).submissions[0]?.clientMessageId ?? '' + + await host.close(SESSION, 'evict') + + expect(childCloses).toBe(1) + expect(verdictOf((await settled()).submissions, sent)).toBe('unknown') + }) + it('never delays quitting the app', async () => { await answeredColdSend() diff --git a/src/shared/structured-agent-session-outbox-withdrawn-send.test.ts b/src/shared/structured-agent-session-outbox-withdrawn-send.test.ts new file mode 100644 index 000000000000..214ed97872aa --- /dev/null +++ b/src/shared/structured-agent-session-outbox-withdrawn-send.test.ts @@ -0,0 +1,78 @@ +// A send the host withdrew (a Stop took it back before it ran) leaves the outbox with no notice, so +// the next queued send goes out. One the host left in doubt holds the queue until its Retry. + +import { describe, expect, it } from 'vitest' +import { agentSessionFailureFact } from './agent-session-failure' +import { agentSessionFailureWords } from './agent-session-failure-words' +import type { AgentJournalSubmission } from './agent-session-journal-types' +import { + createStructuredAgentSessionOutboxEntry, + type StructuredAgentSessionOutboxEntry +} from './structured-agent-session-outbox' +import { reconcileStructuredAgentSessionOutbox } from './structured-agent-session-outbox-reconcile' +import { admitStructuredAgentSessionOutboxEntry } from './structured-agent-session-outbox-admission' + +function entry( + clientMessageId: string, + patch: Partial = {} +): StructuredAgentSessionOutboxEntry { + return { + ...createStructuredAgentSessionOutboxEntry({ + clientMessageId, + sessionId: 'session-1', + text: clientMessageId, + attachments: [], + queuedAt: 1 + }), + ...patch + } +} + +function submission(patch: Partial): AgentJournalSubmission { + return { + clientMessageId: 'stopped', + fence: 1, + payloadFingerprint: 'fingerprint', + dispatchState: 'pending', + providerItemId: null, + reason: null, + submittedAt: 1, + resolvedAt: 2, + ...patch + } +} + +const OUTBOX = [entry('stopped', { state: 'dispatching', lastAttemptAt: 1 }), entry('next')] + +describe('the send a Stop ended before its turn opened', () => { + it('leaves the outbox when withdrawn, and the next send goes out', () => { + const withdrawn = submission({ + dispatchState: 'rejected', + recovered: true, + ...agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' }) + }) + + const reconciled = reconcileStructuredAgentSessionOutbox(OUTBOX, [withdrawn], []) + + expect(reconciled.map((each) => each.clientMessageId)).toEqual(['next']) + expect(admitStructuredAgentSessionOutboxEntry(reconciled)).toMatchObject({ + state: 'dispatch', + entry: { clientMessageId: 'next' } + }) + }) + + it('holds the queue when left in doubt', () => { + const inDoubt = submission({ + dispatchState: 'unknown', + recovered: true, + reason: 'provider_closed_before_acknowledgement' + }) + + const reconciled = reconcileStructuredAgentSessionOutbox(OUTBOX, [inDoubt], []) + + expect(admitStructuredAgentSessionOutboxEntry(reconciled)).toMatchObject({ + state: 'blocked', + entry: { clientMessageId: 'stopped' } + }) + }) +})