From ff1d07b0f35f682ee4f8e4fa2bd6785b2bc9c256 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 14 Sep 2026 15:01:23 -0700 Subject: [PATCH] Preserve generation-scoped agent-session settlement work --- ...e-mobile-structured-agent-session.test.tsx | 24 +++ .../use-mobile-structured-agent-session.ts | 31 ++- .../journal-pending-submission-recovery.ts | 10 +- .../agent-session-journal/journal-reducer.ts | 5 +- .../agent-session-journal/journal-store.ts | 13 +- .../structured-agent-session-attach-flow.ts | 6 +- ...ured-agent-session-attach-orchestration.ts | 56 +++++- ...tructured-agent-session-client-delivery.ts | 40 ++++ ...session-dead-generation-settlement.test.ts | 106 +++++++++- ...gent-session-dead-generation-settlement.ts | 98 ++++++++-- ...tured-agent-session-handoff-queue-start.ts | 12 +- .../structured-agent-session-handoff-queue.ts | 21 +- ...ured-agent-session-handoff-revalidation.ts | 11 +- ...structured-agent-session-handoff-status.ts | 18 +- .../structured-agent-session-handoff.ts | 17 +- ...structured-agent-session-history-result.ts | 6 +- .../structured-agent-session-host-lifetime.ts | 24 ++- .../structured-agent-session-host.ts | 12 ++ .../structured-agent-session-lease-release.ts | 4 +- ...red-agent-session-owner-projection.test.ts | 183 ++++++++++++++++++ ...ructured-agent-session-owner-projection.ts | 101 ++++++++++ .../structured-agent-session-prompt-state.ts | 5 +- ...ructured-agent-session-settlement-retry.ts | 47 ++++- .../structured-agent-session-status-feed.ts | 10 +- ...ructured-agent-session-subscribers.test.ts | 83 ++++++++ .../structured-agent-session-subscribers.ts | 35 +++- ...red-agent-session-surface-lifetime.test.ts | 159 ++++++++++++++- ...ured-agent-session-unexpected-exit.test.ts | 10 +- ...tructured-agent-session-unexpected-exit.ts | 11 +- ...red-conversation-command-admission.test.ts | 56 ++++++ ...ructured-conversation-command-admission.ts | 16 +- .../agent-session-lease-transitions.ts | 16 +- .../agent-session-record-store.test.ts | 59 ++++++ ...agent-session-restart-lease-transitions.ts | 3 +- ...gent-session-surface-release-transition.ts | 22 ++- .../structured-mailbox-pointer-host.test.ts | 4 +- .../structured-mailbox-pointer-host.ts | 2 +- ...d-agent-session-message-projection.test.ts | 31 ++- ...ctured-agent-session-message-projection.ts | 6 +- ...e-structured-agent-session-outbox.test.tsx | 65 +++++++ .../use-structured-agent-session-outbox.ts | 26 ++- .../use-structured-agent-session.ts | 9 +- src/shared/agent-session-journal-types.ts | 2 + src/shared/agent-session-record.ts | 4 + src/shared/structured-agent-session-outbox.ts | 27 ++- .../structured-agent-session-projection.ts | 21 +- ...ructured-agent-session-send-disposition.ts | 5 +- 47 files changed, 1396 insertions(+), 136 deletions(-) create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-owner-projection.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-owner-projection.ts diff --git a/mobile/src/session/use-mobile-structured-agent-session.test.tsx b/mobile/src/session/use-mobile-structured-agent-session.test.tsx index bcc24c5ed32..06d556fef33 100644 --- a/mobile/src/session/use-mobile-structured-agent-session.test.tsx +++ b/mobile/src/session/use-mobile-structured-agent-session.test.tsx @@ -243,6 +243,30 @@ describe('useMobileStructuredAgentSession', () => { sendRequest, subscribe } as unknown as RpcClient + + it('ignores old-owner prompts and running turns while retaining the transcript', async () => { + act(() => { + renderer = create(createElement(Harness)) + }) + await vi.waitFor(() => expect(listener).toEqual(expect.any(Function))) + act(() => + listener?.({ + ...snapshotEvent(4), + page: { + ...snapshotEvent(4).page, + items: [ + { ...approvalItem(), ownerFence: 3 }, + { ...questionItem(), ownerFence: 3 }, + { ...runningStatusItem(), ownerFence: 3 } + ] + } + }) + ) + expect(hook!.permission).toBeNull() + expect(hook!.question).toBeNull() + expect(hook!.turnId).toBeNull() + expect(hook!.isWorking).toBe(false) + }) let storedOperations: Map function Harness({ diff --git a/mobile/src/session/use-mobile-structured-agent-session.ts b/mobile/src/session/use-mobile-structured-agent-session.ts index 2a24d38a38e..3a7a5150bfc 100644 --- a/mobile/src/session/use-mobile-structured-agent-session.ts +++ b/mobile/src/session/use-mobile-structured-agent-session.ts @@ -7,7 +7,10 @@ import { import { encodeNativeChatTranscriptIdentity } from '../../../src/shared/native-chat-transcript-retention' import type { MobileNativeChatSendOutcome } from './mobile-native-chat-send' import { projectStructuredAgentSessionMessages } from '../../../src/shared/structured-agent-session-message-projection' -import { hasUnansweredStructuredAgentSessionDispatch } from '../../../src/shared/structured-agent-session-projection' +import { + hasUnansweredStructuredAgentSessionDispatch, + liveStructuredAgentSessionItems +} from '../../../src/shared/structured-agent-session-projection' import { activeStructuredAgentSessionTurnId, isStructuredAgentSessionThinking @@ -196,8 +199,10 @@ export function useMobileStructuredAgentSession(args: { conversationCommands }, canRun: () => - !activeStructuredAgentSessionTurnId(stateRef.current.items) && - !stateRef.current.items.some( + !activeStructuredAgentSessionTurnId( + liveStructuredAgentSessionItems(stateRef.current.items, stateRef.current.fence) + ) && + !liveStructuredAgentSessionItems(stateRef.current.items, stateRef.current.fence).some( (item) => pendingStructuredApproval(item) || pendingStructuredQuestion(item) ), onError: onSendError, @@ -263,20 +268,26 @@ export function useMobileStructuredAgentSession(args: { () => projectStructuredAgentSessionMessages(state.items, [], state.submissions), [state.items, state.submissions] ) - const turnId = activeStructuredAgentSessionTurnId(state.items) + const ownerItems = useMemo( + () => liveStructuredAgentSessionItems(state.items, state.fence), + [state.items, state.fence] + ) + const turnId = activeStructuredAgentSessionTurnId(ownerItems) const turnTiming = useMobileStructuredAgentTurnTiming(state, turnId) const activityText = - selectStructuredAgentTurnActivity(state.items, turnId, state.activity)?.text ?? null - const thinking = isStructuredAgentSessionThinking(state.items) + selectStructuredAgentTurnActivity(ownerItems, turnId, state.activity)?.text ?? null + const thinking = isStructuredAgentSessionThinking(ownerItems) + // Stable while the readings hold, so a streaming turn does not re-render the + // whole chat surface on every journal batch. const turnIndicator = useMemo(() => ({ thinking, activityText }), [thinking, activityText]) const status = state.status === 'idle' ? 'idle' : state.status const approvalPrompt = useMemo( - () => state.items.find(pendingStructuredApproval) ?? null, - [state.items] + () => ownerItems.find(pendingStructuredApproval) ?? null, + [ownerItems] ) const questionPrompt = useMemo( - () => state.items.find(pendingStructuredQuestion) ?? null, - [state.items] + () => ownerItems.find(pendingStructuredQuestion) ?? null, + [ownerItems] ) return { ...options, diff --git a/src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts b/src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts index c3b9734bb40..a8bdd4c1d4a 100644 --- a/src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts +++ b/src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts @@ -7,14 +7,18 @@ export async function markJournalPendingSubmissionsUnknown( journal: AgentSessionJournal, fence: number, _boundary: { mode: 'death-confirmed' | 'new-owner-not-publishing' }, - reason: string = DISPATCH_DOUBT_HOST_RESTARTED + reason: string = DISPATCH_DOUBT_HOST_RESTARTED, + throughFence: number = fence, + fromFence = 0 ): Promise { const unresolved = journal .submissions() .filter( (entry) => - entry.dispatchState === 'pending' || - (entry.dispatchState === 'unknown' && entry.recovered !== true) + entry.fence >= fromFence && + entry.fence <= throughFence && + (entry.dispatchState === 'pending' || + (entry.dispatchState === 'unknown' && entry.recovered !== true)) ) for (const entry of unresolved) { // An earlier reason already names a sharper fact than "the host restarted". diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index 5c4917bec46..6093446dcdc 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -79,6 +79,7 @@ export function applyJournalRow(state: JournalReducerState, row: JournalRow): vo body: row.body, sequence: row.seq, observedAt: row.ts, + ownerFence: row.fence, ...(row.recovered ? { recovered: row.recovered } : {}) }) return @@ -104,6 +105,7 @@ export function applyJournalRow(state: JournalReducerState, row: JournalRow): vo body: mutation.body, sequence: row.seq, observedAt: row.ts, + ownerFence: row.fence, ...(row.recovered ? { recovered: row.recovered } : {}) }) } else { @@ -218,7 +220,8 @@ function upsertItem( // Provider history may normalize text or omit local attachments from the original send. body: submitted ? existing.body : next.body, sequence: existing.sequence, - observedAt: existing.observedAt + observedAt: existing.observedAt, + ownerFence: existing.ownerFence }) state.tombstones.delete(itemId) } diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index 60b302154a2..ac2db984b3b 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -256,9 +256,18 @@ export class AgentSessionJournal { async markPendingSubmissionsUnknown( fence: number, boundary: { mode: 'death-confirmed' | 'new-owner-not-publishing' }, - reason?: string + reason?: string, + throughFence?: number, + fromFence?: number ): Promise { - return markJournalPendingSubmissionsUnknown(this, fence, boundary, reason) + return markJournalPendingSubmissionsUnknown( + this, + fence, + boundary, + reason, + throughFence, + fromFence + ) } /** The escape hatch for corruption, an unreconcilable prefix, a forked handle, 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 63889a5b8e6..89653467112 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 @@ -33,6 +33,7 @@ import { adapterSupportsCreateIfDeclared } from './structured-agent-session-prov import type { StructuredAgentSessionEventSink } from './structured-agent-session-event-sink' import { resolveAgentSessionReplayOutcome } from './structured-agent-session-replay-outcome' import { readAgentSessionHydrationPage } from './agent-session-history-page' +import { projectStructuredAgentSessionOwnerPage } from './structured-agent-session-owner-projection' import { acquireOwner } from './structured-agent-session-acquisition' import { importAdoptedTranscript, @@ -252,7 +253,10 @@ export async function performAttach( value: { sessionId, fence, - page: readAgentSessionHydrationPage(attached.journal, fence), + page: projectStructuredAgentSessionOwnerPage( + readAgentSessionHydrationPage(attached.journal, fence), + store.getRecord(sessionId) + ), unconfirmedClientMessageIds: attached.unconfirmedClientMessageIds } } 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 a00bbf33020..d9d9795b6bd 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 @@ -21,7 +21,10 @@ import { pinnedAgentSessionLaunchEnv } from './structured-agent-session-launch-env' import { refuseAgentSessionMutation } from './structured-agent-session-mutation-admission' -import { turnVerdictFromDeathEvidence } from './structured-agent-session-stale-turn-verdict' +import { + turnVerdictFromDeathEvidence, + UNVERIFIABLE_TURN_VERDICT +} from './structured-agent-session-stale-turn-verdict' import { captureUnfinishedStructuredAgentSessionWork, settleStructuredAgentSessionDeadGeneration, @@ -120,14 +123,33 @@ export function attachStructuredAgentSession( // The new child's events stay buffered until bindAndDrain, so every pending row // here predates it. A failed settlement cannot withhold the new writer. const verdict = turnVerdictFromDeathEvidence(previousLease?.deathEvidence) - const work = captureUnfinishedStructuredAgentSessionWork(attached.journal) - await settleStructuredAgentSessionDeadGeneration({ + const throughFence = previousLease?.settlementRetryFence ?? fence - 1 + const settlementId = + previousLease?.settlementRetryId ?? + `stale-generation:${sessionId}:${fence}:${acquisitionGeneration ?? 'unknown'}` + const priorSettled = await settleStructuredAgentSessionDeadGeneration({ journal: attached.journal, sessionId, fence, - settlementId: - previousLease?.settlementRetryId ?? - `stale-generation:${sessionId}:${fence}:${acquisitionGeneration ?? 'unknown'}`, + throughFence: throughFence - 1, + settlementId: `${settlementId}:prior`, + verdict: UNVERIFIABLE_TURN_VERDICT, + pendingSubmissionReason: 'provider_exited_before_acknowledgement', + showUnexpectedExitOutcome: false, + onError: (id, error) => context.deps.onEventSinkError?.({ sessionId: id, error }) + }) + const work = captureUnfinishedStructuredAgentSessionWork( + attached.journal, + throughFence, + throughFence + ) + const settled = await settleStructuredAgentSessionDeadGeneration({ + journal: attached.journal, + sessionId, + fence, + throughFence, + fromFence: throughFence, + settlementId, verdict, pendingSubmissionReason: 'provider_exited_before_acknowledgement', submissionRecoveryMode: 'new-owner-not-publishing', @@ -137,7 +159,9 @@ export function attachStructuredAgentSession( unfinishedStructuredAgentSessionWorkWasInterrupted( work, attached.journal, - verdict.completedAt + verdict.completedAt, + throughFence, + throughFence )), // A later generation has cleared the old exit detail; use generic copy then. ...(previousLease?.settlementRetryRequired && previousLease.deathEvidence?.detail @@ -148,6 +172,24 @@ export function attachStructuredAgentSession( console.error('agent-session dead-generation settlement deferred', id, error) } }) + if (settled && priorSettled && previousLease?.settlementRetryRequired) { + try { + await context.deps.store.transitionHandoff(sessionId, (latest) => ({ + ...latest, + lease: + latest.lease.settlementRetryId === previousLease.settlementRetryId + ? { + ...latest.lease, + settlementRetryRequired: undefined, + settlementRetryId: undefined, + settlementRetryFence: undefined + } + : latest.lease + })) + } catch (error) { + context.deps.onEventSinkError?.({ sessionId, error }) + } + } } await bindAndDrain(eventSink, attached.journal, fence, (activity) => context.subscribers.publish(sessionId, attached.journal, activity) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts index 92a4f2653e0..8d85f54f826 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-client-delivery.ts @@ -6,6 +6,10 @@ import type { } from './structured-agent-session-host-types' import { AGENT_SESSION_NOT_ATTACHED } from './structured-agent-session-mutation-admission' import { StructuredAgentSessionSendSettlement } from './structured-agent-session-send-settlement' +import { + projectStructuredAgentSessionOwnerPage, + structuredAgentSessionNeedsOwnerSnapshot +} from './structured-agent-session-owner-projection' import { createStructuredAgentSessionHostStatusFeed, type StructuredAgentSessionStatusSubscriber @@ -17,6 +21,16 @@ export class StructuredAgentSessionClientDelivery { readonly waitForSendSettlement: StructuredAgentSessionSendSettlement['wait'] private readonly statusFeed private readonly sendSettlement + private readonly ownerSnapshotCache = new WeakMap< + AgentSessionJournal, + { + epoch: string + sequence: number + fence: number | null + claimStatus: string | null + required: boolean + } + >() constructor( private readonly sessions: Map, @@ -30,6 +44,11 @@ export class StructuredAgentSessionClientDelivery { this.waitForSendSettlement = this.sendSettlement.wait this.subscribers = new AgentSessionSubscribers({ readCommands: (sessionId) => deps().adapter.readCommands?.(sessionId), + readJournal: (sessionId) => sessions.get(sessionId)?.journal, + needsOwnerSnapshot: (sessionId, journal) => + this.needsOwnerSnapshot(journal, deps().store.getRecord(sessionId)), + projectPage: (sessionId, page) => + projectStructuredAgentSessionOwnerPage(page, deps().store.getRecord(sessionId)), onJournalPublished: (sessionId, journal) => this.publishJournal(sessionId, journal) }) } @@ -65,6 +84,27 @@ export class StructuredAgentSessionClientDelivery { this.sendSettlement.publish(sessionId, journal) } + private needsOwnerSnapshot( + journal: AgentSessionJournal, + record: ReturnType + ): boolean { + const cursor = journal.cursor() + const cached = this.ownerSnapshotCache.get(journal) + const fence = record?.lease.runtimeFence ?? null + const claimStatus = record?.lease.claimStatus ?? null + if ( + cached?.epoch === cursor.epoch && + cached.sequence === cursor.sequence && + cached.fence === fence && + cached.claimStatus === claimStatus + ) { + return cached.required + } + const required = structuredAgentSessionNeedsOwnerSnapshot(journal.snapshot(), record) + this.ownerSnapshotCache.set(journal, { ...cursor, fence, claimStatus, required }) + return required + } + private requireJournal(sessionId: string): AgentSessionJournal { const journal = this.sessions.get(sessionId)?.journal if (!journal) { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts index c8566e71411..4575d3f7efd 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts @@ -1,7 +1,7 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' -import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { openAgentSessionJournal } from '../agent-session-journal/journal-store-factory' import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' @@ -79,6 +79,110 @@ async function seedUnfinishedWork(): Promise { } describe('dead structured-session generation settlement', () => { + it('re-derives uncommitted chunks after an earlier lifecycle chunk committed', async () => { + for (let ordinal = 1; ordinal <= 205; ordinal += 1) { + await journal.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'bulk', ordinal }, + { kind: 'tool-call', name: 'shell', input: {}, state: 'running' }, + { fence: 7 } + ) + } + const append = journal.appendLifecycleBatch.bind(journal) + let calls = 0 + vi.spyOn(journal, 'appendLifecycleBatch').mockImplementation(async (input) => { + calls += 1 + if (calls === 2) { + throw new Error('second chunk unavailable') + } + return append(input) + }) + const settle = () => + settleStructuredAgentSessionDeadGeneration({ + journal, + sessionId: SESSION, + fence: 8, + throughFence: 7, + settlementId: 'bulk-old-owner', + pendingSubmissionReason: 'provider_exited_before_acknowledgement', + verdict: { state: 'unverifiable' }, + showUnexpectedExitOutcome: false + }) + + expect(await settle()).toBe(false) + expect( + journal + .snapshot() + .items.some((item) => item.body.kind === 'tool-call' && item.body.state === 'running') + ).toBe(true) + expect(await settle()).toBe(true) + expect( + journal + .snapshot() + .items.filter((item) => item.body.kind === 'tool-call' && item.body.state === 'failed') + ).toHaveLength(205) + }) + + it('re-derives only the dead owner after a replacement has published new work', async () => { + await seedUnfinishedWork() + await journal.appendSubmission({ + clientMessageId: 'client-new', + payloadFingerprint: 'new-fingerprint', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'new turn' }] }, + fence: 8 + }) + await journal.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-new', ordinal: 1 }, + { kind: 'turn', turnId: 'turn-new', state: 'running' }, + { fence: 8 } + ) + await journal.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-new', ordinal: 2 }, + { + kind: 'question', + question: 'New owner?', + options: [{ id: 'yes', label: 'Yes' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + }, + { fence: 8 } + ) + + expect( + await settleStructuredAgentSessionDeadGeneration({ + journal, + sessionId: SESSION, + fence: 8, + throughFence: 7, + settlementId: 'old-owner', + pendingSubmissionReason: 'provider_exited_before_acknowledgement', + verdict: { state: 'interrupted', completedAt: 1_001 } + }) + ).toBe(true) + expect(journal.submissions()).toEqual( + expect.arrayContaining([ + expect.objectContaining({ clientMessageId: 'client-1', dispatchState: 'unknown' }), + expect.objectContaining({ clientMessageId: 'client-new', dispatchState: 'pending' }) + ]) + ) + expect(journal.snapshot().items).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + ownerFence: 7, + body: expect.objectContaining({ state: 'interrupted' }) + }), + expect.objectContaining({ + ownerFence: 8, + body: expect.objectContaining({ state: 'running' }) + }), + expect.objectContaining({ + ownerFence: 8, + body: expect.objectContaining({ + resolution: expect.objectContaining({ state: 'pending' }) + }) + }) + ]) + ) + }) + it('leaves a replacement prompt pending when it is published after settlement', async () => { await seedUnfinishedWork() await settleStructuredAgentSessionDeadGeneration({ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts index 50bc0cfa4f7..9d07f52b44a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts @@ -1,3 +1,4 @@ +import { createHash } from 'node:crypto' import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key' import type { AgentJournalItemBody, @@ -37,7 +38,7 @@ export function unexpectedProviderExitOutcome(reason?: string): string { type DeadGenerationSubmission = Pick< ReturnType[number], - 'clientMessageId' | 'dispatchState' | 'recovered' + 'clientMessageId' | 'dispatchState' | 'recovered' | 'fence' > export type DeadGenerationJournal = { @@ -54,30 +55,48 @@ export type StructuredAgentSessionUnfinishedWork = { } export function captureUnfinishedStructuredAgentSessionWork( - journal: DeadGenerationJournal + journal: DeadGenerationJournal, + throughFence = Number.MAX_SAFE_INTEGER, + fromFence = 0 ): StructuredAgentSessionUnfinishedWork { return { - items: journal.snapshot().items.filter(isUnfinishedItem), - hadUnsettledSubmissions: hasUnsettledSubmission(journal) + items: journal + .snapshot() + .items.filter( + (item) => + belongsToSettledGeneration(item, fromFence, throughFence) && isUnfinishedItem(item) + ), + hadUnsettledSubmissions: hasUnsettledSubmission(journal, fromFence, throughFence) } } -function hasUnfinishedStructuredAgentSessionWork(journal: DeadGenerationJournal): boolean { - const work = captureUnfinishedStructuredAgentSessionWork(journal) +function hasUnfinishedStructuredAgentSessionWork( + journal: DeadGenerationJournal, + throughFence: number, + fromFence: number +): boolean { + const work = captureUnfinishedStructuredAgentSessionWork(journal, throughFence, fromFence) return work.hadUnsettledSubmissions || work.items.length > 0 } export function unfinishedStructuredAgentSessionWorkWasInterrupted( before: StructuredAgentSessionUnfinishedWork, journal: DeadGenerationJournal, - observedExitAt: number + observedExitAt: number, + throughFence = Number.MAX_SAFE_INTEGER, + fromFence = 0 ): boolean { - const currentSnapshot = journal.snapshot() - if (hasUnsettledSubmission(journal) || currentSnapshot.items.some(isInProgressItem)) { + const currentSnapshot = journal + .snapshot() + .items.filter((item) => belongsToSettledGeneration(item, fromFence, throughFence)) + if ( + hasUnsettledSubmission(journal, fromFence, throughFence) || + currentSnapshot.some(isInProgressItem) + ) { return true } if ( - currentSnapshot.items.some((item) => { + currentSnapshot.some((item) => { const turn = readAgentJournalTurn(item.body) return turn?.state === 'interrupted' && turn.completedAt === observedExitAt }) @@ -88,7 +107,7 @@ export function unfinishedStructuredAgentSessionWorkWasInterrupted( if (inProgressBefore.length === 0) { return false } - const currentItems = new Map(currentSnapshot.items.map((item) => [item.itemId, item])) + const currentItems = new Map(currentSnapshot.map((item) => [item.itemId, item])) const runningTurns = inProgressBefore.filter( (item) => readAgentJournalTurn(item.body)?.state === 'running' ) @@ -100,6 +119,9 @@ export async function settleStructuredAgentSessionDeadGeneration(input: { journal: DeadGenerationJournal sessionId: string fence: number + /** The dead owner's last fence; later owners' rows are never settlement targets. */ + throughFence?: number + fromFence?: number settlementId: string verdict: StructuredAgentSessionTurnVerdict pendingSubmissionReason: string @@ -110,7 +132,13 @@ export async function settleStructuredAgentSessionDeadGeneration(input: { onError?: (sessionId: string, error: unknown) => void }): Promise { try { - const hasUnfinishedWork = hasUnfinishedStructuredAgentSessionWork(input.journal) + const throughFence = input.throughFence ?? input.fence + const fromFence = input.fromFence ?? 0 + const hasUnfinishedWork = hasUnfinishedStructuredAgentSessionWork( + input.journal, + throughFence, + fromFence + ) const showUnexpectedExitOutcome = input.showUnexpectedExitOutcome ?? hasUnfinishedWork if (!showUnexpectedExitOutcome && !hasUnfinishedWork) { return true @@ -118,9 +146,13 @@ export async function settleStructuredAgentSessionDeadGeneration(input: { await input.journal.markPendingSubmissionsUnknown( input.fence, { mode: input.submissionRecoveryMode ?? 'death-confirmed' }, - input.pendingSubmissionReason + input.pendingSubmissionReason, + throughFence, + fromFence ) - const items = input.journal.snapshot().items + const items = input.journal + .snapshot() + .items.filter((item) => belongsToSettledGeneration(item, fromFence, throughFence)) const mutations: JournalLifecycleMutationInput[] = [] if (showUnexpectedExitOutcome) { mutations.push({ @@ -141,9 +173,15 @@ export async function settleStructuredAgentSessionDeadGeneration(input: { } mutations.push(...runningTurnLifecycleRevisions(items, input.verdict)) const batchId = `dead-generation:${input.settlementId}` - for (const chunk of partitionJournalLifecycleMutations(batchId, mutations)) { + const chunks = partitionJournalLifecycleMutations(batchId, mutations) + for (const chunk of chunks) { await input.journal.appendLifecycleBatch({ - settlementId: chunk.settlementId, + // A partial commit changes the next partition; content identity prevents a reused + // chunk index from suppressing still-unsettled rows. + settlementId: + chunks.length === 1 + ? chunk.settlementId + : `${chunk.settlementId}:${createHash('sha256').update(JSON.stringify(chunk.mutations)).digest('hex').slice(0, 16)}`, fence: input.fence, recovered: true, mutations: chunk.mutations @@ -197,13 +235,33 @@ function isCleanlySettled(item: AgentJournalRenderItem | undefined): boolean { return false } -function hasUnsettledSubmission(journal: DeadGenerationJournal): boolean { +function belongsToSettledGeneration( + item: AgentJournalRenderItem, + fromFence: number, + throughFence: number +): boolean { + return ( + item.ownerFence === undefined || + (item.ownerFence >= fromFence && item.ownerFence <= throughFence) + ) +} + +function hasUnsettledSubmission( + journal: DeadGenerationJournal, + fromFence: number, + throughFence: number +): boolean { const submissions = journal.submissions?.() return submissions ? submissions.some( (submission) => - submission.dispatchState === 'pending' || - (submission.dispatchState === 'unknown' && submission.recovered !== true) + submission.fence >= fromFence && + submission.fence <= throughFence && + (submission.dispatchState === 'pending' || + (submission.dispatchState === 'unknown' && submission.recovered !== true)) ) - : (journal.pendingSubmissions?.().length ?? 0) > 0 + : (journal + .pendingSubmissions?.() + .some((submission) => submission.fence >= fromFence && submission.fence <= throughFence) ?? + false) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-queue-start.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-queue-start.ts index f8d6db2bace..e3413cde873 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-queue-start.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-queue-start.ts @@ -1,5 +1,8 @@ import type { AgentSessionHandoffRequest } from '../../../shared/agent-session-wire' -import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection' +import { + activeStructuredAgentSessionTurnId, + liveStructuredAgentSessionItems +} from '../../../shared/structured-agent-session-projection' import type { StructuredAgentSessionHandoffQueue } from './structured-agent-session-handoff-queue' import type { StructuredAgentSessionHandoffDeps, @@ -34,7 +37,12 @@ export function queueStructuredHandoffAfterTurn(input: { sessionId, async (signal) => { if (params.direction === 'to-tui') { - return !activeStructuredAgentSessionTurnId(deps.session(sessionId).journal.snapshot().items) + return !activeStructuredAgentSessionTurnId( + liveStructuredAgentSessionItems( + deps.session(sessionId).journal.snapshot().items, + params.envelope.expectedRuntimeFence + ) + ) } tuiReadiness = tuiOwner ? ((await deps.transport?.waitForTuiIdleOrExit(tuiOwner, signal)) ?? null) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-queue.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-queue.ts index 9ea3d7ff08f..5d175cf65fb 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-queue.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-queue.ts @@ -3,7 +3,10 @@ import type { AgentSessionHandoffStatus } from '../../../shared/agent-session-wire' import type { AgentSessionRecord } from '../../../shared/agent-session-record' -import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection' +import { + activeStructuredAgentSessionTurnId, + liveStructuredAgentSessionItems +} from '../../../shared/structured-agent-session-projection' import type { StructuredAgentSessionHandoffDeps, StructuredTuiOwner @@ -102,7 +105,12 @@ export function enqueueStructuredHandoffAfterTurn(input: { sessionId, async (signal) => { if (params.direction === 'to-tui') { - return !activeStructuredAgentSessionTurnId(deps.session(sessionId).journal.snapshot().items) + return !activeStructuredAgentSessionTurnId( + liveStructuredAgentSessionItems( + deps.session(sessionId).journal.snapshot().items, + params.envelope.expectedRuntimeFence + ) + ) } if (!observedTuiQueue) { observedTuiQueue = true @@ -114,7 +122,14 @@ export function enqueueStructuredHandoffAfterTurn(input: { if (tuiReadiness === 'exited') { return true } - if (!activeStructuredAgentSessionTurnId(deps.session(sessionId).journal.snapshot().items)) { + if ( + !activeStructuredAgentSessionTurnId( + liveStructuredAgentSessionItems( + deps.session(sessionId).journal.snapshot().items, + params.envelope.expectedRuntimeFence + ) + ) + ) { tuiReadiness = 'idle' return true } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-revalidation.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-revalidation.ts index 66005959378..98ffc5c4029 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-revalidation.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-revalidation.ts @@ -1,6 +1,9 @@ import type { AgentSessionRecord } from '../../../shared/agent-session-record' import type { AgentSessionHandoffRequest } from '../../../shared/agent-session-wire' -import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection' +import { + activeStructuredAgentSessionTurnId, + liveStructuredAgentSessionItems +} from '../../../shared/structured-agent-session-projection' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { structuredHandoffRetryResumesStoppedOwner } from './structured-agent-session-handoff-admission' import { structuredSessionHasPendingPrompt } from './structured-agent-session-handoff-status' @@ -28,13 +31,15 @@ export function assertScheduledStructuredHandoffIsAdmissible(input: { ) { throw new Error('agent_session_checkpoint_stale') } - if (structuredSessionHasPendingPrompt(input.journal)) { + if (structuredSessionHasPendingPrompt(input.journal, record.lease.runtimeFence)) { throw new Error('Resolve the pending question or approval before switching.') } if (params.mode !== 'stop-turn' && input.journal.cursor().sequence !== input.journalSequence) { throw new Error('The session changed before the handoff started.') } - const activeTurn = activeStructuredAgentSessionTurnId(input.journal.snapshot().items) + const activeTurn = activeStructuredAgentSessionTurnId( + liveStructuredAgentSessionItems(input.journal.snapshot().items, record.lease.runtimeFence) + ) if (params.direction === 'to-tui') { const expectedTurn = params.mode === 'stop-turn' ? input.turnId : null if (activeTurn !== expectedTurn) { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-status.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-status.ts index 7d519a558b5..a4490c7b886 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-status.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-status.ts @@ -4,6 +4,7 @@ import type { AgentSessionHandoffStatus } from '../../../shared/agent-session-wire' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { liveStructuredAgentSessionItems } from '../../../shared/structured-agent-session-projection' import type { StructuredAgentSessionHandoffTransport, StructuredTuiOwner @@ -89,14 +90,15 @@ function persistedFailedStructuredHandoffStatus( } } -export function structuredSessionHasPendingPrompt(journal: AgentSessionJournal): boolean { - return journal - .snapshot() - .items.some( - (item) => - (item.body.kind === 'approval' || item.body.kind === 'question') && - item.body.resolution.state === 'pending' - ) +export function structuredSessionHasPendingPrompt( + journal: AgentSessionJournal, + fence: number +): boolean { + return liveStructuredAgentSessionItems(journal.snapshot().items, fence).some( + (item) => + (item.body.kind === 'approval' || item.body.kind === 'question') && + item.body.resolution.state === 'pending' + ) } export function switchingStructuredHandoffStatus( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.ts index 57333d1c90c..1803c63fd63 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.ts @@ -6,7 +6,10 @@ import type { AgentSessionMutationResult, AgentSessionWireRefusal } from '../../../shared/agent-session-wire' -import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection' +import { + activeStructuredAgentSessionTurnId, + liveStructuredAgentSessionItems +} from '../../../shared/structured-agent-session-projection' import { admitStructuredHandoffRequest, refuseAdmittedStructuredHandoff, @@ -161,7 +164,12 @@ export class StructuredAgentSessionHandoffCoordinator { `The ${expectedOwner} runtime does not own this session.` ) } - if (structuredSessionHasPendingPrompt(this.deps.session(record.sessionId).journal)) { + if ( + structuredSessionHasPendingPrompt( + this.deps.session(record.sessionId).journal, + record.lease.runtimeFence + ) + ) { return this.refuseAdmitted( callerKey, params, @@ -170,7 +178,10 @@ export class StructuredAgentSessionHandoffCoordinator { ) } const turnId = activeStructuredAgentSessionTurnId( - this.deps.session(record.sessionId).journal.snapshot().items + liveStructuredAgentSessionItems( + this.deps.session(record.sessionId).journal.snapshot().items, + record.lease.runtimeFence + ) ) const tuiOwner = this.state.owner(record.sessionId) const busy = diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts index b8e198c9b6b..7fc78eaa635 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-history-result.ts @@ -7,6 +7,7 @@ import type { } from '../../../shared/agent-session-wire' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { readAgentSessionHistory } from './agent-session-history-page' +import { projectStructuredAgentSessionOwnerPage } from './structured-agent-session-owner-projection' export function structuredAgentSessionProviderSessionMetadata( record: AgentSessionRecord | null @@ -29,11 +30,12 @@ export function readStructuredAgentSessionHistoryResult(input: { const fence = input.record?.lease.runtimeFence const providerSession = structuredAgentSessionProviderSessionMetadata(input.record) if (fence === undefined) { - return providerSession ? { ...result, providerSession } : result + const page = projectStructuredAgentSessionOwnerPage(result.page, input.record) + return providerSession ? { ...result, page, providerSession } : { ...result, page } } return { ...result, - page: { ...result.page, fence }, + page: projectStructuredAgentSessionOwnerPage({ ...result.page, fence }, input.record), ...(result.ok ? {} : { fence }), ...(providerSession ? { providerSession } : {}) } 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 28b38d22555..82d6d89eb78 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 @@ -5,7 +5,10 @@ // bookkeeping that decides when to run it than buried among the twenty other things a session can // do. -import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection' +import { + activeStructuredAgentSessionTurnId, + liveStructuredAgentSessionItems +} from '../../../shared/structured-agent-session-projection' import { evictStructuredAgentSession, STRUCTURED_AGENT_SESSION_EVICTION_STEPS, @@ -72,6 +75,7 @@ export async function evictHeldStructuredAgentSession( // reading "no child here" and skipping the settlement and the lease release it still owes. const owesWindDown = owesProviderChildWindDown(session) session.owesProviderChildWindDown = owesWindDown + let settlementFailed = false const eviction: StructuredAgentSessionEvictionContext = { sessionId, // The retry must not re-stop a child the adapter already proved gone, so this stays honest. @@ -89,10 +93,11 @@ export async function evictHeldStructuredAgentSession( }, discardSink: () => context.runtimeState.discardEventSink(sessionId), settleWork: async () => { - await settleStructuredAgentSessionDeadGeneration({ + const settled = await settleStructuredAgentSessionDeadGeneration({ journal: session.journal, sessionId, fence: session.fence, + fromFence: session.fence, settlementId: `expected-close:${sessionId}:${session.fence}:${session.acquisitionGeneration ?? 'unknown'}`, pendingSubmissionReason: 'provider_closed_before_acknowledgement', verdict: { state: 'interrupted', completedAt: context.now() }, @@ -102,6 +107,7 @@ export async function evictHeldStructuredAgentSession( console.error('agent-session close settlement deferred', id, error) } }) + settlementFailed = !settled }, releaseLease: async () => { await releaseStoredStructuredAgentSessionOwner({ @@ -109,7 +115,15 @@ export async function evictHeldStructuredAgentSession( sessionId, hasProviderChild: owesWindDown, expectedFence: session.fence, - now: context.now() + now: context.now(), + ...(settlementFailed + ? { + settlementRetry: { + settlementId: `expected-close:${sessionId}:${session.fence}:${session.acquisitionGeneration ?? 'unknown'}`, + detail: 'the last surface holding this session released it' + } + } + : {}) }) session.owesProviderChildWindDown = false context.forgetStatus(sessionId) @@ -194,7 +208,9 @@ export function createStructuredAgentSessionHolds( isTurnActive: (sessionId) => { const session = context.sessions.get(sessionId) return session - ? activeStructuredAgentSessionTurnId(session.journal.snapshot().items) !== null + ? activeStructuredAgentSessionTurnId( + liveStructuredAgentSessionItems(session.journal.snapshot().items, session.fence) + ) !== null : false }, onError: (error) => context.deps.onEventSinkError?.(error), 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 8fcdef95480..048b8cd0107 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 @@ -5,6 +5,8 @@ import { StructuredConversationCommandController } from './structured-conversati // Mutations share one durable admission path and serialize per session. import type { AgentJournalSnapshot } from '../../../shared/agent-session-journal-types' +import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' +import { liveStructuredAgentSessionItems } from '../../../shared/structured-agent-session-projection' import type { AgentSessionExecutionLocation } from '../../../shared/agent-session-record' import type * as SessionWire from '../../../shared/agent-session-wire' import type { AgentSessionAttachParams } from './structured-agent-session-attach' @@ -317,6 +319,16 @@ export class StructuredAgentSessionHost { journalSnapshot = (sessionId: string): AgentJournalSnapshot => this.requireSession(sessionId).journal.snapshot() + currentOwnerJournalItems = (sessionId: string): AgentJournalRenderItem[] => { + const record = this.deps.store.getRecord(sessionId) + if (!record) { + throw new Error('agent_session_ownership_unknown') + } + const fence = + record.lease.claimStatus === 'live' ? record.lease.runtimeFence : Number.MAX_SAFE_INTEGER + return liveStructuredAgentSessionItems(this.journalSnapshot(sessionId).items, fence) + } + subscribe = (input: AgentSessionSubscribeInput): (() => void) => this.backgroundTasks.subscribe(input) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-lease-release.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-lease-release.ts index bacc56fbcaf..f4baebefc73 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-lease-release.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-lease-release.ts @@ -23,6 +23,7 @@ export async function releaseStoredStructuredAgentSessionOwner(input: { hasProviderChild: boolean expectedFence: number now: number + settlementRetry?: { settlementId: string; detail: string } }): Promise { if (!input.hasProviderChild) { return @@ -38,7 +39,8 @@ export async function releaseStoredStructuredAgentSessionOwner(input: { await releaseStoredAgentSessionOwnerAfterSurfaceClose(input.store, { sessionId: input.sessionId, expectedFence: input.expectedFence, - now: input.now + now: input.now, + ...(input.settlementRetry ? { settlementRetry: input.settlementRetry } : {}) }) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-owner-projection.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-owner-projection.test.ts new file mode 100644 index 00000000000..5b5197bf7cf --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-owner-projection.test.ts @@ -0,0 +1,183 @@ +import { describe, expect, it } from 'vitest' +import type { AgentSessionHistoryPage } from '../../../shared/agent-session-wire' +import { + EMPTY_STRUCTURED_AGENT_SESSION, + reduceStructuredAgentSession +} from '../../../shared/structured-agent-session-reducer' +import { + agentSessionLeaseFixture, + agentSessionRecordFixture +} from '../../../shared/agent-session-record.test-fixture' +import { projectStructuredAgentSessionOwnerPage } from './structured-agent-session-owner-projection' + +const cursor = { epoch: 'epoch-1', sequence: 5 } + +function page(): AgentSessionHistoryPage { + return { + sessionId: 'session-alpha-1', + epoch: cursor.epoch, + direction: 'tail', + items: [ + { + itemId: 'old-turn', + ownerFence: 7, + revision: 1, + sequence: 1, + observedAt: 1, + body: { kind: 'turn', turnId: 'old', state: 'running' } + }, + { + itemId: 'old-prompt', + ownerFence: 7, + revision: 1, + sequence: 2, + observedAt: 2, + body: { + kind: 'approval', + title: 'Old approval', + detail: null, + options: [], + resolution: { + state: 'pending', + selectedOptionId: null, + resolvedBy: null, + resolvedAt: null + } + } + }, + { + itemId: 'new-turn', + ownerFence: 9, + revision: 1, + sequence: 3, + observedAt: 3, + body: { kind: 'turn', turnId: 'new', state: 'running' } + }, + { + itemId: 'new-prompt', + ownerFence: 9, + revision: 1, + sequence: 4, + observedAt: 4, + body: { + kind: 'question', + question: 'New question', + options: [], + resolution: { + state: 'pending', + selectedOptionId: null, + resolvedBy: null, + resolvedAt: null + } + } + } + ], + submissions: [7, 9].map((fence) => ({ + clientMessageId: `send-${fence}`, + fence, + payloadFingerprint: 'fingerprint', + dispatchState: 'pending' as const, + providerItemId: null, + reason: null, + submittedAt: 1, + resolvedAt: null + })), + removedItemIds: [], + window: { oldest: null, newest: null, nextCursor: cursor }, + liveCursor: cursor, + hasOlder: false, + hasNewer: false + } +} + +describe('host owner projection for old paired clients', () => { + it('makes an unsettled prior owner inert without ending the live child', () => { + const record = agentSessionRecordFixture(agentSessionLeaseFixture({ runtimeFence: 9 })) + const result = projectStructuredAgentSessionOwnerPage(page(), record) + expect(result.items.map((item) => item.body)).toMatchObject([ + { kind: 'turn', state: 'unverifiable' }, + { kind: 'approval', resolution: { state: 'cancelled' } }, + { kind: 'turn', state: 'running' }, + { kind: 'question', resolution: { state: 'pending' } } + ]) + expect( + result.submissions.map((submission) => [submission.dispatchState, submission.recovered]) + ).toEqual([ + ['unknown', true], + ['pending', undefined] + ]) + }) + + it('treats transcript import at a released fence as inert', () => { + const record = agentSessionRecordFixture( + agentSessionLeaseFixture({ + runtimeFence: 9, + claimStatus: 'released', + ownerProcess: null + }) + ) + const result = projectStructuredAgentSessionOwnerPage(page(), record) + expect(result.items.at(-1)?.body).toMatchObject({ resolution: { state: 'cancelled' } }) + expect(result.items[2]?.body).toMatchObject({ state: 'unverifiable' }) + expect(result.submissions.every((submission) => submission.recovered === true)).toBe(true) + }) + + it('does not apply the latest witnessed exit to an earlier failed generation', () => { + const record = agentSessionRecordFixture( + agentSessionLeaseFixture({ + runtimeFence: 9, + settlementRetryRequired: true, + settlementRetryFence: 8, + settlementRetryId: 'second-exit', + deathEvidence: { kind: 'exit-observed', observedAt: 200, detail: 'provider exited' } + }) + ) + const current = page() + current.items.splice(1, 1, { + itemId: 'second-turn', + ownerFence: 8, + revision: 1, + sequence: 2, + observedAt: 2, + body: { kind: 'turn', turnId: 'second', state: 'running' } + }) + const result = projectStructuredAgentSessionOwnerPage(current, record) + expect(result.items.map((item) => item.body)).toMatchObject([ + { state: 'unverifiable' }, + { state: 'interrupted', completedAt: 200 }, + { state: 'running' }, + { resolution: { state: 'pending' } } + ]) + }) + + it('lets a later durable revision replace a projected fallback', () => { + const record = agentSessionRecordFixture(agentSessionLeaseFixture({ runtimeFence: 9 })) + const projected = projectStructuredAgentSessionOwnerPage(page(), record) + const synthetic = reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, { + type: 'history-page', + page: projected + }) + const durable = reduceStructuredAgentSession(synthetic, { + type: 'event', + event: { + type: 'batch', + sessionId: projected.sessionId, + fence: 9, + hostNow: 300, + batch: { + cursor: { epoch: projected.epoch, sequence: 6 }, + removedItemIds: [], + submissions: [], + items: [ + { + ...projected.items[0]!, + revision: 2, + body: { kind: 'turn', turnId: 'old', state: 'interrupted', completedAt: 200 } + } + ] + } + } + }) + expect(durable.items[0]?.body).toMatchObject({ state: 'interrupted', completedAt: 200 }) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-owner-projection.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-owner-projection.ts new file mode 100644 index 00000000000..95c277b1153 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-owner-projection.ts @@ -0,0 +1,101 @@ +import type { + AgentJournalRenderItem, + AgentJournalSubmission, + AgentJournalSnapshot +} from '../../../shared/agent-session-journal-types' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentSessionHistoryPage } from '../../../shared/agent-session-wire' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' +import { cancelledJournalPromptBody } from '../agent-session-journal/journal-prompt-body-bounds' + +function isStaleFence(record: AgentSessionRecord | null, fence: number | undefined): boolean { + return ( + fence !== undefined && + (!record || + (fence <= record.lease.runtimeFence && + (record.lease.claimStatus !== 'live' || fence < record.lease.runtimeFence))) + ) +} + +export function structuredAgentSessionNeedsOwnerSnapshot( + snapshot: AgentJournalSnapshot, + record: AgentSessionRecord | null +): boolean { + return ( + snapshot.items.some((item) => { + if (!isStaleFence(record, item.ownerFence)) { + return false + } + const body = item.body + return ( + readAgentJournalTurn(body)?.state === 'running' || + (body.kind === 'tool-call' && body.state === 'running') || + ((body.kind === 'approval' || body.kind === 'question') && + body.resolution.state === 'pending') + ) + }) || + snapshot.submissions.some( + (submission) => + isStaleFence(record, submission.fence) && + (submission.dispatchState === 'pending' || + (submission.dispatchState === 'unknown' && submission.recovered !== true)) + ) + ) +} + +/** The lease, not an unsettled journal write, decides what can still ask or work. */ +export function projectStructuredAgentSessionOwnerPage( + page: AgentSessionHistoryPage, + record: AgentSessionRecord | null +): AgentSessionHistoryPage { + const items = page.items.map((item): AgentJournalRenderItem => { + if (!isStaleFence(record, item.ownerFence)) { + return item + } + const body = item.body + const turn = readAgentJournalTurn(body) + if (turn?.state === 'running' && (body.kind === 'turn' || body.kind === 'status')) { + const evidence = record?.lease.deathEvidence + const retryFence = record?.lease.settlementRetryFence + const witnessed = + evidence?.kind === 'exit-observed' && + retryFence !== undefined && + (item.ownerFence === retryFence || + (record?.lease.claimStatus === 'released' && + record.lease.handoffStage === 'old-owner-stopped' && + item.ownerFence === retryFence + 1)) + const lifecycle = witnessed + ? { ...turn, state: 'interrupted' as const, completedAt: evidence.observedAt } + : { ...turn, state: 'unverifiable' as const } + return { + ...item, + body: + body.kind === 'turn' + ? { kind: 'turn', ...lifecycle } + : { ...body, turnLifecycle: lifecycle } + } + } + if (body.kind === 'tool-call' && body.state === 'running') { + return { ...item, body: { ...body, state: 'failed' } } + } + const cancelled = cancelledJournalPromptBody(body) + return cancelled && + (body.kind === 'approval' || body.kind === 'question') && + body.resolution.state === 'pending' + ? { ...item, body: cancelled } + : item + }) + const submissions = page.submissions.map((submission): AgentJournalSubmission => + isStaleFence(record, submission.fence) && + (submission.dispatchState === 'pending' || + (submission.dispatchState === 'unknown' && submission.recovered !== true)) + ? { + ...submission, + dispatchState: 'unknown', + recovered: true, + reason: 'provider_exited_before_acknowledgement' + } + : submission + ) + return { ...page, items, submissions } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-state.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-state.ts index 7f71a9c28ce..b06fd02c9cc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-state.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-state.ts @@ -16,7 +16,7 @@ function invalid(message: string): PendingPromptValidation { } export function validatePendingPrompt( - ctx: Pick, + ctx: Pick, input: { itemId: string expectedRevision: number @@ -27,6 +27,9 @@ export function validatePendingPrompt( if (!item) { return invalid(`No item ${input.itemId} in session ${ctx.sessionId}.`) } + if (item.ownerFence !== undefined && item.ownerFence !== ctx.fence) { + return invalid(`Item ${input.itemId} belongs to a previous owner.`) + } const prompt = item.body.kind === 'approval' || item.body.kind === 'question' ? item.body : null if (!prompt || (input.kind !== undefined && prompt.kind !== input.kind)) { return invalid( 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 9dab7a07baf..d5c1f02ee28 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 @@ -5,7 +5,10 @@ import type { StructuredAgentSessionHostSession } 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 { + turnVerdictFromDeathEvidence, + UNVERIFIABLE_TURN_VERDICT +} from './structured-agent-session-stale-turn-verdict' import { captureUnfinishedStructuredAgentSessionWork, settleStructuredAgentSessionDeadGeneration, @@ -31,7 +34,8 @@ export async function retryPendingStructuredAgentSessionSettlement(input: { record, params: input.params, journalRoot: input.deps.journalRoot, - adapter: input.deps.adapter + adapter: input.deps.adapter, + recoverPending: false }) ).journal } catch (error) { @@ -78,11 +82,37 @@ export async function retryLoadedStructuredAgentSessionSettlement(input: { } // Only an observed exit earns an end time; a probe-proven death never saw one. const verdict = turnVerdictFromDeathEvidence(record.lease.deathEvidence) + const throughFence = + record.lease.settlementRetryFence ?? + (record.lease.claimStatus === 'released' + ? record.lease.runtimeFence - 1 + : record.lease.runtimeFence) + // Transcript catch-up after a stopped TUI can be stamped at the released fence. + // This fence has no child; a subsequent live owner's fence must stay untouched. + const lastDeadFence = + record.lease.claimStatus === 'released' && + record.lease.handoffStage === 'old-owner-stopped' && + record.lease.runtimeFence === throughFence + 1 + ? record.lease.runtimeFence + : throughFence + const priorSettled = await settleStructuredAgentSessionDeadGeneration({ + journal: retrySession.journal, + sessionId: input.sessionId, + fence: retrySession.fence, + throughFence: throughFence - 1, + settlementId: `${record.lease.settlementRetryId}:prior`, + pendingSubmissionReason: 'provider_exited_before_acknowledgement', + verdict: UNVERIFIABLE_TURN_VERDICT, + showUnexpectedExitOutcome: false, + onError + }) const ok = await settleStructuredAgentSessionDeadGeneration({ journal: retrySession.journal, sessionId: input.sessionId, fence: retrySession.fence, settlementId: record.lease.settlementRetryId, + throughFence: lastDeadFence, + fromFence: throughFence, pendingSubmissionReason: 'provider_exited_before_acknowledgement', verdict, // The same evidence decides the copy: only a witnessed death is worth telling the user @@ -92,16 +122,22 @@ export async function retryLoadedStructuredAgentSessionSettlement(input: { showUnexpectedExitOutcome: verdict.state === 'interrupted' && unfinishedStructuredAgentSessionWorkWasInterrupted( - captureUnfinishedStructuredAgentSessionWork(retrySession.journal), + captureUnfinishedStructuredAgentSessionWork( + retrySession.journal, + lastDeadFence, + throughFence + ), retrySession.journal, - verdict.completedAt + verdict.completedAt, + lastDeadFence, + throughFence ), ...(record.lease.deathEvidence?.detail ? { unexpectedExitReason: record.lease.deathEvidence.detail } : {}), onError }) - if (!ok) { + if (!ok || !priorSettled) { return false } try { @@ -122,6 +158,7 @@ export async function retryLoadedStructuredAgentSessionSettlement(input: { handoffOperationId: preserveHandoff ? latest.lease.handoffOperationId : null, settlementRetryRequired: undefined, settlementRetryId: undefined, + settlementRetryFence: undefined, lastRenewedAt: input.now() } } 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 220a8116160..587d5e6c1f9 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 @@ -210,7 +210,14 @@ 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) + const lease = record?.lease + // A released fence has no execution owner, even if a late transcript import used it. + const fence = lease + ? lease.claimStatus === 'live' + ? lease.runtimeFence + : Number.MAX_SAFE_INTEGER + : session.fence let projection = this.journalProjections.get(journal) if ( !projection || @@ -234,7 +241,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 an owner // handoff or a mid-session switch lands, so the row follows whichever is in force. diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts index 4f22b56397c..6d8c13c34c1 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts @@ -19,6 +19,14 @@ import type { JournalRow } from '../agent-session-journal/journal-row-schema' import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' import { StructuredAgentSessionStatusFeed } from './structured-agent-session-status-feed' import { AgentSessionSubscribers } from './structured-agent-session-subscribers' +import { + projectStructuredAgentSessionOwnerPage, + structuredAgentSessionNeedsOwnerSnapshot +} from './structured-agent-session-owner-projection' +import { + agentSessionLeaseFixture, + agentSessionRecordFixture +} from '../../../shared/agent-session-record.test-fixture' const SESSION = 'subscriber-session' @@ -35,6 +43,81 @@ afterEach(async () => { }) describe('AgentSessionSubscribers', () => { + it('replaces an old paired client’s raw pending prompt when ownership changes', async () => { + const journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } + }, + journalDir: join(root, 'owner-journal') + }) + await journal.appendItem( + { provider: 'orca', clientMessageId: 'prompt' }, + { + kind: 'approval', + title: 'Old request', + detail: null, + options: [], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + }, + { fence: 7 } + ) + let record = agentSessionRecordFixture(agentSessionLeaseFixture({ runtimeFence: 7 })) + const events: AgentSessionSubscribeEvent[] = [] + const subscribers = new AgentSessionSubscribers({ + readJournal: () => journal, + projectPage: (_sessionId, page) => projectStructuredAgentSessionOwnerPage(page, record), + needsOwnerSnapshot: () => structuredAgentSessionNeedsOwnerSnapshot(journal.snapshot(), record) + }) + subscribers.open({ + id: 'old-client', + sessionId: SESSION, + journal, + fence: 7, + emit: (event) => events.push(event) + }) + expect(events[0]).toMatchObject({ + type: 'snapshot', + page: { items: [{ body: { resolution: { state: 'pending' } } }] } + }) + + record = agentSessionRecordFixture( + agentSessionLeaseFixture({ + runtimeFence: 8, + claimStatus: 'released', + ownerProcess: null + }) + ) + subscribers.handoff(SESSION, 8, { + owner: 'none', + direction: 'to-native', + phase: 'failed', + stage: null, + operationId: null + }) + expect(events.at(-1)).toMatchObject({ + type: 'snapshot', + page: { items: [{ body: { resolution: { state: 'cancelled' } } }] } + }) + + const reconnected: AgentSessionSubscribeEvent[] = [] + subscribers.open({ + id: 'reconnected', + sessionId: SESSION, + journal, + fence: 8, + cursor: journal.cursor(), + emit: (event) => reconnected.push(event) + }) + expect(reconnected[0]).toMatchObject({ + type: 'snapshot', + page: { items: [{ body: { resolution: { state: 'cancelled' } } }] } + }) + }) + it('publishes the current fence when a resumed cursor is already caught up', async () => { const journal = await journals.open({ identity: { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts index 24415df1d3c..1ff21187729 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts @@ -13,6 +13,7 @@ import { type AgentSessionBackgroundTaskState, type AgentSessionSlashCommand, type AgentSessionHandoffStatus, + type AgentSessionHistoryPage, type AgentSessionSubscribeEvent, type AgentSessionTurnActivity } from '../../../shared/agent-session-wire' @@ -47,6 +48,9 @@ export type AgentSessionSubscribersHooks = { onJournalPublished?: (sessionId: string, journal: AgentSessionJournal) => void /** Host wall clock, stamped once per published frame as `hostNow`. */ now?: () => number + projectPage?: (sessionId: string, page: AgentSessionHistoryPage) => AgentSessionHistoryPage + readJournal?: (sessionId: string) => AgentSessionJournal | undefined + needsOwnerSnapshot?: (sessionId: string, journal: AgentSessionJournal) => boolean } export class AgentSessionSubscribers { @@ -80,10 +84,13 @@ export class AgentSessionSubscribers { this.bySession.set(input.sessionId, session) const hostNow = this.now() - if (input.cursor) { + if (input.cursor && !this.hooks.needsOwnerSnapshot?.(input.sessionId, input.journal)) { this.deliver(subscriber, input.journal, hostNow, input.handoff, true, input.backgroundTasks) } else { - const page = readAgentSessionHydrationPage(input.journal, input.fence) + const page = this.ownerPage( + input.sessionId, + readAgentSessionHydrationPage(input.journal, input.fence) + ) this.emit(subscriber, { type: 'snapshot', sessionId: input.sessionId, @@ -163,7 +170,7 @@ export class AgentSessionSubscribers { backgroundTasks: AgentSessionBackgroundTaskState | null | undefined, frame: { type: 'snapshot' } | { type: 'reset'; reset: AgentJournalResetReason } ): void { - const page = readAgentSessionHydrationPage(journal, fence) + const page = this.ownerPage(sessionId, readAgentSessionHydrationPage(journal, fence)) const hostNow = this.now() for (const subscriber of this.subscribers(sessionId)) { this.emit(subscriber, { @@ -183,7 +190,18 @@ export class AgentSessionSubscribers { handoff(sessionId: string, fence: number, handoff: AgentSessionHandoffStatus): void { const hostNow = this.now() + let projectedPage: AgentSessionHistoryPage | undefined for (const subscriber of this.subscribers(sessionId)) { + const journal = subscriber.fence !== fence ? this.hooks.readJournal?.(sessionId) : undefined + if (journal) { + const page = + projectedPage ?? this.ownerPage(sessionId, readAgentSessionHydrationPage(journal, fence)) + projectedPage = page + this.emit(subscriber, { type: 'snapshot', sessionId, page, fence, handoff, hostNow }) + subscriber.cursor = page.liveCursor ?? page.window.nextCursor + subscriber.fence = fence + continue + } this.emit(subscriber, { type: 'batch', sessionId, @@ -241,7 +259,10 @@ export class AgentSessionSubscribers { limit: AGENT_SESSION_HISTORY_MAX_LIMIT }) if (!result.ok) { - const page = { ...result.page, fence: subscriber.fence } + const page = this.ownerPage(subscriber.sessionId, { + ...result.page, + fence: subscriber.fence + }) this.emit(subscriber, { type: 'reset', sessionId: subscriber.sessionId, @@ -256,7 +277,7 @@ export class AgentSessionSubscribers { subscriber.cursor = page.liveCursor ?? page.window.nextCursor return } - const page = result.page + const page = this.ownerPage(subscriber.sessionId, result.page) const advanced = page.window.nextCursor.sequence > subscriber.cursor.sequence if (!advanced) { const commandsChanged = @@ -300,6 +321,10 @@ export class AgentSessionSubscribers { private now = (): number => this.hooks.now?.() ?? Date.now() + private ownerPage(sessionId: string, page: AgentSessionHistoryPage): AgentSessionHistoryPage { + return this.hooks.projectPage?.(sessionId, page) ?? page + } + private isActive = (subscriber: Subscriber): boolean => this.bySession.get(subscriber.sessionId)?.get(subscriber.id) === subscriber 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 acd40dd882b..19a97cf2ed2 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 @@ -9,7 +9,13 @@ import { join } from 'node:path' import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest' import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' import { hasUnansweredStructuredAgentSessionDispatch } from '../../../shared/structured-agent-session-projection' +import { + activeStructuredAgentSessionTurnId, + liveStructuredAgentSessionItems, + projectStructuredAgentSessionStatus +} from '../../../shared/structured-agent-session-projection' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' import type { AgentSessionMutationEnvelope, @@ -25,10 +31,7 @@ import type { StructuredTuiOwner } from './structured-agent-session-handoff-types' import { StructuredHandoffTestRequests } from './structured-agent-session-handoff-test-requests' -import { - unexpectedProviderExitOutcome, - UNEXPECTED_PROVIDER_EXIT_OUTCOME -} from './structured-agent-session-dead-generation-settlement' +import { unexpectedProviderExitOutcome } from './structured-agent-session-dead-generation-settlement' import { AgentSessionJournal } from '../agent-session-journal/journal-store' import type { StructuredAgentSessionStatusSink } from './structured-agent-session-status-feed' import { @@ -266,6 +269,48 @@ afterEach(async () => { }) describe('a chat that closes', () => { + it('retries a failed close settlement on cold read without a new provider child', async () => { + await attach() + emitTurnLifecycle('running', 1) + sink?.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 2 }, + { + kind: 'question', + question: 'Before close?', + options: [{ id: 'yes', label: 'Yes' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + } + ) + await host.flushStreamedEvents(SESSION) + const appendSettlement = vi.spyOn(AgentSessionJournal.prototype, 'appendLifecycleBatch') + appendSettlement.mockRejectedValueOnce(new Error('close journal unavailable')) + + await host.close(SESSION) + expect(store.getRecord(SESSION)?.lease).toMatchObject({ + claimStatus: 'released', + settlementRetryRequired: true, + settlementRetryFence: 1 + }) + appendSettlement.mockRestore() + + await reboot() + await host.restoreReadableSessions() + expect(acquire).not.toHaveBeenCalled() + expect(store.getRecord(SESSION)?.lease.settlementRetryRequired).toBeUndefined() + const history = host.history({ sessionId: SESSION, direction: 'tail' }) + expect(history.ok).toBe(true) + if (history.ok) { + expect( + liveStructuredAgentSessionItems(history.page.items, 2).filter( + (item) => + (item.body.kind === 'approval' || item.body.kind === 'question') && + item.body.resolution.state === 'pending' + ) + ).toEqual([]) + expect(activeStructuredAgentSessionTurnId(history.page.items)).toBeNull() + } + }) + it('releases the provider child it was holding', async () => { await attach() await host.hold(SESSION, SURFACE) @@ -756,6 +801,25 @@ describe('an unexpected provider exit', () => { await attach() await host.hold(SESSION, SURFACE) emitTurnLifecycle('running', 1) + sink?.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 2 }, + { + kind: 'approval', + title: 'Old approval', + detail: null, + options: [{ id: 'yes', label: 'Allow' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + } + ) + sink?.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 3 }, + { + kind: 'question', + question: 'Old question', + options: [{ id: 'yes', label: 'Yes' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + } + ) await host.flushStreamedEvents(SESSION) const runtimeState = ( host as unknown as { @@ -783,10 +847,31 @@ describe('an unexpected provider exit', () => { expect(store.getRecord(SESSION)?.lease).toMatchObject({ claimStatus: 'live', handoffStage: null, - settlementRetryRequired: undefined, + settlementRetryRequired: true, runtimeFence: exitedFence + 2 }) expect(acquire).toHaveBeenCalledTimes(2) + const replacementFence = store.getRecord(SESSION)?.lease.runtimeFence ?? 0 + const replacement = host.history({ sessionId: SESSION, direction: 'tail' }) + expect(replacement.ok).toBe(true) + if (!replacement.ok) { + throw new Error('replacement history unreadable') + } + expect( + liveStructuredAgentSessionItems(replacement.page.items, replacementFence).filter( + (item) => + (item.body.kind === 'approval' || item.body.kind === 'question') && + item.body.resolution.state === 'pending' + ) + ).toEqual([]) + expect( + activeStructuredAgentSessionTurnId( + liveStructuredAgentSessionItems(replacement.page.items, replacementFence) + ) + ).toBeNull() + expect(projectStructuredAgentSessionStatus(replacement.page.items, [], replacementFence)).toBe( + 'idle' + ) dispatch.mockResolvedValueOnce({ state: 'accepted', providerIdentity: { provider: 'codex', threadId: THREAD, turnId: 'after-exit', ordinal: 1 } @@ -795,8 +880,15 @@ describe('an unexpected provider exit', () => { expect( await host.send(CALLER, { envelope: envelope('agentSession.send', { body }), body }) ).toMatchObject({ ok: true, value: { submission: { dispatchState: 'accepted' } } }) - await host.close(SESSION) - expect(host.hasSession(SESSION)).toBe(false) + vi.useFakeTimers() + try { + host.release(SESSION, SURFACE) + await vi.advanceTimersByTimeAsync(GRACE_MS * 3) + expect(closeSession).toHaveBeenCalled() + } finally { + vi.useRealTimers() + } + await vi.waitFor(() => expect(host.hasSession(SESSION)).toBe(false)) appendSettlement.mockRestore() expect(await host.attach(CALLER, hostTestAttachParams(exitedFence + 3))).toMatchObject({ @@ -813,11 +905,62 @@ describe('an unexpected provider exit', () => { history.ok && history.page.items.find( (item) => - item.body.kind === 'status' && item.body.text === UNEXPECTED_PROVIDER_EXIT_OUTCOME + item.body.kind === 'status' && + item.body.text === unexpectedProviderExitOutcome('provider exited') )?.recovered ).toBe(true) expect(acquire).toHaveBeenCalledTimes(3) }) + + it('keeps the newest witnessed verdict distinct after two failed generations', async () => { + await attach() + await host.hold(SESSION, SURFACE) + emitTurnLifecycle('running', 1) + await host.flushStreamedEvents(SESSION) + const appendSettlement = vi + .spyOn(AgentSessionJournal.prototype, 'appendLifecycleBatch') + .mockRejectedValue(new Error('journal unavailable')) + await host.handleAdapterEvent({ + type: 'ended', + sessionId: SESSION, + reason: 'first exit', + cause: 'unexpected-exit', + fence: 1, + acquisitionGeneration: 'generation-1', + observedAt: NOW - 1 + }) + expect(store.getRecord(SESSION)?.lease.runtimeFence).toBe(3) + sink?.appendItem( + { provider: 'codex', threadId: THREAD, turnId: 'turn-second', ordinal: 1 }, + { kind: 'turn', turnId: 'turn-second', state: 'running', startedAt: NOW } + ) + await host.flushStreamedEvents(SESSION) + await host.close(SESSION) + expect(store.getRecord(SESSION)?.lease).toMatchObject({ + settlementRetryRequired: true, + settlementRetryFence: 3, + deathEvidence: { kind: 'exit-observed', observedAt: NOW } + }) + + appendSettlement.mockRestore() + expect(await host.attach(CALLER, hostTestAttachParams(4))).toMatchObject({ ok: true }) + const history = host.history({ sessionId: SESSION, direction: 'tail' }) + expect(history.ok).toBe(true) + if (history.ok) { + const turns = history.page.items.flatMap((item) => { + const turn = readAgentJournalTurn(item.body) + return turn ? [turn] : [] + }) + expect(turns).toEqual( + expect.arrayContaining([ + expect.objectContaining({ turnId: 'turn-1', state: 'unverifiable' }), + expect.objectContaining({ turnId: 'turn-second', state: 'interrupted', completedAt: NOW }) + ]) + ) + expect(turns.find((turn) => turn.turnId === 'turn-1')).not.toHaveProperty('completedAt') + } + expect(store.getRecord(SESSION)?.lease.settlementRetryRequired).toBeUndefined() + }) }) describe('a chat handed to a terminal and taken back', () => { 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 b37c97b2954..3872d028d28 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 @@ -247,7 +247,9 @@ describe('provider-exit recovery tickets', () => { expect(session.journal.markPendingSubmissionsUnknown).toHaveBeenCalledWith( 7, { mode: 'death-confirmed' }, - 'provider_exited_before_acknowledgement' + 'provider_exited_before_acknowledgement', + 7, + 7 ) expect(session.hasProviderChild).toBe(false) // The running row is revised to interrupted at exit receipt, never tombstoned. @@ -369,7 +371,7 @@ describe('provider-exit recovery tickets', () => { snapshot: () => ({ items: [] }), appendLifecycleBatch: vi.fn(async () => ({ epoch: 'epoch-1', sequence: 1 })), markPendingSubmissionsUnknown, - submissions: () => [{ clientMessageId: 'client-1', dispatchState: 'pending' }] + submissions: () => [{ clientMessageId: 'client-1', dispatchState: 'pending', fence: 7 }] } } @@ -395,7 +397,9 @@ describe('provider-exit recovery tickets', () => { expect(markPendingSubmissionsUnknown).toHaveBeenCalledWith( 7, { mode: 'death-confirmed' }, - 'provider_exited_before_acknowledgement' + 'provider_exited_before_acknowledgement', + 7, + 7 ) expect(session.journal.appendLifecycleBatch).toHaveBeenCalledWith( expect.objectContaining({ 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 89509901a67..e1bbfd4c3bc 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 @@ -77,7 +77,11 @@ export async function settleUnexpectedStructuredAgentSessionExit< let settlementFailed = false const stableSettlementId = providerExitSettlementId(unexpectedEvent) - const unfinishedWork = captureUnfinishedStructuredAgentSessionWork(session.journal) + const unfinishedWork = captureUnfinishedStructuredAgentSessionWork( + session.journal, + session.fence, + session.fence + ) let released: Awaited< ReturnType > | null = null @@ -99,7 +103,9 @@ export async function settleUnexpectedStructuredAgentSessionExit< showUnexpectedExitOutcome: unfinishedStructuredAgentSessionWorkWasInterrupted( unfinishedWork, session.journal, - observedAt + observedAt, + session.fence, + session.fence ) })) } finally { @@ -189,6 +195,7 @@ async function retryUnexpectedExitSettlement(input: { journal: input.session.journal, sessionId: input.event.sessionId, fence: input.session.fence, + fromFence: input.session.fence, settlementId: input.stableSettlementId, verdict: input.verdict, pendingSubmissionReason: 'provider_exited_before_acknowledgement', diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.test.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.test.ts index 5a3370c910f..cae3bd5cc1a 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.test.ts @@ -1,6 +1,10 @@ import { describe, expect, it } from 'vitest' import type { AgentSessionRecord } from '../../../shared/agent-session-record' import type { AgentSessionBackgroundTaskState } from '../../../shared/agent-session-wire' +import { + agentSessionLeaseFixture, + agentSessionRecordFixture +} from '../../../shared/agent-session-record.test-fixture' import { conversationCommandBlocked } from './structured-conversation-command-admission' import type { AgentSessionTurnContext } from './structured-agent-session-turns' @@ -67,4 +71,56 @@ describe('conversationCommandBlocked background tasks', () => { 'Wait for the current turn to finish before using this command.' ) }) + + it('does not let an earlier owner block a conversation command', () => { + const ctx = contextWith(null) + ctx.journal.snapshot = () => ({ + sessionId: 'session-1', + cursor: { epoch: 'epoch', sequence: 1 }, + submissions: [], + items: [ + { + itemId: 'old-turn', + revision: 1, + sequence: 1, + observedAt: 1, + ownerFence: 7, + body: { kind: 'turn', turnId: 'old', state: 'running' } + }, + { + itemId: 'old-prompt', + revision: 1, + sequence: 2, + observedAt: 2, + ownerFence: 7, + body: { + kind: 'approval', + title: 'Old', + detail: null, + options: [], + resolution: { + state: 'pending', + selectedOptionId: null, + resolvedBy: null, + resolvedAt: null + } + } + } + ] + }) + ctx.journal.submissions = () => [ + { + clientMessageId: 'old', + fence: 7, + payloadFingerprint: 'fp', + dispatchState: 'pending', + providerItemId: null, + reason: null, + submittedAt: 1, + resolvedAt: null + } + ] + const record = agentSessionRecordFixture(agentSessionLeaseFixture({ runtimeFence: 9 })) + expect(conversationCommandBlocked(ctx, record)).toBeNull() + }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.ts index 78b9f4a59d0..f123ff3a1f6 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.ts +++ b/src/main/native-chat/agent-session-wire/structured-conversation-command-admission.ts @@ -1,12 +1,18 @@ import type { AgentSessionRecord } from '../../../shared/agent-session-record' -import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection' +import { + activeStructuredAgentSessionTurnId, + liveStructuredAgentSessionItems +} from '../../../shared/structured-agent-session-projection' import type { AgentSessionTurnContext } from './structured-agent-session-turns' export function conversationCommandBlocked( ctx: AgentSessionTurnContext, record: AgentSessionRecord ): string | null { - const items = ctx.journal.snapshot().items + const items = liveStructuredAgentSessionItems( + ctx.journal.snapshot().items, + record.lease.runtimeFence + ) if (record.rewind?.phase === 'prepared' || record.rewind?.phase === 'provider-succeeded') { return 'agent_session_rewind:outcome-unknown' } @@ -50,7 +56,11 @@ export function conversationCommandBlocked( if ( ctx.journal .submissions() - .some((entry) => entry.dispatchState === 'pending' || entry.dispatchState === 'unknown') + .some( + (entry) => + entry.fence === record.lease.runtimeFence && + (entry.dispatchState === 'pending' || entry.dispatchState === 'unknown') + ) ) { return 'Resolve pending or unconfirmed messages before using this command.' } diff --git a/src/main/runtime/agent-session-lease-transitions.ts b/src/main/runtime/agent-session-lease-transitions.ts index 28617817647..03e72892603 100644 --- a/src/main/runtime/agent-session-lease-transitions.ts +++ b/src/main/runtime/agent-session-lease-transitions.ts @@ -90,9 +90,16 @@ export function reserveAgentSessionOwner(args: { handoffOperationId: reservation.handoffOperationId, claimKeyId: reservation.claimKeyId, claimStatus: 'reserved', - settlementRetryRequired: undefined, - settlementRetryId: undefined, - deathEvidence: null + // A prior generation's journal obligation survives the new reservation. + settlementRetryRequired: record.lease.settlementRetryRequired, + settlementRetryId: record.lease.settlementRetryId, + settlementRetryFence: record.lease.settlementRetryRequired + ? (record.lease.settlementRetryFence ?? + (record.lease.claimStatus === 'released' + ? record.lease.runtimeFence - 1 + : record.lease.runtimeFence)) + : undefined, + deathEvidence: record.lease.deathEvidence }) } } @@ -249,7 +256,8 @@ export function evictAgentSessionOwner(args: { settlementRetryRequired: settlementRequired ? true : undefined, settlementRetryId: settlementRequired ? agentSessionRestartEvictionSettlementId(record.lease, adjudication) - : undefined + : undefined, + settlementRetryFence: settlementRequired ? record.lease.runtimeFence : undefined }) } diff --git a/src/main/runtime/agent-session-record-store.test.ts b/src/main/runtime/agent-session-record-store.test.ts index 4b325fa0ff3..2c0bb8a7465 100644 --- a/src/main/runtime/agent-session-record-store.test.ts +++ b/src/main/runtime/agent-session-record-store.test.ts @@ -19,6 +19,7 @@ import { AGENT_SESSION_STORE_FILE_NAME } from './agent-session-record-store-file' import type { AgentSessionReserveRequest } from './agent-session-reservation-admission' +import { releaseStoredAgentSessionOwnerAfterSurfaceClose } from './agent-session-surface-release-transition' const NOW = 1_800_000_000_000 @@ -651,6 +652,64 @@ describe('restart reconciliation', () => { expect(reacquired.disposition).toBe('reserved') }) + it('retains witnessed settlement across reservation and a crash before provider bind', async () => { + const store = await open() + await establishOwner(store) + await releaseStoredAgentSessionOwnerAfterSurfaceClose(store, { + sessionId: 'session-alpha', + expectedFence: 1, + now: NOW + 1, + settlementRetry: { settlementId: 'provider-exit:session-alpha:1:owner-a', detail: 'exit 17' } + }) + const reserved = await store.reserveOwner( + reserveRequest({ + expectedFence: 2, + probe: { outcome: 'pid-absent' }, + operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' } + }) + ) + expect(reserved.record.lease).toMatchObject({ + claimStatus: 'reserved', + settlementRetryRequired: true, + settlementRetryFence: 1, + deathEvidence: { kind: 'exit-observed', observedAt: NOW + 1, detail: 'exit 17' } + }) + + const restarted = await open() + expect(restarted.getRecord('session-alpha')?.lease).toMatchObject({ + settlementRetryId: 'provider-exit:session-alpha:1:owner-a', + settlementRetryFence: 1, + deathEvidence: { kind: 'exit-observed', observedAt: NOW + 1 } + }) + }) + + it('derives a pre-migration settlement boundary before reserving a replacement', async () => { + const store = await open() + await establishOwner(store) + await releaseStoredAgentSessionOwnerAfterSurfaceClose(store, { + sessionId: 'session-alpha', + expectedFence: 1, + now: NOW + 1, + settlementRetry: { settlementId: 'legacy-exit', detail: 'observed' } + }) + await store.transitionHandoff('session-alpha', (record) => ({ + ...record, + lease: { ...record.lease, settlementRetryFence: undefined } + })) + const reserved = await store.reserveOwner( + reserveRequest({ + expectedFence: 2, + probe: { outcome: 'pid-absent' }, + operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' } + }) + ) + expect(reserved.record.lease).toMatchObject({ + runtimeFence: 3, + settlementRetryFence: 1, + settlementRetryId: 'legacy-exit' + }) + }) + it('frees a reservation that provably never spawned', async () => { const first = await open() await first.reserveOwner(reserveRequest()) diff --git a/src/main/runtime/agent-session-restart-lease-transitions.ts b/src/main/runtime/agent-session-restart-lease-transitions.ts index a50fd4cd94e..f512ee26731 100644 --- a/src/main/runtime/agent-session-restart-lease-transitions.ts +++ b/src/main/runtime/agent-session-restart-lease-transitions.ts @@ -80,7 +80,8 @@ export function applyAgentSessionRestartAdjudication(args: { handoffOperationId: null, deathEvidence: adjudication.evidence, settlementRetryRequired: true, - settlementRetryId: agentSessionRestartEvictionSettlementId(record.lease, adjudication) + settlementRetryId: agentSessionRestartEvictionSettlementId(record.lease, adjudication), + settlementRetryFence: record.lease.runtimeFence }) } const stage: AgentSessionHandoffStage = diff --git a/src/main/runtime/agent-session-surface-release-transition.ts b/src/main/runtime/agent-session-surface-release-transition.ts index 02000aaad9e..4513a79aab1 100644 --- a/src/main/runtime/agent-session-surface-release-transition.ts +++ b/src/main/runtime/agent-session-surface-release-transition.ts @@ -47,14 +47,22 @@ export function releaseAgentSessionOwnerAfterSurfaceClose(args: { processlessAt: null, claimStatus: 'released', handoffStage: null, - settlementRetryRequired: args.settlementRetry ? true : undefined, - settlementRetryId: args.settlementRetry?.settlementId, + settlementRetryRequired: + args.settlementRetry || record.lease.settlementRetryRequired ? true : undefined, + settlementRetryId: args.settlementRetry?.settlementId ?? record.lease.settlementRetryId, + settlementRetryFence: args.settlementRetry + ? args.expectedFence + : record.lease.settlementRetryFence, lastRenewedAt: args.now, - deathEvidence: { - kind: 'exit-observed', - detail: args.settlementRetry?.detail ?? 'the last surface holding this session released it', - observedAt: args.exitObservedAt ?? args.now - } + deathEvidence: + record.lease.settlementRetryRequired && !args.settlementRetry + ? record.lease.deathEvidence + : { + kind: 'exit-observed', + detail: + args.settlementRetry?.detail ?? 'the last surface holding this session released it', + observedAt: args.exitObservedAt ?? args.now + } }) } diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts index 91bf1158e07..a50482f5124 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-host.test.ts @@ -45,7 +45,7 @@ describe('structured mailbox pointer host', () => { // then delivered mid-turn, which Codex coalesces into the running turn and Claude queues behind // it -- either way folded into work already in flight rather than read as a new instruction. const items = [runningTurn(), ...transcript(500)] - hostRef.current = { journalSnapshot: () => ({ items }) } + hostRef.current = { currentOwnerJournalItems: () => items } expect(createStructuredMailboxPointerHost().readGateFacts('s1')).toEqual({ turnRunning: true, awaitingHuman: false @@ -57,7 +57,7 @@ describe('structured mailbox pointer host', () => { // runtime cannot see at all. expect(createStructuredMailboxPointerHost().readGateFacts('s1')).toBeNull() hostRef.current = { - journalSnapshot: () => { + currentOwnerJournalItems: () => { throw new Error('agent_session_ownership_unknown') } } diff --git a/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts b/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts index b722952df92..044b19aa94f 100644 --- a/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts +++ b/src/main/runtime/orchestration/structured-mailbox-pointer-host.ts @@ -46,7 +46,7 @@ export function readStructuredSessionGateFacts( return null } try { - return structuredSessionGateFacts(host.journalSnapshot(sessionId).items) + return structuredSessionGateFacts(host.currentOwnerJournalItems(sessionId)) } catch (error) { // Not attached is a retain reason, not a failure; anything else is still unreadable. if ((error as Error)?.message !== AGENT_SESSION_NOT_ATTACHED.code) { diff --git a/src/renderer/src/components/native-chat/structured-agent-session-message-projection.test.ts b/src/renderer/src/components/native-chat/structured-agent-session-message-projection.test.ts index 745aa5b4314..1e026dccdd6 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-message-projection.test.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-message-projection.test.ts @@ -5,7 +5,10 @@ import type { } from '../../../../shared/agent-session-journal-types' import { agentJournalSubmissionKey } from '../../../../shared/agent-session-journal-item-key' import { createStructuredAgentSessionOutboxEntry } from '../../../../shared/structured-agent-session-outbox' -import { projectStructuredAgentSessionMessages } from './structured-agent-session-message-projection' +import { + pendingStructuredSessionPrompts, + projectStructuredAgentSessionMessages +} from './structured-agent-session-message-projection' function submission(index: number): AgentJournalSubmission { return { @@ -31,6 +34,32 @@ function item(index: number): AgentJournalRenderItem { } describe('structured agent session message projection', () => { + it('shows historical prompts without letting them replace the current owner composer', () => { + const approval: AgentJournalRenderItem = { + ...item(1), + ownerFence: 1, + body: { + kind: 'approval', + title: 'Old approval', + detail: null, + options: [{ id: 'yes', label: 'Allow' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + } + } + const current: AgentJournalRenderItem = { + ...item(2), + ownerFence: 3, + body: { + kind: 'question', + question: 'Current question', + options: [{ id: 'yes', label: 'Yes' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + } + } + expect(pendingStructuredSessionPrompts([approval], 3)).toEqual([]) + expect(pendingStructuredSessionPrompts([approval, current], 3)).toEqual([current]) + }) + it.each([5, 10])('renders %i rapid accepted desktop sends exactly once', (sendCount) => { const outbox = Array.from({ length: sendCount }, (_, index) => createStructuredAgentSessionOutboxEntry({ diff --git a/src/renderer/src/components/native-chat/structured-agent-session-message-projection.ts b/src/renderer/src/components/native-chat/structured-agent-session-message-projection.ts index 0ea6333a0da..94ecaa8c5f4 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-message-projection.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-message-projection.ts @@ -1,4 +1,5 @@ import type { AgentJournalRenderItem } from '../../../../shared/agent-session-journal-types' +import { liveStructuredAgentSessionItems } from '../../../../shared/structured-agent-session-projection' export { projectStructuredAgentSessionMessages } from '../../../../shared/structured-agent-session-message-projection' @@ -7,9 +8,10 @@ export type StructuredPromptItem = AgentJournalRenderItem & { } export function pendingStructuredSessionPrompts( - items: AgentJournalRenderItem[] + items: AgentJournalRenderItem[], + currentFence?: number | null ): StructuredPromptItem[] { - return items.filter( + return liveStructuredAgentSessionItems(items, currentFence).filter( (item): item is StructuredPromptItem => (item.body.kind === 'approval' || item.body.kind === 'question') && item.body.resolution.state === 'pending' diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.test.tsx index c57ebe1b384..fc00eeee6a6 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.test.tsx @@ -165,6 +165,37 @@ describe('useStructuredAgentSessionOutbox', () => { await waitFor(() => expect(result.current.outbox).toHaveLength(0)) }) + it('makes Retry on an old-fence head dispatchable under a new identity', async () => { + mocks.call.mockReturnValue(new Promise(() => {})) + const emptySubmissions: readonly AgentJournalSubmission[] = [] + const { result, rerender } = renderHook( + ({ fence, submissions }: { fence: number; submissions: readonly AgentJournalSubmission[] }) => + useStructuredAgentSessionOutbox({ + sessionId: 'session-1', + target: LOCAL_TARGET, + fence, + submissions + }), + { initialProps: { fence: 1, submissions: emptySubmissions } } + ) + act(() => expect(result.current.send('retry me')).toBe(true)) + await waitFor(() => expect(mocks.call).toHaveBeenCalledOnce()) + const oldId = result.current.outbox[0]!.clientMessageId + rerender({ fence: 2, submissions: [pendingResultFor(oldId, 10).value.submission] }) + await waitFor(() => expect(result.current.outbox[0]?.state).toBe('unconfirmed')) + expect(mocks.call).toHaveBeenCalledOnce() + + act(() => result.current.retry(oldId)) + await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(2)) + expect(mocks.call.mock.calls[1]?.[2]).toMatchObject({ + envelope: { + expectedRuntimeFence: 2, + clientOperationId: result.current.outbox[0]?.clientMessageId + } + }) + expect(result.current.outbox[0]?.clientMessageId).not.toBe(oldId) + }) + it.each(['agent_session_operation_conflict', 'agent_session_operation_expired'] as const)( 'rotates a send operation after %s', async (code) => { @@ -238,6 +269,40 @@ describe('useStructuredAgentSessionOutbox', () => { expect(result.current.error).toBeNull() }) + it('keeps an old pending send visible while dispatching a new owner send', async () => { + const initialSubmissions: readonly AgentJournalSubmission[] = [] + mocks.call.mockImplementation(async (_target, _method, params) => { + return params.body.blocks[0]?.text === 'old' + ? pendingResultFor(params.envelope.clientOperationId, 10) + : acceptedResultFor(params.envelope.clientOperationId, 3) + }) + const { result, rerender } = renderHook( + ({ fence, submissions }: { fence: number; submissions: readonly AgentJournalSubmission[] }) => + useStructuredAgentSessionOutbox({ + sessionId: 'session-1', + target: LOCAL_TARGET, + fence, + submissions + }), + { initialProps: { fence: 1, submissions: initialSubmissions } } + ) + + act(() => expect(result.current.send('old')).toBe(true)) + await waitFor(() => expect(result.current.outbox[0]?.state).toBe('dispatching')) + const oldId = result.current.outbox[0]!.clientMessageId + rerender({ fence: 3, submissions: [pendingResultFor(oldId, 10).value.submission] }) + await waitFor(() => expect(result.current.outbox[0]?.state).toBe('unconfirmed')) + + act(() => expect(result.current.send('new')).toBe(true)) + await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(2)) + expect(mocks.call.mock.calls[1]?.[2]).toMatchObject({ + envelope: { expectedRuntimeFence: 3 }, + body: { blocks: [{ text: 'new' }] } + }) + await waitFor(() => expect(result.current.outbox).toHaveLength(1)) + expect(result.current.outbox[0]).toMatchObject({ clientMessageId: oldId, state: 'unconfirmed' }) + }) + it('lets no transport error reopen a send the journal already settled', async () => { // The RPC fails while the host has already accepted: the journal is the // authority, so the entry leaves the outbox and no Retry is offered for it. diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.ts b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.ts index 7934fc4759d..63bf65f7db5 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.ts +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox.ts @@ -8,6 +8,7 @@ import { createStructuredAgentSessionOperationId } from '../../../../shared/stru import { createStructuredAgentSessionOutboxEntry, reconcileStructuredAgentSessionOutbox, + structuredAgentSessionDispatchHead, structuredAgentSessionSendRequest, type StructuredAgentSessionOutboxEntry } from '../../../../shared/structured-agent-session-outbox' @@ -99,11 +100,12 @@ export function useStructuredAgentSessionOutbox(args: { (submission) => submission.clientMessageId === current[0]?.clientMessageId ) const hostOwnsHead = - headSubmission?.dispatchState === 'pending' || headSubmission?.dispatchState === 'accepted' + headSubmission?.fence === fence && + (headSubmission.dispatchState === 'pending' || headSubmission.dispatchState === 'accepted') const hostSettledHeadError = current[0]?.state === 'unconfirmed' || blockedIdRef.current === headSubmission?.clientMessageId - const next = reconcileStructuredAgentSessionOutbox(current, submissions) + const next = reconcileStructuredAgentSessionOutbox(current, submissions, fence) if (next.some((entry, index) => entry !== current[index]) || next.length !== current.length) { outboxRef.current = next setOutbox(next) @@ -121,7 +123,7 @@ export function useStructuredAgentSessionOutbox(args: { setError(null) } } - }, [sessionId, submissions]) + }, [fence, sessionId, submissions]) // The one place that owns the refs, the React state and the storage write. const applyDisposition = useCallback( @@ -137,7 +139,8 @@ export function useStructuredAgentSessionOutbox(args: { ) useEffect(() => { - const next = outbox[0] + const next = + fence === null ? undefined : structuredAgentSessionDispatchHead(outbox, submissions, fence) if ( !next || next.sessionId !== sessionId || @@ -150,10 +153,11 @@ export function useStructuredAgentSessionOutbox(args: { } dispatchingRef.current = true const dispatchGeneration = dispatchGenerationRef.current - const staged = [ - { ...next, state: 'dispatching' as const, lastAttemptAt: Date.now() }, - ...outbox.slice(1) - ] + const staged = outbox.map((entry) => + entry.clientMessageId === next.clientMessageId + ? { ...entry, state: 'dispatching' as const, lastAttemptAt: Date.now() } + : entry + ) if (!writeOutbox(sessionId, staged)) { dispatchingRef.current = false blockedIdRef.current = next.clientMessageId @@ -200,7 +204,7 @@ export function useStructuredAgentSessionOutbox(args: { dispatchingRef.current = false } }) - }, [applyDisposition, fence, outbox, sessionId, target]) + }, [applyDisposition, fence, outbox, sessionId, submissions, target]) // A transport-side unknown may never have reached the host, and nothing else // moves it out of `unconfirmed`, so one wedges the whole FIFO queue. Re-issuing @@ -208,7 +212,8 @@ export function useStructuredAgentSessionOutbox(args: { // replays a recorded outcome, or the host performs a genuine first delivery. // A host-confirmed unknown stays parked until the user explicitly asks Retry // to replay the same operation. - const head = outbox[0] + const head = + fence === null ? undefined : structuredAgentSessionDispatchHead(outbox, submissions, fence) // Depend on primitives: `submissions` is rebuilt on every streaming batch, so an // array-identity dep would reset the backoff forever while the agent is working. // A non-null `retryAfterUnknownSubmittedAt` means the user already retried, so @@ -281,6 +286,7 @@ export function useStructuredAgentSessionOutbox(args: { if ( current && (submission?.dispatchState === 'rejected' || + (fence !== null && submission !== undefined && submission.fence < fence) || retryWithFreshClientMessageIdRef.current === clientMessageId) ) { retryWithFreshClientMessageIdRef.current = null diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session.ts b/src/renderer/src/components/native-chat/use-structured-agent-session.ts index 020647de089..afc4bfaa978 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session.ts +++ b/src/renderer/src/components/native-chat/use-structured-agent-session.ts @@ -25,7 +25,8 @@ import { } from '../../../../shared/structured-agent-session-options' import { activeStructuredAgentSessionTurnId, - hasUnansweredStructuredAgentSessionDispatch + hasUnansweredStructuredAgentSessionDispatch, + liveStructuredAgentSessionItems } from '../../../../shared/structured-agent-session-projection' import type { RuntimeClientTarget } from '@/runtime/runtime-rpc-client' import { @@ -104,7 +105,9 @@ export function useStructuredAgentSession(args: { }, [agent, sessionId, state.fence]) // Refresh options each turn to confirm which model the provider actually selected. - const turnId = activeStructuredAgentSessionTurnId(state.items) + const turnId = activeStructuredAgentSessionTurnId( + liveStructuredAgentSessionItems(state.items, state.fence) + ) // A dispatch the provider has not answered is already work; Claude's running row trails the // send by seconds, and only a provider-minted turn is cancellable, so the two stay separate. const isWorking = @@ -239,7 +242,7 @@ export function useStructuredAgentSession(args: { [optionSnapshot, setOption] ) - const prompts = pendingStructuredSessionPrompts(state.items) + const prompts = pendingStructuredSessionPrompts(state.items, state.fence) const { outbox } = outboxController const messages = useStructuredAgentSessionMessages(state.items, outbox, state.submissions) return { diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index fc603fdca08..746bb6bccf9 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -224,6 +224,8 @@ export type AgentJournalRenderItem = { body: AgentJournalItemBody sequence: number observedAt: number + /** Fence of the first row, unchanged by later revisions or settlement. */ + ownerFence?: number /** Set when the row was appended by crash reconciliation rather than live. */ recovered?: true } diff --git a/src/shared/agent-session-record.ts b/src/shared/agent-session-record.ts index a8add5c1128..2e07231fe9c 100644 --- a/src/shared/agent-session-record.ts +++ b/src/shared/agent-session-record.ts @@ -120,6 +120,8 @@ export type AgentSessionLease = { settlementRetryRequired?: boolean /** Stable lifecycle batch id used when retrying the terminal settlement. */ settlementRetryId?: string + /** Largest journal owner fence covered by the pending settlement. */ + settlementRetryFence?: number } export type AgentSessionRecord = { @@ -328,6 +330,8 @@ function isAgentSessionLease(value: unknown): value is AgentSessionLease { typeof lease.settlementRetryRequired === 'boolean') && (lease.settlementRetryId === undefined || isBoundedString(lease.settlementRetryId, MAX_ID_LENGTH)) && + (lease.settlementRetryFence === undefined || + (Number.isSafeInteger(lease.settlementRetryFence) && lease.settlementRetryFence >= 0)) && (lease.deathEvidence === null || isAgentSessionDeathEvidence(lease.deathEvidence)) ) } diff --git a/src/shared/structured-agent-session-outbox.ts b/src/shared/structured-agent-session-outbox.ts index 80c862f9f02..361f5906ee7 100644 --- a/src/shared/structured-agent-session-outbox.ts +++ b/src/shared/structured-agent-session-outbox.ts @@ -95,7 +95,8 @@ export function requeueStructuredAgentSessionSendRefusal( export function reconcileStructuredAgentSessionOutbox( entries: readonly StructuredAgentSessionOutboxEntry[], - submissions: readonly AgentJournalSubmission[] + submissions: readonly AgentJournalSubmission[], + currentFence?: number | null ): StructuredAgentSessionOutboxEntry[] { const settled = new Map(submissions.map((entry) => [entry.clientMessageId, entry])) return entries.flatMap((entry) => { @@ -109,6 +110,13 @@ export function reconcileStructuredAgentSessionOutbox( ) { return [] } + if (submission && currentFence != null && submission.fence < currentFence) { + return [ + entry.state === 'unconfirmed' && entry.retryAfterUnknownSubmittedAt === -1 + ? entry + : { ...entry, state: 'unconfirmed' as const, retryAfterUnknownSubmittedAt: -1 } + ] + } if (submission?.dispatchState === 'pending') { return entry.state === 'dispatching' ? [entry] : [{ ...entry, state: 'dispatching' as const }] } @@ -123,6 +131,23 @@ export function reconcileStructuredAgentSessionOutbox( }) } +/** A dead owner's unresolved send remains visible but cannot own the new writer's FIFO. */ +export function structuredAgentSessionDispatchHead( + entries: readonly StructuredAgentSessionOutboxEntry[], + submissions: readonly AgentJournalSubmission[], + currentFence: number +): StructuredAgentSessionOutboxEntry | undefined { + const byId = new Map(submissions.map((submission) => [submission.clientMessageId, submission])) + return entries.find((entry) => { + const submission = byId.get(entry.clientMessageId) + return ( + !submission || + (submission.fence >= currentFence && + !(submission.dispatchState === 'unknown' && submission.recovered === true)) + ) + }) +} + export function parseStructuredAgentSessionOutboxEntry( value: unknown, sessionId: string diff --git a/src/shared/structured-agent-session-projection.ts b/src/shared/structured-agent-session-projection.ts index 68e8caf22eb..e5d42e7f8a0 100644 --- a/src/shared/structured-agent-session-projection.ts +++ b/src/shared/structured-agent-session-projection.ts @@ -207,6 +207,17 @@ export function hasUnansweredStructuredAgentSessionDispatch( export type StructuredAgentSessionProjectedStatus = 'working' | 'attention' | 'idle' +/** Historical items remain visible, but only the execution owner can ask or work now. */ +export function liveStructuredAgentSessionItems( + items: readonly AgentJournalRenderItem[], + currentFence: number | null | undefined +): AgentJournalRenderItem[] { + return items.filter( + (item) => + currentFence == null || item.ownerFence === undefined || item.ownerFence === currentFence + ) +} + export function structuredAgentSessionTabId(sessionId: string): string { return `structured-agent-session-${sessionId}` } @@ -216,8 +227,9 @@ export function projectStructuredAgentSessionStatus( submissions: readonly AgentJournalSubmission[] = [], currentFence?: number | null ): StructuredAgentSessionProjectedStatus { + const ownerItems = liveStructuredAgentSessionItems(items, currentFence) if ( - items.some( + ownerItems.some( (item) => (item.body.kind === 'approval' || item.body.kind === 'question') && item.body.resolution.state === 'pending' @@ -225,7 +237,7 @@ export function projectStructuredAgentSessionStatus( ) { return 'attention' } - return activeStructuredAgentSessionTurnId(items) || + return activeStructuredAgentSessionTurnId(ownerItems) || hasUnansweredStructuredAgentSessionDispatch(submissions, currentFence) ? 'working' : 'idle' @@ -300,7 +312,10 @@ export function projectStructuredAgentSessionStatusSummary( return { status: null, latestPrompt: '' } } const status = projectStructuredAgentSessionStatus(items, submissions, currentFence) - const activeToolCall = status === 'working' ? activeStructuredAgentSessionToolCall(items) : null + const activeToolCall = + status === 'working' + ? activeStructuredAgentSessionToolCall(liveStructuredAgentSessionItems(items, currentFence)) + : null const toolName = activeToolCall ? normalizeOptionalField(activeToolCall.name, AGENT_STATUS_TOOL_NAME_MAX_LENGTH) : undefined diff --git a/src/shared/structured-agent-session-send-disposition.ts b/src/shared/structured-agent-session-send-disposition.ts index a2490b8eea0..bb21a83181d 100644 --- a/src/shared/structured-agent-session-send-disposition.ts +++ b/src/shared/structured-agent-session-send-disposition.ts @@ -124,7 +124,10 @@ export function disposeStructuredAgentSessionSendResult( return { entries, error: result.refusal.message, - blockedClientMessageId: entries[0]?.clientMessageId ?? null, + blockedClientMessageId: + entries[ + input.entries.findIndex((entry) => entry.clientMessageId === input.entry.clientMessageId) + ]?.clientMessageId ?? null, retryWithFreshClientMessageId: null } }