From bdefc1edd039d4cb45afe4f26353743b74ff1e62 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 1 Oct 2026 19:54:58 -0700 Subject: [PATCH 1/2] fix(native-chat): a host stop decides from the agent's live state, with no wait A close, an eviction or the host stopping a start must decide, before it kills the child, whether it ends work and so must write its Stop event. It waited up to a second for the event sink to flush the provider's rows into the journal, and when the wait ran long it guessed from the latest accepted send. The journal trails the child: an acceptance lands straight in it while the turn row that follows queues behind the sink, and a turn's end row can queue while the journal still shows the turn running. Each adapter now reports what its child has in flight, read off the provider's own frames as they arrive (`liveWork`): Claude's open turn or unechoed send, Codex's running turn (by the journal id of a command it carries), a send on its way, or a send answered into a turn not yet open. The host stop reads that, plus the sends the journal holds unanswered, in its own serialized step. The event names the journal's running turn, or the live one when its row has not landed. A person's Stop still holds against a later host stop while every send in flight is one it stopped, now counting a send accepted whose turn row has not landed. Removes the drain, its one-second bound, and the slow-drain fallback. --- .../claude/claude-request-end-wait.test.ts | 16 ++ src/main/claude/claude-request-end-wait.ts | 10 +- .../claude-structured-session-adapter.ts | 6 +- .../codex/codex-command-turn-claim.test.ts | 9 + src/main/codex/codex-journal-command-turn.ts | 6 + .../codex-structured-journal-contracts.ts | 3 + .../codex-structured-journal-translation.ts | 1 + .../codex/codex-structured-session-adapter.ts | 6 + .../codex-structured-turn-open-wait.test.ts | 39 ++++ .../codex/codex-structured-turn-open-wait.ts | 23 ++ .../journal-stop-marks.ts | 11 +- .../journal-stop-turn-end.ts | 28 ++- ...tured-agent-session-adapter-router.test.ts | 11 + ...structured-agent-session-adapter-router.ts | 3 + .../structured-agent-session-adapter.ts | 6 + ...ctured-agent-session-close-verdict.test.ts | 5 + .../structured-agent-session-host-lifetime.ts | 7 +- ...tructured-agent-session-host-stop-event.ts | 73 +++--- ...-agent-session-host-stop-live-view.test.ts | 215 ++++++++++++++++++ ...session-queued-message-rig.test-fixture.ts | 8 + ...d-agent-session-stop-event-entries.test.ts | 129 +---------- ...red-agent-session-surface-lifetime.test.ts | 12 +- 22 files changed, 423 insertions(+), 204 deletions(-) create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-live-view.test.ts diff --git a/src/main/claude/claude-request-end-wait.test.ts b/src/main/claude/claude-request-end-wait.test.ts index 7cca7a190dcd..732d3df4bf72 100644 --- a/src/main/claude/claude-request-end-wait.test.ts +++ b/src/main/claude/claude-request-end-wait.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { awaitClaudeRequestEnd, CLAUDE_STOP_GRACE_MS, + claudeLiveWork, claudeStoppedRequestEndWait, settleClaudeTurnEndWaiters } from './claude-request-end-wait' @@ -90,3 +91,18 @@ describe("a Stop's wait for Claude to wind down what it has in flight", () => { expect(await settledWithin(wait('session-2', Date.now()), 0)).toBe(true) }) }) + +// What a host stop reads, in its own step and with no wait, to judge whether it ends work. +describe("Claude's live view of what its child has in flight", () => { + it('is its open turn', () => { + expect(claudeLiveWork(session('turn-1'))).toEqual({ turnId: 'turn-1' }) + }) + + it('is a send with no turn yet before its echo', () => { + expect(claudeLiveWork(claudeWith(null, 1).claude)).toEqual({ turnId: null }) + }) + + it('is nothing at rest', () => { + expect(claudeLiveWork(session(null))).toBeUndefined() + }) +}) diff --git a/src/main/claude/claude-request-end-wait.ts b/src/main/claude/claude-request-end-wait.ts index a176c2f96f18..3f58db9e4743 100644 --- a/src/main/claude/claude-request-end-wait.ts +++ b/src/main/claude/claude-request-end-wait.ts @@ -3,6 +3,7 @@ // ends with its own result, and Claude writes both to its transcript. Keyed on Claude's own request, // not on a journal turn: a Stop pressed before the echo has no turn row yet. +import type { StructuredAgentSessionLiveWork } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { ClaudeSession } from './claude-structured-session-state' /** One budget from the Stop's interrupt: for Claude to answer it and wind its request down. */ @@ -10,9 +11,10 @@ export const CLAUDE_STOP_GRACE_MS = 3_000 const waiters = new WeakMap void>>() -/** Claude has a turn open, or a send it has not answered yet. */ -function claudeRequestInFlight(session: ClaudeSession): boolean { - return (session.translator?.currentTurnId ?? null) !== null || session.dispatchWaiters.length > 0 +/** What Claude has in flight (`liveWork`): its open turn, or a send it has not answered yet. */ +export function claudeLiveWork(session: ClaudeSession): StructuredAgentSessionLiveWork | undefined { + const turnId = session.translator?.currentTurnId ?? null + return turnId !== null || session.dispatchWaiters.length > 0 ? { turnId } : undefined } function nextSettle(session: ClaudeSession): { settled: Promise; forget: () => void } { @@ -43,7 +45,7 @@ export async function awaitClaudeRequestEnd(session: ClaudeSession, ms: number): }) try { // A settle that leaves something in flight, such as a subagent's result, waits on. - while (!elapsed && claudeRequestInFlight(session)) { + while (!elapsed && claudeLiveWork(session) !== undefined) { const next = nextSettle(session) try { await Promise.race([next.settled, budget]) diff --git a/src/main/claude/claude-structured-session-adapter.ts b/src/main/claude/claude-structured-session-adapter.ts index 746d603308f1..4aeccbef53c2 100644 --- a/src/main/claude/claude-structured-session-adapter.ts +++ b/src/main/claude/claude-structured-session-adapter.ts @@ -27,7 +27,7 @@ import { type ClaudeStructuredSessionEvent } from './claude-structured-session-state' import { closeAllClaudeSessions, closeClaudeSession } from './claude-structured-session-close' -import { claudeStoppedRequestEndWait } from './claude-request-end-wait' +import { claudeLiveWork, claudeStoppedRequestEndWait } from './claude-request-end-wait' import { drainClaudeObservedExits, observeClaudeSessionExit, @@ -239,6 +239,10 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda const session = this.sessions.get(sessionId) return session ? claudeHoldsDispatch(session) : false } + liveWork: NonNullable = (sessionId) => { + const session = this.sessions.get(sessionId) + return session ? claudeLiveWork(session) : undefined + } answerPrompt: StructuredAgentSessionAdapter['answerPrompt'] = (request) => answerClaudeStructuredPrompt({ request, sessions: this.sessions }) setOption: StructuredAgentSessionAdapter['setOption'] = (input) => diff --git a/src/main/codex/codex-command-turn-claim.test.ts b/src/main/codex/codex-command-turn-claim.test.ts index 078060497f2c..c39fbba3558a 100644 --- a/src/main/codex/codex-command-turn-claim.test.ts +++ b/src/main/codex/codex-command-turn-claim.test.ts @@ -349,6 +349,15 @@ describe('a Codex turn a conversation command claims', () => { expect(translator.commandProviderTurnId('ordinary')).toBe('ordinary') }) + it("maps the provider turn carrying the command to the command's journal turn", () => { + const { translator, emit } = harness() + translator.beginCommand(COMMAND) + emit(notification('turn/started', { turn: { id: PROVIDER_TURN } })) + + expect(translator.commandJournalTurnId(PROVIDER_TURN)).toBe(COMMAND.turnId) + expect(translator.commandJournalTurnId('ordinary')).toBe('ordinary') + }) + it('claims the same provider turn when the refused start is retried', () => { const { writes, translator, emit } = harness(1) translator.beginCommand(COMMAND) diff --git a/src/main/codex/codex-journal-command-turn.ts b/src/main/codex/codex-journal-command-turn.ts index 90e2f556b845..a615a34da9b8 100644 --- a/src/main/codex/codex-journal-command-turn.ts +++ b/src/main/codex/codex-journal-command-turn.ts @@ -78,6 +78,12 @@ export class CodexJournalCommandTurn { return turnId } + /** The journal turn provider turn `providerTurnId` writes to: the command's it carries, or its + * own. */ + journalTurnId(providerTurnId: string): string { + return this.carried.get(providerTurnId)?.command.turnId ?? providerTurnId + } + /** Codex reported the compaction, in the turn carrying the command. */ compacted(providerTurnId: string): void { const carried = this.carried.get(providerTurnId) diff --git a/src/main/codex/codex-structured-journal-contracts.ts b/src/main/codex/codex-structured-journal-contracts.ts index af7c1ed3035d..9e8a6aa4ee92 100644 --- a/src/main/codex/codex-structured-journal-contracts.ts +++ b/src/main/codex/codex-structured-journal-contracts.ts @@ -44,6 +44,9 @@ export type CodexJournalTranslator = { forgetCommand: (turnId: string) => void /** The provider turn a Stop naming `turnId` interrupts; undefined while a command has none. */ commandProviderTurnId: (turnId: string) => string | undefined + /** The journal turn provider turn `providerTurnId` writes to: the command's it carries, or its + * own. */ + commandJournalTurnId: (providerTurnId: string) => string cancelPrompt: (journalItemId: string) => CodexJournalTranslationAdmission restoreThread: ( threadId: string, diff --git a/src/main/codex/codex-structured-journal-translation.ts b/src/main/codex/codex-structured-journal-translation.ts index 20fd8bf58df3..7b91cc5a4afd 100644 --- a/src/main/codex/codex-structured-journal-translation.ts +++ b/src/main/codex/codex-structured-journal-translation.ts @@ -287,6 +287,7 @@ export function createCodexJournalTranslator( beginCommand: (command) => commands.begin(command), forgetCommand: (turnId) => commands.forget(turnId), commandProviderTurnId: (turnId) => commands.providerTurnId(turnId), + commandJournalTurnId: (providerTurnId) => commands.journalTurnId(providerTurnId), cancelPrompt: (journalItemId) => prompts.cancel(journalItemId), resolvePrompt: (journalItemId) => prompts.resolve(journalItemId), flush: () => { diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index 13a687a72cdc..6389f7e49ca9 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -14,6 +14,7 @@ import type { } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { CodexJournalTranslationAdmission } from './codex-structured-journal-translation' import { dispatchCodexTurn, isCodexTurnOptionKey } from './codex-structured-turn-start' +import { codexLiveWork } from './codex-structured-turn-open-wait' import { agentSessionFailureFact, providerDiagnosticOf } from '../../shared/agent-session-failure' import { supportsCodexStructuredLocation } from './codex-structured-location-support' import { CodexStructuredSessionTeardown } from './codex-structured-session-teardown' @@ -175,6 +176,11 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap ) => this.sessions.has(sessionId) ? { supportsTaskStop: false, supportsStopAll: false } : undefined + liveWork: NonNullable = (sessionId) => { + const session = this.sessions.get(sessionId) + return session ? codexLiveWork(session) : undefined + } + bindPromptItemId = ( sessionId: string, journalItemId: string, 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..c384764a720f 100644 --- a/src/main/codex/codex-structured-turn-open-wait.test.ts +++ b/src/main/codex/codex-structured-turn-open-wait.test.ts @@ -145,3 +145,42 @@ describe("a no-turn Stop in the window between Codex's answer and its turn openi expect(rig.interrupts()).toEqual([]) }) }) + +// What a host stop reads, in its own step and with no wait, to judge whether it ends work. +describe("Codex's live view of what its child has in flight", () => { + const liveWork = (rig: Rig) => rig.adapter.liveWork('session-1') + + it('is nothing at rest', async () => { + const rig = await codexTurnLifecycleRig() + + expect(liveWork(rig)).toBeUndefined() + }) + + it('is a send with no turn yet from the answer until its turn opens, then that turn', async () => { + const rig = await codexTurnLifecycleRig() + await answeredColdSend(rig) + + expect(liveWork(rig)).toEqual({ turnId: null }) + rig.turns.start() + expect(liveWork(rig)).toEqual({ turnId: 'turn-1' }) + }) + + it('is nothing once the turn ends', async () => { + const rig = await codexTurnLifecycleRig() + await answeredColdSend(rig) + rig.turns.start() + + rig.turns.end('completed') + + expect(liveWork(rig)).toBeUndefined() + }) + + it('is nothing once the answered turn ends without opening', async () => { + const rig = await codexTurnLifecycleRig() + await answeredColdSend(rig) + + rig.turns.end('interrupted') + + expect(liveWork(rig)).toBeUndefined() + }) +}) diff --git a/src/main/codex/codex-structured-turn-open-wait.ts b/src/main/codex/codex-structured-turn-open-wait.ts index 9f30b292ff0c..04ef7ae69de5 100644 --- a/src/main/codex/codex-structured-turn-open-wait.ts +++ b/src/main/codex/codex-structured-turn-open-wait.ts @@ -1,7 +1,9 @@ // 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 { StructuredAgentSessionLiveWork } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { CodexDispatchEchoes } from './codex-structured-dispatch-echo' +import type { CodexJournalTranslator } from './codex-structured-journal-contracts' import { codexThreadStoppedRunning, readCodexThreadId, @@ -61,6 +63,27 @@ export function createCodexTurnOpenWaits(): CodexTurnOpenWaits { } } +/** What Codex has in flight (`liveWork`), with no wait: the turn it reports running, by the journal + * id of the command it carries when it carries one; or none yet for a send it has not answered, or + * answered into a turn that has not opened (`codexRunningOrOpeningTurn` waits that one out). */ +export function codexLiveWork(session: { + threadId: string + activeTurnIds?: ReadonlySet + dispatchPending?: boolean + dispatchEchoes: Pick + translator: Pick | null +}): StructuredAgentSessionLiveWork | undefined { + const open = session.activeTurnIds ?? new Set() + const running = [...open].at(-1) + if (running) { + return { turnId: session.translator?.commandJournalTurnId(running) ?? running } + } + return session.dispatchPending === true || + session.dispatchEchoes.answeredUnopenedTurn(session.threadId, open) !== null + ? { turnId: null } + : undefined +} + /** * 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 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..cdcf7048bc6f 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 @@ -2,11 +2,7 @@ // the turn-end rule reads (`journal-stop-turn-end.ts`), and whether a person's still decides. import type { JournalReducerState } from './journal-reducer' -import { - latestAcceptedSendUnopened, - personStopDecidesTurn, - type JournalLatestStop -} from './journal-stop-turn-end' +import { personStopDecidesTurn, type JournalLatestStop } from './journal-stop-turn-end' export class JournalStopMarks { constructor(private readonly deps: { state: () => JournalReducerState }) {} @@ -15,11 +11,6 @@ export class JournalStopMarks { return this.deps.state().queuePauseMarks.latestStop } - /** `latestAcceptedSendUnopened`: the latest accepted send's turn row may still be on its way. */ - latestAcceptedSendUnopened(): boolean { - return latestAcceptedSendUnopened(this.deps.state()) - } - /** `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) 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..1eb6f760bf7e 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 @@ -54,10 +54,10 @@ function openingSubmission( : undefined } -/** Whether the latest send the agent accepted to open a turn has opened none the journal holds: its - * turn's row may still be on its way. A send delivered into a running turn (a steer, a fold) +/** The latest send the agent accepted to open a turn, while the journal holds no turn it opened: + * that turn's row may still be on its way. A send delivered into a running turn (a steer, a fold) * opens none, and its item carries that turn's scope (`placeHandedOverMessage`). */ -export function latestAcceptedSendUnopened(state: TurnEndState): boolean { +function latestAcceptedSendUnopened(state: TurnEndState): AgentJournalSubmission | undefined { let latest: { submission: AgentJournalSubmission; sequence: number } | undefined for (const submission of state.submissions.values()) { const item = state.items.get(agentJournalSubmissionKey(submission.clientMessageId)) @@ -71,16 +71,16 @@ export function latestAcceptedSendUnopened(state: TurnEndState): boolean { } } if (!latest) { - return false + return undefined } for (const item of state.items.values()) { if ( openingSubmission(state, readAgentJournalTurn(item.body)?.userItemId) === latest.submission ) { - return false + return undefined } } - return true + return latest.submission } /** A send a Stop that named no turn stopped: one already handed to the agent at the Stop's @@ -136,14 +136,18 @@ function stopEndsTurnAsCancellation( ) } -/** 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) => +/** With no turn row running, the work in flight is the person's Stop's: every send still + * unanswered, and one accepted whose turn's row has yet to land, is one it stopped. */ +function sendsInFlightAreStopTargets(state: TurnEndState, stop: JournalLatestStop): boolean { + const inFlight = [...state.submissions.values()].filter((submission) => isUnansweredStructuredAgentSessionDispatch(submission) ) + const unopened = latestAcceptedSendUnopened(state) + if (unopened) { + inFlight.push(unopened) + } return ( - unanswered.length > 0 && unanswered.every((submission) => isStopTarget(state, stop, submission)) + inFlight.length > 0 && inFlight.every((submission) => isStopTarget(state, stop, submission)) ) } @@ -165,7 +169,7 @@ export function personStopDecidesTurn( return false } if (turnId === null) { - return stop.event.turnId === undefined && unansweredSendsAreStopTargets(state, stop) + return stop.event.turnId === undefined && sendsInFlightAreStopTargets(state, stop) } const turn = [...state.items.values()] .map((item) => readAgentJournalTurn(item.body)) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts index e22dd1bd2890..22e297758c32 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.test.ts @@ -189,6 +189,17 @@ describe('StructuredAgentSessionAdapterRouter.stopEndsSession', () => { expect(router.stopEndsSession('session-1')).toBe(true) }) + it("reports the live owner's work in flight, and none with no live owner", async () => { + const claude = adapterOf(vi.fn(async () => true)) + claude.liveWork = () => ({ turnId: 'turn-1' }) + const codex = adapterOf(vi.fn(async () => false)) + const router = new StructuredAgentSessionAdapterRouter({ claude, codex }, async () => {}) + + expect(router.liveWork('session-1')).toBeUndefined() + await router.acquire({ identity: claudeIdentity('session-1'), fence: 1, spawnToken: 'spawn-1' }) + expect(router.liveWork('session-1')).toEqual({ turnId: 'turn-1' }) + }) + it("waits on the session's live owner to wind the Stop down, and on nothing with none", async () => { const claude = adapterOf(vi.fn(async () => true)) claude.awaitStoppedRequestEnd = vi.fn(async () => undefined) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts index b5d431c859a1..bb92622b782b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter-router.ts @@ -116,6 +116,9 @@ export class StructuredAgentSessionAdapterRouter implements StructuredAgentSessi holdsDispatch = (sessionId: string): boolean => this.liveOwnerOrNull(sessionId)?.holdsDispatch?.(sessionId) ?? false + liveWork: NonNullable = (sessionId) => + this.liveOwnerOrNull(sessionId)?.liveWork?.(sessionId) + stopEndsSession = (sessionId: string): boolean => this.liveOwnerOrNull(sessionId)?.stopEndsSession?.(sessionId) ?? false diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index 318bada1fe52..1d61a3dec0b9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -240,6 +240,8 @@ export type StructuredAgentSessionSetOptionInput = { fence: number } +export type StructuredAgentSessionLiveWork = { turnId: string | null } + export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop & { /** Provider-aware capability check for hosts that route more than one adapter. */ supportsCreate?(location: AgentSessionExecutionLocation, agent: string): boolean @@ -335,6 +337,10 @@ export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop & /** The provider reported taking a send it has neither answered nor ended, as a queued follow-up * or a silent retry does. Derived from the live child; false with none. */ holdsDispatch?(sessionId: string): boolean + /** What the live child has in flight, read off the provider's own frames as they arrive, ahead + * of the journal rows they become: the turn it has open, by its journal id, or null for a send + * it has not answered with no turn open yet. Undefined with nothing in flight. */ + liveWork?(sessionId: string): StructuredAgentSessionLiveWork | undefined /** The `/` surface the running provider reports for itself. Undefined when the * provider never reports one, which is what keeps the client on its catalog. */ readCommands?(sessionId: string): AgentSessionSlashCommand[] | undefined diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-close-verdict.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-close-verdict.test.ts index 33ef8080c030..27a8861351d0 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-close-verdict.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-close-verdict.test.ts @@ -40,18 +40,22 @@ let providerEnd: | null /** The provider already saw its own exit when the close arrived. */ let exitObservedFirst: boolean +/** The provider's live view of its child: the cut turn open, whatever rows the journal holds yet. */ +let providerInFlight: boolean let closeCalls = 0 beforeEach(() => { const state = hostTestState() providerEnd = 'mapped' exitObservedFirst = false + providerInFlight = true closeCalls = 0 host = new StructuredAgentSessionHost({ logger: createStructuredAgentSessionLogger(), store: state.store, adapter: { ...adapter(), + liveWork: () => (providerInFlight ? { turnId: 'cut-turn' } : undefined), closeSession: async () => { closeCalls += 1 const events = state.acquire.mock.calls.at(-1)?.[0].events @@ -194,6 +198,7 @@ describe('a turn cut short by closing its provider', () => { it('leaves a turn cut off before the close as news', async () => { providerEnd = null + providerInFlight = false await attach() const events = hostTestState().acquire.mock.calls[0]?.[0].events // An earlier death the user already saw as Interrupted, then closed. 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..3e78d7c2adf1 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 @@ -37,7 +37,7 @@ import type { StructuredAgentSessionStopCause } from './structured-agent-session export type { StructuredAgentSessionStopEnding } from './structured-agent-session-host-stop-event' import { recordStopEvent, - stopEndsWork, + workStopEnds, type StructuredAgentSessionStopEnding } from './structured-agent-session-host-stop-event' @@ -138,8 +138,9 @@ export async function stopStructuredAgentSessionAgentUnderSerialize( return } // Judged before the kill: a stop that ends nothing writes nothing. - const recorded = (await stopEndsWork(context, sessionId, session, ending)) - ? recordStopEvent(context, sessionId, session, ending) + const stopped = workStopEnds(context, sessionId, session, ending) + const recorded = stopped + ? recordStopEvent(context, sessionId, session, stopped) : Promise.resolve() // 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 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..370ecc14b041 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 @@ -1,8 +1,7 @@ // The Stop event a host stop writes (`JournalStopEvent`): whether it ends work its event must // record, and the write itself, issued before the kill. -import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/structured-agent-session-main-agent-working' -import { withTimeout } from '../../../shared/promise-timeout-fallback' +import { hasUnansweredStructuredAgentSessionDispatch } from '../../../shared/structured-agent-session-unanswered-dispatch' import type { StructuredAgentSessionStopCause } from './structured-agent-session-adapter' import type { StructuredAgentSessionLifetimeContext } from './structured-agent-session-host-lifetime' import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' @@ -26,48 +25,48 @@ export type StructuredAgentSessionStopEnding = ( retry?: true } -/** How long a host stop waits for the session's sink before it judges whether the stop ends work. */ -const STOP_EVENT_DRAIN_TIMEOUT_MS = 1_000 +/** The work a host stop ends, as its event records it: why, and the turn it cuts (none for a send). */ +export type StructuredAgentSessionStoppedWork = { + reason: Exclude + turnId: string | null +} /** - * Whether this stop ends work its event must record: a running turn or an unanswered send, a start's - * own included, read once the sink drained what the provider already said. A start that carries - * no send ends nothing. A person's Stop wrote its own event, and quit, the idle sweep's rest and a - * retry of a stop already owed write none. + * The work this stop ends, which its event must record, read from the host's live view in this + * serialized step, with no wait: a turn or send the provider's child has in flight, or a send the + * journal holds unanswered, a start's own included. Null when it ends none: a start that carries no + * send, or work a person's Stop is already ending. A person's Stop wrote its own event, and quit, + * the idle sweep's rest and a retry of a stop already owed write none. */ -export async function stopEndsWork( +export function workStopEnds( context: StructuredAgentSessionLifetimeContext, sessionId: string, session: StructuredAgentSessionHostSession, ending: StructuredAgentSessionStopEnding -): Promise { +): StructuredAgentSessionStoppedWork | null { const { child, journal } = session if ('recorded' in ending || ending.quit || ending.resting || ending.retry || !child) { - return false + return null + } + const { adapter } = context.deps + const journalTurnId = journal.activeTurnId() + // The child's frames lead the rows they become; a provider with no live view has only the rows. + const live = adapter.liveWork + ? adapter.liveWork(sessionId) + : journalTurnId === null + ? undefined + : { turnId: journalTurnId } + if ( + live === undefined && + !hasUnansweredStructuredAgentSessionDispatch(journal.submissions(), child.fence) + ) { + return null } - // A failed drain has nothing more to deliver, so the journal's read as it stands holds. One - // still running past its bound may hold the turn row of a send already accepted: that reads - // working. - const drain = await withTimeout( - context.runtimeState.flushEventSink(sessionId).then( - () => 'drained' as const, - () => 'failed' as const - ), - STOP_EVENT_DRAIN_TIMEOUT_MS, - 'slow' as const - ) - const working = - (drain === 'slow' && journal.stopMarks.latestAcceptedSendUnopened()) || - isStructuredAgentSessionMainAgentWorking( - journal.activeTurnId(), - 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' || !journal.stopMarks.personStopDecides(journal.activeTurnId())) - ) + if (ending.cause !== 'user-close' && journal.stopMarks.personStopDecides(journalTurnId)) { + return null + } + return { reason: ending.cause, turnId: journalTurnId ?? live?.turnId ?? null } } /** Writes this stop's event (`JournalStopEvent`). Issued before the kill and never awaited by it: @@ -76,15 +75,11 @@ export function recordStopEvent( context: StructuredAgentSessionLifetimeContext, sessionId: string, session: StructuredAgentSessionHostSession, - ending: StructuredAgentSessionStopEnding + { reason, turnId }: StructuredAgentSessionStoppedWork ): Promise { - if ('recorded' in ending) { - return Promise.resolve() - } - const turnId = session.journal.activeTurnId() return session.journal .appendStopEvent( - { reason: ending.cause, ...(turnId !== null ? { turnId } : {}) }, + { reason, ...(turnId !== null ? { turnId } : {}) }, structuredAgentSessionConversationFence(context.deps.store, sessionId) ) .then( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-live-view.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-live-view.test.ts new file mode 100644 index 000000000000..5ed3b89c4a39 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-stop-live-view.test.ts @@ -0,0 +1,215 @@ +// A host stop (a close, an eviction, the host stopping a start) decides whether it ends work from +// the host's live view in its own serialized step, with no wait: the provider child's own report of +// what it has in flight, and the sends the journal holds unanswered. The journal's turn rows trail +// the child's frames, so they never decide it alone. + +import { afterEach, describe, expect, it } from 'vitest' +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import type { JournalStopEvent } from '../agent-session-journal/journal-row-schema' +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()) + +type HostStopKind = 'user-close' | 'evict' | 'host-stop' + +const KINDS: HostStopKind[] = ['user-close', 'evict', 'host-stop'] + +/** A rig whose provider reports its child's work, and the host stop of `kind` on it. The host stops + * a start that never landed from the idle sweep, so that rig's children all stay starting. */ +async function rigFor(kind: HostStopKind): Promise<() => Promise> { + if (kind === 'host-stop') { + rig = await createQueuedMessageTestRig({ + liveView: true, + starting: true, + restartable: true, + idleSweep: { idleMs: 0, intervalMs: 60 * 60 * 1000 } + }) + return () => rig.host.collaboratorsForTests().lifetime.idleSweep.tick() + } + rig = await createQueuedMessageTestRig({ liveView: true }) + return () => rig.host.close(HOST_TEST_SESSION, kind) +} + +function journal() { + const open = rig.host.collaboratorsForTests().sessions.get(HOST_TEST_SESSION)?.journal + if (!open) { + throw new Error('expected the conversation open') + } + return open +} + +function stopEvents(): JournalStopEvent[] { + 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) => + row.kind === 'tombstone' && row.stopEvent ? [row.stopEvent] : [] + ) +} + +/** The Stop events as the provider's close finds them, or null when no close ran. */ +function stopEventsAtClose(): { events: JournalStopEvent[] | null } { + const seen: { events: JournalStopEvent[] | null } = { events: null } + rig.closeSession.mockImplementationOnce(async () => { + seen.events = stopEvents() + return true + }) + return seen +} + +/** A send the provider accepted, with its turn's running row in the journal or still on its way. */ +async function acceptedSend(turnRow: boolean): Promise { + const sent = await rig.workingSend() + await rig.settleAccepted(sent, 'turn-1') + if (turnRow) { + await journal().appendItem( + { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal: 999 }, + { + kind: 'turn', + turnId: 'turn-1', + state: 'running', + startedAt: 1, + userItemId: agentJournalSubmissionKey(sent) + }, + { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + } + return sent +} + +describe.each(KINDS)('a host stop (%s) decides from the live view, with no wait', (kind) => { + it('writes its event, naming the turn, for a turn the provider reports running', async () => { + const hostStop = await rigFor(kind) + await acceptedSend(true) + rig.liveWork.mockReturnValue({ turnId: 'turn-1' }) + const atClose = stopEventsAtClose() + + await hostStop() + + expect(atClose.events).toEqual([{ reason: kind, turnId: 'turn-1', at: expect.any(Number) }]) + }) + + // The acceptance lands straight in the journal; the turn row trails it through the sink. + it('writes its event, naming the turn, for a turn opened whose row has not landed', async () => { + const hostStop = await rigFor(kind) + await acceptedSend(false) + expect(journal().activeTurnId()).toBeNull() + rig.liveWork.mockReturnValue({ turnId: 'turn-1' }) + const atClose = stopEventsAtClose() + + await hostStop() + + expect(atClose.events).toEqual([{ reason: kind, turnId: 'turn-1', at: expect.any(Number) }]) + }) + + // Codex answered the send into a turn it has not opened yet. + it('writes its event, naming no turn, for an accepted send whose turn has not opened', async () => { + const hostStop = await rigFor(kind) + await acceptedSend(false) + rig.liveWork.mockReturnValue({ turnId: null }) + const atClose = stopEventsAtClose() + + await hostStop() + + expect(atClose.events).toEqual([{ reason: kind, at: expect.any(Number) }]) + }) + + // The provider ended the turn; its end row trails through the sink, so the journal still reads + // it running. Nothing is in flight, so this stop ends nothing. + it("writes nothing at rest, while the turn's end row is still on its way", async () => { + const hostStop = await rigFor(kind) + await acceptedSend(true) + expect(journal().activeTurnId()).toBe('turn-1') + const atClose = stopEventsAtClose() + + await hostStop() + + expect(rig.liveWork).toHaveBeenCalled() + expect(atClose.events).toEqual([]) + }) + + it('writes nothing at rest', async () => { + const hostStop = await rigFor(kind) + const atClose = stopEventsAtClose() + + await hostStop() + + expect(atClose.events).toEqual([]) + }) + + // A send whose reply was lost may still run: loss of contact is no proof it ended. + it('writes its event for a send still unanswered, whatever the provider reports', async () => { + const hostStop = await rigFor(kind) + await rig.workingSend() + const atClose = stopEventsAtClose() + + await hostStop() + + expect(atClose.events).toEqual([{ reason: kind, at: expect.any(Number) }]) + }) +}) + +// A person's own close always writes: it is theirs. Any other host stop keeps their Stop's reason. +// A person's Stop of a start ends that child, so the host stops only a later start, which a send +// after the Stop made. +describe.each(['user-close', 'evict'] as const)( + "a person's Stop holds against a later close (%s)", + (kind) => { + const later = kind === 'user-close' ? ['user-stop', kind] : ['user-stop'] + + it('of the turn it named, still winding down', async () => { + const hostStop = await rigFor(kind) + await acceptedSend(true) + rig.liveWork.mockReturnValue({ turnId: 'turn-1' }) + expect(await rig.stop()).toMatchObject({ ok: true }) + expect(stopEvents()).toEqual([expect.objectContaining({ turnId: 'turn-1' })]) + const atClose = stopEventsAtClose() + + await hostStop() + + expect(atClose.events?.map((event) => event.reason)).toEqual(later) + }) + + // Pressed before the turn opened; the send it stopped is then accepted, its turn row still on + // its way: the work in flight is still the Stop's. + it('naming no turn, over the send it stopped whose turn row has not landed', async () => { + const hostStop = await rigFor(kind) + const sent = await rig.workingSend() + rig.liveWork.mockReturnValue({ turnId: 'turn-1' }) + expect(await rig.stop()).toMatchObject({ ok: true }) + expect(stopEvents()).toEqual([expect.not.objectContaining({ turnId: expect.anything() })]) + await rig.settleAccepted(sent, 'turn-1') + expect(journal().activeTurnId()).toBeNull() + const atClose = stopEventsAtClose() + + await hostStop() + + expect(atClose.events?.map((event) => event.reason)).toEqual(later) + }) + } +) + +describe.each(KINDS)("a person's Stop does not hold a later host stop (%s)", (kind) => { + it('over a send made after it', async () => { + const hostStop = await rigFor(kind) + const stopped = await rig.workingSend() + expect(await rig.stop()).toMatchObject({ ok: true }) + await rig.settleAccepted(stopped, 'stopped') + const after = await rig.workingSend() + await rig.settleAccepted(after, 'turn-2') + rig.liveWork.mockReturnValue({ turnId: 'turn-2' }) + const atClose = stopEventsAtClose() + + await hostStop() + + expect(atClose.events?.map((event) => event.reason)).toEqual(['user-stop', kind]) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts index 54cfbb470128..135e37717b81 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-message-rig.test-fixture.ts @@ -46,6 +46,9 @@ export async function createQueuedMessageTestRig( idleSweep?: { idleMs: number; intervalMs: number } /** The provider's Stop ends its child, as Claude's does. */ stopEndsSession?: true + /** The provider reports what its child has in flight (`liveWork`), as both real ones do; + * without it the host reads the journal's rows alone. */ + liveView?: true } = {} ) { const root = await mkdtemp(join(tmpdir(), 'orca-queued-messages-')) @@ -69,6 +72,9 @@ export async function createQueuedMessageTestRig( const closeSession: Mock> = vi.fn( async () => true ) + const liveWork: Mock> = vi.fn( + () => undefined + ) let events: StructuredAgentSessionEventSink | undefined const store = await openTestAgentSessionRecordStore(root) const makeHost = () => @@ -106,6 +112,7 @@ export async function createQueuedMessageTestRig( compact, cancelTurn, ...(options.stopEndsSession ? { stopEndsSession: () => true } : {}), + ...(options.liveView ? { liveWork } : {}), answerPrompt: vi.fn(async () => undefined), setOption: vi.fn(async () => undefined) }, @@ -313,6 +320,7 @@ export async function createQueuedMessageTestRig( dispatch, cancelTurn, closeSession, + liveWork, awaitStarted, compact, finishCompact, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-entries.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-entries.test.ts index dd0acf82b371..2e26adff4ac8 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-entries.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-entries.test.ts @@ -3,9 +3,8 @@ // nothing, quit writes nothing (its resume marker records why), and any later Stop event ends a // person's Stop pause. -import { afterEach, describe, expect, it, vi } from 'vitest' +import { afterEach, describe, expect, it } from 'vitest' import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' -import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import type { JournalStopEvent } from '../agent-session-journal/journal-row-schema' import { HOST_TEST_SESSION } from './structured-agent-session-host-test-data' @@ -160,132 +159,6 @@ describe('every Stop entry writes its event, with its reason, before it ends the }) }) - // A drain still running after its bound may hold the turn's row: the agent reads working. - it("writes a person's close while the running turn's row waits behind a slow sink", async () => { - rig = await createQueuedMessageTestRig() - const sent = await rig.workingSend() - const open = journal() - const append = open.appendItem.bind(open) - let held = false - vi.spyOn(open, 'appendItem').mockImplementation(async (...args: Parameters) => { - if (!held && args[1].kind === 'turn' && args[1].state === 'running') { - held = true - await new Promise((resolve) => setTimeout(resolve, 1_500)) - } - return append(...args) - }) - rig.host['runtimeState'].eventSinkFor(HOST_TEST_SESSION).sink.appendItem( - { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal: 999 }, - { - kind: 'turn', - turnId: 'turn-1', - state: 'running', - startedAt: Date.now(), - userItemId: agentJournalSubmissionKey(sent) - }, - { turnScope: AGENT_JOURNAL_THREAD_SCOPE } - ) - // The echo's acceptance lands straight in the journal, ahead of the turn row. - await rig.settleAccepted(sent, 'turn-1') - const atClose = stopEventsAtClose() - - await rig.host.close(HOST_TEST_SESSION, 'user-close') - - expect(atClose.events?.map((event) => event.reason)).toEqual(['user-close']) - }, 20_000) - - // A slow drain counts as working only for a send accepted with no turn row yet. - // A steer delivered into the running turn opens no turn of its own, so it is never owed one. - it.each([ - ['', false], - [', its last send a steer into the stopped turn', true] - ])( - 'writes nothing when the drain runs long as it evicts a chat at rest%s', - async (_label, steered) => { - rig = await createQueuedMessageTestRig() - const working = await rig.workingSend() - const opener = agentJournalSubmissionKey(working) - const identity = { - provider: 'codex' as const, - threadId: 'thread-1', - turnId: 'turn-1', - ordinal: 999 - } - const scope = { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } - await journal().appendItem( - identity, - { kind: 'turn', turnId: 'turn-1', state: 'running', startedAt: 1, userItemId: opener }, - scope - ) - if (steered) { - const steer = rig.send('steer the running turn') - await steer.result - await eventually(async () => - expect((await rig.submission(steer.id))?.handedOverAt).toBeDefined() - ) - await rig.settleAccepted(steer.id, 'turn-1') - expect( - journal() - .snapshot() - .items.find((item) => item.itemId === agentJournalSubmissionKey(steer.id))?.turnScope - ).toMatchObject({ kind: 'turn' }) - } - await queuedDraft('queued behind the turn') - expect(await rig.stop()).toMatchObject({ ok: true }) - await rig.settleAccepted(working, 'turn-1') - await journal().appendItem( - identity, - { - kind: 'turn', - turnId: 'turn-1', - state: 'interrupted', - completedAt: Date.now(), - userItemId: opener - }, - scope - ) - expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) - const sink = rig.host['runtimeState'].eventSinkFor(HOST_TEST_SESSION) - const drained = sink.drained.bind(sink) - vi.spyOn(sink, 'drained') - .mockImplementationOnce( - () => new Promise((resolve) => setTimeout(() => resolve({ ok: true }), 1_500)) - ) - .mockImplementation(drained) - const atClose = stopEventsAtClose() - - await rig.host.close(HOST_TEST_SESSION, 'evict') - - expect(atClose.events?.map((event) => event.reason)).toEqual(['user-stop']) - expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) - }, - 20_000 - ) - - // The drain is best effort: when it fails, the journal as it stands says the agent rests. - it('writes nothing when the drain fails as it evicts a chat at rest', async () => { - rig = await createQueuedMessageTestRig() - const working = await runningTurn() - expect(await rig.stop()).toMatchObject({ ok: true }) - await rig.settleAccepted(working, 'stopped') - await journal().appendItem( - { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal: 999 }, - { kind: 'turn', turnId: 'turn-1', state: 'interrupted', completedAt: Date.now() }, - { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } - ) - expect(journal().activeTurnId()).toBeNull() - const sink = rig.host['runtimeState'].eventSinkFor(HOST_TEST_SESSION) - const drained = sink.drained.bind(sink) - vi.spyOn(sink, 'drained') - .mockResolvedValueOnce({ ok: false, error: new Error('drain lost once') }) - .mockImplementation(drained) - const atClose = stopEventsAtClose() - - await rig.host.close(HOST_TEST_SESSION, 'evict') - - expect(atClose.events?.map((event) => event.reason)).toEqual(['user-stop']) - }) - // The idle sweep finishes a stop whose exit was unproven: the same stop, so its event stands alone. it.each([['user-close' as const], ['evict' as const]])( 'writes one event for a close (%s) whose exit was unproven, and none for its retry', diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts index 1f8fc80c863d..35dab4b01d29 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts @@ -145,14 +145,12 @@ function waitOutSeveralSweeps(): Promise { return new Promise((resolve) => setTimeout(resolve, SWEEP_MS * 20)) } -/** Fails the next eviction at `drain-published`, which leaves the session indexed for a retry. The - * stop drains once before it, to judge whether it ends work. */ +/** Fails the next eviction at `drain-published`, which leaves the session indexed for a retry. */ function failEvictionDrain(): void { - const sink = host['runtimeState'].eventSinkFor(SESSION) - const drained = sink.drained.bind(sink) - vi.spyOn(sink, 'drained') - .mockImplementationOnce(drained) - .mockResolvedValueOnce({ ok: false, error: new Error('drain barrier lost') }) + vi.spyOn(host['runtimeState'].eventSinkFor(SESSION), 'drained').mockResolvedValueOnce({ + ok: false, + error: new Error('drain barrier lost') + }) } /** The submissions as they stood when the session was forgotten; its journal is gone after that. */ From 3c286da0aff5c2c0ef546b1a944043bdf5862ca8 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 1 Oct 2026 19:55:40 -0700 Subject: [PATCH 2/2] test(native-chat): Codex's live view names a command's turn by its journal id --- .../codex-structured-turn-open-wait.test.ts | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) 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 c384764a720f..e7c4ae8d5a64 100644 --- a/src/main/codex/codex-structured-turn-open-wait.test.ts +++ b/src/main/codex/codex-structured-turn-open-wait.test.ts @@ -9,7 +9,7 @@ import { codexTurnLifecycleRig, settledWithin } from './codex-structured-dispatch-test-support' -import { CODEX_TURN_OPEN_WAIT_MS } from './codex-structured-turn-open-wait' +import { CODEX_TURN_OPEN_WAIT_MS, codexLiveWork } from './codex-structured-turn-open-wait' type Rig = Awaited> @@ -165,6 +165,20 @@ describe("Codex's live view of what its child has in flight", () => { expect(liveWork(rig)).toEqual({ turnId: 'turn-1' }) }) + // Codex carries a command out as a turn of its own, whose rows join the command's turn. + it("names a turn carrying a command by the command's journal turn", () => { + const work = codexLiveWork({ + threadId: CODEX_TEST_THREAD_ID, + activeTurnIds: new Set(['provider-turn']), + dispatchEchoes: { answeredUnopenedTurn: () => null }, + translator: { + commandJournalTurnId: (turnId) => (turnId === 'provider-turn' ? 'command-turn' : turnId) + } + }) + + expect(work).toEqual({ turnId: 'command-turn' }) + }) + it('is nothing once the turn ends', async () => { const rig = await codexTurnLifecycleRig() await answeredColdSend(rig)