diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index 7dc02dd197b..2eb32a3ed48 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -26,7 +26,7 @@ type JournalRowBase = { fence: number /** Observed (provider or host) timestamp. Ordering is by `seq`, not by this. */ ts: number - /** Set when crash reconciliation appended the row after the fact. */ + /** Set when host lifecycle reconciliation appends a row instead of a live provider event. */ recovered?: true } 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 df86c182f35..e54fb712144 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -317,12 +317,18 @@ export class AgentSessionJournal { capturePrecedingPendingSubmissions: () => string[] ) => Promise> ): Promise { - const settlesTurn = bodies.some((body) => { - const turn = readAgentJournalTurn(body) - return turn !== null && turn.state !== 'running' - }) + const terminalTurnIds = new Set( + bodies.flatMap((body) => { + const turn = readAgentJournalTurn(body) + return turn !== null && turn.state !== 'running' ? [turn.turnId] : [] + }) + ) const capturePrecedingPendingSubmissions = (): string[] => { - if (!settlesTurn) { + if (terminalTurnIds.size === 0) { + return [] + } + const activeTurnId = this.activeTurnId() + if (activeTurnId === null || !terminalTurnIds.has(activeTurnId)) { return [] } return this.submissions() @@ -332,7 +338,7 @@ export class AgentSessionJournal { .map((submission) => submission.clientMessageId) } const result = await append(capturePrecedingPendingSubmissions) - if (!result.appended || !settlesTurn) { + if (!result.appended || result.precedingPendingSubmissionIds.length === 0) { return result.value } await Promise.all( @@ -342,7 +348,8 @@ export class AgentSessionJournal { clientMessageId, state: 'unknown', reason: DISPATCH_DOUBT_TURN_SETTLED, - fence + fence, + recovered: true }) } catch (error) { console.warn( diff --git a/src/main/native-chat/agent-session-journal/journal-turn-settlement.test.ts b/src/main/native-chat/agent-session-journal/journal-turn-settlement.test.ts index ddd924f45b8..ca112575596 100644 --- a/src/main/native-chat/agent-session-journal/journal-turn-settlement.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-turn-settlement.test.ts @@ -121,12 +121,13 @@ describe('turn settlement dispatch resolution', () => { expect.objectContaining({ clientMessageId: 'message-1', dispatchState: 'unknown', - reason: 'turn_settled_before_acknowledgement' + reason: 'turn_settled_before_acknowledgement', + recovered: true }) ]) const afterSettlement = journal.readSince(settlement) expect(afterSettlement.ok && afterSettlement.rows).toEqual([ - expect.objectContaining({ kind: 'dispatch', state: 'unknown' }) + expect.objectContaining({ kind: 'dispatch', state: 'unknown', recovered: true }) ]) }) @@ -153,7 +154,8 @@ describe('turn settlement dispatch resolution', () => { expect(journal.submissions()[0]).toMatchObject({ dispatchState: 'unknown', - reason: 'turn_settled_before_acknowledgement' + reason: 'turn_settled_before_acknowledgement', + recovered: true }) }) @@ -233,6 +235,77 @@ describe('turn settlement dispatch resolution', () => { }) }) + it('does not let a terminal record for another turn settle the active turn submission', async () => { + const journal = await open() + await openTurn(journal, 'old-turn') + await journal.appendLifecycleBatch(terminalTurn('old-turn', 'completed')) + await openTurn(journal, 'active-turn') + await appendPending(journal, 'active-message') + + await journal.appendItem( + turnIdentity('old-turn'), + { kind: 'turn', turnId: 'old-turn', state: 'completed' }, + { fence: 1 } + ) + + expect(journal.activeTurnId()).toBe('active-turn') + expect(journal.submissions()[0]).toMatchObject({ dispatchState: 'pending' }) + }) + + it('does not let a repeated direct terminal record settle a later submission', async () => { + const journal = await open() + await openTurn(journal, 'turn-repeated') + await appendPending(journal, 'original') + const identity = turnIdentity('turn-repeated') + const completed = { + kind: 'turn' as const, + turnId: 'turn-repeated', + state: 'completed' as const + } + await journal.appendItem(identity, completed, { fence: 1 }) + await appendPending(journal, 'later') + + await journal.appendItem(identity, completed, { fence: 1 }) + + expect(journal.submissions().find((entry) => entry.clientMessageId === 'later')).toMatchObject({ + dispatchState: 'pending' + }) + }) + + it.each(['accepted', 'rejected'] as const)( + 'lets late %s evidence narrow a turn-settled unknown', + async (state) => { + const journal = await open() + await openTurn(journal, `turn-late-${state}`) + await appendPending(journal, `late-${state}`) + await journal.appendLifecycleBatch(terminalTurn(`turn-late-${state}`, 'completed')) + + await journal.resolveDispatch( + state === 'accepted' + ? { + clientMessageId: `late-${state}`, + state, + providerIdentity: { + provider: 'codex', + threadId: 'thread-1', + turnId: `turn-late-${state}`, + ordinal: 0 + }, + fence: 1 + } + : { + clientMessageId: `late-${state}`, + state, + reason: 'provider later rejected the send', + fence: 1 + } + ) + + expect(journal.submissions()[0]).toMatchObject({ dispatchState: state }) + expect(journal.submissions()[0]).not.toHaveProperty('recovered') + } + ) + it('does not settle a pending submission from another fence', async () => { const journal = await open() await openTurn(journal, 'turn-fenced') @@ -287,4 +360,26 @@ describe('turn settlement dispatch resolution', () => { expect(journal.activeTurnId()).toBeNull() await expect(appendPending(journal, 'next-send')).resolves.toBeUndefined() }) + + it('retires the crash window through the existing next-attach reconciliation', async () => { + const journal = await open() + await openTurn(journal, 'turn-crash-window') + await appendPending(journal, 'unwritten-settlement') + const resolution = vi + .spyOn(journal, 'resolveDispatch') + .mockRejectedValueOnce(new Error('process exited before dispatch row')) + vi.spyOn(console, 'warn').mockImplementation(() => undefined) + await journal.appendLifecycleBatch(terminalTurn('turn-crash-window', 'completed')) + resolution.mockRestore() + await journal.close() + + const restarted = await open() + expect(restarted.submissions()[0]).toMatchObject({ dispatchState: 'pending' }) + await restarted.markPendingSubmissionsUnknown(2) + + expect(restarted.submissions()[0]).toMatchObject({ + dispatchState: 'unknown', + recovered: true + }) + }) }) 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..5bd8b32ee40 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,24 +1,55 @@ import { describe, expect, it } from 'vitest' import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { + AgentJournalSnapshot, + AgentJournalSubmission +} from '../../../shared/agent-session-journal-types' import type { AgentSessionBackgroundTaskState } from '../../../shared/agent-session-wire' import { conversationCommandBlocked } from './structured-conversation-command-admission' -import type { AgentSessionTurnContext } from './structured-agent-session-turns' -function contextWith( - backgroundTasks: AgentSessionBackgroundTaskState | null -): AgentSessionTurnContext { +function snapshot(items: AgentJournalSnapshot['items'] = []): AgentJournalSnapshot { return { sessionId: 'session-1', + cursor: { epoch: 'epoch-1', sequence: 0 }, + items, + submissions: [] + } +} + +function contextWith( + backgroundTasks: AgentSessionBackgroundTaskState | null, + submissions: AgentJournalSubmission[] = [] +): Parameters[0] { + return { + sessionId: 'session-1', + fence: 1, journal: { - snapshot: () => ({ items: [] }), - submissions: () => [] + snapshot: () => snapshot(), + submissions: () => submissions }, adapter: { backgroundTaskState: () => backgroundTasks } - } as unknown as AgentSessionTurnContext + } } const RECORD = { lease: {} } as unknown as AgentSessionRecord +function submission( + dispatchState: AgentJournalSubmission['dispatchState'], + overrides: Partial = {} +): AgentJournalSubmission { + return { + clientMessageId: 'message-1', + fence: 1, + payloadFingerprint: 'fingerprint', + dispatchState, + providerItemId: null, + reason: null, + submittedAt: 1, + resolvedAt: dispatchState === 'pending' ? null : 2, + ...overrides + } +} + describe('conversationCommandBlocked background tasks', () => { it('admits the command when nothing is being monitored', () => { expect(conversationCommandBlocked(contextWith(null), RECORD)).toBeNull() @@ -52,19 +83,44 @@ describe('conversationCommandBlocked background tasks', () => { // so a live fan-out never re-labels the reason or blocks anything new. const ctx = contextWith({ state: 'monitoring', supportsTaskStop: true }) ctx.journal.snapshot = () => - ({ - items: [ - { - id: 'turn-1', - body: { - kind: 'status', - turnLifecycle: { turnId: 'turn-1', state: 'running' } - } + snapshot([ + { + itemId: 'turn-1', + revision: 1, + sequence: 1, + observedAt: 1, + body: { + kind: 'status', + text: 'Working', + turnLifecycle: { turnId: 'turn-1', state: 'running' } } - ] - }) as unknown as ReturnType + } + ]) expect(conversationCommandBlocked(ctx, RECORD)).toBe( 'Wait for the current turn to finish before using this command.' ) }) }) + +describe('conversationCommandBlocked dispatch ownership', () => { + it.each(['pending', 'unknown'] as const)('blocks a live %s dispatch', (dispatchState) => { + expect(conversationCommandBlocked(contextWith(null, [submission(dispatchState)]), RECORD)).toBe( + 'Resolve pending or unconfirmed messages before using this command.' + ) + }) + + it('admits a command after turn settlement retires an unconfirmed dispatch', () => { + const retired = submission('unknown', { + reason: 'turn_settled_before_acknowledgement', + recovered: true + }) + + expect(conversationCommandBlocked(contextWith(null, [retired]), RECORD)).toBeNull() + }) + + it('does not let an unanswered dispatch from an older owner block the current fence', () => { + expect( + conversationCommandBlocked(contextWith(null, [submission('pending', { fence: 0 })]), 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..9f374590aef 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,9 +1,19 @@ import type { AgentSessionRecord } from '../../../shared/agent-session-record' -import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection' +import { + activeStructuredAgentSessionTurnId, + hasUnansweredStructuredAgentSessionDispatch +} from '../../../shared/structured-agent-session-projection' import type { AgentSessionTurnContext } from './structured-agent-session-turns' +type ConversationCommandAdmissionContext = { + sessionId: AgentSessionTurnContext['sessionId'] + fence: AgentSessionTurnContext['fence'] + journal: Pick + adapter: Pick +} + export function conversationCommandBlocked( - ctx: AgentSessionTurnContext, + ctx: ConversationCommandAdmissionContext, record: AgentSessionRecord ): string | null { const items = ctx.journal.snapshot().items @@ -47,11 +57,7 @@ export function conversationCommandBlocked( ? 'Stop background tasks before using this command.' : 'Wait for background tasks to finish before using this command.' } - if ( - ctx.journal - .submissions() - .some((entry) => entry.dispatchState === 'pending' || entry.dispatchState === 'unknown') - ) { + if (hasUnansweredStructuredAgentSessionDispatch(ctx.journal.submissions(), ctx.fence)) { return 'Resolve pending or unconfirmed messages before using this command.' } return null diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index fc603fdca08..37e50cb0f12 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -247,8 +247,8 @@ export type AgentJournalSubmission = { reason: string | null submittedAt: number resolvedAt: number | null - /** Set when crash reconciliation resolved the dispatch, not the provider. A live - * `unknown` is a send still outstanding; a recovered one outlived its writer. */ + /** Set when host lifecycle evidence proves the dispatch has no live owner. A live + * `unknown` is still outstanding; a recovered one remains delivery-ambiguous but retired. */ recovered?: true } diff --git a/src/shared/structured-agent-session-projection.test.ts b/src/shared/structured-agent-session-projection.test.ts index 928180e7a6c..03c4044cbec 100644 --- a/src/shared/structured-agent-session-projection.test.ts +++ b/src/shared/structured-agent-session-projection.test.ts @@ -225,6 +225,25 @@ describe('structured agent session status projection', () => { }) }) + it('reads a turn-settled unknown as retired work for every shared status projection', () => { + const asked = item('asked', 1, { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'go' }] + }) + const retired = { + ...submission('m1', 'unknown'), + reason: 'turn_settled_before_acknowledgement', + recovered: true as const + } + + expect(hasUnansweredStructuredAgentSessionDispatch([retired], 1)).toBe(false) + expect(projectStructuredAgentSessionStatus([asked], [retired], 1)).toBe('idle') + expect(projectStructuredAgentSessionStatusSummary([asked], [retired], 1)).toMatchObject({ + status: 'idle' + }) + }) + it('carries the running tool and the newest assistant prose the sidebar row shows', () => { const ask = item('ask', 1, { kind: 'message', diff --git a/src/shared/structured-agent-session-projection.ts b/src/shared/structured-agent-session-projection.ts index 68e8caf22eb..040be298c67 100644 --- a/src/shared/structured-agent-session-projection.ts +++ b/src/shared/structured-agent-session-projection.ts @@ -187,8 +187,8 @@ export function hasPersistedStructuredAgentSessionTurn( * every session list, so the send itself is the evidence. * * A live `unknown` still counts because an ambiguous adapter reply does not prove the provider - * stopped. A recovered `unknown` does not — it outlived the host generation that sent it, so - * there is nothing still running to report. + * stopped. A recovered `unknown` does not — host lifecycle evidence retired its execution owner, + * so there is nothing still running to report. */ export function hasUnansweredStructuredAgentSessionDispatch( submissions: readonly AgentJournalSubmission[],