diff --git a/config/scripts/vitest-sqlite-runtime-files.mjs b/config/scripts/vitest-sqlite-runtime-files.mjs index 85a0698f4731..0006a1ea7441 100644 --- a/config/scripts/vitest-sqlite-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-runtime-files.mjs @@ -53,12 +53,14 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts', 'src/main/native-chat/agent-session-wire/agent-session-stop-note-projection.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-acquire-isolation.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-runtime-stamp.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-adopted-import.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-append-delivery.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-at-rest-commands.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-attach-reconciliation.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-attach-send-recovery.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-attachment-admission.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-claude-compact-stop.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-claude-echo-working.test.ts', @@ -155,6 +157,7 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/native-chat/agent-session-wire/structured-agent-session-subagent-recency.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-subscriber-background-tasks.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-target-reconcile.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-thread-goal.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-transition.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts', @@ -169,6 +172,8 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/orcad/orcad-local-serve-selection.test.ts', 'src/main/orcad/serve-profile-ssh-targets.test.ts', 'src/main/pi/rpc-launch-resolution.test.ts', + 'src/main/runtime/agent-session-closed-owner-acquisition.test.ts', + 'src/main/runtime/agent-session-closed-owner-history.test.ts', 'src/main/runtime/agent-session-conversation-clear-commit.test.ts', 'src/main/runtime/agent-session-conversation-name-store.test.ts', 'src/main/runtime/agent-session-death-evidence-persistence.test.ts', diff --git a/src/main/acp/acp-recovery-history.test.ts b/src/main/acp/acp-recovery-history.test.ts index 402b9e2f76d1..3d14063abb49 100644 --- a/src/main/acp/acp-recovery-history.test.ts +++ b/src/main/acp/acp-recovery-history.test.ts @@ -100,7 +100,10 @@ describe('ACP restart recovery from the agent store', () => { ) expect(window).toMatchObject({ boundaryConsistent: false, turnInFlight: true }) expect( - reconcileSubmissions({ submissions: [...load.state.submissions.values()], history: window! }) + reconcileSubmissions({ + submissions: [...load.state.submissions.values()], + history: { ...window!, start: null } + }) ).toEqual([ { clientMessageId: 'held', @@ -119,7 +122,10 @@ describe('ACP restart recovery from the agent store', () => { const { window } = await windowFor(load, [stored('msg_1', 'continue', 105)]) expect(window?.items).toEqual([]) expect( - reconcileSubmissions({ submissions: [...load.state.submissions.values()], history: window! }) + reconcileSubmissions({ + submissions: [...load.state.submissions.values()], + history: { ...window!, start: null } + }) ).toMatchObject([{ clientMessageId: 'held', outcome: 'unknown' }]) }) @@ -153,7 +159,10 @@ describe('ACP restart recovery from the agent store', () => { stored('msg_2', 'yes', 405) ]) expect( - reconcileSubmissions({ submissions: [...load.state.submissions.values()], history: window! }) + reconcileSubmissions({ + submissions: [...load.state.submissions.values()], + history: { ...window!, start: null } + }) ).toMatchObject([{ clientMessageId: 'held', outcome: 'accepted', providerItemId: 'msg_2' }]) }) diff --git a/src/main/claude/claude-structured-history-window.test.ts b/src/main/claude/claude-structured-history-window.test.ts index f0a2e78841c7..9c2b87ae7dbc 100644 --- a/src/main/claude/claude-structured-history-window.test.ts +++ b/src/main/claude/claude-structured-history-window.test.ts @@ -327,7 +327,7 @@ describe('a crash between Claude saving a prompt and Orca recording its echo', ( expect(window.items.map((item) => item.providerItemId)).toEqual(['bravo']) const [verdict] = reconcileSubmissions({ - history: window, + history: { ...window, start: { fence: 1, movedAt: 0 } }, submissions: [ { clientMessageId: 'bravo-send', diff --git a/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts b/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts index 278fea939737..c1a0532b6dc6 100644 --- a/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts @@ -23,7 +23,7 @@ import { digestPayload } from './journal-payload-bounds' import { reconcileSubmissions, type ProviderHistoryItem, - type ProviderHistoryWindow + type PlacedProviderHistoryWindow } from './journal-submission-reconciler' import { createTrackedJournalOpener } from './journal-host-database-test-support' import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' @@ -85,11 +85,18 @@ function history(input: { } } +/** Read from a resume point the sends' own owner, at fence 1, set before handing them over. */ function window( items: ProviderHistoryItem[], - overrides: Partial = {} -): ProviderHistoryWindow { - return { items, boundaryConsistent: true, turnInFlight: false, ...overrides } + overrides: Partial = {} +): PlacedProviderHistoryWindow { + return { + items, + boundaryConsistent: true, + turnInFlight: false, + start: { fence: 1, movedAt: 0 }, + ...overrides + } } beforeEach(async () => { @@ -324,7 +331,8 @@ describe('reconciliation matching', () => { providerItemId: null, reason: null, submittedAt: 1, - resolvedAt: null + resolvedAt: null, + firstHandover: { fence: 1, at: 1 } }, { clientMessageId: 'cm_2', @@ -334,7 +342,8 @@ describe('reconciliation matching', () => { providerItemId: null, reason: null, submittedAt: 2, - resolvedAt: null + resolvedAt: null, + firstHandover: { fence: 1, at: 2 } } ] @@ -471,6 +480,101 @@ describe('reconciliation matching', () => { }) }) + it('leaves a send unknown once a later owner moved where the history starts', () => { + // The owner at fence 2 finished a turn after this send: its resume point may sit past it. + const [outcome] = reconcileSubmissions({ + submissions: [submissions[0]!], + history: window([], { start: { fence: 2, movedAt: 0 } }) + }) + expect(outcome).toEqual({ + clientMessageId: 'cm_1', + outcome: 'unknown', + reason: 'history_start_unproven' + }) + }) + + it('stays unknown when no recorded resume point bounds the read', () => { + const [outcome] = reconcileSubmissions({ + submissions: [submissions[0]!], + history: window([], { start: null }) + }) + expect(outcome).toMatchObject({ outcome: 'unknown', reason: 'history_start_unproven' }) + }) + + it('still accepts a send a later-started history holds', () => { + const [outcome] = reconcileSubmissions({ + submissions: [submissions[0]!], + history: window( + [history({ itemId: 'item-1', clientId: 'cm_1', text: 'same text', ordinal: 0 })], + { start: { fence: 2, movedAt: 0 } } + ) + }) + expect(outcome).toMatchObject({ outcome: 'accepted', providerItemId: 'item-1' }) + }) + + it('rejects an absent send handed over at a fence after the start last moved', () => { + const [outcome] = reconcileSubmissions({ + submissions: [{ ...submissions[0]!, firstHandover: { fence: 3, at: 1 } }], + history: window([], { start: { fence: 2, movedAt: 50 } }) + }) + expect(outcome).toMatchObject({ outcome: 'rejected', reason: 'not_delivered' }) + }) + + it('rejects an absent send its own owner handed over after a turn moved the start', () => { + // Turn 1 ended at 10; the send went out at 20 and the host died before the agent took it. + const [outcome] = reconcileSubmissions({ + submissions: [{ ...submissions[0]!, firstHandover: { fence: 1, at: 20 } }], + history: window([], { start: { fence: 1, movedAt: 10 } }) + }) + expect(outcome).toMatchObject({ outcome: 'rejected', reason: 'not_delivered' }) + }) + + it.each([ + ['after', 30], + ['at the same instant as', 20] + ])('leaves a send unknown when its own owner moved the start %s its handover', (_, movedAt) => { + // The agent ran the send without its echo being recorded, and the turn's end moved the start. + const [outcome] = reconcileSubmissions({ + submissions: [{ ...submissions[0]!, firstHandover: { fence: 1, at: 20 } }], + history: window([], { start: { fence: 1, movedAt } }) + }) + expect(outcome).toEqual({ + clientMessageId: 'cm_1', + outcome: 'unknown', + reason: 'history_start_unproven' + }) + }) + + it('dates a send handed over twice by its first handover', async () => { + // A build from before handover was one-way could hand one send over again. + const journal = await open() + await journal.appendSubmission({ + clientMessageId: 'cm_1', + payloadFingerprint: digestPayload('deploy the thing'), + body: userMessage('deploy the thing'), + fence: 1, + handoverRecorded: true + }) + for (const at of [100, 300]) { + clock = at + await journal.resolveDispatch({ + clientMessageId: 'cm_1', + state: 'pending', + turnScope: AGENT_JOURNAL_THREAD_SCOPE, + fence: 1 + }) + } + const restarted = await open() + await restarted.markPendingSubmissionsUnknown(2) + + // The point moved between the two handovers: the first may have delivered it before the start. + const [outcome] = reconcileSubmissions({ + submissions: restarted.submissions(), + history: window([], { start: { fence: 1, movedAt: 200 } }) + }) + expect(outcome).toMatchObject({ outcome: 'unknown', reason: 'history_start_unproven' }) + }) + it('leaves settled submissions alone', () => { expect( reconcileSubmissions({ diff --git a/src/main/native-chat/agent-session-journal/journal-database.ts b/src/main/native-chat/agent-session-journal/journal-database.ts index ba961541a7e9..7dff6f827034 100644 --- a/src/main/native-chat/agent-session-journal/journal-database.ts +++ b/src/main/native-chat/agent-session-journal/journal-database.ts @@ -15,6 +15,7 @@ import { import { JournalUnreleasedSchemaError } from './journal-open-failure' import { ensureQueuedMessagesTable } from './queued-message-schema' import { ensureAgentSessionAttachmentClaimTables } from '../agent-session-attachments/agent-session-attachment-claims' +import { ensureAgentSessionClosedOwnersTable } from '../../runtime/agent-session-closed-owner-rows' export const JOURNAL_BUSY_TIMEOUT_MS = 5000 /** Bounds the WAL a checkpoint leaves behind; SQLite truncates it back to this after a reset. */ @@ -82,6 +83,8 @@ export function openJournalDatabase(dbPath: string): OpenJournalDatabase { // writable open with no `user_version` bump (see `ensureQueuedMessagesTable`). ensureQueuedMessagesTable(probe) ensureAgentSessionAttachmentClaimTables(probe) + // One transaction: creating the table is what marks its one-time migration done. + runJournalTransaction(probe, () => ensureAgentSessionClosedOwnersTable(probe)) hardenSqliteDatabaseFiles(dbPath) transferred = true return { db: probe, readOnly: false } diff --git a/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts b/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts index 305b7383ad9e..eb81910a3070 100644 --- a/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts @@ -50,6 +50,8 @@ export function applyJournalDispatchRow( submission.resolvedAt = row.state === 'pending' ? null : row.ts if (row.state === 'pending') { submission.handedOverAt = row.ts + // The first handover: what it delivered may sit before anything a later one did. + submission.firstHandover ??= { fence: row.fence, at: row.ts } placeHandedOverMessage(state, submission, row) } else if (row.state === 'rejected') { placeRejectedMessage(state, submission, row) diff --git a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts index 8e541f110b00..30cd0ec141f0 100644 --- a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts @@ -14,7 +14,10 @@ import type { } from '../../../shared/agent-session-journal-types' import { digestPayload } from './journal-payload-bounds' import { reconcileJournalSubmissionsAgainstHistory } from './journal-restart-reconciliation' -import type { ProviderHistoryItem, ProviderHistoryWindow } from './journal-submission-reconciler' +import type { + PlacedProviderHistoryWindow, + ProviderHistoryItem +} from './journal-submission-reconciler' import { createTrackedJournalOpener } from './journal-host-database-test-support' import { agentSessionFailureFact } from '../../../shared/agent-session-failure' import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' @@ -65,11 +68,18 @@ function history(uuid: string, text: string): ProviderHistoryItem { } } +/** Read from a resume point the sends' own owner, at fence 1, set before handing them over. */ function window( items: ProviderHistoryItem[], - overrides: Partial = {} -): ProviderHistoryWindow { - return { items, boundaryConsistent: true, turnInFlight: false, ...overrides } + overrides: Partial = {} +): PlacedProviderHistoryWindow { + return { + items, + boundaryConsistent: true, + turnInFlight: false, + start: { fence: 1, movedAt: 0 }, + ...overrides + } } /** A host that wrote the submission row and died before learning its outcome. */ @@ -103,13 +113,12 @@ describe('reconcileJournalSubmissionsAgainstHistory', () => { it('settles a message found in provider history as accepted on its provider identity', async () => { const journal = await reopenAfterCrash() - const settled = await reconcileJournalSubmissionsAgainstHistory({ + await reconcileJournalSubmissionsAgainstHistory({ journal, fence: 2, history: window([history('uuid-1', 'deploy the thing')]) }) - expect(settled).toEqual(['cm_1']) const submission = journal.submissions()[0] expect(submission?.dispatchState).toBe('accepted') expect(submission?.providerItemId).toBe(agentJournalItemKey(claudeIdentity('uuid-1'))) @@ -118,13 +127,12 @@ describe('reconcileJournalSubmissionsAgainstHistory', () => { it('settles a message provably absent from history as rejected: not_delivered', async () => { const journal = await reopenAfterCrash() - const settled = await reconcileJournalSubmissionsAgainstHistory({ + await reconcileJournalSubmissionsAgainstHistory({ journal, fence: 2, history: window([]) }) - expect(settled).toEqual(['cm_1']) const submission = journal.submissions()[0] expect(submission?.dispatchState).toBe('rejected') // A sentence, since released clients print the reason as it is, and the fact beside it. @@ -144,26 +152,24 @@ describe('reconcileJournalSubmissionsAgainstHistory', () => { it('leaves a submission unknown while the provider reports a turn in flight', async () => { const journal = await reopenAfterCrash() - const settled = await reconcileJournalSubmissionsAgainstHistory({ + await reconcileJournalSubmissionsAgainstHistory({ journal, fence: 2, history: window([], { turnInFlight: true }) }) - expect(settled).toEqual([]) expect(journal.submissions()[0]?.dispatchState).toBe('unknown') }) it('leaves a submission unknown when the history boundary is inconsistent', async () => { const journal = await reopenAfterCrash() - const settled = await reconcileJournalSubmissionsAgainstHistory({ + await reconcileJournalSubmissionsAgainstHistory({ journal, fence: 2, history: window([], { boundaryConsistent: false }) }) - expect(settled).toEqual([]) expect(journal.submissions()[0]?.dispatchState).toBe('unknown') }) @@ -180,13 +186,12 @@ describe('reconcileJournalSubmissionsAgainstHistory', () => { 'look at this' ) - const settled = await reconcileJournalSubmissionsAgainstHistory({ + await reconcileJournalSubmissionsAgainstHistory({ journal, fence: 2, history: window([]) }) - expect(settled).toEqual([]) expect(journal.submissions()[0]?.dispatchState).toBe('unknown') }) @@ -203,13 +208,12 @@ describe('reconcileJournalSubmissionsAgainstHistory', () => { 'first\nsecond' ) - const settled = await reconcileJournalSubmissionsAgainstHistory({ + await reconcileJournalSubmissionsAgainstHistory({ journal, fence: 2, history: window([history('uuid-1', 'first\nsecond')]) }) - expect(settled).toEqual([]) expect(journal.submissions()[0]?.dispatchState).toBe('unknown') }) diff --git a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts index cc59c95893cd..51ef108c82f2 100644 --- a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts +++ b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts @@ -23,7 +23,10 @@ import { } from '../../../shared/agent-session-journal-item-key' import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission' import type { AgentSessionJournal } from './journal-store' -import { reconcileSubmissions, type ProviderHistoryWindow } from './journal-submission-reconciler' +import { + reconcileSubmissions, + type PlacedProviderHistoryWindow +} from './journal-submission-reconciler' /** * Only a text-only body can be compared against provider content. A submission @@ -68,8 +71,8 @@ function comparableSubmissions(journal: AgentSessionJournal): AgentJournalSubmis * claimable would let an undelivered message match an older identical one. */ function unseenHistory( journal: AgentSessionJournal, - history: ProviderHistoryWindow -): ProviderHistoryWindow { + history: PlacedProviderHistoryWindow +): PlacedProviderHistoryWindow { const snapshot = journal.snapshot() const committed = new Set(snapshot.items.map((item) => item.itemId)) // Accepted submissions alias their provider item to the optimistic `orca:*` @@ -90,20 +93,17 @@ function unseenHistory( } } -/** - * Decide what the crash boundary could only doubt. Returns the client message - * ids this pass settled, so the attach result stops reporting them unconfirmed. - */ +/** Decide what the crash boundary could only doubt. Settles one send at a time, so a failed write + * leaves it and every send after it as the crash boundary wrote them. */ export async function reconcileJournalSubmissionsAgainstHistory(input: { journal: AgentSessionJournal fence: number - history: ProviderHistoryWindow -}): Promise { + history: PlacedProviderHistoryWindow +}): Promise { const submissions = comparableSubmissions(input.journal) if (submissions.length === 0) { - return [] + return } - const settled: string[] = [] for (const outcome of reconcileSubmissions({ submissions, history: unseenHistory(input.journal, input.history) @@ -136,7 +136,5 @@ export async function reconcileJournalSubmissionsAgainstHistory(input: { recovered: true } ) - settled.push(outcome.clientMessageId) } - return settled } diff --git a/src/main/native-chat/agent-session-journal/journal-submission-fold.ts b/src/main/native-chat/agent-session-journal/journal-submission-fold.ts index 7b03b4f457f8..2d128aabf401 100644 --- a/src/main/native-chat/agent-session-journal/journal-submission-fold.ts +++ b/src/main/native-chat/agent-session-journal/journal-submission-fold.ts @@ -26,7 +26,9 @@ export function applyJournalSubmission( submittedAt: row.ts, resolvedAt: null, submittedSequence: row.seq, - ...(row.handoverRecorded ? { handoverRecorded: true, acceptedSequence: row.seq } : {}), + ...(row.handoverRecorded + ? { handoverRecorded: true, acceptedSequence: row.seq } + : { firstHandover: { fence: row.fence, at: row.ts } }), // A malformed stored link is dropped, never the row. ...(typeof row.queuedMessageId === 'string' && row.queuedMessageId.length > 0 ? { queuedMessageId: row.queuedMessageId } diff --git a/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts b/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts index 91522f76efef..3bbfb2c38be2 100644 --- a/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts +++ b/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts @@ -36,10 +36,10 @@ export type ProviderHistoryItem = { } export type ProviderHistoryWindow = { - /** Provider items observed at or after the journal's last committed item. */ + /** Provider items the read found: with a consistent boundary, all of them from the resume point on. */ items: readonly ProviderHistoryItem[] /** - * The history read actually started at the journal's last committed item. A + * The history read actually started at the session's recorded resume point. A * fork, a compacted provider log, or a truncated read makes absence * meaningless, so a missing submission cannot be called "not delivered". */ @@ -48,6 +48,16 @@ export type ProviderHistoryWindow = { turnInFlight: boolean } +/** When the resume point the read started at last moved: the fence its link was written at, and a + * host-clock stamp (the journal's clock) taken no earlier than the move itself. */ +export type ProviderHistoryWindowStart = { fence: number; movedAt: number } + +/** A window placed against the sends it is asked about. */ +export type PlacedProviderHistoryWindow = ProviderHistoryWindow & { + /** null when no recorded resume point bounds the read. */ + start: ProviderHistoryWindowStart | null +} + export type SubmissionReconciliation = | { clientMessageId: string @@ -64,6 +74,7 @@ export type SubmissionUnknownReason = | 'history_boundary_inconsistent' | 'turn_in_flight' | 'ambiguous_match' + | 'history_start_unproven' /** * Resolve every unsettled submission against provider history. @@ -74,7 +85,7 @@ export type SubmissionUnknownReason = */ export function reconcileSubmissions(input: { submissions: readonly AgentJournalSubmission[] - history: ProviderHistoryWindow + history: PlacedProviderHistoryWindow }): SubmissionReconciliation[] { const unsettled = input.submissions.filter( (submission) => submission.dispatchState === 'pending' || submission.dispatchState === 'unknown' @@ -164,7 +175,7 @@ function resolveOne( submission: AgentJournalSubmission, matched: Map, ambiguous: Set, - history: ProviderHistoryWindow + history: PlacedProviderHistoryWindow ): SubmissionReconciliation { const item = matched.get(submission.clientMessageId) if (item) { @@ -196,6 +207,13 @@ function resolveOne( reason: 'turn_in_flight' } } + if (!windowStartsBeforeHandover(submission, history)) { + return { + clientMessageId: submission.clientMessageId, + outcome: 'unknown', + reason: 'history_start_unproven' + } + } // Absent from a history we can trust the boundary of, with nothing running: // the provider never took it. return { @@ -204,3 +222,20 @@ function resolveOne( reason: DISPATCH_REJECTED_NOT_DELIVERED } } + +/** Absence proves non-delivery only past where the read starts. Moves at an earlier fence precede + * any handover at a later one; at the same fence, the host clock orders them, ties unproven. */ +function windowStartsBeforeHandover( + submission: AgentJournalSubmission, + history: PlacedProviderHistoryWindow +): boolean { + const { start } = history + const handover = submission.firstHandover + if (!start || !handover) { + return false + } + if (handover.fence !== start.fence) { + return handover.fence > start.fence + } + return start.movedAt < handover.at +} diff --git a/src/main/native-chat/agent-session-journal/journal-submission-writer.ts b/src/main/native-chat/agent-session-journal/journal-submission-writer.ts index 1fc0ac21d0da..f2ddedb30380 100644 --- a/src/main/native-chat/agent-session-journal/journal-submission-writer.ts +++ b/src/main/native-chat/agent-session-journal/journal-submission-writer.ts @@ -53,7 +53,7 @@ export class JournalSubmissionWriter { } /** - * Record a dispatch transition, including a proven retry returning to pending. + * Record a dispatch transition: the handover of a send not yet handed over, or a settlement. * * Accepting REQUIRES the provider identity rather than a free-form id: the * adopted key is what the provider's echo will upsert into, so a mismatched diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-acquire-isolation.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-acquire-isolation.test.ts new file mode 100644 index 000000000000..f04c3f49b135 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-acquire-isolation.test.ts @@ -0,0 +1,46 @@ +import { expect, it, vi } from 'vitest' +import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import { HOST_TEST_NOW, HOST_TEST_THREAD } from './structured-agent-session-host-test-data' +import { + CALLER, + attach, + attachParams, + hostTestState, + replaceHostTestState +} from './structured-agent-session-host-test-harness' + +it("creates a chat while another saved chat's owner check fails", async () => { + await attach() + const prior = hostTestState() + prior.acquire.mockImplementation(async ({ fence, spawnToken }) => ({ + process: { hostId: 'local', pid: 4242, processStartTimeMs: HOST_TEST_NOW, spawnToken }, + link: { + linkId: 'new-link', + handle: codexProviderHandle(HOST_TEST_THREAD), + origin: 'created', + mintedAtFence: fence, + observedAt: HOST_TEST_NOW + } + })) + const peerId = attachParams().envelope.sessionId + const store = await openTestAgentSessionRecordStore(prior.root) + const probeOwners = vi.fn(async () => { + throw new Error('owner check failed') + }) + const host = new StructuredAgentSessionHost({ ...prior.host.deps, store, probeOwners }) + replaceHostTestState({ store, host }) + const request = attachParams() + const params = attachParams({ + envelope: { ...request.envelope, sessionId: 'new-chat' }, + accountHome: { variable: 'CODEX_HOME', path: prior.root } + }) + + const result = await host.attach(CALLER, params) + + expect(result.ok, JSON.stringify(result)).toBe(true) + expect(probeOwners).not.toHaveBeenCalled() + expect(store.getRecord(peerId)?.lease.unreconciled).toBe(true) + expect(store.getRecord('new-chat')?.lease.claimStatus).toBe('live') +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts index 495b7c629989..a9ab36a7dc8b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts @@ -144,7 +144,8 @@ describe('structured session acquisition options', () => { handle: codexProviderHandle('legacy-thread'), origin, mintedAtFence: input.fence, - observedAt: NOW + // Before the crashed send: journal rows are stamped on the real clock, not NOW. + observedAt: 0 } } }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index d753547d57ee..9fef9531eaad 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -422,8 +422,8 @@ export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop & | undefined /** Option keys skipped after a provider rejected their persisted restore value. */ readOptionRestoreFailures?(sessionId: string): readonly string[] - /** Provider history for restart reconciliation, bounded to what the provider - * recorded after the journal's last committed item. Only the adapter can say + /** Provider history for restart reconciliation, read from the resume point + * `identity.providerHandle` names. Only the adapter can say * whether the read has a proven start and whether a turn is still running, so * it owns both flags. Omit where the provider records no boundary-consistent * history; an omitted window leaves every unsettled submission `unknown`. */ 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 1563146f7e4a..bb7c8f2537f2 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 @@ -45,8 +45,8 @@ import { withAgentSessionCreatePhase, type AgentSessionCreatePhaseRecorder } from '../../observability/agent-session-instrumentation' -import type { ProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' -import { readProviderHistoryWindow } from './structured-agent-session-provider-history-window' +import type { PlacedProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' +import { sampleProviderHistoryWindow } from './structured-agent-session-history-sample' import type { StructuredAgentSessionStartupAttempt } from './structured-agent-session-startup-attempt-contract' import type { StructuredAgentSessionStartupProgress } from './structured-agent-session-startup-attempt' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' @@ -133,7 +133,7 @@ export async function performAttach( let reservedRecord: AgentSessionRecord | null = null let unsupportedReservationSettlementAttempted = false let replayed = false - let providerHistoryWindow: ProviderHistoryWindow | null = null + let providerHistoryWindow: PlacedProviderHistoryWindow | null = null const preparedTranscript = store.getRecord(sessionId) ? { ok: true as const, items: null } : await prepareAdoptedTranscript(params) @@ -187,10 +187,10 @@ export async function performAttach( // Sample provider history before a new child is acquired. Once acquireOwner // starts the child, the adapter's liveness signal intentionally becomes // conservative and an absent prompt can no longer prove non-delivery. - providerHistoryWindow = await readProviderHistoryWindow({ + providerHistoryWindow = await sampleProviderHistoryWindow({ adapter: input.adapter, identity: journalIdentityFor(record, params), - accountHome: record.accountHome, + record, ownerAlreadyAdmitted: agentSessionLeaseAdmitsWriter(record.lease) }) if (!agentSessionLeaseAdmitsWriter(record.lease)) { @@ -255,9 +255,8 @@ export async function performAttach( await input.beforeJournalOpen?.() attached = await attachJournal({ record, - params, - adapter: input.adapter, openConversation: input.openConversation, + logger: input.logger, providerHistoryWindow }) await importAdoptedTranscript(params, attached, record, preparedTranscript.items) 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 cd419133389c..045979341f16 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 @@ -145,6 +145,7 @@ async function runAttachUnderAbort( await withAgentSessionCreatePhase('resolve_recovery', recordPhase, () => context.runtimeState.resolveRecovery(sessionId) ) + const probedOwner = context.deps.store.getRecord(sessionId)?.lease.ownerProcess ?? null const probe = await withAgentSessionCreatePhase('probe_owner', recordPhase, () => context.runtimeState.probeOwner(sessionId) ) @@ -190,6 +191,7 @@ async function runAttachUnderAbort( claimKeyId: context.deps.claimKeyId, handoffOperationId: params.envelope.clientOperationId, probe, + probedOwner, ...(await pinnedAgentSessionLaunchArgs(context.deps.resolveLaunchArgs, params)), ...(await pinnedAgentSessionLaunchEnv(context.deps.resolveLaunchEnv, params)) }, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-reconciliation.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-reconciliation.test.ts index e22c5dea9fdc..0a5d2e235d28 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-reconciliation.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-reconciliation.test.ts @@ -1,6 +1,6 @@ // Attach is where the restart reconciler runs. These cover the wiring itself: -// that the window the adapter reports reaches the journal, that what it settles -// stops being reported unconfirmed, and that deciding never sends. +// that the sampled window reaches the journal, and that what it settles stops +// being reported unconfirmed. import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -10,18 +10,21 @@ import { agentSessionRecordFixture } from '../../../shared/agent-session-record. import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' import { projectStructuredAgentSessionStatusState } from '../../../shared/structured-agent-session-projection' import { digestPayload } from '../agent-session-journal/journal-payload-bounds' -import type { ProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' +import { claudeProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import type { PlacedProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' import { createTrackedJournalOpener, openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' -import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' import { openTestAttachConversation } from './structured-agent-session-attach-test-conversation' +import { recordingStructuredAgentSessionLogger } from './structured-agent-session-logger-test-support' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { attachJournal, journalIdentityFor, type AgentSessionAttachParams } from './structured-agent-session-attach' +import { sampleProviderHistoryWindow } from './structured-agent-session-history-sample' const RECORD = agentSessionRecordFixture() @@ -40,6 +43,7 @@ const PARAMS = { } as unknown as AgentSessionAttachParams const IDENTITY = journalIdentityFor(RECORD, PARAMS) +const SESSION_ID = 'provider-session-alpha-1' let root: string const journals = createTrackedJournalOpener() @@ -48,22 +52,15 @@ function userMessage(text: string): AgentJournalMessageItem { return { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] } } -function window(overrides: Partial = {}): ProviderHistoryWindow { - return { items: [], boundaryConsistent: true, turnInFlight: false, ...overrides } -} - -/** Only the surface `attachJournal` touches; every send-shaped method is a spy - * so a re-delivery would be visible rather than silent. */ -function adapterWith(providerHistoryWindow?: () => Promise): { - adapter: StructuredAgentSessionAdapter - dispatch: ReturnType -} { - const dispatch = vi.fn() - const adapter = { - dispatch, - ...(providerHistoryWindow ? { providerHistoryWindow } : {}) - } as unknown as StructuredAgentSessionAdapter - return { adapter, dispatch } +/** Read from the resume point the record's own owner set before the crashed send went out. */ +function window(overrides: Partial = {}): PlacedProviderHistoryWindow { + return { + items: [], + boundaryConsistent: true, + turnInFlight: false, + start: { fence: RECORD.lease.runtimeFence, movedAt: 0 }, + ...overrides + } } /** A previous process wrote the submission row and died before its outcome. */ @@ -81,12 +78,12 @@ async function crashedJournal(clientMessageId = 'cm_1', text = 'deploy the thing await journal.close() } -async function attach(adapter: StructuredAgentSessionAdapter) { +async function attach(providerHistoryWindow: PlacedProviderHistoryWindow | null) { const attached = await attachJournal({ record: RECORD, - params: PARAMS, openConversation: openTestAttachConversation(openTestJournalHostDatabase(root)), - adapter + logger: recordingStructuredAgentSessionLogger().logger, + providerHistoryWindow }) journals.track(attached.journal) return attached @@ -102,25 +99,67 @@ afterEach(async () => { }) describe('attachJournal restart reconciliation', () => { + it('keeps committed settlements after a later write fails and re-derives the remainder', async () => { + await crashedJournal('cm_1', 'first') + await crashedJournal('cm_2', 'second') + const journal = journals.track( + await openTestAttachConversation(openTestJournalHostDatabase(root))(RECORD) + ) + const providerHistoryWindow = window({ + items: ['cm_1', 'cm_2'].map((clientMessageId, index) => ({ + clientMessageId, + providerItemId: `item-${index}`, + payloadFingerprint: null, + identity: { + provider: 'claude', + sessionId: 'provider-session-alpha-1', + uuid: `item-${index}` + } + })) + }) + const logging = recordingStructuredAgentSessionLogger() + const resolve = journal.resolveDispatch + const writes = vi.spyOn(journal, 'resolveDispatch') + writes.mockImplementation(function (this: AgentSessionJournal, input, hook) { + if (input.clientMessageId === 'cm_2' && input.state === 'accepted') { + return Promise.reject(new Error('disk full on second settlement')) + } + return resolve.call(this, input, hook) + }) + const input = { + record: RECORD, + logger: logging.logger, + openConversation: async () => journal, + providerHistoryWindow + } + + expect((await attachJournal(input)).unconfirmedClientMessageIds).toEqual(['cm_2']) + expect(journal.submissions().map((send) => send.dispatchState)).toEqual(['accepted', 'unknown']) + expect(logging.scopes()).toEqual(['attach-send-reconcile']) + writes.mockRestore() + + expect((await attachJournal(input)).unconfirmedClientMessageIds).toEqual([]) + expect(journal.submissions().map((send) => send.dispatchState)).toEqual([ + 'accepted', + 'accepted' + ]) + }) + it('settles a provably undelivered submission and stops reporting it unconfirmed', async () => { await crashedJournal() - const { adapter, dispatch } = adapterWith(async () => window()) - const attached = await attach(adapter) + const attached = await attach(window()) expect(attached.unconfirmedClientMessageIds).toEqual([]) const submission = attached.journal.submissions()[0] expect(submission?.dispatchState).toBe('rejected') expect(submission?.rejection).toEqual({ kind: 'notDelivered' }) - // Deciding is not sending: nothing here puts the message back on the wire. - expect(dispatch).not.toHaveBeenCalled() }) it('gives a send the provider never received no verdict and no listing', async () => { await crashedJournal() - const { adapter } = adapterWith(async () => window()) - const attached = await attach(adapter) + const attached = await attach(window()) // Nobody failed: the crash stranded it, so the chat must not read Failed or be listed by it. const { items, submissions } = attached.journal.snapshot() @@ -131,35 +170,84 @@ describe('attachJournal restart reconciliation', () => { it('still reports a submission unconfirmed when the window cannot decide it', async () => { await crashedJournal() - const { adapter, dispatch } = adapterWith(async () => window({ turnInFlight: true })) - const attached = await attach(adapter) + const attached = await attach(window({ turnInFlight: true })) expect(attached.unconfirmedClientMessageIds).toEqual(['cm_1']) expect(attached.journal.submissions()[0]?.dispatchState).toBe('unknown') - expect(dispatch).not.toHaveBeenCalled() }) - it('leaves the crash boundary untouched for an adapter that reports no history', async () => { + it('leaves the crash boundary untouched when there is no history to use', async () => { await crashedJournal() - const { adapter } = adapterWith() - const attached = await attach(adapter) + const attached = await attach(null) expect(attached.unconfirmedClientMessageIds).toEqual(['cm_1']) expect(attached.journal.submissions()[0]?.dispatchState).toBe('unknown') }) - it('does not fail the attach when reading provider history throws', async () => { + // The send went out under fence 7 with the resume point at leaf-a; history read from the head + // holds it either way. What decides is whether that head was set before the send. + const AFTER_THE_SEND = Number.MAX_SAFE_INTEGER + it.each([ + [ + 'still rejects it after a later owner re-proved the same point', + 1, + 'leaf-a', + 2_000, + 'rejected' + ], + ['leaves it unknown once a later owner moved the point', 1, 'leaf-b', 2_000, 'unknown'], + [ + 'rejects it when its own owner moved the point before the send', + 0, + 'leaf-b', + 2_000, + 'rejected' + ], + [ + 'leaves it unknown when its own owner moved the point after the send', + 0, + 'leaf-b', + AFTER_THE_SEND, + 'unknown' + ] + ])('%s', async (_, laterFence, leaf, movedAt, expected) => { await crashedJournal() - const { adapter } = adapterWith(async () => { - throw new Error('transcript unreadable') + const fence = RECORD.lease.runtimeFence + const first = { + ...RECORD.providerHandleChain[0]!, + handle: claudeProviderHandle(SESSION_ID, 'leaf-a') + } + const head = { + ...first, + linkId: `link-at-${fence + laterFence}`, + origin: laterFence ? ('resumed' as const) : first.origin, + mintedAtFence: fence + laterFence, + observedAt: movedAt, + handle: claudeProviderHandle(SESSION_ID, leaf) + } + const record = { + ...RECORD, + lease: { ...RECORD.lease, runtimeFence: fence + 2 }, + providerHandleChain: laterFence ? [first, head] : [head] + } + const providerHistoryWindow = await sampleProviderHistoryWindow({ + adapter: { providerHistoryWindow: async () => window() }, + identity: journalIdentityFor(record, PARAMS), + record, + ownerAlreadyAdmitted: false }) - const attached = await attach(adapter) + const attached = await attachJournal({ + record, + openConversation: openTestAttachConversation(openTestJournalHostDatabase(root)), + logger: recordingStructuredAgentSessionLogger().logger, + providerHistoryWindow + }) + journals.track(attached.journal) - expect(attached.unconfirmedClientMessageIds).toEqual(['cm_1']) - expect(attached.journal.submissions()[0]?.dispatchState).toBe('unknown') + expect(attached.journal.submissions()[0]?.dispatchState).toBe(expected) }) it('leaves a message the open conversation still has queued alone (W4′e)', async () => { @@ -175,18 +263,15 @@ describe('attachJournal restart reconciliation', () => { handoverRecorded: true }) // History that holds nothing: absence would prove a handed-over message undelivered. - const { adapter, dispatch } = adapterWith(async () => window()) - const attached = await attachJournal({ record: RECORD, - params: PARAMS, - adapter, - openConversation: async () => journal + logger: recordingStructuredAgentSessionLogger().logger, + openConversation: async () => journal, + providerHistoryWindow: window() }) expect(attached.journal).toBe(journal) expect(journal.submissions()[0]).toMatchObject({ dispatchState: 'pending' }) expect(journal.submissions()[0]?.handedOverAt).toBeUndefined() - expect(dispatch).not.toHaveBeenCalled() }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-send-recovery.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-send-recovery.test.ts new file mode 100644 index 000000000000..c7b6a0f0b2c8 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-send-recovery.test.ts @@ -0,0 +1,299 @@ +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import Database from '../../sqlite/sync-database' +import { journalDatabasePath } from '../agent-session-journal/journal-host-database' +import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' +import { replayJournal } from '../agent-session-journal/journal-open' +import { parseJournalRow } from '../agent-session-journal/journal-row-schema' +import type { ProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' +import { claudeProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import { reviseAgentSessionProviderResumePoint } from '../../runtime/agent-session-provider-handle-transition' +import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import type { AgentSessionAttachParams } from './structured-agent-session-attach' +import { hostTestAttachParams } from './structured-agent-session-host-test-data' +import { recordingStructuredAgentSessionLogger } from './structured-agent-session-logger-test-support' +import { + createRestTestRig, + restTestSend, + REST_TEST_CALLER as CALLER, + REST_TEST_SESSION as SESSION, + REST_TEST_THREAD as THREAD, + type RestTestRig +} from './structured-agent-session-rest-test-rig' + +let rig: RestTestRig +let logging: ReturnType +let history: ProviderHistoryWindow +let ownerLive = false +let resumeParams: AgentSessionAttachParams +const retired: StructuredAgentSessionHost[] = [] +const settlementError = Object.assign(new Error('earlier send settlement failed'), { + code: 'SQLITE_BUSY' +}) + +beforeEach(async () => { + logging = recordingStructuredAgentSessionLogger() + history = { items: [], boundaryConsistent: true, turnInFlight: false } + ownerLive = false + rig = await createRestTestRig({ + logger: logging.logger, + probeOwner: async () => + ownerLive + ? { outcome: 'identity-matched', matchedOn: ['spawn-token'] } + : { outcome: 'pid-absent' }, + idleSweep: { intervalMs: 3_600_000 } + }) + const acquire = rig.adapter.acquire.getMockImplementation()! + // The rig's clock is not the journal's: a link set before any send must read as earlier. + rig.adapter.acquire.mockImplementation(async (input) => { + const acquired = await acquire(input) + return { ...acquired, link: { ...acquired.link, observedAt: 0 } } + }) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await Promise.all(retired.splice(0).map((host) => host.flushAllStreamedEvents())) + await rig.dispose() +}) + +async function crashRestart(): Promise { + ownerLive = false + retired.push(rig.host) + rig.store = await openTestAgentSessionRecordStore(rig.root) + rig.host = new StructuredAgentSessionHost({ + ...rig.host.deps, + store: rig.store, + adapter: { ...rig.host.deps.adapter, providerHistoryWindow: async () => history } + }) + await rig.host.reconcileRestartLeases() +} + +async function crashWithSends(count: number, overrides: Partial = {}) { + const attached = await rig.host.attach(CALLER, hostTestAttachParams(null, overrides)) + if (!attached.ok) { + throw new Error(attached.refusal.message) + } + rig.adapter.dispatch.mockResolvedValue({ state: 'admitted' }) + const sends: ReturnType[] = [] + for (let index = 0; index < count; index += 1) { + const send = restTestSend(`message ${index}`, attached.fence) + expect(await rig.host.send(CALLER, send)).toMatchObject({ ok: true }) + sends.push(send) + } + await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledTimes(count)) + await crashRestart() + return sends +} + +const CLAUDE_SESSION = 'claude-session-1' +const claudeHandle = claudeProviderHandle(CLAUDE_SESSION, 'leaf-before') + +function claudeHistory(clientMessageIds: string[]): ProviderHistoryWindow { + return { + boundaryConsistent: true, + turnInFlight: false, + items: clientMessageIds.map((clientMessageId) => ({ + clientMessageId, + providerItemId: `uuid-${clientMessageId}`, + payloadFingerprint: null, + identity: { provider: 'claude', sessionId: CLAUDE_SESSION, uuid: `uuid-${clientMessageId}` } + })) + } +} + +function acceptedHistory(clientMessageIds: string[]): ProviderHistoryWindow { + return { + boundaryConsistent: true, + turnInFlight: false, + items: clientMessageIds.map((clientMessageId, ordinal) => ({ + clientMessageId, + providerItemId: `item-${ordinal}`, + payloadFingerprint: null, + identity: { provider: 'codex', threadId: THREAD, turnId: 'old', ordinal } + })) + } +} + +/** The next recovered `state` verdict fails inside its own SQLite transaction, once: the INSERT + * runs and the real rollback undoes it. */ +function failNextSettlementWrite(state: 'accepted' | 'rejected'): void { + const connection = openTestJournalHostDatabase(rig.root).db + const prepare = connection.prepare.bind(connection) + let armed = true + // The database caches statements, so each is wrapped once. + const wrapped = new WeakSet() + vi.spyOn(connection, 'prepare').mockImplementation((sql: string) => { + const statement = prepare(sql) + if (!sql.startsWith('INSERT INTO journal_rows') || wrapped.has(statement)) { + return statement + } + wrapped.add(statement) + const run = statement.run.bind(statement) + vi.spyOn(statement, 'run').mockImplementation((...params) => { + const result = run(...params) + const json = params.at(-1) + const parsed = typeof json === 'string' ? parseJournalRow(json) : null + if ( + armed && + parsed?.ok && + parsed.row.kind === 'dispatch' && + parsed.row.recovered === true && + parsed.row.state === state + ) { + armed = false + throw settlementError + } + return result + }) + return statement + }) +} + +/** What a fresh process would read: the committed rows on disk, not this host's memory. */ +function dispatchStateOnDisk(clientMessageId: string): string | undefined { + const disk = new Database(journalDatabasePath(rig.root), { readonly: true }) + try { + return replayJournal(disk, SESSION)?.state.submissions.get(clientMessageId)?.dispatchState + } finally { + disk.close() + } +} + +async function resume(overrides: Partial = {}) { + resumeParams = hostTestAttachParams(rig.store.getRecord(SESSION)!.lease.runtimeFence, overrides) + const result = await rig.host.attach(CALLER, resumeParams) + ownerLive = result.ok + return result +} + +it.each(['accepted', 'rejected'] as const)( + 'resumes when persisting an earlier send as %s fails, without resending it', + async (state) => { + const [send] = await crashWithSends(1) + const clientMessageId = send!.envelope.clientOperationId + if (state === 'accepted') { + history = acceptedHistory([clientMessageId]) + } + failNextSettlementWrite(state) + + const attached = await resume() + + expect(attached, JSON.stringify(attached)).toMatchObject({ + ok: true, + value: { unconfirmedClientMessageIds: [clientMessageId] } + }) + expect(rig.host.collaboratorsForTests().sessions.get(SESSION)?.child?.phase).toBe('ready') + expect(rig.store.getRecord(SESSION)?.lease.claimStatus).toBe('live') + expect((await rig.host.journalSnapshot(SESSION)).submissions[0]).toMatchObject({ + dispatchState: 'unknown' + }) + expect(dispatchStateOnDisk(clientMessageId)).toBe('unknown') + expect(logging.entries).toContainEqual({ + level: 'warn', + message: 'settling earlier sends against provider history failed', + fields: { scope: 'attach-send-reconcile', sessionId: SESSION, error: settlementError } + }) + expect(await rig.host.send(CALLER, send!)).toMatchObject({ ok: true, replayed: true }) + expect(rig.adapter.dispatch).toHaveBeenCalledTimes(1) + + await crashRestart() + // The owner in between re-proved the resume point without moving it, so absence still counts. + expect(await resume()).toMatchObject({ ok: true, value: { unconfirmedClientMessageIds: [] } }) + expect((await rig.host.journalSnapshot(SESSION)).submissions[0]?.dispatchState).toBe(state) + expect(rig.adapter.dispatch).toHaveBeenCalledTimes(1) + } +) + +it('keeps a delivered send unconfirmed after a later turn moves the history past it', async () => { + const claude: Partial = { + provider: 'claude', + agent: 'claude', + accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/dev/.claude' }, + providerHandle: { kind: 'claude', sessionId: CLAUDE_SESSION, leafUuid: 'leaf-before' } + } + // Each acquisition re-proves the stored leaf, as a resumed Claude child does. + rig.adapter.acquire.mockImplementation(async ({ fence, spawnToken }) => ({ + acquisitionGeneration: `claude-${fence}`, + process: { hostId: 'local', pid: 4242, processStartTimeMs: 1_700_000_000_000, spawnToken }, + link: { + linkId: `claude-link-${fence}`, + handle: rig.store.getRecord(SESSION)?.providerHandleChain.at(-1)?.handle ?? claudeHandle, + origin: fence === 1 ? 'created' : 'resumed', + mintedAtFence: fence, + observedAt: 0 + } + })) + const [send] = await crashWithSends(1, claude) + const clientMessageId = send!.envelope.clientOperationId + history = claudeHistory([clientMessageId]) + failNextSettlementWrite('accepted') + expect(await resume(claude)).toMatchObject({ + ok: true, + value: { unconfirmedClientMessageIds: [clientMessageId] } + }) + + // The user keeps chatting; that turn ends and moves the resume point past both messages. + const fence = rig.store.getRecord(SESSION)!.lease.runtimeFence + rig.adapter.dispatch.mockResolvedValueOnce({ + state: 'accepted', + providerIdentity: { provider: 'claude', sessionId: CLAUDE_SESSION, uuid: 'next-uuid' } + }) + const next = restTestSend('after the restart', fence) + expect(await rig.host.send(CALLER, next)).toMatchObject({ ok: true }) + await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledTimes(2)) + await rig.store.transitionHandoff(SESSION, (record) => + reviseAgentSessionProviderResumePoint({ + record, + fence, + handle: claudeProviderHandle(CLAUDE_SESSION, 'leaf-after-next'), + now: Date.now() + }) + ) + history = { items: [], boundaryConsistent: true, turnInFlight: false } + + await crashRestart() + + // Absent only because the history now starts after it: calling it undelivered would offer a + // Retry that sends it twice. + expect(await resume(claude)).toMatchObject({ + ok: true, + value: { unconfirmedClientMessageIds: [clientMessageId] } + }) + const submissions = (await rig.host.journalSnapshot(SESSION)).submissions + expect(submissions.map((entry) => [entry.clientMessageId, entry.dispatchState])).toEqual([ + [clientMessageId, 'unknown'], + [next.envelope.clientOperationId, 'accepted'] + ]) + expect(rig.adapter.dispatch).toHaveBeenCalledTimes(2) +}) + +it('keeps the running agent when saving an earlier send fails on re-attach', async () => { + const [send] = await crashWithSends(1) + const clientMessageId = send!.envelope.clientOperationId + // The first resume cannot decide it, so the send outlives that attach unconfirmed. + history = { items: [], boundaryConsistent: false, turnInFlight: false } + expect(await resume()).toMatchObject({ + ok: true, + value: { unconfirmedClientMessageIds: [clientMessageId] } + }) + const fence = rig.store.getRecord(SESSION)!.lease.runtimeFence + history = acceptedHistory([clientMessageId]) + failNextSettlementWrite('accepted') + + expect(await rig.host.attach(CALLER, resumeParams)).toMatchObject({ + ok: true, + value: { fence, unconfirmedClientMessageIds: [clientMessageId] } + }) + + expect(logging.scopes()).toContain('attach-send-reconcile') + expect(rig.adapter.acquire).toHaveBeenCalledTimes(2) + expect(rig.host.deps.adapter.releaseAcquisition).not.toHaveBeenCalled() + expect(rig.adapter.closeSession).not.toHaveBeenCalled() + expect(rig.host.collaboratorsForTests().sessions.get(SESSION)?.child?.phase).toBe('ready') + expect(rig.store.getRecord(SESSION)?.lease).toMatchObject({ + claimStatus: 'live', + runtimeFence: fence + }) + expect(dispatchStateOnDisk(clientMessageId)).toBe('unknown') +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts index 9138fa20f8b6..5e8cabd9d620 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts @@ -43,9 +43,9 @@ import { import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { reconcileJournalSubmissionsAgainstHistory } from '../agent-session-journal/journal-restart-reconciliation' -import type { ProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' -import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import type { PlacedProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' import { structuredAgentSessionRefusalMessage } from './structured-agent-session-refusal-message' +import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' /** * Everything a client may declare about the session it wants. Deliberately no @@ -92,6 +92,7 @@ export type AgentSessionAttachAuthority = { claimKeyId: string handoffOperationId: string | null probe: AgentSessionOwnerProbe + probedOwner?: AgentSessionRecord['lease']['ownerProcess'] launchArgs?: AgentSessionLaunchArgs launchEnv?: AgentSessionLaunchEnv } @@ -181,7 +182,8 @@ export function journalIdentityFor( export type AttachedJournal = { journal: AgentSessionJournal - /** Submissions the crash boundary left `unknown` that provider history could not decide. */ + /** Sends the crash boundary left `unknown` that this attach did not settle: history could not + * decide them, or saving the verdict failed. */ unconfirmedClientMessageIds: string[] } @@ -189,10 +191,8 @@ export type AttachedJournal = { * The conversation's journal — opened, and its crash boundary settled, by the conversation's own * open — with provider history deciding the submissions that boundary could only doubt. * - * Why the reconciliation belongs HERE and nowhere else: this runs after the - * record store handed this host the lease and before `onAttached` starts a - * provider child, so nothing can be appending to the provider's history while it - * is read, and the window stays valid until the resume consumes it. Every other + * The attach samples history before acquiring a new child, then applies that window before + * draining the child's buffered events. Every other * settlement site — a proven child exit — runs while the host * may still start another child, and a read there could be overtaken before it * is acted on. Orca still never re-sends: this decides state only. A queued @@ -200,77 +200,38 @@ export type AttachedJournal = { */ export async function attachJournal(input: { record: AgentSessionRecord - params: AgentSessionAttachParams - adapter: StructuredAgentSessionAdapter + logger: StructuredAgentSessionLogger /** The host's open conversation, whose journal the attach adopts. */ openConversation: (record: AgentSessionRecord) => Promise - /** Provider history sampled before a new child is acquired. `null` means the - * adapter had no usable history; omit to read lazily for direct callers. */ - providerHistoryWindow?: ProviderHistoryWindow | null + /** Provider history sampled before a new child is acquired; null when there is none to use. */ + providerHistoryWindow: PlacedProviderHistoryWindow | null }): Promise { - const identity = journalIdentityFor(input.record, input.params) - const fence = input.record.lease.runtimeFence const journal = await input.openConversation(input.record) - const settled = await reconcileAgainstProviderHistory({ - adapter: input.adapter, - identity, - journal, - fence, - accountHome: input.record.accountHome, - ...(Object.hasOwn(input, 'providerHistoryWindow') - ? { history: input.providerHistoryWindow } - : {}) - }) + if (input.providerHistoryWindow) { + try { + await reconcileJournalSubmissionsAgainstHistory({ + journal, + fence: input.record.lease.runtimeFence, + history: input.providerHistoryWindow + }) + } catch (error) { + // Best effort: a send whose verdict did not save stays `unknown`, for a later attach's window. + input.logger.warn('settling earlier sends against provider history failed', { + scope: 'attach-send-reconcile', + sessionId: input.record.sessionId, + error + }) + } + } return { journal, unconfirmedClientMessageIds: journal .submissions() - .filter( - (entry) => - entry.dispatchState === 'unknown' && - entry.recovered === true && - !settled.includes(entry.clientMessageId) - ) + .filter((entry) => entry.dispatchState === 'unknown' && entry.recovered === true) .map((entry) => entry.clientMessageId) } } -/** Reading provider history is best effort: a provider that reports none, or a - * read that fails, leaves every submission exactly as the crash boundary wrote - * it. The journal writes the outcome implies are NOT caught here — a failed - * write must reach the caller that retains the journal handle. */ -async function reconcileAgainstProviderHistory(input: { - adapter: StructuredAgentSessionAdapter - identity: AgentSessionJournalIdentity - journal: AgentSessionJournal - fence: number - accountHome: AgentSessionAccountHome - history?: ProviderHistoryWindow | null -}): Promise { - let history = input.history - if (history === undefined) { - if (!input.adapter.providerHistoryWindow) { - return [] - } - try { - history = await input.adapter.providerHistoryWindow({ - identity: input.identity, - accountHome: input.accountHome - }) - } catch { - return [] - } - } - if (!history) { - return [] - } - return reconcileJournalSubmissionsAgainstHistory({ - journal: input.journal, - fence: input.fence, - history - }) -} - /** * The first link of an adopting session's chain. * @@ -336,6 +297,7 @@ export function reserveRequestFor(input: { claimKeyId: authority.claimKeyId, handoffOperationId: authority.handoffOperationId, probe: authority.probe, + ...(authority.probedOwner === undefined ? {} : { probedOwner: authority.probedOwner }), operation: { callerKey: input.callerKey, operationId: params.envelope.clientOperationId, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-history-sample.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-history-sample.test.ts new file mode 100644 index 000000000000..6c9df86b3c52 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-history-sample.test.ts @@ -0,0 +1,90 @@ +import { describe, expect, it } from 'vitest' +import { agentSessionRecordFixture } from '../../../shared/agent-session-record.test-fixture' +import type { AgentSessionProviderHandleLink } from '../../../shared/agent-session-provider-handle' +import { claudeProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import type { ProviderHistoryWindow } from '../agent-session-journal/journal-submission-reconciler' +import { + resumePointStart, + sampleProviderHistoryWindow +} from './structured-agent-session-history-sample' + +const RECORD = agentSessionRecordFixture() +const IDENTITY = { + sessionId: RECORD.sessionId, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'claude', + providerHandle: RECORD.providerHandleChain[0]!.handle +} +const WINDOW: ProviderHistoryWindow = { items: [], boundaryConsistent: true, turnInFlight: false } + +function link(fence: number, leaf: string, observedAt: number): AgentSessionProviderHandleLink { + return { + linkId: `link-${fence}`, + origin: 'resumed', + mintedAtFence: fence, + observedAt, + handle: claudeProviderHandle('provider-session-alpha-1', leaf) + } +} + +describe('resumePointStart', () => { + it('dates the start from the link that set the head leaf, not from links that re-proved it', () => { + expect(resumePointStart([link(3, 'a', 30), link(4, 'b', 40), link(5, 'b', 50)])).toEqual({ + fence: 4, + movedAt: 40 + }) + }) + + it('dates it from the head when the head moved the leaf', () => { + expect(resumePointStart([link(3, 'a', 30), link(4, 'b', 40), link(5, 'c', 50)])).toEqual({ + fence: 5, + movedAt: 50 + }) + }) + + it('has no start without a chain', () => { + expect(resumePointStart([])).toBeNull() + }) +}) + +describe('sampleProviderHistoryWindow', () => { + const sample = ( + read: () => Promise, + ownerAlreadyAdmitted = false + ) => + sampleProviderHistoryWindow({ + adapter: { providerHistoryWindow: read }, + identity: IDENTITY, + record: { ...RECORD, providerHandleChain: [link(7, 'a', 70), link(8, 'a', 80)] }, + ownerAlreadyAdmitted + }) + + it('places the window at where its resume point was set', async () => { + await expect(sample(async () => WINDOW)).resolves.toEqual({ + ...WINDOW, + start: { fence: 7, movedAt: 70 } + }) + }) + + it('reads a live owner as a turn in flight', async () => { + await expect(sample(async () => WINDOW, true)).resolves.toMatchObject({ turnInFlight: true }) + }) + + it('has no window when the read fails or finds nothing', async () => { + await expect( + sample(async () => { + throw new Error('transcript unreadable') + }) + ).resolves.toBeNull() + await expect(sample(async () => null)).resolves.toBeNull() + await expect( + sampleProviderHistoryWindow({ + adapter: {}, + identity: IDENTITY, + record: RECORD, + ownerAlreadyAdmitted: false + }) + ).resolves.toBeNull() + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-history-sample.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-history-sample.ts new file mode 100644 index 000000000000..53f393de569e --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-history-sample.ts @@ -0,0 +1,61 @@ +// Provider history for an attach's restart reconciliation, placed at the resume point it was read +// from. Reading is best effort: no reader, a failed read, or no history leaves every send exactly +// as the crash boundary wrote it. + +import type { AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' +import { + agentSessionProviderHandlesEqual, + type AgentSessionProviderHandleChain +} from '../../../shared/agent-session-provider-handle' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { + PlacedProviderHistoryWindow, + ProviderHistoryWindow, + ProviderHistoryWindowStart +} from '../agent-session-journal/journal-submission-reconciler' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' + +export async function sampleProviderHistoryWindow(input: { + adapter: Pick + identity: AgentSessionJournalIdentity + /** The record `identity` came from, read before this attach acquires a child. */ + record: AgentSessionRecord + ownerAlreadyAdmitted: boolean +}): Promise { + const read = input.adapter.providerHistoryWindow + if (!read) { + return null + } + let history: ProviderHistoryWindow | null + try { + history = await read({ identity: input.identity, accountHome: input.record.accountHome }) + } catch { + return null + } + if (!history) { + return null + } + return { + ...history, + // A lease that was already live may belong to a provider child this process + // has not indexed yet. Preserve the safe unknown outcome in that case. + turnInFlight: history.turnInFlight || input.ownerAlreadyAdmitted, + start: resumePointStart(input.record.providerHandleChain) + } +} + +/** Where the head's resume point was set: each acquisition re-proves an unmoved point as a new link, + * so it is the first of the trailing links that carry the head's handle, leaf included. */ +export function resumePointStart( + chain: AgentSessionProviderHandleChain +): ProviderHistoryWindowStart | null { + let start = chain.at(-1) + for (let index = chain.length - 2; start && index >= 0; index -= 1) { + const link = chain[index]! + if (!agentSessionProviderHandlesEqual(link.handle, start.handle)) { + break + } + start = link + } + return start ? { fence: start.mintedAtFence, movedAt: start.observedAt } : null +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-harness.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-harness.ts index ab748f866b5b..9a56eb81ec40 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-harness.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-harness.ts @@ -146,7 +146,8 @@ beforeEach(async () => { handle: codexProviderHandle(THREAD), origin: store.getRecord(SESSION)?.providerHandleChain.length ? 'resumed' : 'created', mintedAtFence: fence, - observedAt: NOW + // Before any send: journal rows are stamped on the real clock, not NOW. + observedAt: 0 } })) releaseAcquisition = vi.fn(async () => true) 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 251ccc88e32b..847784b4dc1f 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 @@ -130,6 +130,7 @@ export class StructuredAgentSessionHost { }) this.restore = reveal.createStructuredAgentSessionHostRestore(deps, { reconcileLeases: this.reconcileLeases, + reconcileRestartLeases: () => this.reconcileLeases(), resolveRecovery: (sessionId) => this.runtimeState.resolveRecovery(sessionId), serialize: (sessionId, task) => this.serialize(sessionId, task), hasSession: this.hasSession, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts index b6c8444a28bc..e717d5a63f4c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts @@ -39,6 +39,7 @@ describe('StructuredAgentSessionReadableRestorer', () => { journalDatabase: openTestJournalHostDatabase(stateDirectory), logger: recordingStructuredAgentSessionLogger().logger }, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-resolution.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-resolution.ts index 257072f7c23b..2834f8d10a91 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-resolution.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-resolution.ts @@ -82,6 +82,7 @@ export async function resolveStructuredSessionRecovery( sessionId, expectedFence: record.lease.runtimeFence, probe, + probedOwner: record.lease.ownerProcess, now: deps.now() })) return 'resolved' diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts index e61587a6598c..595fc8672f15 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts @@ -27,7 +27,7 @@ describe('createRestartReconciler', () => { expect(reconcileOnRestart).toHaveBeenCalledTimes(2) }) - it('passes every pending record through the batch owner probe', async () => { + it('passes only the addressed record through the batch owner probe', async () => { let records = [ { sessionId: 'session-1', lease: { unreconciled: true } }, { sessionId: 'session-2', lease: { unreconciled: true } } @@ -40,14 +40,18 @@ describe('createRestartReconciler', () => { }) const reconcileOnRestart = vi.fn( async (args: { + sessionId?: string probeMany?: ( pending: readonly AgentSessionRecord[] ) => Promise> }) => { - await args.probeMany?.(records) + await args.probeMany?.(records.filter((record) => record.sessionId === args.sessionId)) records = records.map((record) => ({ ...record, - lease: { ...record.lease, unreconciled: false } + lease: { + ...record.lease, + unreconciled: record.sessionId !== args.sessionId + } })) return new Map() } @@ -64,10 +68,8 @@ describe('createRestartReconciler', () => { ).resolves.toBeNull() expect(probeMany).toHaveBeenCalledOnce() - expect(probeMany.mock.calls[0]?.[0].map((record) => record.sessionId)).toEqual([ - 'session-1', - 'session-2' - ]) + expect(probeMany.mock.calls[0]?.[0].map((record) => record.sessionId)).toEqual(['session-1']) + expect(records[1]?.lease.unreconciled).toBe(true) expect(probe).not.toHaveBeenCalled() }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts index 29f0c9f1047d..15ba7cfe9f9f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts @@ -21,27 +21,27 @@ export function createRestartReconciler(deps: { records: readonly AgentSessionRecord[] ) => Promise> now: () => number -}): (sessionId: string) => Promise { - let pending: Promise | null = null +}): (sessionId?: string) => Promise { + const pending = new Map>() return async (sessionId) => { - if (!deps.store.listRecords().some((record) => record.lease.unreconciled)) { + if (!hasUnreconciledLease(deps.store, sessionId)) { return null } - if (!pending) { - const run = reconcileCurrentLeases(deps) - pending = run.finally(() => { - pending = null - }) + if (!pending.has(sessionId)) { + const run = reconcileCurrentLease(deps, sessionId) + pending.set( + sessionId, + run.finally(() => { + pending.delete(sessionId) + }) + ) } try { - await pending + await pending.get(sessionId) return null } catch (error) { - return classifyStoreFailure( - error, - deps.store.getRecord(sessionId)?.lease.runtimeFence ?? null, - deps.store.getRecord(sessionId) - ) + const record = sessionId === undefined ? undefined : deps.store.getRecord(sessionId) + return classifyStoreFailure(error, record?.lease.runtimeFence ?? null, record) } } } @@ -79,15 +79,15 @@ export function reportEachFailureOnce( /** The reconcile a reader runs, at startup and before each restored read: it never throws, since * an unreconciled lease grants no writer and the next send reconciles again before it acts. - * Answers whether every lease is settled. */ -export function createReaderReconcile( - reconcile: (sessionId: string) => Promise, + * Answers whether the leases it checked are settled. */ +export function createReaderReconcile( + reconcile: (...args: Args) => Promise, failures: ReaderBookkeepingFailures -): (sessionId: string) => Promise { - return async (sessionId) => { +): (...args: Args) => Promise { + return async (...args) => { let failure: unknown try { - const refusal = await reconcile(sessionId) + const refusal = await reconcile(...args) if (!refusal) { failures.clear() return true @@ -113,24 +113,34 @@ function failureKey(failure: unknown): string { return String(failure) } -async function reconcileCurrentLeases(deps: { - store: AgentSessionRecordStore - probe: (record: AgentSessionRecord) => Promise - probeMany?: ( - records: readonly AgentSessionRecord[] - ) => Promise> - now: () => number -}): Promise { +async function reconcileCurrentLease( + deps: { + store: AgentSessionRecordStore + probe: (record: AgentSessionRecord) => Promise + probeMany?: ( + records: readonly AgentSessionRecord[] + ) => Promise> + now: () => number + }, + sessionId?: string +): Promise { for (let pass = 0; pass < MAX_RECONCILIATION_PASSES; pass += 1) { await deps.store.reconcileOnRestart({ + ...(sessionId === undefined ? {} : { sessionId }), probe: deps.probe, ...(deps.probeMany ? { probeMany: deps.probeMany } : {}), now: deps.now() }) - if (!deps.store.listRecords().some((record) => record.lease.unreconciled)) { + if (!hasUnreconciledLease(deps.store, sessionId)) { return } } // An outgoing runtime can still be writing during restart; preserve the record and retry later. throw new Error('execution_owner_reconciling') } + +function hasUnreconciledLease(store: AgentSessionRecordStore, sessionId?: string): boolean { + return sessionId === undefined + ? store.listRecords().some((record) => record.lease.unreconciled) + : store.getRecord(sessionId)?.lease.unreconciled === true +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts index 7df838d04e82..1f0c1cb949d0 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts @@ -62,6 +62,7 @@ describe('restart journal restoration', () => { const restoration = restoreStructuredAgentSessionsOnRestart({ openDeps: NO_OPEN_DEPS, records, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), @@ -103,6 +104,7 @@ describe('restart journal restoration', () => { await restoreStructuredAgentSessionsOnRestart({ openDeps: NO_OPEN_DEPS, records, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), @@ -149,6 +151,7 @@ describe('restart journal restoration', () => { // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the restore reads only the record's session id here. records: [{ sessionId: 'session-1' } as AgentSessionRecord], openDeps: NO_OPEN_DEPS, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => { calls.push('resolveRecovery') @@ -174,7 +177,7 @@ describe('restart journal restoration', () => { const restore = ( bookkeeping: Pick< Parameters[0], - 'reconcile' | 'resolveRecovery' + 'reconcileAll' | 'reconcile' | 'resolveRecovery' > ) => restoreStructuredAgentSessionsOnRestart({ @@ -194,13 +197,28 @@ describe('restart journal restoration', () => { beforeEach(() => restoreRead.mockResolvedValue(null)) + it('opens with one whole-host check, then checks each chat while it holds', async () => { + const reconcileAll = vi.fn(async () => true) + const reconcile = vi.fn(async (_sessionId: string) => true) + + await restore({ reconcileAll, reconcile, resolveRecovery: async () => true }) + + expect(reconcileAll).toHaveBeenCalledOnce() + expect(reconcileAll).toHaveBeenCalledWith() + expect(reconcile.mock.calls.map(([sessionId]) => sessionId).sort()).toEqual( + records.map((record) => record.sessionId).sort() + ) + }) + it('skips it for every chat when the pass check fails, and still opens them all', async () => { - const reconcile = vi.fn(slowFailure) + const reconcileAll = vi.fn(slowFailure) + const reconcile = vi.fn(async () => true) const resolveRecovery = vi.fn(async () => true) - await restore({ reconcile, resolveRecovery }) + await restore({ reconcileAll, reconcile, resolveRecovery }) - expect(reconcile).toHaveBeenCalledOnce() + expect(reconcileAll).toHaveBeenCalledOnce() + expect(reconcile).not.toHaveBeenCalled() expect(resolveRecovery).not.toHaveBeenCalled() expect(restoreRead).toHaveBeenCalledTimes(records.length) }) @@ -208,7 +226,11 @@ describe('restart journal restoration', () => { it('starts no more after the first failed recovery, and still opens every chat', async () => { const resolveRecovery = vi.fn(slowFailure) - await restore({ reconcile: async () => true, resolveRecovery }) + await restore({ + reconcileAll: async () => true, + reconcile: async () => true, + resolveRecovery + }) // Only those already started when the first failed: at most one per chat open at once. expect(resolveRecovery.mock.calls.length).toBeLessThanOrEqual(4) @@ -235,6 +257,7 @@ describe('restart journal restoration', () => { restoreStructuredAgentSessionsOnRestart({ openDeps: { ...NO_OPEN_DEPS, logger: log.logger }, records, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), @@ -262,6 +285,7 @@ describe('restart journal restoration', () => { // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the restore reads only the record's session id here. records: [{ sessionId: 'session-1' } as AgentSessionRecord], openDeps: NO_OPEN_DEPS, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts index 6c6efc49d728..f301e519b21c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts @@ -28,7 +28,9 @@ export type StructuredAgentSessionReadRestoreDeps = { store: Pick } // Lease bookkeeping. Neither throws: a read grants no writer, so bookkeeping must not block it. - /** Whether every lease is settled. */ + /** Whether every lease on this host is settled. */ + reconcileAll: () => Promise + /** Whether this chat's lease is settled. */ reconcile: (sessionId: string) => Promise /** False when its store write failed; the next attach or send resolves it again. */ resolveRecovery: (sessionId: string) => Promise @@ -74,14 +76,12 @@ async function restoreOneStructuredAgentSessionReadUnderSerialize( export async function restoreStructuredAgentSessionsOnRestart( input: StructuredAgentSessionReadRestoreDeps & { records: AgentSessionRecord[] } ): Promise { - const [first] = input.records - if (!first) { + if (input.records.length === 0) { return } - // One check for the pass. Each chat checks again while it holds, since another writer can mark - // leases unreconciled mid-pass; after the first failure, retrying per chat only waits on the - // same store again, and the next attach or send settles those chats instead. - let settled = await input.reconcile(first.sessionId) + // One whole-host check for the pass, which also retries a startup pass that failed. After a + // failure, per-chat retries only hit the same store again; each chat's own action reconciles it. + let settled = await input.reconcileAll() const settleLeases = async (sessionId: string): Promise => { // A session latched in recovery exits here at startup, without waiting for a client. if ( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts index 9a6e3aca5917..5e85f35a43b4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts @@ -58,20 +58,23 @@ export function createStructuredAgentSessionHostRestore( deps: StructuredAgentSessionHostDeps, wiring: Omit< ConstructorParameters[0], - 'openDeps' | 'reconcile' | 'resolveRecovery' + 'openDeps' | 'reconcileAll' | 'reconcile' | 'resolveRecovery' > & { reconcileLeases: (sessionId: string) => Promise + reconcileRestartLeases: () => Promise resolveRecovery: (sessionId: string) => Promise } ): { reconcileRestartLeases: () => Promise restoreReadableSessions: (sessionIds?: readonly string[]) => Promise } { - const { reconcileLeases, resolveRecovery, ...rest } = wiring + const { reconcileLeases, reconcileRestartLeases, resolveRecovery, ...rest } = wiring const failures = reportEachFailureOnce(deps.logger) const reconcile = createReaderReconcile(reconcileLeases, failures) + const reconcileStartup = createReaderReconcile(reconcileRestartLeases, failures) const restorer = new StructuredAgentSessionReadableRestorer({ openDeps: deps, + reconcileAll: reconcileStartup, reconcile, // The next attach or send resolves recovery again, strictly, before it acts. resolveRecovery: (sessionId) => @@ -87,7 +90,7 @@ export function createStructuredAgentSessionHostRestore( const gate = new StructuredAgentSessionRestartRestoreGate() return { reconcileRestartLeases: async () => { - await reconcile('startup') + await reconcileStartup() }, restoreReadableSessions: (sessionIds) => gate.run(() => restorer.restore(sessionIds)) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-target-reconcile.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-target-reconcile.test.ts new file mode 100644 index 000000000000..7ecb84031949 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-target-reconcile.test.ts @@ -0,0 +1,199 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import type { AgentSessionReserveRequest } from '../../runtime/agent-session-reservation-admission' +import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' +import { closeTestJournalHostDatabases } from '../agent-session-journal/journal-host-database-test-support' +import { createRestartReconciler } from './structured-agent-session-restart-reconcile' + +const NOW = 1_800_000_000_000 +const UNUSED: AgentSessionOwnerProbe = { outcome: 'reservation-unused' } +let root: string +let store: AgentSessionRecordStore +let operation = 0 + +function reservation(sessionId: string): AgentSessionReserveRequest { + operation += 1 + return { + sessionId, + location: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'folder-workspace', + workspaceKind: 'folder' as const + }, + provider: 'codex' as const, + accountHome: { variable: 'CODEX_HOME' as const, path: root }, + expectedFence: null, + spawnToken: sessionId, + claimKeyId: 'key-1', + handoffOperationId: null, + probe: UNUSED, + operation: { + callerKey: 'test', + operationId: `${NOW}-${operation.toString(16).padStart(32, '0')}`, + fingerprint: sessionId + }, + now: NOW + } +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-targeted-reconcile-')) + operation = 0 + const previous = await openTestAgentSessionRecordStore(root) + await previous.reserveOwner(reservation('session-1')) + await previous.reserveOwner(reservation('session-2')) + for (const sessionId of ['session-1', 'session-2']) { + await previous.commitProcessIdentity({ + sessionId, + fence: previous.getRecord(sessionId)?.lease.runtimeFence ?? 0, + process: { + hostId: 'local', + pid: 4242, + processStartTimeMs: NOW, + spawnToken: sessionId + }, + now: NOW + }) + } + store = await openTestAgentSessionRecordStore(root) +}) + +afterEach(async () => { + closeTestJournalHostDatabases() + await rm(root, { recursive: true, force: true }) +}) + +describe('targeted restart reconciliation', () => { + it("does not fail an action because another chat's owner check fails", async () => { + const probe = vi.fn(async (record: AgentSessionRecord) => { + if (record.sessionId === 'session-2') { + throw new Error('owner check failed') + } + return UNUSED + }) + const reconcile = createRestartReconciler({ store, probe, now: () => NOW }) + + expect(await reconcile('session-1')).toBeNull() + expect(probe).toHaveBeenCalledOnce() + expect(store.getRecord('session-2')?.lease.unreconciled).toBe(true) + await expect(reconcile('session-2')).rejects.toThrow('owner check failed') + }) + + it("settles a chat while another chat's record changes under every check of it", async () => { + let renames = 0 + const probe = vi.fn(async (record: AgentSessionRecord) => { + if (record.sessionId === 'session-2') { + renames += 1 + // Each check of session-2 is stale by the time it applies. + await store.setConversationName('session-2', `renamed ${renames}`) + } + return UNUSED + }) + const reconcile = createRestartReconciler({ store, probe, now: () => NOW }) + + expect(await reconcile('session-1')).toBeNull() + expect(store.getRecord('session-1')?.lease.unreconciled).toBe(false) + // The whole-host pass does run out of passes on session-2. + expect(await reconcile()).not.toBeNull() + expect(store.getRecord('session-2')?.lease.unreconciled).toBe(true) + }) + + it("does not wait on another chat's stalled owner check in the startup batch", async () => { + let release: (probe: AgentSessionOwnerProbe) => void = () => undefined + let started: () => void = () => undefined + const probing = new Promise((resolve) => { + started = resolve + }) + const waiting = new Promise((resolve) => { + release = resolve + }) + const probe = vi.fn(async (record: AgentSessionRecord) => { + if (record.sessionId === 'session-2') { + started() + return waiting + } + return UNUSED + }) + const reconcile = createRestartReconciler({ store, probe, now: () => NOW }) + const startup = reconcile() + await probing + const target = reconcile('session-1') + let timeout: ReturnType | undefined + try { + expect( + await Promise.race([ + target, + new Promise((resolve) => { + timeout = setTimeout(() => resolve('blocked'), 1_000) + }) + ]) + ).toBeNull() + expect(store.getRecord('session-2')?.lease.unreconciled).toBe(true) + } finally { + clearTimeout(timeout) + release(UNUSED) + await Promise.all([startup, target]) + } + }) + + it('joins simultaneous checks of the same session', async () => { + const probe = vi.fn(async () => UNUSED) + const reconcile = createRestartReconciler({ store, probe, now: () => NOW }) + expect(await Promise.all([reconcile('session-1'), reconcile('session-1')])).toEqual([ + null, + null + ]) + expect(probe).toHaveBeenCalledOnce() + }) + + it('does no reconciliation for a new chat or a session already reconciled', async () => { + const probe = vi.fn(async () => UNUSED) + const reconcile = createRestartReconciler({ store, probe, now: () => NOW }) + expect(await reconcile('new-session')).toBeNull() + expect(probe).not.toHaveBeenCalled() + expect(await reconcile('session-1')).toBeNull() + expect(await reconcile('session-1')).toBeNull() + expect(probe).toHaveBeenCalledOnce() + }) + + it('keeps an unproven target in recovery and refuses a new writer', async () => { + const reconcile = createRestartReconciler({ + store, + probe: async () => ({ outcome: 'indeterminate', reason: 'owner runs on another host' }), + now: () => NOW + }) + expect(await reconcile('session-1')).toBeNull() + const fence = store.getRecord('session-1')?.lease.runtimeFence ?? 0 + await expect( + store.reserveOwner({ + ...reservation('session-1'), + expectedFence: fence, + probe: { outcome: 'indeterminate', reason: 'owner runs on another host' } + }) + ).rejects.toThrow('agent_session_ownership_unknown') + }) + + it('still batches every pending record for explicit startup reconciliation', async () => { + const probeMany = vi.fn( + async (records: readonly AgentSessionRecord[]) => + new Map(records.map((record) => [record.sessionId, UNUSED])) + ) + const reconcile = createRestartReconciler({ + store, + probe: async () => UNUSED, + probeMany, + now: () => NOW + }) + expect(await reconcile()).toBeNull() + expect(probeMany.mock.calls[0]?.[0].map((record) => record.sessionId)).toEqual([ + 'session-1', + 'session-2' + ]) + }) +}) diff --git a/src/main/runtime/agent-session-closed-owner-acquisition.test.ts b/src/main/runtime/agent-session-closed-owner-acquisition.test.ts new file mode 100644 index 000000000000..1d6fc948ca0d --- /dev/null +++ b/src/main/runtime/agent-session-closed-owner-acquisition.test.ts @@ -0,0 +1,246 @@ +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 { claudeProviderHandle } from '../../shared/agent-session-provider-handle-encoding' +import { agentSessionRecordFixture } from '../../shared/agent-session-record.test-fixture' +import { + closeTestJournalHostDatabases, + openTestJournalHostDatabase +} from '../native-chat/agent-session-journal/journal-host-database-test-support' +import { openTestAgentSessionRecordStore } from './agent-session-record-store-test-harness' +import type { AgentSessionAcquisitionExitProof } from './agent-session-acquisition-failure-settlement' + +const SESSION = 'session-alpha-1' +const NOW = 1_800_000_000_000 +const OPERATION_ID = `${NOW}-${'1'.repeat(32)}` +let directory: string + +beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), 'orca-closed-acquisition-')) +}) + +afterEach(async () => { + closeTestJournalHostDatabases() + await rm(directory, { recursive: true, force: true }) +}) + +async function acquisition(proved: boolean, spawned: boolean) { + const store = await openTestAgentSessionRecordStore(directory) + const fixture = agentSessionRecordFixture() + const { record } = await store.reserveOwner({ + sessionId: SESSION, + location: fixture.location, + provider: 'claude', + accountHome: fixture.accountHome, + expectedFence: null, + spawnToken: 'spawn-1', + claimKeyId: 'key-1', + handoffOperationId: OPERATION_ID, + operation: { callerKey: 'caller-1', operationId: OPERATION_ID, fingerprint: 'fp-1' }, + probe: { outcome: 'reservation-unused' }, + now: NOW + }) + const fence = record.lease.runtimeFence + if (spawned) { + await store.commitProcessIdentity({ + sessionId: SESSION, + fence, + now: NOW, + process: { hostId: 'local', pid: 4242, processStartTimeMs: NOW, spawnToken: 'spawn-1' } + }) + } + if (proved) { + await store.proveOwner({ + sessionId: SESSION, + fence, + now: NOW, + link: { + linkId: 'link-1', + handle: claudeProviderHandle('provider-1', null), + mintedAtFence: fence, + observedAt: NOW, + origin: 'created' + } + }) + } + return { store, fence } +} + +function failure(fence: number, exitProof: AgentSessionAcquisitionExitProof) { + return { + sessionId: SESSION, + fence, + spawnToken: 'spawn-1', + callerKey: 'caller-1', + operationId: OPERATION_ID, + outcome: { + status: 'failed' as const, + code: 'agent_session_operation_invalid' as const, + message: 'failed' + }, + exitProof, + now: NOW + 1 + } +} + +describe('closed-owner proofs during failed acquisition', () => { + it.each(['exit-proven', 'root-exit-observed', 'processless'] as const)( + 'keeps a failed reservation proof for %s', + async (exitProof) => { + const { store, fence } = await acquisition(false, exitProof !== 'processless') + await store.settleFailedAcquisition(failure(fence, exitProof)) + expect(store.closedOwners(SESSION)).toEqual([ + expect.objectContaining({ + deadOwnerFence: fence, + evidence: expect.objectContaining({ ownerFence: fence }), + process: exitProof === 'processless' ? null : expect.objectContaining({ pid: 4242 }) + }) + ]) + expect((await openTestAgentSessionRecordStore(directory)).closedOwners(SESSION)).toEqual( + store.closedOwners(SESSION) + ) + } + ) + + it.each(['exit-proven', 'root-exit-observed'] as const)( + 'keeps a failed post-acquisition attachment proof for %s', + async (exitProof) => { + const { store, fence } = await acquisition(true, true) + await store.settleFailedPostAcquisitionAttachment(failure(fence, exitProof)) + expect(store.closedOwners(SESSION)[0]).toMatchObject({ + deadOwnerFence: fence, + process: { pid: 4242 }, + evidence: { kind: 'exit-observed', ownerFence: fence } + }) + } + ) + + it.each([false, true])( + 'creates no fact for an unproven failed acquisition (proved handle: %s)', + async (proved) => { + const { store, fence } = await acquisition(proved, true) + await (proved + ? store.settleFailedPostAcquisitionAttachment(failure(fence, 'unproven')) + : store.settleFailedAcquisition(failure(fence, 'unproven'))) + expect(store.closedOwners(SESSION)).toEqual([]) + expect(store.getRecord(SESSION)?.lease.handoffStage).toBe('recovering') + } + ) + + it('atomically rolls back the failed acquisition, its operation answer, and its closure', async () => { + const { store, fence } = await acquisition(false, true) + openTestJournalHostDatabase(directory).db.exec(`CREATE TEMP TRIGGER refuse_closure + BEFORE INSERT ON agent_session_closed_owners BEGIN SELECT RAISE(ABORT, 'closure refused'); END`) + await expect(store.settleFailedAcquisition(failure(fence, 'exit-proven'))).rejects.toThrow( + 'closure refused' + ) + expect(store.closedOwners(SESSION)).toEqual([]) + expect(store.getRecord(SESSION)?.lease.runtimeFence).toBe(fence) + expect(store.getOperationRow('caller-1', OPERATION_ID)?.outcome.status).toBe('pending') + }) + + it('captures the positive unused-reservation proof at restart, with no process identity', async () => { + await acquisition(false, false) + const restarted = await openTestAgentSessionRecordStore(directory) + await restarted.reconcileOnRestart({ + probe: async () => ({ outcome: 'reservation-unused' }), + now: NOW + 1 + }) + expect(restarted.closedOwners(SESSION)[0]).toMatchObject({ + deadOwnerFence: 1, + process: null, + evidence: { kind: 'pid-absent', detail: 'reservation never spawned' } + }) + }) + + it('keeps a fresh unused-reservation proof when the next reservation clears it', async () => { + const { store, fence } = await acquisition(false, false) + await store.transitionHandoff(SESSION, (record) => ({ + ...record, + lease: { ...record.lease, handoffOperationId: null } + })) + const fixture = agentSessionRecordFixture() + await store.reserveOwner({ + sessionId: SESSION, + location: fixture.location, + provider: 'claude', + accountHome: fixture.accountHome, + expectedFence: fence, + spawnToken: 'spawn-2', + claimKeyId: 'key-1', + handoffOperationId: null, + operation: { + callerKey: 'caller-1', + operationId: `${NOW}-${'2'.repeat(32)}`, + fingerprint: 'fp-2' + }, + probe: { outcome: 'reservation-unused' }, + now: NOW + 1 + }) + expect(store.getRecord(SESSION)?.lease.deathEvidence).toBeNull() + expect(store.closedOwners(SESSION)[0]).toMatchObject({ + deadOwnerFence: fence, + process: null, + evidence: { kind: 'pid-absent', detail: 'reservation never spawned' } + }) + }) + + it('creates no fact when restart drops an unverified processless reservation', async () => { + await acquisition(false, false) + const restarted = await openTestAgentSessionRecordStore(directory) + await restarted.reconcileOnRestart({ + probe: async () => ({ outcome: 'indeterminate', reason: 'no answer' }), + now: NOW + 1 + }) + expect(restarted.closedOwners(SESSION)).toEqual([]) + expect(restarted.getRecord(SESSION)?.lease.claimStatus).toBe('released') + }) + + it('rejects a stale processless probe after the owner identity arrives at the same fence', async () => { + const { store, fence } = await acquisition(false, false) + await store.transitionHandoff(SESSION, (record) => ({ + ...record, + lease: { ...record.lease, handoffOperationId: null } + })) + await store.commitProcessIdentity({ + sessionId: SESSION, + fence, + now: NOW, + process: { hostId: 'local', pid: 4242, processStartTimeMs: NOW, spawnToken: 'spawn-1' } + }) + const fixture = agentSessionRecordFixture() + await expect( + store.reserveOwner({ + sessionId: SESSION, + location: fixture.location, + provider: 'claude', + accountHome: fixture.accountHome, + expectedFence: fence, + spawnToken: 'spawn-2', + claimKeyId: 'key-1', + handoffOperationId: null, + operation: { + callerKey: 'caller-1', + operationId: `${NOW}-${'2'.repeat(32)}`, + fingerprint: 'fp-2' + }, + probe: { outcome: 'pid-absent' }, + probedOwner: null, + now: NOW + 1 + }) + ).rejects.toThrow() + await expect( + store.evictProvenDeadOwner({ + sessionId: SESSION, + expectedFence: fence, + probe: { outcome: 'pid-absent' }, + probedOwner: null, + now: NOW + 1 + }) + ).rejects.toThrow() + expect(store.closedOwners(SESSION)).toEqual([]) + expect(store.getRecord(SESSION)?.lease.ownerProcess?.pid).toBe(4242) + expect(store.getRecord(SESSION)?.lease.runtimeFence).toBe(fence) + }) +}) diff --git a/src/main/runtime/agent-session-closed-owner-history.test.ts b/src/main/runtime/agent-session-closed-owner-history.test.ts new file mode 100644 index 000000000000..f5e944ef6447 --- /dev/null +++ b/src/main/runtime/agent-session-closed-owner-history.test.ts @@ -0,0 +1,587 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication' +import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../shared/agent-session-journal-types' +import type { AgentSessionRecord } from '../../shared/agent-session-record' +import { + agentSessionLeaseFixture, + agentSessionRecordFixture +} from '../../shared/agent-session-record.test-fixture' +import { + closeTestJournalHostDatabases, + openTestJournalHostDatabase +} from '../native-chat/agent-session-journal/journal-host-database-test-support' +import { + insertJournalRow, + publishJournalSessionEpoch +} from '../native-chat/agent-session-journal/journal-row-table' +import { JournalRowWriter } from '../native-chat/agent-session-journal/journal-row-writer' +import type { JournalRow } from '../native-chat/agent-session-journal/journal-row-schema' +import { JOURNAL_DB_SCHEMA_VERSION } from '../native-chat/agent-session-journal/journal-database-schema' +import { loadAgentSessionStoreRows } from './agent-session-record-rows' +import { AgentSessionStoreTransactions } from './agent-session-store-transactions' +import type { AgentSessionRecordStore } from './agent-session-record-store' +import { + openTestAgentSessionRecordStore, + seedTestAgentSessionRecordStore +} from './agent-session-record-store-test-harness' +import { releaseStoredAgentSessionOwnerAfterSurfaceClose } from './agent-session-surface-release-transition' +import { releaseUnprovenAgentSessionOwner } from './agent-session-lease-transitions' +import { createStructuredAgentSessionOwnerProbe } from './structured-agent-session-owner-probe' +import { + agentSessionClosedOwnerKey, + type AgentSessionClosedOwner +} from './agent-session-closed-owner' +import { beginAgentSessionRuntimeIncarnationForTest } from './agent-session-runtime-attribution' +import { recordAgentSessionRuntimeEnd } from './agent-session-runtime-end-record' + +const SESSION = 'session-alpha-1' +const NOW = 1_800_000_000_000 +const MATCHED: AgentSessionOwnerProbe = { outcome: 'identity-matched', matchedOn: ['spawn-token'] } +let directory: string +let serial = 0 + +beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), 'orca-closed-owners-')) +}) + +afterEach(async () => { + closeTestJournalHostDatabases() + vi.restoreAllMocks() + await rm(directory, { recursive: true, force: true }) +}) + +function hostDatabase() { + return openTestJournalHostDatabase(directory) +} + +function journalRow(seq: number, epoch = 'epoch-1'): JournalRow { + return { + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + epoch, + seq, + fence: 7, + ts: NOW + seq, + kind: 'epoch', + reason: 'session_created', + providerHandle: { kind: 'claude', sessionId: 'provider-session-alpha-1', leafUuid: null } + } +} + +function seedJournal(): void { + hostDatabase().transaction((db) => { + publishJournalSessionEpoch(db, { sessionId: SESSION, workspaceId: 'workspace-1' }, 'epoch-1') + insertJournalRow(db, SESSION, journalRow(1)) + }) +} + +async function loadedOwner(record = agentSessionRecordFixture()): Promise { + await seedTestAgentSessionRecordStore(directory, { records: [record] }) + return openTestAgentSessionRecordStore(directory) +} + +async function liveOwner(): Promise { + const store = await loadedOwner() + await store.reconcileOnRestart({ probe: async () => MATCHED, now: NOW }) + await store.transitionHandoff(SESSION, (record) => ({ + ...record, + lease: { ...record.lease, handoffStage: null } + })) + seedJournal() + return store +} + +function reservation(record: AgentSessionRecord, probe: AgentSessionOwnerProbe) { + serial += 1 + const operationId = `${NOW}-${String(serial).padStart(32, '0')}` + return { + sessionId: record.sessionId, + location: record.location, + provider: 'claude' as const, + accountHome: record.accountHome, + expectedFence: record.lease.runtimeFence, + spawnToken: `spawn-${serial}`, + claimKeyId: 'key-1', + handoffOperationId: operationId, + operation: { callerKey: 'caller-1', operationId, fingerprint: `fp-${serial}` }, + probe, + now: NOW + 100 + } +} + +function requiredRecord(store: AgentSessionRecordStore): AgentSessionRecord { + const record = store.getRecord(SESSION) + if (!record) { + throw new Error('missing test owner') + } + return record +} + +function requiredFact(store: AgentSessionRecordStore): AgentSessionClosedOwner { + const fact = store.closedOwners(SESSION)[0] + if (!fact) { + throw new Error('missing test closed owner') + } + return fact +} + +function factCount(): number { + return Number( + hostDatabase().db.prepare('SELECT count(*) AS n FROM agent_session_closed_owners').get()?.n + ) +} + +const proofTransitions = [ + 'surface exit', + 'explicit eviction', + 'restart eviction', + 'reservation probe' +] as const + +async function closeOwner( + store: AgentSessionRecordStore, + transition: (typeof proofTransitions)[number] +) { + if (transition === 'surface exit') { + await releaseStoredAgentSessionOwnerAfterSurfaceClose(store, { + sessionId: SESSION, + expectedFence: 7, + now: NOW + 10, + exitObservedAt: NOW + 1, + exitReason: 'root exited' + }) + } else if (transition === 'explicit eviction') { + await store.evictProvenDeadOwner({ + sessionId: SESSION, + expectedFence: 7, + probe: { outcome: 'identity-mismatch', field: 'process-start-time' }, + now: NOW + 10 + }) + } else if (transition === 'restart eviction') { + await store.transitionHandoff(SESSION, (record) => ({ + ...record, + lease: { ...record.lease, unreconciled: true } + })) + await store.reconcileOnRestart({ + probe: async () => ({ outcome: 'pid-absent' }), + now: NOW + 10 + }) + } else { + await store.reserveOwner(reservation(requiredRecord(store), { outcome: 'pid-absent' })) + } +} + +describe('committed closed-owner facts', () => { + it.each(proofTransitions)( + 'keeps the %s proof after replacement and restart', + async (transition) => { + const store = await liveOwner() + const original = requiredRecord(store) + const committed: AgentSessionClosedOwner[] = [] + const legacy = vi.fn() + store.onDeathEvidence(legacy) + store.onClosedOwner((fact) => { + expect(factCount()).toBe(1) + committed.push(fact) + }) + await closeOwner(store, transition) + const fact = requiredFact(store) + expect(fact).toMatchObject({ + schemaVersion: 1, + sessionId: SESSION, + deadOwnerFence: 7, + location: original.location, + process: { hostId: 'local', pid: 4242, processStartTimeMs: 1_700_000_000_000 }, + evidence: { ownerFence: 7 } + }) + expect(fact.process).not.toHaveProperty('spawnToken') + if ('path' in original.accountHome) { + expect(JSON.stringify(fact)).not.toContain(original.accountHome.path) + } + expect(committed).toEqual([fact]) + if (transition === 'reservation probe') { + expect(legacy).not.toHaveBeenCalled() + } else { + expect(legacy).toHaveBeenCalledWith(SESSION, fact) + await store.reserveOwner( + reservation(requiredRecord(store), { outcome: 'reservation-unused' }) + ) + } + expect(requiredRecord(store).lease.deathEvidence).toBeNull() + expect(store.closedOwners(SESSION)).toEqual([fact]) + closeTestJournalHostDatabases() + expect((await openTestAgentSessionRecordStore(directory)).closedOwners(SESSION)).toEqual([ + fact + ]) + } + ) + + it.each(proofTransitions)('rolls back the %s and its fact together', async (transition) => { + const store = await liveOwner() + const listener = vi.fn() + store.onClosedOwner(listener) + hostDatabase().db + .exec(`CREATE TEMP TRIGGER refuse_closure BEFORE INSERT ON agent_session_closed_owners + BEGIN SELECT RAISE(ABORT, 'closure refused'); END`) + await expect(closeOwner(store, transition)).rejects.toThrow('closure refused') + expect(requiredRecord(store).lease.runtimeFence).toBe(7) + expect(requiredRecord(store).lease.deathEvidence).toBeNull() + expect(store.closedOwners(SESSION)).toEqual([]) + expect(factCount()).toBe(0) + expect(listener).not.toHaveBeenCalled() + expect( + hostDatabase().db.prepare('SELECT count(*) AS n FROM agent_session_operations').get()?.n + ).toBe(0) + }) + + it('orders immutable facts by generation even if the clock moves back', async () => { + const store = await liveOwner() + await closeOwner(store, 'surface exit') + const first = requiredFact(store) + const request = reservation(requiredRecord(store), { outcome: 'reservation-unused' }) + const { record } = await store.reserveOwner(request) + await store.commitProcessIdentity({ + sessionId: SESSION, + fence: record.lease.runtimeFence, + process: { + hostId: 'local', + pid: 4243, + processStartTimeMs: NOW, + spawnToken: request.spawnToken + }, + now: NOW + }) + await store.evictProvenDeadOwner({ + sessionId: SESSION, + expectedFence: record.lease.runtimeFence, + probe: { outcome: 'pid-absent' }, + now: NOW - 1 + }) + const facts = store.closedOwners(SESSION) + expect(facts.map((fact) => fact.deadOwnerFence)).toEqual([7, 9]) + expect(facts[0]).toBe(first) + expect(facts[1]?.evidence.lastProvenAliveAt).toBe(NOW - 1) + expect(() => Object.assign(first.evidence, { detail: 'changed' })).toThrow(TypeError) + }) + + it.each(['restart eviction', 'reservation probe'] as const)( + 'keeps the earlier runtime end cause in the %s fact', + async (transition) => { + const store = await liveOwner() + const original = requiredRecord(store) + const owner = original.lease.ownerProcess + if (!owner) { + throw new Error('missing fixture process') + } + await store.transitionHandoff(SESSION, (record) => ({ + ...record, + lease: { ...record.lease, ownerProcess: { ...owner, runtime: undefined } } + })) + recordAgentSessionRuntimeEnd('update', NOW) + beginAgentSessionRuntimeIncarnationForTest() + const restarted = await openTestAgentSessionRecordStore(directory) + if (transition === 'restart eviction') { + await restarted.reconcileOnRestart({ + probe: async () => ({ outcome: 'pid-absent' }), + now: NOW + 1 + }) + } else { + await restarted.reconcileOnRestart({ probe: async () => MATCHED, now: NOW + 1 }) + await restarted.transitionHandoff(SESSION, (record) => ({ + ...record, + lease: { ...record.lease, handoffStage: null } + })) + await closeOwner(restarted, transition) + } + expect(requiredFact(restarted).evidence.runtimeEnd).toBe('update') + } + ) +}) + +describe('proof requirements and stale observations', () => { + it('records nothing for an expired lease and an indeterminate reservation probe', async () => { + const store = await liveOwner() + await expect( + store.reserveOwner( + reservation(requiredRecord(store), { + outcome: 'indeterminate', + reason: 'disconnected' + }) + ) + ).rejects.toThrow() + expect(store.closedOwners(SESSION)).toEqual([]) + expect(factCount()).toBe(0) + }) + + it('records nothing for an unproven release', async () => { + const store = await loadedOwner() + await store.reconcileOnRestart({ + probe: async () => ({ outcome: 'indeterminate', reason: 'disconnected' }), + now: NOW + }) + await store.transitionHandoff(SESSION, (record) => + releaseUnprovenAgentSessionOwner({ + record, + expectedFence: 7, + now: NOW + 1 + }) + ) + await store.reserveOwner(reservation(requiredRecord(store), { outcome: 'reservation-unused' })) + expect(store.closedOwners(SESSION)).toEqual([]) + }) + + it('records nothing when a local host cannot verify an SSH owner', async () => { + const remote = agentSessionRecordFixture() + const owner = remote.lease.ownerProcess + if (!owner) { + throw new Error('missing fixture process') + } + const store = await loadedOwner({ + ...remote, + location: { ...remote.location, executionHostId: 'ssh:host' }, + lease: { ...remote.lease, ownerProcess: { ...owner, hostId: 'ssh:host' } } + }) + await store.reconcileOnRestart({ + probe: createStructuredAgentSessionOwnerProbe('local'), + now: NOW + }) + expect(requiredRecord(store).lease.handoffStage).toBe('recovering') + expect(store.closedOwners(SESSION)).toEqual([]) + }) + + it('rejects a stale reservation probe and delayed exit without closing the successor', async () => { + const store = await liveOwner() + const stale = reservation(requiredRecord(store), { outcome: 'pid-absent' }) + await closeOwner(store, 'reservation probe') + const fact = requiredFact(store) + await expect(store.reserveOwner(stale)).rejects.toThrow() + await expect( + releaseStoredAgentSessionOwnerAfterSurfaceClose(store, { + sessionId: SESSION, + expectedFence: 7, + now: NOW + 1 + }) + ).rejects.toThrow() + expect(store.closedOwners(SESSION)).toEqual([fact]) + expect(factCount()).toBe(1) + }) + + it('ignores a delayed restart probe after the compared owner changed', async () => { + const store = await loadedOwner() + let answer: (probe: AgentSessionOwnerProbe) => void = () => { + throw new Error('probe not started') + } + const began = Promise.withResolvers() + const reconciliation = store.reconcileOnRestart({ + probe: () => { + began.resolve() + return new Promise((resolve) => { + answer = resolve + }) + }, + now: NOW + }) + await began.promise + await store.reconcileOnRestart({ probe: async () => MATCHED, now: NOW + 1 }) + await store.transitionHandoff(SESSION, (record) => ({ + ...record, + lease: { ...record.lease, unreconciled: true } + })) + answer({ outcome: 'pid-absent' }) + expect((await reconciliation).size).toBe(0) + expect(store.closedOwners(SESSION)).toEqual([]) + }) + + it('captures a folder workspace in its WSL execution scope without needing Git', async () => { + const record = agentSessionRecordFixture() + const store = await loadedOwner({ + ...record, + location: { ...record.location, wslDistro: 'Ubuntu', workspaceKind: 'folder' } + }) + await store.reconcileOnRestart({ probe: async () => ({ outcome: 'pid-absent' }), now: NOW }) + expect(requiredFact(store).location).toMatchObject({ + wslDistro: 'Ubuntu', + workspaceKind: 'folder' + }) + expect(agentSessionClosedOwnerKey(requiredFact(store))).toContain('Ubuntu') + }) +}) + +describe('retiring a fact', () => { + function writer() { + return new JournalRowWriter({ + sessionId: SESSION, + now: () => NOW, + serialize: async (run) => run(), + database: hostDatabase, + readOnly: () => false, + highestFence: () => 7, + nextSequence: () => 2, + commit: () => {} + }) + } + + it('deletes only when the journal write commits and never reconstructs a retired fact', async () => { + const store = await liveOwner() + await closeOwner(store, 'surface exit') + const receipt = store.closedOwnerReceipt(requiredFact(store)) + await writer().enqueue((seq) => journalRow(seq), undefined, { + write: (db) => { + receipt.write(db) + expect(store.closedOwners(SESSION)).toHaveLength(1) + }, + committed: receipt.committed + }) + expect(store.closedOwners(SESSION)).toEqual([]) + expect(factCount()).toBe(0) + closeTestJournalHostDatabases() + expect((await openTestAgentSessionRecordStore(directory)).closedOwners(SESSION)).toEqual([]) + }) + + it('keeps the fact in memory and on disk when the journal write rolls back after the receipt', async () => { + const store = await liveOwner() + await closeOwner(store, 'surface exit') + const fact = requiredFact(store) + const receipt = store.closedOwnerReceipt(fact) + await expect( + writer().enqueue((seq) => journalRow(seq), undefined, { + write: (db) => { + receipt.write(db) + throw new Error('journal transaction failed') + }, + committed: receipt.committed + }) + ).rejects.toThrow('journal transaction failed') + expect(store.closedOwners(SESSION)).toEqual([fact]) + expect(factCount()).toBe(1) + expect(hostDatabase().db.prepare('SELECT MAX(seq) AS n FROM journal_rows').get()?.n).toBe(1) + await store.setConversationName(SESSION, 'after rollback') + closeTestJournalHostDatabases() + expect((await openTestAgentSessionRecordStore(directory)).closedOwners(SESSION)).toEqual([fact]) + }) + + it('removes facts with a session record in the store transaction', async () => { + const store = await liveOwner() + await closeOwner(store, 'surface exit') + const database = hostDatabase() + const transactions = new AgentSessionStoreTransactions( + database, + loadAgentSessionStoreRows(database.db) + ) + await transactions.transact((draft) => { + draft.records.delete(SESSION) + }) + expect(transactions.state.closedOwners.size).toBe(0) + expect(factCount()).toBe(0) + expect( + database.db.prepare('SELECT 1 FROM agent_session_records WHERE session_id = ?').get(SESSION) + ).toBeUndefined() + }) + + it('also cascades when the session row itself is deleted', async () => { + const store = await liveOwner() + await closeOwner(store, 'surface exit') + hostDatabase().transaction((db) => { + db.prepare('DELETE FROM agent_session_records WHERE session_id = ?').run(SESSION) + }) + expect(factCount()).toBe(0) + }) +}) + +describe('host schema and legacy proof migration', () => { + const legacyProof = agentSessionRecordFixture( + agentSessionLeaseFixture({ + claimStatus: 'released', + ownerProcess: null, + reservedSpawnToken: null, + runtimeFence: 8, + deathEvidence: { kind: 'pid-absent', detail: 'old proof', observedAt: NOW, ownerFence: 7 } + }) + ) + + /** A database a build without the closed-owner table wrote, at `userVersion`. */ + async function databaseWithoutClosedOwners( + record: AgentSessionRecord, + userVersion = JOURNAL_DB_SCHEMA_VERSION + ): Promise { + await seedTestAgentSessionRecordStore(directory, { records: [record] }) + hostDatabase().db.exec('DROP TABLE agent_session_closed_owners') + hostDatabase().db.pragma(`user_version = ${userVersion}`) + closeTestJournalHostDatabases() + } + + function closedOwnersTableExists(): boolean { + return ( + hostDatabase() + .db.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?") + .get('agent_session_closed_owners') !== undefined + ) + } + + it('migrates an attributed old proof once without changing the schema version', async () => { + await databaseWithoutClosedOwners(legacyProof) + const migrated = await openTestAgentSessionRecordStore(directory) + expect(hostDatabase().db.pragma('user_version', { simple: true })).toBe( + JOURNAL_DB_SCHEMA_VERSION + ) + expect(requiredFact(migrated)).toMatchObject({ deadOwnerFence: 7, process: null }) + const receipt = migrated.closedOwnerReceipt(requiredFact(migrated)) + hostDatabase().transaction(receipt.write) + receipt.committed() + closeTestJournalHostDatabases() + // The old proof is still on the lease; a second migration would bring the retired fact back. + const reopened = await openTestAgentSessionRecordStore(directory) + expect(requiredRecord(reopened).lease.deathEvidence?.ownerFence).toBe(7) + expect(reopened.closedOwners(SESSION)).toEqual([]) + expect(factCount()).toBe(0) + }) + + it('never migrates into a table that already exists', async () => { + await seedTestAgentSessionRecordStore(directory, { records: [legacyProof] }) + expect(closedOwnersTableExists()).toBe(true) + closeTestJournalHostDatabases() + const store = await openTestAgentSessionRecordStore(directory) + expect(store.closedOwners(SESSION)).toEqual([]) + expect(factCount()).toBe(0) + }) + + it('neither creates nor migrates on a read-only open of a newer schema', async () => { + await databaseWithoutClosedOwners(legacyProof, JOURNAL_DB_SCHEMA_VERSION + 1) + const olderHost = await openTestAgentSessionRecordStore(directory) + expect(olderHost.readOnly).toBe(true) + expect(olderHost.closedOwners(SESSION)).toEqual([]) + expect(closedOwnersTableExists()).toBe(false) + }) + + it('leaves legacy evidence without a named generation alone', async () => { + await databaseWithoutClosedOwners( + agentSessionRecordFixture( + agentSessionLeaseFixture({ + claimStatus: 'released', + ownerProcess: null, + reservedSpawnToken: null, + deathEvidence: { kind: 'pid-absent', detail: 'unknown owner', observedAt: NOW } + }) + ) + ) + const store = await openTestAgentSessionRecordStore(directory) + expect(store.closedOwners(SESSION)).toEqual([]) + expect(requiredRecord(store).lease.deathEvidence?.detail).toBe('unknown owner') + }) + + it('refuses owner mutations and delete receipts on a newer host schema', async () => { + const store = await liveOwner() + await closeOwner(store, 'surface exit') + const fact = requiredFact(store) + hostDatabase().db.pragma(`user_version = ${JOURNAL_DB_SCHEMA_VERSION + 1}`) + closeTestJournalHostDatabases() + const olderHost = await openTestAgentSessionRecordStore(directory) + expect(olderHost.readOnly).toBe(true) + expect(olderHost.closedOwners(SESSION)).toEqual([fact]) + await expect(olderHost.setConversationName(SESSION, 'no write')).rejects.toThrow() + expect(() => olderHost.closedOwnerReceipt(fact).write(hostDatabase().db)).toThrow() + expect(factCount()).toBe(1) + }) +}) diff --git a/src/main/runtime/agent-session-closed-owner-history.ts b/src/main/runtime/agent-session-closed-owner-history.ts new file mode 100644 index 000000000000..55bc4bf16d79 --- /dev/null +++ b/src/main/runtime/agent-session-closed-owner-history.ts @@ -0,0 +1,76 @@ +import type { + AgentSessionDeathEvidence, + AgentSessionRecord +} from '../../shared/agent-session-record' +import { agentSessionClosedOwnerKey, closedAgentSessionOwner } from './agent-session-closed-owner' +import type { AgentSessionStoreState } from './agent-session-store-state' +import { agentSessionRuntimeIncarnation } from './agent-session-runtime-attribution' +import { + adjudicateAgentSessionRestart, + type AgentSessionOwnerProbe +} from '../../shared/agent-session-lease-adjudication' + +export function retainAgentSessionReservationClosure( + state: AgentSessionStoreState, + record: AgentSessionRecord, + probe: AgentSessionOwnerProbe, + observedAt: number +): void { + const proof = adjudicateAgentSessionRestart({ lease: record.lease, probe, observedAt }) + if (proof.disposition === 'evicted' && proof.evidence) { + retainAgentSessionClosedOwner(state, record, proof.evidence) + } +} + +export function retainAgentSessionClosedOwner( + state: AgentSessionStoreState, + record: AgentSessionRecord, + evidence: AgentSessionDeathEvidence +): void { + if (evidence.ownerFence !== record.lease.runtimeFence) { + return + } + const fact = closedAgentSessionOwner(record, evidence) + if (fact) { + const key = agentSessionClosedOwnerKey(fact) + if (!state.closedOwners.has(key)) { + state.closedOwners.set(key, fact) + } + } +} + +/** Captures every proof writer after runtime attribution, before the draft's rows are written. */ +export function captureAgentSessionClosedOwners( + published: AgentSessionStoreState, + draft: AgentSessionStoreState +): void { + for (const [sessionId, record] of draft.records) { + const before = published.records.get(sessionId) + const evidence = record.lease.deathEvidence + if (before && evidence && evidence !== before.lease.deathEvidence) { + retainAgentSessionClosedOwner(draft, before, evidence) + } + } + for (const [key, fact] of draft.closedOwners) { + if (!draft.records.has(fact.sessionId) && !draft.unreadableRecords.has(fact.sessionId)) { + draft.closedOwners.delete(key) + continue + } + if (published.closedOwners.has(key)) { + continue + } + const before = published.records.get(fact.sessionId) + const runtime = before?.lease.ownerProcess?.runtime + const runtimeEnd = + runtime && runtime !== agentSessionRuntimeIncarnation() + ? draft.runtimeEnds?.get(runtime) + : undefined + if ( + fact.evidence.runtimeEnd === undefined && + runtimeEnd && + before?.lease.claimStatus !== 'conflicted' + ) { + draft.closedOwners.set(key, { ...fact, evidence: { ...fact.evidence, runtimeEnd } }) + } + } +} diff --git a/src/main/runtime/agent-session-closed-owner-rows.ts b/src/main/runtime/agent-session-closed-owner-rows.ts new file mode 100644 index 000000000000..dcb015ed27f7 --- /dev/null +++ b/src/main/runtime/agent-session-closed-owner-rows.ts @@ -0,0 +1,110 @@ +import type Database from '../sqlite/sync-database' +import { isPersistedAgentSessionRecord } from '../../shared/agent-session-record' +import { decodePersistedAgentSessionRecord } from '../../shared/agent-session-record-stored-form' +import { + agentSessionClosedOwnerKey, + closedAgentSessionOwner, + isReadableAgentSessionClosedOwner, + type AgentSessionClosedOwner +} from './agent-session-closed-owner' + +function parsed(value: unknown): unknown { + try { + return typeof value === 'string' ? JSON.parse(value) : null + } catch { + return null + } +} + +function closedOwnersTableExists(db: Database.Database): boolean { + return ( + db + .prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?") + .get('agent_session_closed_owners') !== undefined + ) +} + +/** + * Created at EVERY writable open with no `user_version` bump (see `ensureQueuedMessagesTable`), and + * run inside one transaction: no older build creates the table, so its absence is what says the + * one-time migration has not run. Running it again would bring back a fact a receipt retired. + */ +export function ensureAgentSessionClosedOwnersTable(db: Database.Database): void { + const migrated = closedOwnersTableExists(db) + db.exec(` +CREATE TABLE IF NOT EXISTS agent_session_closed_owners ( + owner_key TEXT PRIMARY KEY, + session_id TEXT NOT NULL REFERENCES agent_session_records(session_id) ON DELETE CASCADE, + fact_json TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS agent_session_closed_owners_session + ON agent_session_closed_owners (session_id); +`) + if (!migrated) { + migrateAgentSessionClosedOwnerRows(db) + } +} + +export function loadAgentSessionClosedOwnerRows( + db: Database.Database +): Map { + const facts = new Map() + if (!closedOwnersTableExists(db)) { + return facts + } + for (const row of db + .prepare( + 'SELECT owner_key, session_id, fact_json FROM agent_session_closed_owners ORDER BY rowid' + ) + .all()) { + const fact = parsed(row.fact_json) + if ( + typeof row.owner_key === 'string' && + isReadableAgentSessionClosedOwner(row.owner_key, fact) && + row.session_id === fact.sessionId + ) { + facts.set(row.owner_key, fact) + } + } + return facts +} + +export function writeAgentSessionClosedOwnerRows( + db: Database.Database, + writes: { upsert: [string, string][]; remove: string[] } +): void { + for (const [key, json] of writes.upsert) { + const fact = parsed(json) + if (!isReadableAgentSessionClosedOwner(key, fact)) { + throw new Error('agent_session_store_write_invalid') + } + db.prepare( + 'INSERT INTO agent_session_closed_owners (owner_key, session_id, fact_json) VALUES (?, ?, ?)' + ).run(key, fact.sessionId, json) + } + for (const key of writes.remove) { + db.prepare('DELETE FROM agent_session_closed_owners WHERE owner_key = ?').run(key) + } +} + +/** An older proof cannot recover a cleared process or past cutoff. */ +function migrateAgentSessionClosedOwnerRows(db: Database.Database): void { + for (const row of db.prepare('SELECT session_id, record_json FROM agent_session_records').all()) { + const stored = parsed(row.record_json) + if (!isPersistedAgentSessionRecord(stored) || stored.sessionId !== row.session_id) { + continue + } + const { record } = decodePersistedAgentSessionRecord(stored) + const evidence = record.lease.deathEvidence + const fact = + evidence?.ownerFence !== undefined && evidence.ownerFence <= record.lease.runtimeFence + ? closedAgentSessionOwner(record, evidence) + : null + if (fact) { + writeAgentSessionClosedOwnerRows(db, { + upsert: [[agentSessionClosedOwnerKey(fact), JSON.stringify(fact)]], + remove: [] + }) + } + } +} diff --git a/src/main/runtime/agent-session-closed-owner.ts b/src/main/runtime/agent-session-closed-owner.ts new file mode 100644 index 000000000000..e277d1c40f44 --- /dev/null +++ b/src/main/runtime/agent-session-closed-owner.ts @@ -0,0 +1,125 @@ +import { + agentSessionScopeKey, + isAgentSessionDeathEvidence, + isAgentSessionExecutionLocation, + isAgentSessionId, + type AgentSessionDeathEvidence, + type AgentSessionExecutionLocation, + type AgentSessionProcessIdentity, + type AgentSessionRecord +} from '../../shared/agent-session-record' + +/** A proven-ended owner generation whose leftover work is not yet settled, scoped to its execution + * host; it dies with its settlement's journal commit (receipt) or its chat record. */ +export type AgentSessionClosedOwner = { + schemaVersion: 1 + sessionId: string + location: AgentSessionExecutionLocation + deadOwnerFence: number + evidence: AgentSessionDeathEvidence & { ownerFence: number } + process: Omit | null +} + +export function agentSessionClosedOwnerKey( + owner: Pick +): string { + return JSON.stringify([ + agentSessionScopeKey(owner.location), + owner.sessionId, + owner.deadOwnerFence + ]) +} + +export function closedAgentSessionOwner( + record: AgentSessionRecord, + evidence: AgentSessionDeathEvidence +): AgentSessionClosedOwner | null { + if (evidence.ownerFence === undefined) { + return null + } + const owner = record.lease.runtimeFence === evidence.ownerFence ? record.lease.ownerProcess : null + return { + schemaVersion: 1, + sessionId: record.sessionId, + location: { + executionHostId: record.location.executionHostId, + wslDistro: record.location.wslDistro, + workspaceId: record.location.workspaceId, + workspaceKind: record.location.workspaceKind + }, + deadOwnerFence: evidence.ownerFence, + evidence: { + kind: evidence.kind, + detail: evidence.detail, + observedAt: evidence.observedAt, + ownerFence: evidence.ownerFence, + ...(evidence.lastProvenAliveAt === undefined + ? {} + : { lastProvenAliveAt: evidence.lastProvenAliveAt }), + ...(evidence.runtimeEnd === undefined ? {} : { runtimeEnd: evidence.runtimeEnd }) + }, + process: owner + ? { + hostId: owner.hostId, + pid: owner.pid, + processStartTimeMs: owner.processStartTimeMs, + ...(owner.runtime ? { runtime: owner.runtime } : {}) + } + : null + } +} + +function natural(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 +} + +function bounded(value: unknown): value is string { + return typeof value === 'string' && value.length > 0 && value.length <= 512 +} + +function isClosedOwnerProcess(value: unknown): value is AgentSessionClosedOwner['process'] { + return ( + value === null || + (typeof value === 'object' && + 'hostId' in value && + bounded(value.hostId) && + 'pid' in value && + natural(value.pid) && + value.pid > 0 && + 'processStartTimeMs' in value && + (value.processStartTimeMs === null || natural(value.processStartTimeMs)) && + (!('runtime' in value) || bounded(value.runtime)) && + Object.keys(value).every((key) => + ['hostId', 'pid', 'processStartTimeMs', 'runtime'].includes(key) + )) + ) +} + +export function isReadableAgentSessionClosedOwner( + key: string, + value: unknown +): value is AgentSessionClosedOwner { + return ( + typeof value === 'object' && + value !== null && + 'schemaVersion' in value && + value.schemaVersion === 1 && + 'sessionId' in value && + isAgentSessionId(value.sessionId) && + 'location' in value && + isAgentSessionExecutionLocation(value.location) && + 'deadOwnerFence' in value && + natural(value.deadOwnerFence) && + 'evidence' in value && + isAgentSessionDeathEvidence(value.evidence) && + value.evidence.ownerFence === value.deadOwnerFence && + 'process' in value && + isClosedOwnerProcess(value.process) && + key === + agentSessionClosedOwnerKey({ + sessionId: value.sessionId, + location: value.location, + deadOwnerFence: value.deadOwnerFence + }) + ) +} diff --git a/src/main/runtime/agent-session-reconciliation-target.ts b/src/main/runtime/agent-session-reconciliation-target.ts index 03fd77be7dd9..f51883bcaa28 100644 --- a/src/main/runtime/agent-session-reconciliation-target.ts +++ b/src/main/runtime/agent-session-reconciliation-target.ts @@ -1,5 +1,18 @@ import { isDeepStrictEqual } from 'node:util' -import type { AgentSessionRecord } from '../../shared/agent-session-record' +import type { + AgentSessionProcessIdentity, + AgentSessionRecord +} from '../../shared/agent-session-record' +import { agentSessionRefusalError } from '../../shared/agent-session-wire-refusals' + +export function assertAgentSessionProbedOwner( + current: AgentSessionRecord, + probedOwner: AgentSessionProcessIdentity | null | undefined +): void { + if (probedOwner !== undefined && !isDeepStrictEqual(current.lease.ownerProcess, probedOwner)) { + throw agentSessionRefusalError('agent_session_checkpoint_stale', { reason: 'leaseMoved' }) + } +} export function agentSessionReconciliationTargetMatches( current: AgentSessionRecord, diff --git a/src/main/runtime/agent-session-record-rows.ts b/src/main/runtime/agent-session-record-rows.ts index ecc9fd9d72c3..e10536915773 100644 --- a/src/main/runtime/agent-session-record-rows.ts +++ b/src/main/runtime/agent-session-record-rows.ts @@ -21,6 +21,10 @@ import { isReadableRetiredAgentSessionClaimKey } from './agent-session-store-row-rules' import { AgentSessionTabTable, type PersistedAgentSessionTab } from './agent-session-tab-table' +import { + loadAgentSessionClosedOwnerRows, + writeAgentSessionClosedOwnerRows +} from './agent-session-closed-owner-rows' /** "Never recorded" and "recorded, now empty" differ, and #17439's legacy fallback needs the first; * no row count can say which. */ @@ -81,6 +85,7 @@ function unreadableRecordReason(value: unknown): string { export function loadAgentSessionStoreRows(db: Database.Database): AgentSessionStoreState { const state: AgentSessionStoreState = { records: new Map(), + closedOwners: loadAgentSessionClosedOwnerRows(db), operations: new Map(), retiredClaimKeys: [], unreadableRecords: new Map(), @@ -192,6 +197,7 @@ export function writeAgentSessionStoreRows( for (const sessionId of writes.records.remove) { db.prepare('DELETE FROM agent_session_records WHERE session_id = ?').run(sessionId) } + writeAgentSessionClosedOwnerRows(db, writes.closedOwners) const upsertOperation = db.prepare( 'INSERT INTO agent_session_operations (operation_key, row_json) VALUES (?, ?) ON CONFLICT(operation_key) DO UPDATE SET row_json = excluded.row_json' ) diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index 1b4dd7bd9d60..114ccbe192bf 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -33,6 +33,7 @@ import { agentSessionScopeKey, type AgentSessionExecutionLocation, type AgentSessionOptionsReplacement, + type AgentSessionProcessIdentity, type AgentSessionRecord } from '../../shared/agent-session-record' import { @@ -78,6 +79,12 @@ import { type CompareAndSetConversationName, setAgentSessionRecordConversationName } from './agent-session-record-conversation-name' +import { + agentSessionClosedOwnerKey, + type AgentSessionClosedOwner +} from './agent-session-closed-owner' +import { AgentSessionStoreNotifications } from './agent-session-store-notifications' +import { assertAgentSessionProbedOwner } from './agent-session-reconciliation-target' type AgentSessionOperationSettlement = Parameters[1] @@ -85,8 +92,7 @@ export const AGENT_SESSION_LEASE_TTL_MS = 30_000, AGENT_SESSION_LEASE_RENEW_INTERVAL_MS = 10_000 export class AgentSessionRecordStore { - private readonly deathEvidenceListeners = new Set<(sessionId: string) => void>() - private readonly firstRecordListeners = new Set<() => void>() + private readonly notifications = new AgentSessionStoreNotifications() readonly conversationReceipts: ReturnType private constructor( @@ -120,6 +126,11 @@ export class AgentSessionRecordStore { listRecords = (): AgentSessionRecord[] => [...this.state.records.values()] + closedOwners = (sessionId: string): AgentSessionClosedOwner[] => + [...this.state.closedOwners.values()] + .filter((fact) => fact.sessionId === sessionId) + .sort((left, right) => left.deadOwnerFence - right.deadOwnerFence) + /** Every chat this host holds a row for, readable or not. */ listHeldSessionIds = (): string[] => heldAgentSessionIds(this.state) @@ -247,9 +258,13 @@ export class AgentSessionRecordStore { sessionId: string expectedFence: number probe: AgentSessionOwnerProbe + probedOwner?: AgentSessionProcessIdentity | null now: number }): Promise { - return this.mutate(args.sessionId, (record) => evictAgentSessionOwner({ ...args, record })) + return this.mutate(args.sessionId, (record) => { + assertAgentSessionProbedOwner(record, args.probedOwner) + return evictAgentSessionOwner({ ...args, record }) + }) } async transitionHandoff( @@ -260,15 +275,14 @@ export class AgentSessionRecordStore { } /** - * Adjudicate every lease this host loaded. No lease grants a writer until it appears here. On a - * database a newer Orca wrote, the verdicts are kept in memory only: they are re-derived at every + * Adjudicate loaded leases, or only the addressed session. No lease grants a writer until it + * appears here. On a newer database, verdicts stay in memory only: they are re-derived at every * start, and none grants a writer there, since every grant is a write. */ async reconcileOnRestart( args: AgentSessionRestartProbeArgs ): Promise> { - const pending = this.listRecords().filter((record) => record.lease.unreconciled) - const probes = await collectAgentSessionRestartProbes(pending, args) + const probes = await collectAgentSessionRestartProbes(this, args) return this.transact((draft) => applyAgentSessionRestartProbes(draft, probes, args.now), { inMemoryWhenReadOnly: true }) @@ -315,6 +329,12 @@ export class AgentSessionRecordStore { operationOutcomeReceipt = (args: AgentSessionOperationSettlement): JournalOperationReceipt => this.transactions.receipt((draft) => settleAgentSessionOperationInto(draft, args)) + /** Retires a closed owner in the same transaction as the journal write that settles its work. */ + closedOwnerReceipt = (fact: AgentSessionClosedOwner): JournalOperationReceipt => + this.transactions.receipt((draft) => { + draft.closedOwners.delete(agentSessionClosedOwnerKey(fact)) + }) + replaceSessionOptions = (args: AgentSessionOptionsReplacement): Promise => this.mutate(args.sessionId, (record) => replaceAgentSessionRecordOptions(record, args)) @@ -339,47 +359,25 @@ export class AgentSessionRecordStore { }) } - /** Told, once committed, of each session a transaction wrote a proof of death for — whichever - * transition wrote it, since every one lands here. Must not throw. */ - onDeathEvidence(listener: (sessionId: string) => void): () => void { - this.deathEvidenceListeners.add(listener) - return () => this.deathEvidenceListeners.delete(listener) - } + /** Existing proof notifications carry the committed fact without changing their timing. */ + onDeathEvidence = this.notifications.onDeathEvidence - /** Told, once committed, when the store records its first chat. Must not throw. */ - onFirstRecord(listener: () => void): () => void { - this.firstRecordListeners.add(listener) - return () => this.firstRecordListeners.delete(listener) - } + /** Includes fresh reservation proofs as well as every other committed closure. */ + onClosedOwner = this.notifications.onClosedOwner + + onFirstRecord = this.notifications.onFirstRecord - /** Serializes every mutation. `apply` changes only the draft it is given; readers see the change - * once its rows have committed. */ private transact = async ( apply: (draft: AgentSessionStoreState) => T, options?: { inMemoryWhenReadOnly?: boolean } ): Promise => { - let proven: string[] = [] - let heldBefore = true + let publish: (durable: boolean) => void = () => {} const result = await this.transactions.transact((draft) => { - heldBefore = this.holdsRecords() - if (this.deathEvidenceListeners.size === 0) { - return apply(draft) - } - const before = new Map( - [...draft.records].map(([id, record]) => [id, record.lease.deathEvidence]) - ) - const applied = apply(draft) - proven = [...draft.records] - .filter(([id, { lease }]) => lease.deathEvidence && lease.deathEvidence !== before.get(id)) - .map(([id]) => id) - return applied + const staged = this.notifications.stage(draft, apply) + publish = staged.publish + return staged.result }, options) - for (const sessionId of proven) { - this.deathEvidenceListeners.forEach((listener) => listener(sessionId)) - } - if (!heldBefore && this.holdsRecords()) { - this.firstRecordListeners.forEach((listener) => listener()) - } + publish(!this.readOnly) return result } } diff --git a/src/main/runtime/agent-session-reservation-admission.test.ts b/src/main/runtime/agent-session-reservation-admission.test.ts index 3602aa29cd74..796980a57b7e 100644 --- a/src/main/runtime/agent-session-reservation-admission.test.ts +++ b/src/main/runtime/agent-session-reservation-admission.test.ts @@ -68,6 +68,7 @@ function reserveRequest( function storeState(records: readonly AgentSessionRecord[] = []): AgentSessionStoreState { return { records: new Map(records.map((record) => [record.sessionId, record])), + closedOwners: new Map(), operations: new Map(), retiredClaimKeys: [], unreadableRecords: new Map(), diff --git a/src/main/runtime/agent-session-reservation-admission.ts b/src/main/runtime/agent-session-reservation-admission.ts index ff840c755b35..3d426b9591de 100644 --- a/src/main/runtime/agent-session-reservation-admission.ts +++ b/src/main/runtime/agent-session-reservation-admission.ts @@ -32,6 +32,7 @@ import { type AgentSessionExecutionLocation, type AgentSessionLaunchArgs, type AgentSessionLaunchEnv, + type AgentSessionProcessIdentity, type AgentSessionRecord } from '../../shared/agent-session-record' import { isAgentSessionLaunchArgs } from '../../shared/agent-session-launch-args' @@ -48,6 +49,8 @@ import { import type { AgentSessionStoreState } from './agent-session-store-state' import { agentSessionRecordIdentityFields } from './agent-session-record-founding' import { agentSessionAccountHomesEqual } from '../../shared/agent-session-account-home' +import { retainAgentSessionReservationClosure } from './agent-session-closed-owner-history' +import { assertAgentSessionProbedOwner } from './agent-session-reconciliation-target' export type AgentSessionReserveRequest = { /** Host-resolved floating directory committed with the first owner reservation. */ @@ -75,6 +78,8 @@ export type AgentSessionReserveRequest = { claimKeyId: string handoffOperationId: string | null probe: AgentSessionOwnerProbe + /** Host-only snapshot taken before the asynchronous owner probe. */ + probedOwner?: AgentSessionProcessIdentity | null operation: { callerKey: string; operationId: string; fingerprint: string } now: number leaseTtlMs?: number @@ -202,17 +207,22 @@ export function applyAgentSessionReservation( throw agentSessionRefusalError('agent_session_conflict', { reason: 'sessionExists' }) } assertReservedTabUnheld(state, request) + assertAgentSessionProbedOwner(existing, request.probedOwner) const pinned = { ...existing, ...(!existing.launchArgs && request.launchArgs ? { launchArgs: [...request.launchArgs] } : {}), ...(!existing.launchArgs && request.launchArgs ? { updatedAt: request.now } : {}) } - return reserveAgentSessionOwner({ + const result = reserveAgentSessionOwner({ record: pinned, expectedFence: request.expectedFence ?? existing.lease.runtimeFence, probe: request.probe, reservation }) + if (result.disposition === 'reserved') { + retainAgentSessionReservationClosure(state, existing, request.probe, request.now) + } + return result } /** diff --git a/src/main/runtime/agent-session-restart-reconciliation.ts b/src/main/runtime/agent-session-restart-reconciliation.ts index 3e05914c42e3..e864af337c25 100644 --- a/src/main/runtime/agent-session-restart-reconciliation.ts +++ b/src/main/runtime/agent-session-restart-reconciliation.ts @@ -6,6 +6,7 @@ import { agentSessionReconciliationTargetMatches } from './agent-session-reconci import { applyAgentSessionRestartAdjudication } from './agent-session-restart-lease-transitions' export type AgentSessionRestartProbeArgs = { + sessionId?: string probe: (record: AgentSessionRecord) => Promise probeMany?: ( records: readonly AgentSessionRecord[] @@ -16,10 +17,19 @@ export type AgentSessionRestartProbeArgs = { type RestartProbe = { record: AgentSessionRecord; probe: AgentSessionOwnerProbe } export async function collectAgentSessionRestartProbes( - records: readonly AgentSessionRecord[], + store: { + listRecords: () => AgentSessionRecord[] + getRecord: (sessionId: string) => AgentSessionRecord | null + }, args: AgentSessionRestartProbeArgs ): Promise> { const probes = new Map() + const target = args.sessionId === undefined ? null : store.getRecord(args.sessionId) + const candidates = args.sessionId === undefined ? store.listRecords() : target ? [target] : [] + const records = candidates.filter((record) => record.lease.unreconciled) + if (records.length === 0) { + return probes + } const batched = args.probeMany ? await args.probeMany(records) : null for (const record of records) { probes.set(record.sessionId, { diff --git a/src/main/runtime/agent-session-store-draft.ts b/src/main/runtime/agent-session-store-draft.ts index ac7e13d63772..207e05fc2bb9 100644 --- a/src/main/runtime/agent-session-store-draft.ts +++ b/src/main/runtime/agent-session-store-draft.ts @@ -15,12 +15,14 @@ import { isReadableRetiredAgentSessionClaimKey } from './agent-session-store-row-rules' import type { PersistedAgentSessionTab } from './agent-session-tab-table' +import { isReadableAgentSessionClosedOwner } from './agent-session-closed-owner' /** Unreadable rows are derived at load and never written, so the draft shares them. */ export function draftAgentSessionStoreState(state: AgentSessionStoreState): AgentSessionStoreState { return { ...state, records: new Map(state.records), + closedOwners: new Map(state.closedOwners), operations: new Map(state.operations), retiredClaimKeys: [...state.retiredClaimKeys], sessionTabs: state.sessionTabs?.clone() ?? null @@ -32,6 +34,7 @@ type KeyedRowWrites = { upsert: [key: string, json: string][]; remove: string[] /** Exactly the rows a draft changed, each serialized as it will be stored. */ export type AgentSessionStoreRowWrites = { records: KeyedRowWrites + closedOwners: KeyedRowWrites operations: KeyedRowWrites /** The whole list when it changed, else null. */ retiredClaimKeys: RetiredAgentSessionClaimKey[] | null @@ -96,6 +99,11 @@ export function agentSessionStoreDraftRowWrites( const operations = serializeChangedRows(published.operations, draft.operations, (key, written) => isReadableAgentSessionStoreOperation(key, written) ) + const closedOwners = serializeChangedRows( + published.closedOwners, + draft.closedOwners, + isReadableAgentSessionClosedOwner + ) const retiredClaimKeys = retiredClaimKeysChanged( published.retiredClaimKeys, draft.retiredClaimKeys @@ -124,6 +132,8 @@ export function agentSessionStoreDraftRowWrites( const changed = records.upsert.length > 0 || records.remove.length > 0 || + closedOwners.upsert.length > 0 || + closedOwners.remove.length > 0 || operations.upsert.length > 0 || operations.remove.length > 0 || retiredClaimKeys !== null || @@ -133,6 +143,7 @@ export function agentSessionStoreDraftRowWrites( } return { records, + closedOwners, operations, retiredClaimKeys, sessionTabs: tabsChanged ? { recorded: draft.sessionTabs !== null, tabs } : null diff --git a/src/main/runtime/agent-session-store-notifications.ts b/src/main/runtime/agent-session-store-notifications.ts new file mode 100644 index 000000000000..40320c4bf1ec --- /dev/null +++ b/src/main/runtime/agent-session-store-notifications.ts @@ -0,0 +1,79 @@ +import { + agentSessionClosedOwnerKey, + type AgentSessionClosedOwner +} from './agent-session-closed-owner' +import type { AgentSessionStoreState } from './agent-session-store-state' + +type DeathEvidenceListener = (sessionId: string, fact: AgentSessionClosedOwner | null) => void + +/** Stages notifications with the draft and publishes their captured facts only after commit. */ +export class AgentSessionStoreNotifications { + private readonly deathEvidenceListeners = new Set() + private readonly closedOwnerListeners = new Set<(fact: AgentSessionClosedOwner) => void>() + private readonly firstRecordListeners = new Set<() => void>() + + onDeathEvidence = (listener: DeathEvidenceListener): (() => void) => { + this.deathEvidenceListeners.add(listener) + return () => this.deathEvidenceListeners.delete(listener) + } + + onClosedOwner = (listener: (fact: AgentSessionClosedOwner) => void): (() => void) => { + this.closedOwnerListeners.add(listener) + return () => this.closedOwnerListeners.delete(listener) + } + + onFirstRecord = (listener: () => void): (() => void) => { + this.firstRecordListeners.add(listener) + return () => this.firstRecordListeners.delete(listener) + } + + stage( + draft: AgentSessionStoreState, + apply: (draft: AgentSessionStoreState) => T + ): { result: T; publish: (durable: boolean) => void } { + const heldBefore = draft.records.size > 0 || draft.unreadableRecords.size > 0 + const beforeClosed = + this.closedOwnerListeners.size > 0 ? new Set(draft.closedOwners.keys()) : null + const beforeEvidence = + this.deathEvidenceListeners.size > 0 + ? new Map([...draft.records].map(([id, record]) => [id, record.lease.deathEvidence])) + : null + const result = apply(draft) + const proven = beforeEvidence + ? [...draft.records] + .filter( + ([id, { lease }]) => + lease.deathEvidence && lease.deathEvidence !== beforeEvidence.get(id) + ) + .map(([sessionId, record]) => ({ + sessionId, + key: + record.lease.deathEvidence?.ownerFence === undefined + ? null + : agentSessionClosedOwnerKey({ + ...record, + deadOwnerFence: record.lease.deathEvidence.ownerFence + }) + })) + : [] + return { + result, + publish: (durable) => { + for (const { sessionId, key } of proven) { + const fact = key === null ? null : (draft.closedOwners.get(key) ?? null) + this.deathEvidenceListeners.forEach((listener) => listener(sessionId, fact)) + } + if (durable && beforeClosed) { + for (const [key, fact] of draft.closedOwners) { + if (!beforeClosed.has(key)) { + this.closedOwnerListeners.forEach((listener) => listener(fact)) + } + } + } + if (!heldBefore && (draft.records.size > 0 || draft.unreadableRecords.size > 0)) { + this.firstRecordListeners.forEach((listener) => listener()) + } + } + } + } +} diff --git a/src/main/runtime/agent-session-store-state.ts b/src/main/runtime/agent-session-store-state.ts index cee15a96ac49..2f04dea849a8 100644 --- a/src/main/runtime/agent-session-store-state.ts +++ b/src/main/runtime/agent-session-store-state.ts @@ -5,11 +5,13 @@ import type { AgentSessionOperationRow } from '../../shared/agent-session-operat import type { AgentSessionRecord } from '../../shared/agent-session-record' import type { AgentSessionOrcaStopCause } from '../../shared/agent-session-orca-stop' import type { AgentSessionTabTable } from './agent-session-tab-table' +import type { AgentSessionClosedOwner } from './agent-session-closed-owner' export type RetiredAgentSessionClaimKey = { keyId: string; retiredAt: number } export type AgentSessionStoreState = { records: Map + closedOwners: Map operations: Map retiredClaimKeys: RetiredAgentSessionClaimKey[] /** Rows this build cannot validate, kept with a durable refusal reason. */ diff --git a/src/main/runtime/agent-session-store-transactions.ts b/src/main/runtime/agent-session-store-transactions.ts index 3877da33af42..0f0d59257040 100644 --- a/src/main/runtime/agent-session-store-transactions.ts +++ b/src/main/runtime/agent-session-store-transactions.ts @@ -26,6 +26,7 @@ import { draftAgentSessionStoreState, type AgentSessionStoreRowWrites } from './agent-session-store-draft' +import { captureAgentSessionClosedOwners } from './agent-session-closed-owner-history' // Why: rows are diffed by identity, so a row changed in place would never be written. Tests and // development builds make that a TypeError; packaged builds skip the walk. @@ -56,6 +57,12 @@ function freezeRows( for (const key of operations) { deepFreeze(state.operations.get(key)) } + const closedOwners = writes + ? writes.closedOwners.upsert.map(([key]) => key) + : state.closedOwners.keys() + for (const key of closedOwners) { + deepFreeze(state.closedOwners.get(key)) + } if (!writes || writes.retiredClaimKeys) { state.retiredClaimKeys.forEach(deepFreeze) } @@ -134,7 +141,7 @@ export class AgentSessionStoreTransactions { if (this.journalDatabase.readOnly) { throw readOnlyStoreRefusal() } - staged = this.stage(apply) + staged = this.stage(apply, true) const writes = staged.writes if (writes) { writeAgentSessionStoreRows(db, writes) @@ -152,7 +159,7 @@ export class AgentSessionStoreTransactions { if (readOnly && !inMemoryWhenReadOnly) { throw readOnlyStoreRefusal() } - const staged = this.stage(apply) + const staged = this.stage(apply, !readOnly) const writes = staged.writes if (writes && !readOnly) { this.journalDatabase.transaction((db) => writeAgentSessionStoreRows(db, writes)) @@ -161,11 +168,18 @@ export class AgentSessionStoreTransactions { return staged.result } - private stage(apply: (draft: AgentSessionStoreState) => T): StagedStoreTransaction { + /** Only a write that lands records closed owners: a read-only verdict lives in memory only. */ + private stage( + apply: (draft: AgentSessionStoreState) => T, + persisted: boolean + ): StagedStoreTransaction { const published = this.published const draft = draftAgentSessionStoreState(published) const result = apply(draft) attributeAgentSessionRuntime(published, draft) + if (persisted) { + captureAgentSessionClosedOwners(published, draft) + } const writes = agentSessionStoreDraftRowWrites(published, draft) return { result, diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index 2ff04f88ad4f..3ae2acdb4046 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -497,6 +497,9 @@ export type AgentJournalSubmission = { handoverRecorded?: true /** When the host handed it to the provider (its `dispatch{pending}` row). */ handedOverAt?: number + /** Host-only, for restart reconciliation: its first handover's fence and row time, recomputed on + * every fold. The snapshot still carries it, but the submission schema omits it and no client reads it. */ + firstHandover?: { fence: number; at: number } /** Host-only: the submission row's sequence, which tells which host process accepted it. Set * only on a send accepted for later handover. The snapshot still carries it, but the submission * schema omits it; no released client reads it, and clients read `submittedSequence` instead. */ diff --git a/src/shared/agent-session-record.ts b/src/shared/agent-session-record.ts index 185c62c8333e..3101d07f6684 100644 --- a/src/shared/agent-session-record.ts +++ b/src/shared/agent-session-record.ts @@ -289,7 +289,7 @@ function isAgentSessionJournalCheckpoint(value: unknown): value is AgentSessionJ ) } -function isAgentSessionDeathEvidence(value: unknown): value is AgentSessionDeathEvidence { +export function isAgentSessionDeathEvidence(value: unknown): value is AgentSessionDeathEvidence { if (typeof value !== 'object' || value === null) { return false }