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..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> @@ -145,3 +145,56 @@ 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' }) + }) + + // 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) + 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. */