diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts index 8fc325a57e4..91792f5875b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts @@ -392,6 +392,15 @@ describe('a start the chat needed and did not get', () => { describe('an attach that fails after indexing its child', () => { it('leaves no child behind, so the next send starts one and is delivered', async () => { await host.close(SESSION) + const owned: boolean[] = [] + host.subscribeStatus({ + id: 'list-1', + emit: (event) => { + if (event.type === 'status') { + owned.push(event.session.hostExecutionOwned === true) + } + } + }) const first = await accept('hello') // The attach's own success record is the step after `onAttached` indexed the child. const record = vi.spyOn(store, 'recordOperationOutcome') @@ -403,6 +412,9 @@ describe('an attach that fails after indexing its child', () => { return AgentSessionRecordStore.prototype.recordOperationOutcome.call(store, input) }) await eventually(() => expect(submission(first)?.dispatchState).toBe('rejected')) + // Nothing was indexed, so nothing had to be taken back: no list ever showed a child. + expect(host['sessions'].get(SESSION)?.child).toBeNull() + expect(owned).not.toContain(true) const acquiresBefore = acquire.mock.calls.length const next = await accept('after the failure') diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-failure.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-failure.ts index 533f539466c..89fb4654f66 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-failure.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-failure.ts @@ -24,6 +24,7 @@ export async function settlePostAcquisitionAttachFailure( ? 'root-exit-observed' : 'exit-proven' } + input.onAcquisitionReleased?.(cause) try { await input.store.settleFailedPostAcquisitionAttachment({ sessionId: record.sessionId, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts index 3498b6f5d88..5390e3f0f1e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts @@ -69,6 +69,8 @@ export type AttachFlowInput = { beforeJournalOpen?: () => Promise | void /** The host's open conversation, which the attach adopts rather than opening a second one. */ openConversation?: (sessionId: string) => Promise + /** A failure after acquisition released the session's acquisition; `cause` is that failure. */ + onAcquisitionReleased?: (cause: unknown) => void } export async function performAttach( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts index 6aeb6bc46d5..c93dabd549d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts @@ -24,7 +24,12 @@ import { refuseAgentSessionMutation } from './structured-agent-session-mutation- import { retryPendingStructuredAgentSessionSettlement } from './structured-agent-session-settlement-retry' import { settleStaleSessionStateOnAcquire } from './structured-agent-session-stale-turn-verdict' import type { StructuredAgentSessionAttachContext } from './structured-agent-session-attach-context' -import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' +import type { StructuredAgentSessionProviderChild } from './structured-agent-session-host-types' +import { + endProviderChild, + indexProviderChild, + structuredAgentSessionConversationFence +} from './structured-agent-session-provider-child' import type { DeferredStructuredAgentSessionEventSink } from './structured-agent-session-event-sink' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { @@ -94,6 +99,10 @@ async function runAttach( ): Promise> { const sessionId = params.envelope.sessionId const recordPhase = options.recordPhase + // Readers of a conversation already open are re-baselined when this attach moves its fence. + const fenceBefore = context.sessions.has(sessionId) + ? structuredAgentSessionConversationFence(context.deps.store, sessionId) + : null if (options.admitRecoveryTicket && !options.admitRecoveryTicket()) { return refuseAgentSessionMutation({ code: 'agent_session_checkpoint_stale', @@ -114,9 +123,8 @@ async function runAttach( const settled = await withAgentSessionCreatePhase('settlement_retry', recordPhase, () => retryPendingStructuredAgentSessionSettlement({ deps: context.deps, - sessions: context.sessions, sessionId, - params, + openJournal: async () => (await context.openConversation(sessionId))?.journal ?? null, now: () => context.now() }) ) @@ -130,16 +138,15 @@ async function runAttach( context.runtimeState.probeOwner(sessionId) ) // A child this attach spawns writes through a sink this attempt owns. Only a successful - // attach makes it the session's; any other exit closes it with whatever the child queued. + // attach makes the child and its sink the session's; any other exit closes the sink with + // whatever the child queued, and leaves the conversation's child as it was. const attemptSink = context.runtimeState.mintEventSink(sessionId) - let attemptSinkAdopted = false - let indexed: StructuredAgentSessionHostSession | null = null - let attachedOk = false - const attached = stampFailedCreateOwnerVerdict( - context.deps.store, - callerKey, - params.envelope, - await performAttach({ + const attempt: { candidate: AttachCandidate | null; committed: boolean } = { + candidate: null, + committed: false + } + try { + const attached = await performAttach({ store: context.deps.store, adapter: context.deps.adapter, journalRoot: context.deps.journalRoot, @@ -163,8 +170,6 @@ async function runAttach( params, now: () => context.now(), recordPhase, - // A failed attach leaves the conversation open, with its queued messages, for the failure - // to be written into; only the attempt's own sink is discarded, in the `finally` below. openConversation: async (id) => { const conversation = await context.openConversation(id) if (!conversation) { @@ -172,15 +177,15 @@ async function runAttach( } return conversation.journal }, + // The cleanup released the acquisition, which for a re-attach is the live child itself. + onAcquisitionReleased: (cause) => endReleasedChild(context, sessionId, cause), onAttached: async (attached, acquisitionGeneration, acquiredOwner, providerChildPhase) => { - const fence = context.deps.store.getRecord(sessionId)?.lease.runtimeFence ?? 0 - const previous = context.sessions.get(sessionId) - const previousFence = previous?.fence + const fence = structuredAgentSessionConversationFence(context.deps.store, sessionId) + const current = context.sessions.get(sessionId)?.child ?? null // A re-attach to a live child keeps the sink that child already writes through. const eventSink = acquiredOwner ? attemptSink : (context.runtimeState.currentEventSink(sessionId) ?? attemptSink) - // The journal is the conversation's own open one, so a throw here leaves it indexed. if (acquiredOwner) { // Before the drain: the buffered events are the new child's, never a stale row's. await settleStaleSessionStateOnAcquire({ @@ -193,20 +198,15 @@ async function runAttach( await bindAndDrain(eventSink, attached.journal, fence, (activity) => context.subscribers.publish(sessionId, attached.journal, activity) ) - context.runtimeState.adoptEventSink(sessionId, eventSink) - attemptSinkAdopted = eventSink === attemptSink - indexed = { - journal: attached.journal, - params, - fence, - hasProviderChild: true, - // A re-attach to a live child keeps what that child already proved. - providerChildPhase: acquiredOwner - ? providerChildPhase - : (previous?.providerChildPhase ?? 'ready'), - acquisitionGeneration: acquisitionGeneration ?? previous?.acquisitionGeneration ?? null + attempt.candidate = { + sink: eventSink, + child: { + generation: acquisitionGeneration ?? current?.generation ?? null, + fence, + // A re-attach to a live child keeps what that child already proved. + phase: acquiredOwner ? providerChildPhase : (current?.phase ?? 'ready') + } } - context.sessions.set(sessionId, indexed) await recoverStructuredRewind( context.deps.store, sessionId, @@ -216,45 +216,57 @@ async function runAttach( context.now ) await recoverInterruptedCompaction(context.deps.store, sessionId, attached.journal, fence) - if (previousFence !== undefined && previousFence !== fence) { + if (fenceBefore !== null && fence !== fenceBefore) { context.subscribers.snapshot(sessionId, attached.journal) } else { context.subscribers.publish(sessionId, attached.journal) } } }) - .then((result) => { - attachedOk = result.ok - return result - }) - .finally(() => { - const releasedIndexedChild = - !attachedOk && indexed !== null && context.sessions.get(sessionId) === indexed - if (releasedIndexedChild && indexed) { - // The failure path released the child this attempt indexed; the conversation stays. - indexed.hasProviderChild = false - indexed.providerChildPhase = 'ready' - context.runtimeState.currentEventSink(sessionId)?.close() - context.runtimeState.discardEventSink(sessionId) - context.publishStatus?.(sessionId) - } - if (!attemptSinkAdopted || releasedIndexedChild) { - attemptSink.close() - followRecordFence(context, sessionId) - } - }) - ) - return attached + const { candidate } = attempt + const conversation = context.sessions.get(sessionId) + if (attached.ok && candidate && conversation) { + context.runtimeState.adoptEventSink(sessionId, candidate.sink) + attempt.committed = candidate.sink === attemptSink + indexProviderChild(conversation, candidate.child) + context.publishStatus?.(sessionId) + } + return stampFailedCreateOwnerVerdict(context.deps.store, callerKey, params.envelope, attached) + } finally { + if (!attempt.committed) { + attemptSink.close() + } + } } -/** A failed attempt still moved the record's fence when it reserved; the conversation it leaves - * open writes its failure row and rejections at that fence, not the one it opened at. */ -function followRecordFence(context: StructuredAgentSessionAttachContext, sessionId: string): void { - const conversation = context.sessions.get(sessionId) - const fence = context.deps.store.getRecord(sessionId)?.lease.runtimeFence - if (conversation && !conversation.hasProviderChild && fence !== undefined) { - conversation.fence = Math.max(conversation.fence, fence) +type AttachCandidate = { + child: StructuredAgentSessionProviderChild + sink: DeferredStructuredAgentSessionEventSink +} + +function endReleasedChild( + context: StructuredAgentSessionAttachContext, + sessionId: string, + cause: unknown +): void { + const session = context.sessions.get(sessionId) + const child = session?.child + if ( + !session || + !child || + !endProviderChild(session, { + generation: child.generation, + fence: child.fence, + cause: 'attach-failed', + reason: cause instanceof Error ? cause.message : String(cause), + duringStartup: child.phase === 'starting' + }) + ) { + return } + context.runtimeState.currentEventSink(sessionId)?.close() + context.runtimeState.discardEventSink(sessionId) + context.publishStatus?.(sessionId) } /** Binds the sink to the journal and waits for the barrier the host publishes diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-root-exit.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-root-exit.test.ts index c449711e745..2637db9356d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-root-exit.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-root-exit.test.ts @@ -103,10 +103,11 @@ describe('Claude root-exit eviction', () => { { journal, params, - fence, - hasProviderChild: true, - providerChildPhase: 'ready', - acquisitionGeneration: acquisition.acquisitionGeneration ?? null + child: { + generation: acquisition.acquisitionGeneration ?? null, + fence, + phase: 'ready' + } } ] ]) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-close-retry.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-close-retry.test.ts index 8c99be697c5..303ecbe92fc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-close-retry.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-close-retry.test.ts @@ -78,10 +78,7 @@ function hostSession(journal: AgentSessionJournal): StructuredAgentSessionHostSe return { journal, params: {} as StructuredAgentSessionHostSession['params'], - fence: 1, - hasProviderChild: false, - providerChildPhase: 'ready', - acquisitionGeneration: null + child: null } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-open.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-open.ts index 5b20c537cfc..eb850e1941f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-open.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-open.ts @@ -96,14 +96,7 @@ export async function openStructuredAgentSessionConversationJournal( deps.onEventSinkError?.({ sessionId, error }) } return { - session: { - journal: opened.journal, - params, - fence, - hasProviderChild: false, - providerChildPhase: 'ready', - acquisitionGeneration: null - }, + session: { journal: opened.journal, params, child: null }, reset: opened.recovery?.reset ?? null } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversations.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversations.test.ts index 202ffe50ae4..7014e394728 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversations.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversations.test.ts @@ -37,10 +37,7 @@ function session(journal: AgentSessionJournal) { return { journal, params: hostTestAttachParams(null), - fence: 0, - hasProviderChild: false, - providerChildPhase: 'ready' as const, - acquisitionGeneration: null + child: null } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts index 7bdc65119c6..9d524633959 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-delivery-loop.ts @@ -24,6 +24,8 @@ export type StructuredAgentSessionDeliveryLoopDeps = { serialize: (sessionId: string, task: () => Promise) => Promise /** Gives the session a provider child if it has none; for a caller inside `serialize`. */ ensureProviderChild: (sessionId: string) => Promise + /** The fence the conversation's own writes carry; see `structuredAgentSessionConversationFence`. */ + conversationFence: (sessionId: string) => number /** What the chat says when the session could not be made ready. */ startFailureText: (sessionId: string, cause: AgentSessionWireRefusal) => string onError: (sessionId: string, error: unknown) => void @@ -99,7 +101,7 @@ export class StructuredAgentSessionDeliveryLoop { return this.stop(sessionId) } await session.journal.rejectQueuedSubmissions( - session.fence, + this.deps.conversationFence(sessionId), DISPATCH_REJECTED_HOST_RESTARTED, // A handle closes only with nothing queued, so one an earlier handle wrote is a leftover. (submission) => session.journal.wroteBeforeOpen(submission.acceptedSequence) @@ -116,7 +118,8 @@ export class StructuredAgentSessionDeliveryLoop { return this.stop(sessionId) } // Re-derived here, not carried from the start: the child may have gone since. - if (!session.hasProviderChild) { + const { child } = session + if (!child) { return 'continue' } const next = oldestQueuedSubmission(session) @@ -127,9 +130,9 @@ export class StructuredAgentSessionDeliveryLoop { { sessionId, journal: session.journal, - fence: session.fence, + fence: child.fence, adapter: this.deps.adapter, - providerChildPhase: () => this.deps.sessions.get(sessionId)?.providerChildPhase + providerChildPhase: () => this.deps.sessions.get(sessionId)?.child?.phase }, next ) @@ -139,7 +142,10 @@ export class StructuredAgentSessionDeliveryLoop { private async fail(sessionId: string, text: string): Promise { const session = this.deps.sessions.get(sessionId) if (session) { - await recordStructuredAgentSessionStartFailure(session, text) + await recordStructuredAgentSessionStartFailure( + { journal: session.journal, fence: this.deps.conversationFence(sessionId) }, + text + ) } return this.stop(sessionId) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts index 89a49213088..ac3057b81e8 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-recovery.ts @@ -39,14 +39,13 @@ export class StructuredAgentSessionEventRecovery { this.sinkFailures.add(sessionId) void this.context .serialize(sessionId, async () => { - const session = this.context.sessions.get(sessionId) + const child = this.context.sessions.get(sessionId)?.child const stop = this.context.deps.adapter.forceCloseSession ?? this.context.deps.adapter.closeSession - if (!session?.hasProviderChild || !stop) { + if (!child || !stop) { return null } - const fence = session.fence - const acquisitionGeneration = session.acquisitionGeneration + const { fence, generation: acquisitionGeneration } = child const stopped = await stop(sessionId) if (!stopped || !acquisitionGeneration) { return null diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts index cb0cb2fb233..83f8ade9a4c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.test.ts @@ -30,8 +30,8 @@ function context(): StructuredAgentSessionEvictionContext & { order: string[] } return true }) } as unknown as StructuredAgentSessionEvictionContext['adapter'], - forget: vi.fn(async () => { - order.push('forget') + acknowledgeRelease: vi.fn(() => { + order.push('acknowledgeRelease') }), discardSink: vi.fn(() => order.push('discardSink')), settleWork: vi.fn(async () => { @@ -51,7 +51,7 @@ function runtimeState(): StructuredAgentSessionHostRuntimeState { } describe('structured agent session eviction', () => { - it('stops the child before it lets the sink go, then forgets the session', async () => { + it('stops the child before it lets the sink go, then acknowledges the release', async () => { const ctx = context() await evictStructuredAgentSession(ctx) expect(ctx.order).toEqual([ @@ -62,7 +62,7 @@ describe('structured agent session eviction', () => { 'close', 'discardSink', 'releaseLease', - 'forget' + 'acknowledgeRelease' ]) }) @@ -90,7 +90,7 @@ describe('structured agent session eviction', () => { 'close-sink', 'discard-sink', 'release-lease', - 'forget-session' + 'acknowledge-release' ]) }) @@ -114,7 +114,7 @@ describe('structured agent session eviction', () => { } }) - it('aborts after a failed drain barrier without unbinding or forgetting the session', async () => { + it('aborts after a failed drain barrier without unbinding or acknowledging the release', async () => { const ctx = context() ctx.eventSink.drained = vi.fn(async () => { ctx.order.push('drained') @@ -128,7 +128,7 @@ describe('structured agent session eviction', () => { expect(ctx.eventSink.close).not.toHaveBeenCalled() expect(ctx.discardSink).not.toHaveBeenCalled() expect(ctx.releaseLease).not.toHaveBeenCalled() - expect(ctx.forget).not.toHaveBeenCalled() + expect(ctx.acknowledgeRelease).not.toHaveBeenCalled() expect(ctx.order).toEqual(['closeSession', 'drained']) }) }) @@ -154,9 +154,9 @@ describe('rows the provider emits while closing', () => { return true } } as never, - forget: async () => {}, discardSink: () => state.discardEventSink(sessionId), - releaseLease: async () => {} + releaseLease: async () => {}, + acknowledgeRelease: () => {} }) expect(published).toEqual(['final-flush']) @@ -184,18 +184,18 @@ describe('a child that will not stop', () => { 'close', 'discardSink', 'releaseLease', - 'forget' + 'acknowledgeRelease' ]) }) - it('aborts without forgetting the session, so the next close is a real retry', async () => { + it('aborts without acknowledging the release, so the next stop is a real retry', async () => { const ctx = context() ctx.adapter.closeSession = vi.fn(async () => false) await expect(evictStructuredAgentSession(ctx)).rejects.toMatchObject({ step: 'stop-provider-child' }) - expect(ctx.forget).not.toHaveBeenCalled() + expect(ctx.acknowledgeRelease).not.toHaveBeenCalled() expect(ctx.discardSink).not.toHaveBeenCalled() expect(ctx.order).toEqual([]) }) @@ -210,7 +210,7 @@ describe('a child that will not stop', () => { StructuredAgentSessionEvictionError ) expect(ctx.eventSink.close).not.toHaveBeenCalled() - expect(ctx.forget).not.toHaveBeenCalled() + expect(ctx.acknowledgeRelease).not.toHaveBeenCalled() }) }) @@ -225,9 +225,9 @@ describe('eviction against the real sink cache', () => { sessionId, eventSink: state.eventSinkFor(sessionId), adapter: { closeSession: async () => true } as never, - forget: async () => {}, discardSink: () => state.discardEventSink(sessionId), - releaseLease: async () => {} + releaseLease: async () => {}, + acknowledgeRelease: () => {} }) const published: string[] = [] diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts index 951bbd2ab32..5ff00733dee 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-eviction.ts @@ -1,4 +1,4 @@ -// Releasing one structured session's resources. +// Stopping one structured session's provider child and handing its lease back. // // Teardown is a DATA list, not a method body, for the reason this file exists at all: the host // tracked which sessions were live in a map, and tore them down at three unrelated call sites @@ -30,10 +30,9 @@ export type StructuredAgentSessionEvictionContext = { hasProviderChild?: boolean eventSink: DeferredStructuredAgentSessionEventSink adapter: StructuredAgentSessionAdapter - /** Closes the session's journal handle and drops the map entry. Async and - * awaited: `close()` is ordered behind queued writes, and a delete that - * returns while the close is still queued leaves nothing to retry. */ - forget: () => Promise + /** Tells the adapter the released lease is done with, so it drops this child's route and index. + * The conversation stays: stopping the agent never closes its journal. */ + acknowledgeRelease: () => Promise | void /** Drops the cached sink so a later attach mints a fresh one. */ discardSink: () => void /** Fires right before the stop, while the child's turn and background roster are still live. A @@ -126,11 +125,11 @@ export const STRUCTURED_AGENT_SESSION_EVICTION_STEPS: readonly StructuredAgentSe // these two; eviction has to as well. { name: 'discard-sink', run: (context) => context.discardSink() }, // Why here and not last: the durable lease still names a process this host just stopped, and a - // record left claiming a live owner is one nothing can resume — the next surface to open the - // chat would find a session it may not acquire. Placed BEFORE forget so a release that cannot - // be written aborts while the session is still indexed, which is what makes the retry real. + // record left claiming a live owner is one nothing can resume — the next send would find a + // session it may not acquire. Placed BEFORE the acknowledgement so a release that cannot be + // written aborts while the adapter still routes the session, which is what makes the retry real. { name: 'release-lease', run: (context) => context.releaseLease() }, - { name: 'forget-session', run: (context) => context.forget() } + { name: 'acknowledge-release', run: (context) => context.acknowledgeRelease() } ] export class StructuredAgentSessionEvictionError extends Error { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-forget-status.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-forget-status.test.ts index 304cadf8f86..1ffc08a0249 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-forget-status.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-forget-status.test.ts @@ -135,10 +135,7 @@ async function workingSession(): Promise<{ accountHome: ownerRecord().accountHome, runtimeKind: 'native' }, - fence: 1, - hasProviderChild: true, - providerChildPhase: 'ready', - acquisitionGeneration: null + child: { generation: null, fence: 1, phase: 'ready' } } ] ]) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-holds.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-holds.test.ts index 68027f81887..9b6718e9957 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-holds.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-holds.test.ts @@ -296,7 +296,7 @@ describe('holds', () => { describe('the teardown deadline', () => { it('leaves the child loaded instead of forcing it, and keeps the session indexed', async () => { - const forget = vi.fn() + const acknowledgeRelease = vi.fn() const releaseLease = vi.fn(async () => {}) await expect( @@ -309,7 +309,7 @@ describe('the teardown deadline', () => { close: vi.fn() } as never, adapter: { closeSession: () => new Promise(() => {}) } as never, - forget, + acknowledgeRelease, discardSink: vi.fn(), releaseLease }, @@ -317,7 +317,7 @@ describe('the teardown deadline', () => { ) ).rejects.toMatchObject({ step: 'stop-provider-child' }) - expect(forget).not.toHaveBeenCalled() + expect(acknowledgeRelease).not.toHaveBeenCalled() expect(releaseLease).not.toHaveBeenCalled() }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts index 3065c464329..8b25364c881 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts @@ -15,6 +15,7 @@ import type { StructuredAgentSessionHostDeps, StructuredAgentSessionHostSession } from './structured-agent-session-host-types' +import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' import { structuredAgentSessionStartFailureText } from './structured-agent-session-send-preparation' import { settleInterruptedCompaction } from './structured-compaction-recovery' import { recoverStructuredRewind } from './structured-rewind-recovery' @@ -44,6 +45,8 @@ export function createStructuredAgentSessionConversationDelivery(input: { adapter: deps.adapter, serialize: input.serialize, ensureProviderChild: input.ensureProviderChild, + conversationFence: (sessionId) => + structuredAgentSessionConversationFence(deps.store, sessionId), startFailureText: (sessionId, cause) => structuredAgentSessionStartFailureText(deps.store.getRecord(sessionId), cause), onError: (sessionId, error) => deps.onEventSinkError?.({ sessionId, error }) @@ -82,9 +85,10 @@ async function settleInterruptedCommands( sessionId: string, session: StructuredAgentSessionHostSession ): Promise { + const fence = structuredAgentSessionConversationFence(deps.store, sessionId) try { - await settleInterruptedCompaction(deps.store, sessionId, session.journal, session.fence) - await recoverStructuredRewind(deps.store, sessionId, session.journal, session.fence) + await settleInterruptedCompaction(deps.store, sessionId, session.journal, fence) + await recoverStructuredRewind(deps.store, sessionId, session.journal, fence) } catch (error) { deps.onEventSinkError?.({ sessionId, error }) } 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 e3b284acb5b..91035098257 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 @@ -18,8 +18,13 @@ import { StructuredAgentSessionHolds } from './structured-agent-session-holds' import type { StructuredAgentSessionHostRuntimeState } from './structured-agent-session-host-runtime-state' import type { StructuredAgentSessionHostDeps, - StructuredAgentSessionHostSession + StructuredAgentSessionHostSession, + StructuredAgentSessionProviderChildIdentity } from './structured-agent-session-host-types' +import { + endProviderChild, + structuredAgentSessionConversationFence +} from './structured-agent-session-provider-child' import { releaseStoredStructuredAgentSessionOwner } from './structured-agent-session-lease-release' import { resumeHeldStructuredAgentSession } from './structured-agent-session-hold-resume' import type { StructuredAgentSessionAttachContext } from './structured-agent-session-attach-context' @@ -32,6 +37,9 @@ export type StructuredAgentSessionLifetimeContext = { now: () => number /** Drops the session's row from the agent-status store; see `forgetStructuredAgentSession`. */ forgetStatus: (sessionId: string) => void + /** Re-projects the session's status and fence after its agent stopped and the chat stays. */ + publishStatus?: (sessionId: string) => void + publishFence?: (sessionId: string) => void /** Quit-only snapshot taken immediately before the provider child is stopped. */ restartWitness?: { beforeStop: (sessionId: string) => void @@ -43,12 +51,19 @@ export type StructuredAgentSessionLifetimeContext = { * told, so a caller that only deletes strands a live-looking row no reader can ever decay. A * handle closes with nothing queued: what is still queued now will not be handed over. */ export async function forgetStructuredAgentSession( - context: StructuredAgentSessionLifetimeContext, + context: Pick & { + deps: Pick & { + store: Pick + } + }, sessionId: string ): Promise { const session = context.sessions.get(sessionId) await session?.journal - .rejectQueuedSubmissions(session.fence, DISPATCH_REJECTED_PROVIDER_CLOSED) + .rejectQueuedSubmissions( + structuredAgentSessionConversationFence(context.deps.store, sessionId), + DISPATCH_REJECTED_PROVIDER_CLOSED + ) // Best effort: the next open rejects a leftover itself. .catch((error: unknown) => context.deps.onEventSinkError?.({ sessionId, error })) await session?.journal.close() @@ -60,58 +75,75 @@ function hasProviderChild( context: StructuredAgentSessionLifetimeContext, sessionId: string ): boolean { - return context.sessions.get(sessionId)?.hasProviderChild === true + return (context.sessions.get(sessionId)?.child ?? null) !== null } /** The wind-down this host owes for the session's child. A live child always owes one, whatever a - * previous childless eviction recorded: a remembered `false` must never outrank the child in front - * of it. */ -function owesProviderChildWindDown(session: StructuredAgentSessionHostSession): boolean { - return session.hasProviderChild || session.owesProviderChildWindDown === true + * previous childless eviction recorded: a remembered tombstone must never outrank the child in + * front of it. */ +function owedProviderChildWindDown( + session: StructuredAgentSessionHostSession +): StructuredAgentSessionProviderChildIdentity | undefined { + return session.child + ? { generation: session.child.generation, fence: session.child.fence } + : session.owesProviderChildWindDown } -/** Runs the eviction steps under a deadline. A step that fails — or runs out of time — aborts the - * rest, which leaves the session indexed and the child loaded so the next close is a real retry. */ -export async function evictHeldStructuredAgentSession( +/** + * The agent goes to rest; the conversation stays. Runs the eviction steps under a deadline. A step + * that fails — or runs out of time — aborts the rest and leaves the wind-down owed, so the next + * stop is a real retry. `ending` is how the child's end is told: a Stop, or an eviction whose close + * forgets the conversation next. + */ +export async function stopStructuredAgentSessionAgentUnderSerialize( context: StructuredAgentSessionLifetimeContext, - sessionId: string + sessionId: string, + ending: 'stop' | 'evict' = 'stop' ): Promise { const session = context.sessions.get(sessionId) if (!session) { return } - // The obligation OUTLIVES the child. `hasProviderChild` is retired the instant the adapter - // proves the exit, so a step that aborts after that point would otherwise leave the retry - // reading "no child here" and skipping the settlement and the lease release it still owes. - const owesWindDown = owesProviderChildWindDown(session) - session.owesProviderChildWindDown = owesWindDown + // The obligation OUTLIVES the child. `child` is ended the instant the adapter proves the exit, + // so a step that aborts after that point would otherwise leave the retry reading "no child + // here" and skipping the settlement and the lease release it still owes. + const owed = owedProviderChildWindDown(session) + session.owesProviderChildWindDown = owed + const stopping = session.child let settlementError: unknown const eviction: StructuredAgentSessionEvictionContext = { sessionId, // The retry must not re-stop a child the adapter already proved gone, so this stays honest. - hasProviderChild: session.hasProviderChild, - owesProviderChildWindDown: owesWindDown, + hasProviderChild: stopping !== null, + owesProviderChildWindDown: owed !== undefined, eventSink: context.runtimeState.eventSinkFor(sessionId), adapter: context.deps.adapter, ...(context.restartWitness ? { beforeProviderChildStop: () => context.restartWitness?.beforeStop(sessionId) } : {}), - // Host state must not disagree with the adapter for the seven steps in between. + // Host state must not disagree with the adapter for the steps in between. onProviderChildStopped: () => { - session.hasProviderChild = false + if (stopping) { + endProviderChild(session, { + generation: stopping.generation, + fence: stopping.fence, + cause: ending, + reason: null, + duringStartup: stopping.phase === 'starting' + }) + } context.restartWitness?.stopped(sessionId) }, - forget: async () => { - await forgetStructuredAgentSession(context, sessionId) - context.deps.adapter.acknowledgeSessionRelease?.(sessionId) - }, + acknowledgeRelease: () => context.deps.adapter.acknowledgeSessionRelease?.(sessionId), discardSink: () => context.runtimeState.discardEventSink(sessionId), settleWork: async () => { + const fence = + owed?.fence ?? structuredAgentSessionConversationFence(context.deps.store, sessionId) const settled = await settleStructuredAgentSessionDeadGeneration({ journal: session.journal, sessionId, - fence: session.fence, - settlementId: `expected-close:${sessionId}:${session.fence}:${session.acquisitionGeneration ?? 'unknown'}`, + fence, + settlementId: `expected-close:${sessionId}:${fence}:${owed?.generation ?? 'unknown'}`, pendingSubmissionReason: 'provider_closed_before_acknowledgement', verdict: { state: 'interrupted', completedAt: context.now() }, showUnexpectedExitOutcome: false, @@ -127,15 +159,23 @@ export async function evictHeldStructuredAgentSession( } }, releaseLease: async () => { - await releaseStoredStructuredAgentSessionOwner({ - store: context.deps.store, - sessionId, - hasProviderChild: owesWindDown, - expectedFence: session.fence, - now: context.now() - }) - session.owesProviderChildWindDown = false - context.forgetStatus(sessionId) + if (owed) { + await releaseStoredStructuredAgentSessionOwner({ + store: context.deps.store, + sessionId, + hasProviderChild: true, + expectedFence: owed.fence, + now: context.now() + }) + } + session.owesProviderChildWindDown = undefined + if (ending === 'evict') { + context.forgetStatus(sessionId) + return + } + // The conversation stays open at the fence the release moved it to. + context.publishStatus?.(sessionId) + context.publishFence?.(sessionId) } } await evictStructuredAgentSession( @@ -144,6 +184,19 @@ export async function evictHeldStructuredAgentSession( ) } +/** Ends the conversation's resources, not the conversation: its child stops, and then its handle + * closes and it leaves the map. A stop that fails throws first, leaving it indexed for a retry. */ +export async function evictHeldStructuredAgentSession( + context: StructuredAgentSessionLifetimeContext, + sessionId: string +): Promise { + if (!context.sessions.has(sessionId)) { + return + } + await stopStructuredAgentSessionAgentUnderSerialize(context, sessionId, 'evict') + await forgetStructuredAgentSession(context, sessionId) +} + /** Stops every provider child owned by this host while keeping failed evictions reachable. A * session whose child is already stopped but whose wind-down aborted is still in scope — that is * the retry. */ @@ -154,7 +207,7 @@ export async function evictOwnedStructuredAgentSessions( retainOnFailure: Set ): Promise { const ownedSessionIds = [...context.sessions] - .filter(([, session]) => owesProviderChildWindDown(session)) + .filter(([, session]) => owedProviderChildWindDown(session) !== undefined) .map(([sessionId]) => sessionId) // Retained up front and cleared only once an eviction settles: the quit phase is bounded, and a // timeout leaves these still running. Closing their journals underneath them is the one outcome @@ -215,8 +268,7 @@ export function createStructuredAgentSessionHolds( ? activeStructuredAgentSessionTurnId(session.journal.snapshot().items) !== null || deliveryActive(sessionId) || session.journal.submissions().some(isQueuedAgentJournalSubmission) || - (session.providerChildPhase === 'starting' && - session.journal.pendingSubmissions().length > 0) + (session.child?.phase === 'starting' && session.journal.pendingSubmissions().length > 0) : false }, onError: (error) => context.deps.onEventSinkError?.(error), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts index 971c63aa6fc..5f2e7bd7452 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts @@ -23,6 +23,7 @@ import type { } from '../../../shared/agent-session-wire' import { DISPATCH_REJECTED_CANCELLED } from '../../../shared/structured-agent-session-dispatch-rejection' import { threadGoalPlan } from './structured-agent-session-thread-goal' +import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' import { admitAndRunAgentSessionMutation, type AgentSessionMutationRequest @@ -55,8 +56,8 @@ export type StructuredAgentSessionMutationContext = { openConversation: (sessionId: string) => Promise /** A message was accepted: the session's delivery loop hands it over. */ wakeDelivery: (sessionId: string) => void - /** Stops a provider child that has not proven its start; inside the caller's serialize. */ - stopStartingChild: (sessionId: string) => Promise + /** Stops the session's provider child, keeping its conversation; inside the caller's serialize. */ + stopAgent: (sessionId: string) => Promise now: () => number } @@ -78,7 +79,7 @@ function mutate( prepareSession, publish: (journal) => context.publish(envelope.sessionId, journal), flushStreamedEvents: context.flushStreamedEvents, - providerChildPhase: () => context.sessions.get(envelope.sessionId)?.providerChildPhase, + providerChildPhase: () => context.sessions.get(envelope.sessionId)?.child?.phase, now: () => context.now() }) ) @@ -154,13 +155,13 @@ export function cancelStructuredAgentSessionTurn( ctx.fence, DISPATCH_REJECTED_CANCELLED ) - const session = context.sessions.get(ctx.sessionId) - if (session?.hasProviderChild && session.providerChildPhase === 'starting') { - // A start that may never land is the one thing here Stop has to end. - await context.stopStartingChild(ctx.sessionId) + const child = context.sessions.get(ctx.sessionId)?.child + if (child?.phase === 'starting') { + // A start that may never land is the one thing here Stop has to end; the chat stays. + await context.stopAgent(ctx.sessionId) return { ok: true, value: { turnId: params.turnId, cancelled: true } } } - return session?.hasProviderChild + return child ? plan.run(ctx) : { ok: true, value: { turnId: params.turnId, cancelled: withdrawn.length > 0 } } } @@ -210,7 +211,12 @@ export function readStructuredAgentSessionOptions( if (!context.deps.adapter.readOptions) { throw new Error('structured_agent_session_options_unsupported') } - const options = await context.deps.adapter.readOptions({ sessionId, fence: session.fence }) + const options = await context.deps.adapter.readOptions({ + sessionId, + fence: + session.child?.fence ?? + structuredAgentSessionConversationFence(context.deps.store, sessionId) + }) return { ...options, rewind: @@ -244,6 +250,7 @@ export async function settleStructuredAgentSessionLateDispatch( if (!session) { return } + const fence = structuredAgentSessionConversationFence(context.deps.store, input.sessionId) // The journal queue drains before close; the host queue would defer this past teardown. await session.journal.resolveDispatch( 'providerIdentity' in input @@ -251,13 +258,13 @@ export async function settleStructuredAgentSessionLateDispatch( clientMessageId: input.clientMessageId, state: 'accepted', providerIdentity: input.providerIdentity, - fence: session.fence + fence } : { clientMessageId: input.clientMessageId, state: 'rejected', reason: input.reason, - fence: session.fence + fence } ) } @@ -275,7 +282,9 @@ export async function settleStructuredAgentSessionLateDispatch( * it never makes a send re-deliverable, because the provider may well have run it. */ export async function releaseStructuredAgentSessionUnansweredDispatches( - context: Pick, + context: Pick & { + deps: { store: Pick } + }, input: { sessionId: string; reason: string } ): Promise { const session = context.sessions.get(input.sessionId) @@ -294,7 +303,7 @@ export async function releaseStructuredAgentSessionUnansweredDispatches( state: 'unknown', // The earlier reason names a sharper fact than this one does. reason: entry.reason ?? input.reason, - fence: session.fence, + fence: structuredAgentSessionConversationFence(context.deps.store, input.sessionId), recovered: true }) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts index fd2c911c74e..84fa67b9331 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-types.ts @@ -26,25 +26,43 @@ export type StructuredAgentSessionReveal = { readable: boolean } +/** Which provider child: the adapter acquisition and the lease fence it writes at. */ +export type StructuredAgentSessionProviderChildIdentity = { + readonly generation: string | null + readonly fence: number +} + +/** The provider process behind a conversation. Written only in + * `structured-agent-session-provider-child`. */ +export type StructuredAgentSessionProviderChild = StructuredAgentSessionProviderChildIdentity & { + /** A publish-first acquire is `starting` until the adapter's `started` event; only then are its + * reported options fact. */ + phase: StructuredAgentSessionProviderChildPhase +} + +/** How the conversation's last child ended. In memory only: the delivery loop reads it to tell a + * Stop from a failure. */ +export type StructuredAgentSessionEndedChild = StructuredAgentSessionProviderChildIdentity & { + cause: 'exit' | 'attach-failed' | 'stop' | 'evict' + /** The provider's own diagnostic, when the ending carried one. */ + reason: string | null + duringStartup: boolean +} + +/** The conversation: its journal, params and readers outlive any child that serves it. */ export type StructuredAgentSessionHostSession = { /** Readonly: a new handle enters only through the session map's `set`, which binds its delivery. */ readonly journal: AgentSessionJournal params: AgentSessionAttachParams - fence: number - /** Whether THIS host generation is running the provider process behind the session. A journal - * restored for reading has none — so it may not be evicted to free a child, nor have its lease - * released as an observed exit. */ - hasProviderChild: boolean - /** Whether the child behind `hasProviderChild` has proven its start. A publish-first acquire - * is `starting` until the adapter's `started` event; only then are its reported options fact. */ - providerChildPhase: StructuredAgentSessionProviderChildPhase + /** The child THIS host generation runs for the conversation. A conversation opened for reading + * has none — so it may not be evicted to free a child, nor have its lease released as an + * observed exit. */ + child: StructuredAgentSessionProviderChild | null /** The wind-down this host still owes for a child it started: settling that generation's work - * and handing the lease back. A separate fact from `hasProviderChild`, which goes false the - * moment the adapter proves the exit — an eviction that aborts after that point must still be - * able to finish the wind-down on the next close. */ - owesProviderChildWindDown?: boolean - /** Exact adapter acquisition behind `hasProviderChild`; retained after exit to fence recovery. */ - acquisitionGeneration: string | null + * and handing the lease back. Outlives `child`, which ends the moment the adapter proves the + * exit — an eviction that aborts after that point must still finish it on the next close. */ + owesProviderChildWindDown?: StructuredAgentSessionProviderChildIdentity + lastEndedChild?: StructuredAgentSessionEndedChild } export type StructuredAgentSessionHostDeps = { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index ba059204cb1..bfe626813bd 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -20,6 +20,7 @@ import { attachStructuredAgentSession } from './structured-agent-session-attach- import { createStructuredAgentSessionHolds, evictHeldStructuredAgentSession, + stopStructuredAgentSessionAgentUnderSerialize, type StructuredAgentSessionLifetimeContext } from './structured-agent-session-host-lifetime' import type { @@ -173,7 +174,14 @@ export class StructuredAgentSessionHost { runtimeState: this.runtimeState, sessions: this.sessions, now: () => this.now(), - forgetStatus: this.clientDelivery.forgetStatus + forgetStatus: this.clientDelivery.forgetStatus, + publishStatus: this.clientDelivery.publishStatus, + publishFence: (sessionId) => { + const journal = this.sessions.get(sessionId)?.journal + if (journal) { + this.subscribers.snapshot(sessionId, journal) + } + } } } @@ -261,7 +269,8 @@ export class StructuredAgentSessionHost { serialize: (sessionId, task) => this.serialize(sessionId, task), openConversation: this.conversationDelivery.open, wakeDelivery: (sessionId) => this.conversationDelivery.loop.wake(sessionId), - stopStartingChild: (sessionId) => this.closeUnderSerialize(sessionId), + stopAgent: (sessionId) => + stopStructuredAgentSessionAgentUnderSerialize(this.lifetimeContext(), sessionId), now: () => this.now() } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-handles.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-handles.test.ts index ef08e1c7e93..942c9ef6152 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-journal-handles.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-journal-handles.test.ts @@ -15,11 +15,7 @@ import { journalDatabaseFile } from '../agent-session-journal/journal-paths' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' import { openAgentSessionJournalWithRecovery } from './agent-session-journal-recovery' -import { - evictStructuredAgentSession, - STRUCTURED_AGENT_SESSION_EVICTION_STEPS, - type StructuredAgentSessionEvictionContext -} from './structured-agent-session-eviction' +import { forgetStructuredAgentSession } from './structured-agent-session-host-lifetime' import { tearDownStructuredAgentSessionHost } from './structured-agent-session-host-teardown' import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' @@ -73,30 +69,12 @@ function hostSession(journal: AgentSessionJournal): StructuredAgentSessionHostSe return { journal, params: {} as StructuredAgentSessionHostSession['params'], - fence: 1, - hasProviderChild: false, - providerChildPhase: 'ready', - acquisitionGeneration: null + child: null } } -function evictionContext( - overrides: Partial -): StructuredAgentSessionEvictionContext { - return { - sessionId: SESSION, - hasProviderChild: false, - eventSink: { - drained: async () => ({ ok: true }) as const, - unbind: () => undefined, - close: () => undefined - } as unknown as StructuredAgentSessionEvictionContext['eventSink'], - adapter: {} as StructuredAgentSessionEvictionContext['adapter'], - forget: async () => undefined, - discardSink: () => undefined, - releaseLease: async () => undefined, - ...overrides - } +function forgetContext(sessions: Map) { + return { deps: { store: { getRecord: () => null } }, sessions, forgetStatus: () => undefined } } beforeEach(async () => { @@ -140,46 +118,37 @@ describe('site 6: recovery rehydration', () => { }) }) -describe('sites 9 and 10: the delete and overwrite callbacks', () => { +describe('sites 9 and 10: closing a conversation handle', () => { it('awaits the journal close before dropping the map entry', async () => { const journal = await journals.open({ identity: IDENTITY, journalDir }) const sessions = new Map([[SESSION, hostSession(journal)]]) const order: string[] = [] + const close = journal.close.bind(journal) + journal.close = async () => { + order.push('close-started') + await close() + order.push(sessions.has(SESSION) ? 'closed' : 'dropped-before-close') + } - await evictStructuredAgentSession( - evictionContext({ - forget: async () => { - order.push('close-started') - await sessions.get(SESSION)?.journal.close() - order.push('closed') - sessions.delete(SESSION) - order.push('forgotten') - } - }), - STRUCTURED_AGENT_SESSION_EVICTION_STEPS - ) + await forgetStructuredAgentSession(forgetContext(sessions), SESSION) - expect(order).toEqual(['close-started', 'closed', 'forgotten']) + expect(order).toEqual(['close-started', 'closed']) expect(sessions.size).toBe(0) await expectNothingHoldsTheDirectory(journalDir) }) - it('aborts the eviction with the session still indexed when the close rejects', async () => { + it('keeps the session indexed when the close rejects', async () => { const journal = await journals.open({ identity: IDENTITY, journalDir }) const sessions = new Map([[SESSION, hostSession(journal)]]) + const close = journal.close.bind(journal) + journal.close = () => Promise.reject(new Error('close rejected')) - await expect( - evictStructuredAgentSession( - evictionContext({ - forget: async () => { - await Promise.reject(new Error('close rejected')) - } - }), - STRUCTURED_AGENT_SESSION_EVICTION_STEPS - ) - ).rejects.toMatchObject({ step: 'forget-session' }) + await expect(forgetStructuredAgentSession(forgetContext(sessions), SESSION)).rejects.toThrow( + 'close rejected' + ) // Still indexed, so the next close is a real retry. expect(sessions.has(SESSION)).toBe(true) + journal.close = close }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-record.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-record.test.ts new file mode 100644 index 00000000000..3f21a807673 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-record.test.ts @@ -0,0 +1,278 @@ +// The provider child is its own record on the conversation: stopping it, losing it or failing to +// start it ends the child, never the conversation. Against the real host, store and journal, with a +// live subscriber opened before each action. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' +import type { + AgentSessionStatusSummary, + AgentSessionSubscribeEvent +} from '../../../shared/agent-session-wire' +import { DISPATCH_REJECTED_CANCELLED } from '../../../shared/structured-agent-session-dispatch-rejection' +import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + HOST_TEST_THREAD as THREAD, + hostTestAttachParams, + hostTestMessage, + hostTestOperationId, + resetHostTestOperationIds +} from './structured-agent-session-host-test-data' + +const CALLER = { callerKey: 'client-1' } + +let root: string +let store: AgentSessionRecordStore +let host: StructuredAgentSessionHost +let acquire: Mock +let dispatch: Mock +let adapterExtras: Partial + +function eventually(assertion: () => void | Promise): Promise { + return vi.waitFor(assertion, { timeout: 10_000 }) +} + +function generation(): string { + return `generation-${acquire.mock.calls.length}` +} + +const spawnChild: StructuredAgentSessionAdapter['acquire'] = async ({ fence, spawnToken }) => ({ + process: { hostId: 'local', pid: 4242, processStartTimeMs: 1_700_000_000_000, spawnToken }, + acquisitionGeneration: generation(), + link: { + linkId: `link-${fence}`, + handle: { provider: 'codex' as const, threadId: THREAD }, + origin: store.getRecord(SESSION)?.providerHandleChain.length + ? ('resumed' as const) + : ('created' as const), + mintedAtFence: fence, + observedAt: NOW + } +}) + +/** A Claude-shaped child: published at spawn, so it is `starting` until `started`. */ +const spawnStartingChild: StructuredAgentSessionAdapter['acquire'] = async (input) => ({ + ...(await spawnChild(input)), + providerChildPhase: 'starting' as const +}) + +function startHost(): void { + host = new StructuredAgentSessionHost({ + store, + adapter: { + acquire, + dispatch, + closeSession: vi.fn(async () => true), + releaseAcquisition: vi.fn(async () => true), + cancelTurn: vi.fn(async () => ({ cancelled: false })), + answerPrompt: vi.fn(async () => undefined), + setOption: vi.fn(async () => undefined), + ...adapterExtras + }, + journalRoot: root, + claimKeyId: 'key-1', + mintSpawnToken: () => `spawn-${acquire.mock.calls.length}`, + releaseGraceMs: 60_000, + now: () => NOW + }) +} + +async function restartHost(): Promise { + await host.flushAllStreamedEvents() + startHost() +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-child-record-')) + resetHostTestOperationIds() + adapterExtras = {} + acquire = vi.fn(spawnChild) + dispatch = vi.fn(async (input) => ({ + state: 'accepted' as const, + providerIdentity: { + provider: 'codex' as const, + threadId: THREAD, + turnId: `turn-${input.clientMessageId}`, + ordinal: dispatch.mock.calls.length + } + })) + store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' }) + startHost() + expect(await host.attach(CALLER, hostTestAttachParams(null))).toMatchObject({ ok: true }) + await host.close(SESSION) +}) + +afterEach(async () => { + await host.flushAllStreamedEvents() + await rm(root, { recursive: true, force: true }) +}) + +async function accept(text: string): Promise { + const body = hostTestMessage(text) + const clientOperationId = hostTestOperationId() + const sent = await host.send(CALLER, { + envelope: { + sessionId: SESSION, + clientOperationId, + expectedRuntimeFence: 1, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.send', + sessionId: SESSION, + fields: { body } + }) + }, + body + }) + expect(sent).toMatchObject({ ok: true, value: { submission: { dispatchState: 'pending' } } }) + return clientOperationId +} + +function stop() { + const turnId = 'turn-none' + return host.cancel(CALLER, { + envelope: { + sessionId: SESSION, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: null, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.cancel', + sessionId: SESSION, + fields: { turnId } + }) + }, + turnId + }) +} + +function submission(id: string): AgentJournalSubmission | undefined { + return host.journalSnapshot(SESSION).submissions.find((entry) => entry.clientMessageId === id) +} + +function statusRows(): { text: string; tone?: string }[] { + return host + .journalSnapshot(SESSION) + .items.flatMap((item) => + item.body.kind === 'status' + ? [{ text: item.body.text, ...(item.body.tone ? { tone: item.body.tone } : {}) }] + : [] + ) +} + +function conversation() { + return host['sessions'].get(SESSION) +} + +function subscribe(): AgentSessionSubscribeEvent[] { + const events: AgentSessionSubscribeEvent[] = [] + host.subscribe({ + id: 'sub-1', + sessionId: SESSION, + emit: (event) => events.push(structuredClone(event)) + }) + return events +} + +/** The chat's status row as a session list sees it, frame by frame. */ +function watchStatus(): AgentSessionStatusSummary[] { + const frames: AgentSessionStatusSummary[] = [] + host.subscribeStatus({ + id: 'list-1', + emit: (event) => { + if (event.type === 'status' && event.session.sessionId === SESSION) { + frames.push(event.session) + } + } + }) + return frames +} + +function deferred() { + let resolve!: (value: T) => void + const promise = new Promise((next) => { + resolve = next + }) + return { promise, resolve } +} + +describe('Stop on a child still proving its start', () => { + it('ends the child and keeps the conversation, its holders and its readers (R1)', async () => { + const ended = deferred() + adapterExtras = { + awaitStarted: vi.fn(() => ended.promise), + closeSession: vi.fn(async () => { + ended.resolve() + return true + }) + } + await restartHost() + acquire.mockImplementationOnce(spawnStartingChild) + await host.hold(SESSION, 'surface-1', { resume: false }) + const first = await accept('hello') + const journal = conversation()?.journal + const events = subscribe() + const frames = watchStatus() + await eventually(() => expect(conversation()?.child?.phase).toBe('starting')) + + expect(await stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + + // The same conversation: no reopen, the holder kept, and the chat told it is idle again. + expect(conversation()?.journal).toBe(journal) + expect(conversation()?.child).toBeNull() + expect(host.isHeld(SESSION)).toBe(true) + expect(frames.at(-1)).not.toHaveProperty('hostExecutionPhase') + expect(frames.at(-1)).not.toHaveProperty('hostExecutionOwned') + expect(submission(first)).toMatchObject({ + dispatchState: 'rejected', + reason: DISPATCH_REJECTED_CANCELLED + }) + // A Stop is not a failure: no row, and the loop is gone. + expect(statusRows()).toEqual([]) + await eventually(() => expect(host['conversationDelivery'].loop.isRunning(SESSION)).toBe(false)) + + const next = await accept('after stop') + await eventually(() => expect(submission(next)?.dispatchState).toBe('accepted')) + expect(conversation()?.journal).toBe(journal) + expect(dispatch.mock.calls.map(([input]) => input.clientMessageId)).toEqual([next]) + // The reader opened before the Stop saw the next message delivered on the same stream. + expect( + events.some( + (event) => + event.type === 'batch' && + event.batch.submissions.some( + (entry) => entry.clientMessageId === next && entry.dispatchState === 'accepted' + ) + ) + ).toBe(true) + }) +}) + +describe('a settlement retry for an earlier child inside the attach for the next one', () => { + it('settles the earlier child and leaves the queued message to the new child (R1)', async () => { + // The earlier child's settlement is still owed; the lease is otherwise released. + await store.transitionHandoff(SESSION, (record) => ({ + ...record, + lease: { + ...record.lease, + settlementRetryRequired: true, + settlementRetryId: `provider-exit:${SESSION}:1:generation-1` + } + })) + const retryFence = store.getRecord(SESSION)!.lease.runtimeFence + const id = await accept('for the next child') + + await eventually(() => expect(submission(id)?.dispatchState).toBe('accepted')) + expect(store.getRecord(SESSION)?.lease.settlementRetryRequired).toBeUndefined() + // Handed over at the new child's fence, which the attach reserved after the retry. + const newFence = store.getRecord(SESSION)!.lease.runtimeFence + expect(newFence).toBeGreaterThan(retryFence) + expect(submission(id)?.fence).toBe(newFence) + expect(conversation()?.child).toMatchObject({ generation: generation(), fence: newFence }) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child.ts new file mode 100644 index 00000000000..56be117bc4a --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-child.ts @@ -0,0 +1,66 @@ +// The provider child behind a conversation, kept as its own record on the conversation's entry. +// +// The entry is the conversation and outlives any number of children. A child enters only when an +// attach has fully succeeded, and leaves only through `endProviderChild`, which every ending shares: +// an exit, a failed re-attach, a Stop and an eviction. Each is matched on the child's generation +// and fence, so an ending that arrives late for an older child cannot end a newer one. + +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import type { + StructuredAgentSessionEndedChild, + StructuredAgentSessionHostSession, + StructuredAgentSessionProviderChild, + StructuredAgentSessionProviderChildIdentity +} from './structured-agent-session-host-types' + +type ChildBearer = Pick + +/** The fence a conversation write carries: the record's, which is where the next child starts. A + * child's own writes carry `child.fence`, which equals it while that child holds the lease. */ +export function structuredAgentSessionConversationFence( + store: Pick, + sessionId: string +): number { + return store.getRecord(sessionId)?.lease.runtimeFence ?? 0 +} + +/** For the end of a successful attach only: a failed one never wrote a child to take back. */ +export function indexProviderChild( + session: ChildBearer, + child: StructuredAgentSessionProviderChild +): void { + session.child = child +} + +export function markProviderChildStarted( + session: ChildBearer, + identity: StructuredAgentSessionProviderChildIdentity +): boolean { + const child = matchingChild(session, identity) + if (child) { + child.phase = 'ready' + } + return child !== null +} + +export function endProviderChild( + session: ChildBearer, + ended: StructuredAgentSessionEndedChild +): boolean { + if (!matchingChild(session, ended)) { + return false + } + session.child = null + session.lastEndedChild = ended + return true +} + +function matchingChild( + session: ChildBearer, + identity: StructuredAgentSessionProviderChildIdentity +): StructuredAgentSessionProviderChild | null { + const { child } = session + return child && child.generation === identity.generation && child.fence === identity.fence + ? child + : null +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-started.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-started.ts index 55c718f967b..6d6d910cc55 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-provider-started.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-provider-started.ts @@ -16,6 +16,7 @@ import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' import { nativeSessionOptionsFromReport } from './structured-agent-session-option-restoration' +import { markProviderChildStarted } from './structured-agent-session-provider-child' export type StructuredAgentSessionProviderStartedContext = { deps: StructuredAgentSessionHostDeps @@ -35,13 +36,14 @@ export function settleStructuredAgentSessionProviderStarted( return context.serialize(event.sessionId, async () => { const session = context.sessions.get(event.sessionId) if ( - !session?.hasProviderChild || - session.fence !== event.fence || - session.acquisitionGeneration !== event.acquisitionGeneration + !session || + !markProviderChildStarted(session, { + generation: event.acquisitionGeneration, + fence: event.fence + }) ) { return } - session.providerChildPhase = 'ready' // Prompts held for the start are written now but open a turn only on their echo; a release // tick in between would stop the child before it runs them. context.restartReleaseGrace(event.sessionId) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.test.ts index 21c3d9a5f7f..514008e49f9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-read-restore.test.ts @@ -87,7 +87,7 @@ describe('a session whose journal is still the pre-SQLite format', () => { .items.map((entry) => (entry.body.kind === 'status' ? entry.body.text : '')) expect(disclosed.join('')).toContain(transcript) // Publishing it costs no agent process; acquisition still waits for the user. - expect(restored!.session.hasProviderChild).toBe(false) + expect(restored!.session.child).toBeNull() }) it('is published for a remnant whose log is gone', async () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-continuation.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-continuation.ts index c025c0dc46a..87a56efb7b9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-continuation.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-continuation.ts @@ -44,7 +44,9 @@ export type StructuredAgentSessionContinuationOutcome = { /** The slice of the host one continuation needs. Structural so this module never imports the host. */ export type StructuredAgentSessionContinuationHost = { - sessions: ReadonlyMap + sessions: ReadonlyMap + /** The fence a conversation write carries; null when the session is not open. */ + conversationFence: (sessionId: string) => number | null send: (input: { envelope: AgentSessionMutationEnvelope body: AgentJournalMessageItem @@ -72,7 +74,7 @@ export function restartContinuationDeps( marker: AgentSessionResumeMarker ): StructuredAgentSessionContinuationDeps { return { - currentFence: (sessionId) => host.sessions.get(sessionId)?.fence ?? null, + currentFence: host.conversationFence, send: (input) => host.send({ ...input, @@ -107,17 +109,18 @@ export function noteRestartReattachFailed( /** Writes a host-authored status note into the chat. */ function restartNoteWriter( - host: Pick + host: Pick ): StructuredAgentSessionContinuationDeps['note'] { return async (sessionId, text, tone) => { const session = host.sessions.get(sessionId) - if (!session) { + const fence = host.conversationFence(sessionId) + if (!session || fence === null) { return } await session.journal.appendItem( { provider: 'orca', clientMessageId: `restart-continuation:${sessionId}:${host.now()}` }, { kind: 'status', text, ...(tone ? { tone } : {}) }, - { fence: session.fence } + { fence } ) } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts index 41ae0a1b1a8..61e61debe14 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts @@ -37,8 +37,7 @@ describe('restart journal restoration', () => { session: { journal: {}, params: { location: { workspaceId: 'workspace-1' }, provider: 'codex' }, - fence: 1, - hasProviderChild: false, + child: null, sessionId }, reset: null @@ -94,7 +93,7 @@ describe('restart journal restoration', () => { journal: {}, params, fence: 4, - hasProviderChild: false, + child: null, acquisitionGeneration: null }, reset: null @@ -132,7 +131,7 @@ describe('restart journal restoration', () => { journal: {}, params: {}, fence: 4, - hasProviderChild: false, + child: null, acquisitionGeneration: null }, reset: null diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts index 29453e74652..0fa4ee8534d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-host.ts @@ -41,8 +41,9 @@ import { type StructuredAgentSessionContinuationOutcome } from './structured-agent-session-restart-continuation' import { createStructuredAgentSessionRestartWitnesses } from './structured-agent-session-restart-witnesses' +import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' -type LiveSession = { journal: AgentSessionJournal; hasProviderChild: boolean; fence: number } +type LiveSession = { journal: AgentSessionJournal; child: { fence: number } | null } export type StructuredAgentSessionRestartResumeSurfaces = { revealSession: (sessionId: string) => Promise<{ readable: boolean }> @@ -152,6 +153,10 @@ export function createStructuredAgentSessionRestartResume( const continuationHost: StructuredAgentSessionContinuationHost = { ...surfaces, sessions, + conversationFence: (sessionId) => + sessions.has(sessionId) + ? structuredAgentSessionConversationFence(deps.store, sessionId) + : null, stillResumable: (marker, options) => derive([marker], 'may-be-held', options).candidates.length === 1 } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-test-harness.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-test-harness.ts index e5b65009634..68823c078dc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-test-harness.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-test-harness.ts @@ -145,7 +145,7 @@ export type HarnessJournal = { appendItem: (envelope: unknown, body: { kind: string; text: string }) => Promise } -export type HarnessSession = { journal: HarnessJournal; hasProviderChild: boolean; fence?: number } +export type HarnessSession = { journal: HarnessJournal; child: { fence: number } | null } export function journal( items: AgentJournalRenderItem[], diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume.test.ts index 1d6a8a1fc5e..09a172f0b1d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume.test.ts @@ -48,7 +48,7 @@ describe('deriving what was working at teardown', () => { it('marks a session this host was running a turn for', () => { const markers = markersAtTeardown({ sessions: new Map([ - [SESSION, { journal: journal([turnItem('turn-1', 'running')]), hasProviderChild: true }] + [SESSION, { journal: journal([turnItem('turn-1', 'running')]), child: { fence: 1 } }] ]), getRecord: () => record(), backgroundTasks: () => undefined, @@ -74,7 +74,7 @@ describe('deriving what was working at teardown', () => { it('carries the update trigger so the surface can say the restart was not the user choice', () => { const [recorded] = markersAtTeardown({ sessions: new Map([ - [SESSION, { journal: journal([turnItem('turn-1', 'running')]), hasProviderChild: true }] + [SESSION, { journal: journal([turnItem('turn-1', 'running')]), child: { fence: 1 } }] ]), getRecord: () => record(), backgroundTasks: () => undefined, @@ -89,7 +89,7 @@ describe('deriving what was working at teardown', () => { it('marks nothing for an idle session', () => { expect( markersAtTeardown({ - sessions: new Map([[SESSION, { journal: journal([]), hasProviderChild: true }]]), + sessions: new Map([[SESSION, { journal: journal([]), child: { fence: 1 } }]]), getRecord: () => record(), backgroundTasks: () => undefined, trigger: 'quit', @@ -103,7 +103,7 @@ describe('deriving what was working at teardown', () => { expect( markersAtTeardown({ sessions: new Map([ - [SESSION, { journal: journal([turnItem('turn-1', 'completed')]), hasProviderChild: true }] + [SESSION, { journal: journal([turnItem('turn-1', 'completed')]), child: { fence: 1 } }] ]), getRecord: () => record(), backgroundTasks: () => undefined, @@ -115,12 +115,12 @@ describe('deriving what was working at teardown', () => { }) // The user's stated fear. A journal restored for READING carries whatever `running` row an older - // crash left behind, and it is the live `hasProviderChild` — not that row — that decides. + // crash left behind, and it is the live `child` — not that row — that decides. it('marks nothing for a stale running row this host was not executing', () => { expect( markersAtTeardown({ sessions: new Map([ - [SESSION, { journal: journal([turnItem('turn-1', 'running')]), hasProviderChild: false }] + [SESSION, { journal: journal([turnItem('turn-1', 'running')]), child: null }] ]), getRecord: () => record(), backgroundTasks: () => undefined, @@ -141,7 +141,7 @@ describe('deriving what was working at teardown', () => { SESSION, { journal: journal([turnItem('turn-1', 'running'), pendingApproval()]), - hasProviderChild: true + child: { fence: 1 } } ] ]), @@ -165,7 +165,7 @@ describe('deriving what was working at teardown', () => { it('marks a settled lead whose subagent was still running, anchored on its last turn', () => { const markers = markersAtTeardown({ sessions: new Map([ - [SESSION, { journal: journal([turnItem('turn-1', 'completed')]), hasProviderChild: true }] + [SESSION, { journal: journal([turnItem('turn-1', 'completed')]), child: { fence: 1 } }] ]), getRecord: () => record(), backgroundTasks: () => [ @@ -188,7 +188,7 @@ describe('deriving what was working at teardown', () => { it('records only the live rows of the roster, bounded', () => { const [recorded] = markersAtTeardown({ sessions: new Map([ - [SESSION, { journal: journal([turnItem('turn-1', 'completed')]), hasProviderChild: true }] + [SESSION, { journal: journal([turnItem('turn-1', 'completed')]), child: { fence: 1 } }] ]), getRecord: () => record(), backgroundTasks: () => [ @@ -213,7 +213,7 @@ describe('deriving what was working at teardown', () => { it('marks a settled lead whose only live work is a monitor', () => { const markers = markersAtTeardown({ sessions: new Map([ - [SESSION, { journal: journal([turnItem('turn-1', 'completed')]), hasProviderChild: true }] + [SESSION, { journal: journal([turnItem('turn-1', 'completed')]), child: { fence: 1 } }] ]), getRecord: () => record(), backgroundTasks: () => [{ id: 'task-m', kind: 'monitor', name: 'ci-watch' }], @@ -230,7 +230,7 @@ describe('deriving what was working at teardown', () => { expect( markersAtTeardown({ sessions: new Map([ - [SESSION, { journal: journal([turnItem('turn-1', 'completed')]), hasProviderChild: true }] + [SESSION, { journal: journal([turnItem('turn-1', 'completed')]), child: { fence: 1 } }] ]), getRecord: () => record(), backgroundTasks: () => [ @@ -248,7 +248,7 @@ describe('deriving what was working at teardown', () => { it('records the identity root so an advancing Claude leaf cannot invalidate the marker', () => { const [recorded] = markersAtTeardown({ sessions: new Map([ - [SESSION, { journal: journal([turnItem('turn-1', 'running')]), hasProviderChild: true }] + [SESSION, { journal: journal([turnItem('turn-1', 'running')]), child: { fence: 1 } }] ]), getRecord: () => claudeRecord(null), backgroundTasks: () => undefined, @@ -264,7 +264,7 @@ describe('deriving what was working at teardown', () => { expect( markersAtTeardown({ sessions: new Map([ - [SESSION, { journal: journal([turnItem('turn-1', 'running')]), hasProviderChild: true }] + [SESSION, { journal: journal([turnItem('turn-1', 'running')]), child: { fence: 1 } }] ]), getRecord: () => record({ chain: [] }), backgroundTasks: () => undefined, @@ -285,7 +285,7 @@ describe('deriving what was working at teardown', () => { SESSION, { journal: journal([], false, [submission('msg-1', 'pending')]), - hasProviderChild: true + child: { fence: 1 } } ] ]), @@ -310,7 +310,7 @@ describe('deriving what was working at teardown', () => { journal: journal([turnItem('turn-0', 'completed')], false, [ submission('msg-1', 'pending') ]), - hasProviderChild: true + child: { fence: 1 } } ] ]), @@ -335,7 +335,7 @@ describe('deriving what was working at teardown', () => { journal: journal([turnItem('turn-1', 'running')], false, [ submission('msg-1', 'accepted') ]), - hasProviderChild: true + child: { fence: 1 } } ] ]), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts index 318f2192cc2..8bf3bc2242c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts @@ -70,8 +70,14 @@ export function createStructuredAgentSessionHostRestore( const restorer = new StructuredAgentSessionReadableRestorer({ openDeps: deps, supportsRecord: (record) => adapterSupportsRecord(deps.adapter, record), - retrySettlement: (sessionId, params) => - retryPendingStructuredAgentSessionSettlement({ deps, sessions, sessionId, params, now }), + // Runs once the restorer indexed the conversation, so it retries into that one handle. + retrySettlement: (sessionId) => + retryPendingStructuredAgentSessionSettlement({ + deps, + sessionId, + openJournal: async () => sessions.get(sessionId)?.journal ?? null, + now + }), ...wiring }) const gate = new StructuredAgentSessionRestartRestoreGate() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-settlement-retry.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-settlement-retry.test.ts index 976e9417628..4ea4b4e13a3 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-settlement-retry.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-settlement-retry.test.ts @@ -45,7 +45,7 @@ function retry(settlementId: string, deathEvidence: AgentSessionDeathEvidence) { return retryLoadedStructuredAgentSessionSettlement({ deps: { store: store() }, sessionId: SESSION, - session: { journal, fence: FENCE, acquisitionGeneration: null }, + journal, now: () => 2_000 }) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-settlement-retry.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-settlement-retry.ts index 1000881220e..8d19420feb4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-settlement-retry.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-settlement-retry.ts @@ -1,60 +1,43 @@ -import type { AgentSessionAttachParams } from './structured-agent-session-attach' -import { attachJournal } from './structured-agent-session-attach' -import type { - StructuredAgentSessionHostDeps, - StructuredAgentSessionHostSession -} from './structured-agent-session-host-types' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types' import type { StructuredAgentSessionLeaseStore } from './structured-agent-session-lease-release' import { turnVerdictFromDeathEvidence } from './structured-agent-session-stale-turn-verdict' import { captureUnfinishedStructuredAgentSessionWork, settleStructuredAgentSessionDeadGeneration, - unfinishedStructuredAgentSessionWorkWasInterrupted + unfinishedStructuredAgentSessionWorkWasInterrupted, + type DeadGenerationJournal } from './structured-agent-session-dead-generation-settlement' +/** Retries a durable provider-exit settlement against the conversation's own journal. The retry is + * for an earlier child, so it writes at the record's fence and leaves the conversation — and any + * message queued for the next child — alone. */ export async function retryPendingStructuredAgentSessionSettlement(input: { deps: StructuredAgentSessionHostDeps - sessions: Map sessionId: string - params: AgentSessionAttachParams + /** The conversation's journal, opened through the host's one open when it is closed. */ + openJournal: () => Promise now: () => number }): Promise { const record = input.deps.store.getRecord(input.sessionId) if (!record?.lease.settlementRetryRequired || !record.lease.settlementRetryId) { return true } - let journal = input.sessions.get(input.sessionId)?.journal - if (!journal) { - try { - journal = ( - await attachJournal({ - record, - params: input.params, - journalRoot: input.deps.journalRoot, - adapter: input.deps.adapter - }) - ).journal - } catch (error) { - input.deps.onEventSinkError?.({ sessionId: input.sessionId, error }) - return false - } + let journal: AgentSessionJournal | null + try { + journal = await input.openJournal() + } catch (error) { + input.deps.onEventSinkError?.({ sessionId: input.sessionId, error }) + return false } - const current = input.sessions.get(input.sessionId) - const retrySession = - current ?? - ({ - journal, - params: input.params, - fence: record.lease.runtimeFence, - hasProviderChild: false, - acquisitionGeneration: null - } as StructuredAgentSessionHostSession) - return retryLoadedStructuredAgentSessionSettlement({ - deps: input.deps, - sessionId: input.sessionId, - session: retrySession, - now: input.now - }) + return journal + ? retryLoadedStructuredAgentSessionSettlement({ + deps: input.deps, + sessionId: input.sessionId, + journal, + now: input.now + }) + : false } export async function retryLoadedStructuredAgentSessionSettlement(input: { @@ -63,23 +46,23 @@ export async function retryLoadedStructuredAgentSessionSettlement(input: { onEventSinkError?: StructuredAgentSessionHostDeps['onEventSinkError'] } sessionId: string - session: Pick + journal: DeadGenerationJournal now: () => number }): Promise { const record = input.deps.store.getRecord(input.sessionId) if (!record?.lease.settlementRetryRequired || !record.lease.settlementRetryId) { return true } - const retrySession = input.session - retrySession.fence = record.lease.runtimeFence + const { journal } = input + const fence = record.lease.runtimeFence const onError = (id: string, error: unknown): void => input.deps.onEventSinkError?.({ sessionId: id, error }) // Only an observed exit earns an end time; a probe-proven death never saw one. const verdict = turnVerdictFromDeathEvidence(record.lease.deathEvidence) const ok = await settleStructuredAgentSessionDeadGeneration({ - journal: retrySession.journal, + journal, sessionId: input.sessionId, - fence: retrySession.fence, + fence, settlementId: record.lease.settlementRetryId, pendingSubmissionReason: 'provider_exited_before_acknowledgement', verdict, @@ -90,8 +73,8 @@ export async function retryLoadedStructuredAgentSessionSettlement(input: { showUnexpectedExitOutcome: verdict.state === 'interrupted' && unfinishedStructuredAgentSessionWorkWasInterrupted( - captureUnfinishedStructuredAgentSessionWork(retrySession.journal), - retrySession.journal, + captureUnfinishedStructuredAgentSessionWork(journal), + journal, verdict.completedAt ), ...(record.lease.deathEvidence?.detail diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-row.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-row.ts index 516122c091c..52a1d02f7c2 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-row.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-row.ts @@ -9,7 +9,7 @@ import type { StructuredAgentSessionHostSession } from './structured-agent-sessi * is still queued: a start whose messages Stop withdrew did not fail anyone. */ export async function recordStructuredAgentSessionStartFailure( - session: Pick, + session: Pick & { fence: number }, text: string ): Promise { const oldest = oldestQueuedSubmission(session) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-failure-exit.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-failure-exit.test.ts index c7f61d5c9ae..c8b5b1f40c5 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-startup-failure-exit.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-startup-failure-exit.test.ts @@ -19,9 +19,7 @@ function startedSession(): StructuredAgentSessionUnexpectedExitSession & { journal: { appendLifecycleBatch: ReturnType } } { return { - hasProviderChild: true, - fence: 7, - acquisitionGeneration: GENERATION, + child: { generation: GENERATION, fence: 7, phase: 'ready' }, journal: { // Nothing ran: the start failed before any response or acknowledged prompt. snapshot: () => ({ items: [] }), @@ -105,7 +103,10 @@ describe('a provider that ends before it finished starting', () => { }) it("reads a start that failed off the host's own phase when the provider omits the flag", async () => { - const session = { ...startedSession(), providerChildPhase: 'starting' as const } + const session = { + ...startedSession(), + child: { generation: GENERATION, fence: 7, phase: 'starting' as const } + } const ticket = await settleUnexpectedStructuredAgentSessionExit(contextFor(session), ended) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-test-session.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-test-session.ts index e4fa6ebed1b..c130aaf9d59 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-test-session.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-test-session.ts @@ -2,17 +2,11 @@ import type { AgentSessionJournal } from '../agent-session-journal/journal-store export function indexedStatusFeedSession(session: { journal: AgentSessionJournal - hasProviderChild?: boolean - providerChildPhase?: 'starting' | 'ready' - fence?: number + child?: { phase: 'starting' | 'ready' } | null }) { return { journal: session.journal, - fence: session.fence ?? 1, - ...(session.hasProviderChild !== undefined - ? { hasProviderChild: session.hasProviderChild } - : {}), - ...(session.providerChildPhase ? { providerChildPhase: session.providerChildPhase } : {}), + ...(session.child !== undefined ? { child: session.child } : {}), params: { location: { executionHostId: 'local' as const, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.test.ts index 65c7f54fb30..2c2863707c8 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.test.ts @@ -3,6 +3,7 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { agentSessionRecordFixture } from '../../../shared/agent-session-record.test-fixture' import type { AgentSessionBackgroundTask, AgentSessionStatusEvent, @@ -33,6 +34,8 @@ const USER_IDENTITY = { ordinal: 1 } as const +type Indexed = Parameters[0] + let root: string const journals = createTrackedJournalOpener() @@ -82,7 +85,9 @@ function feedFor( } } } as unknown as ReadonlyMap>, - getRecord: () => record as AgentSessionRecord | null, + // A partial record still has a lease: the feed reads the conversation's fence off it. + getRecord: () => + record && ({ lease: agentSessionRecordFixture().lease, ...record } as AgentSessionRecord), now: () => (now += 1) }) const events: AgentSessionStatusEvent[] = [] @@ -93,17 +98,17 @@ function feedFor( describe('StructuredAgentSessionStatusFeed', () => { it('projects whether the owned child has proven its start, and nothing once it is not owned', async () => { const journal = await openJournal() - const session = { journal, hasProviderChild: true, providerChildPhase: 'starting' as const } + const session = { journal, child: { phase: 'starting' as const } } const sessions = new Map[0]>([[SESSION, session]]) const { feed, events, dispose } = feedFor(sessions) expect(events.at(-1)).toMatchObject({ type: 'snapshot', sessions: [{ hostExecutionOwned: true, hostExecutionPhase: 'starting' }] }) - sessions.set(SESSION, { ...session, providerChildPhase: 'ready' }) + sessions.set(SESSION, { ...session, child: { phase: 'ready' } }) feed.publish(SESSION, journal) expect(events.at(-1)).toMatchObject({ session: { hostExecutionPhase: 'ready' } }) - sessions.set(SESSION, { ...session, hasProviderChild: false }) + sessions.set(SESSION, { ...session, child: null }) feed.publish(SESSION, journal) expect(events.at(-1)).not.toMatchObject({ session: { hostExecutionPhase: expect.any(String) } }) dispose() @@ -111,7 +116,7 @@ describe('StructuredAgentSessionStatusFeed', () => { it('publishes provider ownership transitions without changing journal time', async () => { const journal = await openJournal() - const sessions = new Map([[SESSION, { journal, hasProviderChild: true }]]) + const sessions = new Map([[SESSION, { journal, child: { phase: 'ready' } }]]) const { feed, events, dispose } = feedFor(sessions) events.length = 0 await journal.appendItem( @@ -130,7 +135,7 @@ describe('StructuredAgentSessionStatusFeed', () => { throw new Error('status publication missing') } const journalTime = firstStatus.session.updatedAt - sessions.get(SESSION)!.hasProviderChild = false + sessions.get(SESSION)!.child = null feed.publish(SESSION, journal) expect(events.at(-1)).toEqual({ type: 'status', @@ -167,8 +172,10 @@ describe('StructuredAgentSessionStatusFeed', () => { it('stops projecting an old-host unknown submission after the owner fence advances', async () => { const journal = await openJournal() - const session = { journal, fence: 1 } - const { feed, events } = feedFor(new Map([[SESSION, session]])) + // The conversation's fence is the record's: a child's end moves it. + const lease = agentSessionRecordFixture().lease + const record = agentSessionRecordFixture({ ...lease, runtimeFence: 1 }) + const { feed, events } = feedFor(new Map([[SESSION, { journal }]]), record) await journal.appendSubmission({ clientMessageId: 'old-host', payloadFingerprint: 'fp', @@ -183,7 +190,7 @@ describe('StructuredAgentSessionStatusFeed', () => { }) feed.publish(SESSION) expect(events.at(-1)).toMatchObject({ session: { status: 'working' } }) - session.fence = 2 + record.lease.runtimeFence = 2 feed.publish(SESSION) expect(events.at(-1)).toMatchObject({ session: { status: 'idle' } }) }) @@ -793,7 +800,7 @@ describe('the status sink sees the roster the broadcast cache deliberately lacks it('receives every change once, ownership revocation, and the forget edge', async () => { const journal = await openJournal() - const sessions = new Map([[SESSION, { journal, hasProviderChild: true }]]) + const sessions = new Map([[SESSION, { journal, child: { phase: 'ready' } }]]) const { sink, published, forgotten } = sinkFor() const { feed } = feedFor(sessions, null, undefined, undefined, sink) await journal.appendItem( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts index 29bf29a00f1..5403ec16abe 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.ts @@ -41,9 +41,7 @@ export type StructuredAgentSessionStatusSubscriber = { type StatusFeedSession = { journal: AgentSessionJournal params: { location: AgentSessionRecord['location']; provider: AgentSessionRecord['provider'] } - hasProviderChild?: boolean - providerChildPhase?: StructuredAgentSessionProviderChildPhase - fence?: number + child?: { phase: StructuredAgentSessionProviderChildPhase } | null } export type StructuredAgentSessionStatusFeedDeps = { @@ -238,7 +236,9 @@ export class StructuredAgentSessionStatusFeed { // An unreadable journal projects as "no turn": the chat itself shows the reset. const cursor = journal.cursor() const readOnly = journal.isReadOnly - const fence = session.fence + const record = this.deps.getRecord(sessionId) + // The conversation's fence, which a child's end moves: its unanswered sends stop counting. + const fence = record?.lease.runtimeFence let projection = this.journalProjections.get(journal) if ( !projection || @@ -262,7 +262,6 @@ export class StructuredAgentSessionStatusFeed { } this.journalProjections.set(journal, projection) } - const record = this.deps.getRecord(sessionId) const providerSession = structuredAgentSessionProviderSessionMetadata(record) // The journal has no model: the record's acknowledged options are where a mid-session // switch lands, so the row follows whichever is in force. @@ -277,13 +276,8 @@ export class StructuredAgentSessionStatusFeed { sessionId, workspaceId: session.params.location.workspaceId, agent: session.params.provider, - ...(session.hasProviderChild - ? { - hostExecutionOwned: true as const, - ...(session.providerChildPhase - ? { hostExecutionPhase: session.providerChildPhase } - : {}) - } + ...(session.child + ? { hostExecutionOwned: true as const, hostExecutionPhase: session.child.phase } : {}), ...projection.summary, ...(record?.rewind?.phase === 'prepared' || record?.rewind?.phase === 'provider-succeeded' diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-recency.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-recency.test.ts index cf908932667..4053988233f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-recency.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subagent-recency.test.ts @@ -63,7 +63,9 @@ async function openSession() { const roster: { tasks: AgentSessionBackgroundTask[] } = { tasks: [] } const server = new AgentHookServer() const feed = new StructuredAgentSessionStatusFeed({ - sessions: new Map([[SESSION, indexedStatusFeedSession({ journal, hasProviderChild: true })]]), + sessions: new Map([ + [SESSION, indexedStatusFeedSession({ journal, child: { phase: 'ready' } })] + ]), getRecord: () => null, now: () => 1, readBackgroundTasks: () => ({ state: 'monitoring', tasks: roster.tasks }), 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 c8b8839318f..1e0344a7b07 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 @@ -287,12 +287,10 @@ describe('a chat that closes', () => { }) .mockImplementation(closeJournal) - await expect(host.close(SESSION)).rejects.toMatchObject({ - step: 'forget-session', - cause: expect.objectContaining({ message: 'journal close result lost' }) - }) + // The child stopped and its lease went back; only the conversation's close is left to retry. + await expect(host.close(SESSION)).rejects.toThrow('journal close result lost') expect(host.hasSession(SESSION)).toBe(true) - expect(host['sessions'].get(SESSION)?.hasProviderChild).toBe(false) + expect(host['sessions'].get(SESSION)?.child).toBeNull() expect(store.getRecord(SESSION)?.lease).toMatchObject({ claimStatus: 'released', ownerProcess: null @@ -330,7 +328,7 @@ describe('a chat that closes', () => { await expect(host.close(SESSION)).rejects.toMatchObject({ step: 'drain-published' }) // The child is proven gone, but the wind-down it owes is not done: nothing settled, no release. - expect(session!.hasProviderChild).toBe(false) + expect(session!.child).toBeNull() expect(store.getRecord(SESSION)?.lease.claimStatus).not.toBe('released') await expect(host.close(SESSION)).resolves.toBeUndefined() @@ -784,7 +782,7 @@ describe('a quit over an eviction that never got its retry', () => { failNextDrain() await expect(host.close(SESSION)).rejects.toMatchObject({ step: 'drain-published' }) - expect(host['sessions'].get(SESSION)?.hasProviderChild).toBe(false) + expect(host['sessions'].get(SESSION)?.child).toBeNull() expect(store.getRecord(SESSION)?.lease.claimStatus).not.toBe('released') await host.flushAllStreamedEvents() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts index 79f21a349b4..952fa70657d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unanswered-dispatch-release.test.ts @@ -32,10 +32,12 @@ function contextWith(submissions: AgentJournalSubmission[]) { return { epoch: 'e', sequence: 1 } }) } - // The mutation reads only `journal.submissions`, `journal.resolveDispatch` and `fence`. - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: all three are supplied here; the rest of the host session is unreachable from this mutation. - const session = { journal, fence: FENCE } as unknown as ReleaseSession - const context: ReleaseContext = { sessions: new Map([['s-1', session]]) } + // The mutation reads only `journal.submissions`, `journal.resolveDispatch` and the record fence. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: all three are supplied here; the rest of the session and the record is unreachable from this mutation. + const context = { + sessions: new Map([['s-1', { journal }]]), + deps: { store: { getRecord: () => ({ lease: { runtimeFence: FENCE } }) } } + } as unknown as ReleaseContext return { context, resolved, journal } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts index bba6154af43..5aae86f3291 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts @@ -33,9 +33,14 @@ function recoveryContext(input: { resumeCapable?: boolean }) { const session = { - hasProviderChild: false, - fence: 8, - acquisitionGeneration: input.generation ?? GENERATION + child: null, + lastEndedChild: { + generation: input.generation ?? GENERATION, + fence: 7, + cause: 'exit', + reason: null, + duringStartup: false + } } as StructuredAgentSessionHostSession const record = { lease: { @@ -128,9 +133,7 @@ describe('provider-exit recovery tickets', () => { .mockRejectedValueOnce(new Error('journal unavailable')) .mockResolvedValue({ epoch: 'epoch-1', sequence: 2 }) const session = { - hasProviderChild: true, - fence: 7, - acquisitionGeneration: GENERATION, + child: { generation: GENERATION, fence: 7, phase: 'ready' }, journal: { snapshot: () => ({ items: [lifecycleItem('turn-1', 1, { state: 'running', startedAt: 1_000 })] @@ -174,7 +177,7 @@ describe('provider-exit recovery tickets', () => { retryLoadedStructuredAgentSessionSettlement({ deps: { store } as never, sessionId: SESSION, - session: { journal: session.journal, fence: 8, acquisitionGeneration: null }, + journal: session.journal, now: () => now }) ).resolves.toBe(true) @@ -199,9 +202,7 @@ describe('provider-exit recovery tickets', () => { lifecycleItem('turn-2', 2, { state: 'running', startedAt: 30 }) ] const session = { - hasProviderChild: true, - fence: 7, - acquisitionGeneration: GENERATION, + child: { generation: GENERATION, fence: 7, phase: 'ready' }, journal: { snapshot: () => ({ items }), appendLifecycleBatch, @@ -250,7 +251,7 @@ describe('provider-exit recovery tickets', () => { 7, 'provider_exited_before_acknowledgement' ) - expect(session.hasProviderChild).toBe(false) + expect(session.child).toBeNull() // The running row is revised to interrupted at exit receipt, never tombstoned. expect(appendLifecycleBatch).toHaveBeenCalledExactlyOnceWith({ settlementId: `dead-generation:provider-exit:${SESSION}:7:${GENERATION}`, @@ -307,9 +308,7 @@ describe('provider-exit recovery tickets', () => { sequence: 3 })) const session: StructuredAgentSessionUnexpectedExitSession = { - hasProviderChild: true, - fence: 7, - acquisitionGeneration: GENERATION, + child: { generation: GENERATION, fence: 7, phase: 'ready' }, journal: { snapshot: () => ({ items }), appendLifecycleBatch, @@ -365,9 +364,7 @@ describe('provider-exit recovery tickets', () => { it('settles a submission the dead child never acknowledged', async () => { const markPendingSubmissionsUnknown = vi.fn(async () => ['client-1']) const session: StructuredAgentSessionUnexpectedExitSession = { - hasProviderChild: true, - fence: 7, - acquisitionGeneration: GENERATION, + child: { generation: GENERATION, fence: 7, phase: 'ready' }, journal: { snapshot: () => ({ items: [] }), appendLifecycleBatch: vi.fn(async () => ({ epoch: 'epoch-1', sequence: 1 })), @@ -414,9 +411,7 @@ describe('provider-exit recovery tickets', () => { it('does not release or reacquire while terminal settlement retry is still failing', async () => { const session: StructuredAgentSessionUnexpectedExitSession = { - hasProviderChild: true, - fence: 7, - acquisitionGeneration: GENERATION, + child: { generation: GENERATION, fence: 7, phase: 'ready' }, journal: { markPendingSubmissionsUnknown: vi.fn(async () => []), rejectPendingSubmissions: vi.fn(async () => []), @@ -453,8 +448,7 @@ describe('provider-exit recovery tickets', () => { const result = await settleUnexpectedStructuredAgentSessionExit(context, event) expect(result).toBeNull() - expect(session.hasProviderChild).toBe(false) - expect(session.fence).toBe(8) + expect(session.child).toBeNull() expect(publishFence).toHaveBeenCalledTimes(1) expect(release).toHaveBeenCalledTimes(2) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.ts index 7e714838311..212963b8539 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.ts @@ -1,8 +1,6 @@ -import type { - StructuredAgentSessionEndedEvent, - StructuredAgentSessionProviderChildPhase -} from './structured-agent-session-adapter' +import type { StructuredAgentSessionEndedEvent } from './structured-agent-session-adapter' import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' +import { endProviderChild } from './structured-agent-session-provider-child' import { releaseStoredStructuredAgentSessionOwnerAfterUnexpectedExit, type StructuredAgentSessionLeaseStore @@ -30,13 +28,10 @@ export type StructuredAgentSessionRecoveryTicket = { stableSettlementId: string } -export type StructuredAgentSessionUnexpectedExitSession = { - journal: DeadGenerationJournal - hasProviderChild: boolean - fence: number - acquisitionGeneration: string | null - providerChildPhase?: StructuredAgentSessionProviderChildPhase -} +export type StructuredAgentSessionUnexpectedExitSession = Pick< + StructuredAgentSessionHostSession, + 'child' | 'lastEndedChild' +> & { journal: DeadGenerationJournal } export type StructuredAgentSessionUnexpectedExitContext< TSession extends StructuredAgentSessionUnexpectedExitSession = StructuredAgentSessionHostSession @@ -66,24 +61,35 @@ export async function settleUnexpectedStructuredAgentSessionExit< const observedAt = event.observedAt ?? context.now() return context.serialize(unexpectedEvent.sessionId, async () => { const session = context.sessions.get(unexpectedEvent.sessionId) + const child = session?.child if ( - !session?.hasProviderChild || - session.fence !== unexpectedEvent.fence || - session.acquisitionGeneration !== unexpectedEvent.acquisitionGeneration + !session || + !child || + child.fence !== unexpectedEvent.fence || + child.generation !== unexpectedEvent.acquisitionGeneration ) { return null } - const record = context.store.getRecord(unexpectedEvent.sessionId) - if (!record || record.lease.handoffStage !== null) { - // An acquisition or recovery already owns this lease's transition. - session.hasProviderChild = false - context.publishStatus?.(unexpectedEvent.sessionId) - return null - } // The host's own phase decides, so a provider that omits the flag still gets a start that // failed told as one: the row says so, and nothing resumes into the same failure. const exitedDuringStartup = - unexpectedEvent.startupUnproven === true || session.providerChildPhase === 'starting' + unexpectedEvent.startupUnproven === true || child.phase === 'starting' + const endChild = (): void => { + endProviderChild(session, { + generation: child.generation, + fence: child.fence, + cause: 'exit', + reason: unexpectedEvent.reason, + duringStartup: exitedDuringStartup + }) + context.publishStatus?.(unexpectedEvent.sessionId) + } + const record = context.store.getRecord(unexpectedEvent.sessionId) + if (!record || record.lease.handoffStage !== null) { + // An acquisition or recovery already owns this lease's transition. + endChild() + return null + } let settlementFailed = false const stableSettlementId = providerExitSettlementId(unexpectedEvent) @@ -103,7 +109,8 @@ export async function settleUnexpectedStructuredAgentSessionExit< settlementFailed = !(await retryUnexpectedExitSettlement({ context, event: unexpectedEvent, - session, + journal: session.journal, + fence: child.fence, stableSettlementId, verdict: { state: 'interrupted', completedAt: observedAt }, exitedDuringStartup, @@ -125,7 +132,7 @@ export async function settleUnexpectedStructuredAgentSessionExit< sessionId: unexpectedEvent.sessionId, expectedFence: unexpectedEvent.fence, expectedAcquisitionGeneration: unexpectedEvent.acquisitionGeneration, - acquisitionGeneration: session.acquisitionGeneration, + acquisitionGeneration: child.generation, now: context.now(), exitObservedAt: observedAt, ...(settlementFailed @@ -141,10 +148,8 @@ export async function settleUnexpectedStructuredAgentSessionExit< } catch (error) { context.onBarrierError?.(unexpectedEvent.sessionId, error) } finally { - session.hasProviderChild = false - context.publishStatus?.(unexpectedEvent.sessionId) + endChild() if (released) { - session.fence = released.lease.runtimeFence context.publishFence(unexpectedEvent.sessionId, session) } } @@ -170,10 +175,7 @@ export function isStructuredAgentSessionRecoveryTicketCurrent( store: Pick sessions: Map< string, - Pick< - StructuredAgentSessionUnexpectedExitSession, - 'hasProviderChild' | 'fence' | 'acquisitionGeneration' - > + Pick > hasResumeCapableHolder: (sessionId: string) => boolean }, @@ -182,9 +184,9 @@ export function isStructuredAgentSessionRecoveryTicketCurrent( const session = context.sessions.get(ticket.sessionId) const record = context.store.getRecord(ticket.sessionId) return ( - session?.hasProviderChild === false && - session.fence === ticket.releasedFence && - session.acquisitionGeneration === ticket.deadAcquisitionGeneration && + session !== undefined && + session.child === null && + session.lastEndedChild?.generation === ticket.deadAcquisitionGeneration && record?.lease.runtimeFence === ticket.releasedFence && record.lease.claimStatus === 'released' && record.lease.handoffStage === null && @@ -195,16 +197,17 @@ export function isStructuredAgentSessionRecoveryTicketCurrent( async function retryUnexpectedExitSettlement(input: { context: Pick event: UnexpectedExitLifecycleEvent - session: Pick + journal: DeadGenerationJournal + fence: number stableSettlementId: string verdict: StructuredAgentSessionTurnVerdict exitedDuringStartup: boolean showUnexpectedExitOutcome?: boolean }): Promise { return settleStructuredAgentSessionDeadGeneration({ - journal: input.session.journal, + journal: input.journal, sessionId: input.event.sessionId, - fence: input.session.fence, + fence: input.fence, settlementId: input.stableSettlementId, verdict: input.verdict, pendingSubmissionReason: 'provider_exited_before_acknowledgement', diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-working-at-teardown.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-working-at-teardown.ts index 5a3c2497c34..820f4c3c95d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-working-at-teardown.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-working-at-teardown.ts @@ -138,8 +138,7 @@ function liveTasks( type WorkingCandidateSession = { journal: AgentSessionJournal /** Only this host generation's own child counts. A restored-for-reading journal has none. */ - hasProviderChild: boolean - fence?: number + child: { fence: number } | null } /** The offer one session is owed, taken right before teardown stops its provider child; null when @@ -157,12 +156,12 @@ export function structuredAgentSessionWorkingAtStop(input: { }): AgentSessionResumeMarker | null { const { sessionId, session } = input // A journal this host cannot read tells us nothing about what the turn was doing. - if (!session?.hasProviderChild || session.journal.isReadOnly) { + if (!session?.child || session.journal.isReadOnly) { return null } const snapshot = session.journal.snapshot() const roster = input.backgroundTasks(sessionId) - const status = structuredAgentSessionShownStatus(snapshot, roster, session.fence) + const status = structuredAgentSessionShownStatus(snapshot, roster, session.child.fence) if (status.state === 'done') { return null } diff --git a/src/main/runtime/runtime-worktree-structured-agent-rows-liveness.test.ts b/src/main/runtime/runtime-worktree-structured-agent-rows-liveness.test.ts index b1e43a9025d..27d5ba48b29 100644 --- a/src/main/runtime/runtime-worktree-structured-agent-rows-liveness.test.ts +++ b/src/main/runtime/runtime-worktree-structured-agent-rows-liveness.test.ts @@ -89,7 +89,7 @@ async function awaitingApproval() { SESSION, { journal, - hasProviderChild: true, + child: { phase: 'ready' as const }, params: { location: { executionHostId: 'local' as const, diff --git a/tests/e2e/codex-child-approval-activity-row.unit.test.ts b/tests/e2e/codex-child-approval-activity-row.unit.test.ts index b9210bf79e8..71bb0fddce1 100644 --- a/tests/e2e/codex-child-approval-activity-row.unit.test.ts +++ b/tests/e2e/codex-child-approval-activity-row.unit.test.ts @@ -143,7 +143,9 @@ async function openHost() { journalDir: join(root, SESSION) }) const feed = new StructuredAgentSessionStatusFeed({ - sessions: new Map([[SESSION, indexedStatusFeedSession({ journal, hasProviderChild: true })]]), + sessions: new Map([ + [SESSION, indexedStatusFeedSession({ journal, child: { phase: 'ready' } })] + ]), getRecord: () => null, now: () => 1, readBackgroundTasks: () => ({ state: 'monitoring', tasks: [] })