diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts b/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts new file mode 100644 index 00000000000..5b84b8417e5 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts @@ -0,0 +1,229 @@ +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 { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import type { + AgentJournalItemBody, + AgentJournalItemIdentity, + AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import { AGENT_SESSION_HISTORY_MAX_LIMIT } from '../../../shared/agent-session-wire' +import { + openAgentSessionJournal, + type AgentSessionJournal +} from '../agent-session-journal/journal-store' +import { projectJournalBatch } from './agent-session-journal-batch' +import { readAgentSessionHistory, resolveHistoryLimit } from './agent-session-history-page' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: 'session-1', + workspaceId: 'ws-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } +} + +let root: string +let clock = 1_000 +let epochs = 0 +let journal: AgentSessionJournal + +function tick(): number { + clock += 1 + return clock +} + +function item(ordinal: number): AgentJournalItemIdentity { + return { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal } +} + +function body(text: string): AgentJournalItemBody { + return { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text }] } +} + +async function appendItems(count: number): Promise { + for (let ordinal = 1; ordinal <= count; ordinal += 1) { + await journal.appendItem(item(ordinal), body(`item-${ordinal}`), { fence: 1 }) + } +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-wire-history-')) + clock = 1_000 + epochs = 0 + journal = await openAgentSessionJournal({ + identity: IDENTITY, + journalDir: root, + now: tick, + mintEpoch: () => { + epochs += 1 + return `epoch-${epochs}` + } + }) +}) + +afterEach(async () => { + await rm(root, { recursive: true, force: true }) +}) + +describe('resolveHistoryLimit', () => { + it('clamps rather than rejecting so a mid-scroll client keeps paging', () => { + expect(resolveHistoryLimit(undefined)).toBe(40) + expect(resolveHistoryLimit(0)).toBe(1) + expect(resolveHistoryLimit(-5)).toBe(1) + expect(resolveHistoryLimit(10_000)).toBe(AGENT_SESSION_HISTORY_MAX_LIMIT) + expect(resolveHistoryLimit(Number.NaN)).toBe(40) + }) +}) + +describe('readAgentSessionHistory', () => { + it('serves the newest page on tail and pages backward from it', async () => { + await appendItems(5) + const tail = readAgentSessionHistory(journal, { + sessionId: 'session-1', + direction: 'tail', + limit: 2 + }) + if (!tail.ok) { + throw new Error(`expected a page, got reset ${tail.reset}`) + } + expect(tail.page.items.map((entry) => entry.body)).toEqual([body('item-4'), body('item-5')]) + expect(tail.page.hasOlder).toBe(true) + expect(tail.page.hasNewer).toBe(false) + + const older = readAgentSessionHistory(journal, { + sessionId: 'session-1', + direction: 'before', + cursor: tail.page.window.nextCursor, + limit: 2 + }) + if (!older.ok) { + throw new Error(`expected a page, got reset ${older.reset}`) + } + expect(older.page.items.map((entry) => entry.body)).toEqual([body('item-2'), body('item-3')]) + expect(older.page.hasNewer).toBe(true) + }) + + it('catches a live reader up from its cursor and stops at the limit', async () => { + await appendItems(2) + const cursor = journal.cursor() + await appendItems(5) + const page = readAgentSessionHistory(journal, { + sessionId: 'session-1', + direction: 'after', + cursor, + limit: 2 + }) + if (!page.ok) { + throw new Error(`expected a page, got reset ${page.reset}`) + } + expect(page.page.items).toHaveLength(2) + expect(page.page.hasNewer).toBe(true) + expect(page.page.window.nextCursor.sequence).toBeGreaterThan(cursor.sequence) + }) + + it('reports a forward read with no cursor as cursor_ahead rather than serving the tail', async () => { + await appendItems(1) + expect( + readAgentSessionHistory(journal, { sessionId: 'session-1', direction: 'after' }) + ).toMatchObject({ ok: false, reset: 'cursor_ahead' }) + }) + + it('resets a cursor from a previous epoch', async () => { + await appendItems(1) + const stale = journal.cursor() + await journal.rollEpoch('legacy_import', 2) + for (const direction of ['before', 'after'] as const) { + expect( + readAgentSessionHistory(journal, { sessionId: 'session-1', direction, cursor: stale }) + ).toMatchObject({ ok: false, reset: 'epoch_changed' }) + } + }) + + it('resets a cursor ahead of the journal', async () => { + await appendItems(1) + const ahead = { epoch: journal.epoch, sequence: journal.cursor().sequence + 10 } + expect( + readAgentSessionHistory(journal, { + sessionId: 'session-1', + direction: 'before', + cursor: ahead + }) + ).toMatchObject({ ok: false, reset: 'cursor_ahead' }) + }) + + it('carries the submission for a message on the page', async () => { + await journal.appendSubmission({ + clientMessageId: 'msg-1', + payloadFingerprint: 'a'.repeat(64), + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] }, + fence: 1 + }) + const page = readAgentSessionHistory(journal, { sessionId: 'session-1', direction: 'tail' }) + if (!page.ok) { + throw new Error(`expected a page, got reset ${page.reset}`) + } + expect(page.page.items[0]?.itemId).toBe(agentJournalSubmissionKey('msg-1')) + expect(page.page.submissions).toHaveLength(1) + expect(page.page.submissions[0]).toMatchObject({ + clientMessageId: 'msg-1', + dispatchState: 'pending' + }) + }) +}) + +describe('projectJournalBatch', () => { + it('reports a hole in the row sequence as journal_gap', async () => { + await appendItems(3) + const since = journal.readSince({ epoch: journal.epoch, sequence: 0 }) + if (!since.ok) { + throw new Error(`expected rows, got reset ${since.reset}`) + } + const withHole = since.rows.filter((row) => row.seq !== since.rows[1]?.seq) + expect( + projectJournalBatch({ rows: withHole, snapshot: journal.snapshot(), afterSequence: 0 }) + ).toEqual({ ok: false, reset: 'journal_gap' }) + }) + + it('publishes touched items at their current reduced state, not as a delta', async () => { + await appendItems(1) + const cursor = journal.cursor() + await journal.appendItem(item(1), body('revised'), { fence: 1 }) + const since = journal.readSince(cursor) + if (!since.ok) { + throw new Error(`expected rows, got reset ${since.reset}`) + } + const projected = projectJournalBatch({ + rows: since.rows, + snapshot: journal.snapshot(), + afterSequence: cursor.sequence + }) + if (!projected.ok) { + throw new Error(`expected a batch, got reset ${projected.reset}`) + } + expect(projected.batch.items).toHaveLength(1) + expect(projected.batch.items[0]).toMatchObject({ body: body('revised'), revision: 2 }) + expect(projected.batch.cursor).toEqual(journal.cursor()) + }) + + it('lists a tombstoned item as removed', async () => { + await appendItems(1) + const cursor = journal.cursor() + await journal.appendTombstone(item(1), { fence: 1 }) + const since = journal.readSince(cursor) + if (!since.ok) { + throw new Error(`expected rows, got reset ${since.reset}`) + } + const projected = projectJournalBatch({ + rows: since.rows, + snapshot: journal.snapshot(), + afterSequence: cursor.sequence + }) + if (!projected.ok) { + throw new Error(`expected a batch, got reset ${projected.reset}`) + } + expect(projected.batch.removedItemIds).toHaveLength(1) + expect(projected.batch.items).toHaveLength(0) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-page.ts b/src/main/native-chat/agent-session-wire/agent-session-history-page.ts new file mode 100644 index 00000000000..c3c621bb1f0 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-history-page.ts @@ -0,0 +1,143 @@ +// Paged history over one journal. +// +// `tail` and `before` read the REDUCED timeline, so backward paging keeps +// working after compaction — the folded snapshot still holds every live item. +// `after` is the catch-up direction and must read rows instead: an item created +// early and revised late orders by its creation sequence, so an item-window +// read would silently skip that revision. Rows carry the revision, which is why +// `after` is the only direction that can answer `cursor_compacted`. + +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import type { + AgentJournalCursor, + AgentJournalRenderItem, + AgentJournalSnapshot +} from '../../../shared/agent-session-journal-types' +import { + AGENT_SESSION_HISTORY_DEFAULT_LIMIT, + AGENT_SESSION_HISTORY_MAX_LIMIT, + type AgentSessionHistoryDirection, + type AgentSessionHistoryPage, + type AgentSessionHistoryRequest, + type AgentSessionHistoryResult +} from '../../../shared/agent-session-wire' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { projectJournalBatch } from './agent-session-journal-batch' + +/** Clamped, never rejected: a client asking for more than the host will serve + * should get a smaller page and keep paging, not an error mid-scroll. */ +export function resolveHistoryLimit(limit: number | undefined): number { + if (limit === undefined || !Number.isFinite(limit)) { + return AGENT_SESSION_HISTORY_DEFAULT_LIMIT + } + return Math.min(AGENT_SESSION_HISTORY_MAX_LIMIT, Math.max(1, Math.floor(limit))) +} + +export function readAgentSessionHistory( + journal: AgentSessionJournal, + request: AgentSessionHistoryRequest +): AgentSessionHistoryResult { + const snapshot = journal.snapshot() + if (journal.isReadOnly) { + return { ok: false, reset: 'schema_unreadable', snapshot } + } + const limit = resolveHistoryLimit(request.limit) + if (request.direction === 'after') { + return readForward(journal, snapshot, request.cursor, limit) + } + const cursor = request.direction === 'before' ? request.cursor : undefined + if (cursor) { + if (cursor.epoch !== snapshot.cursor.epoch) { + return { ok: false, reset: 'epoch_changed', snapshot } + } + if (cursor.sequence > snapshot.cursor.sequence) { + return { ok: false, reset: 'cursor_ahead', snapshot } + } + } + const older = cursor + ? snapshot.items.filter((item) => item.sequence < cursor.sequence) + : snapshot.items + const items = older.slice(Math.max(0, older.length - limit)) + return { + ok: true, + page: buildPage({ + snapshot, + direction: request.direction, + items, + hasOlder: older.length > items.length, + hasNewer: older.length < snapshot.items.length, + fallbackCursor: cursor ?? { epoch: snapshot.cursor.epoch, sequence: 0 }, + nextCursor: items[0] + ? { epoch: snapshot.cursor.epoch, sequence: items[0].sequence } + : undefined + }) + } +} + +function readForward( + journal: AgentSessionJournal, + snapshot: AgentJournalSnapshot, + cursor: AgentJournalCursor | undefined, + limit: number +): AgentSessionHistoryResult { + if (!cursor) { + // Why: forward paging replays rows after a position; without one there is + // nothing to be after, and silently serving the tail would hand the client + // a page it cannot place. + return { ok: false, reset: 'cursor_ahead', snapshot } + } + const since = journal.readSince(cursor) + if (!since.ok) { + return { ok: false, reset: since.reset, snapshot } + } + const rows = since.rows.slice(0, limit) + const projected = projectJournalBatch({ rows, snapshot, afterSequence: cursor.sequence }) + if (!projected.ok) { + return { ok: false, reset: projected.reset, snapshot } + } + const lastSequence = rows.at(-1)?.seq ?? cursor.sequence + return { + ok: true, + page: buildPage({ + snapshot, + direction: 'after', + items: projected.batch.items, + // Reading after a position means there is something before it. + hasOlder: cursor.sequence > 0, + hasNewer: since.rows.length > rows.length, + fallbackCursor: cursor, + nextCursor: { epoch: cursor.epoch, sequence: lastSequence } + }) + } +} + +function buildPage(input: { + snapshot: AgentJournalSnapshot + direction: AgentSessionHistoryDirection + items: AgentJournalRenderItem[] + hasOlder: boolean + hasNewer: boolean + fallbackCursor: AgentJournalCursor + nextCursor: AgentJournalCursor | undefined +}): AgentSessionHistoryPage { + const epoch = input.snapshot.cursor.epoch + const pageItemIds = new Set(input.items.map((item) => item.itemId)) + const oldest = input.items[0] + const newest = input.items.at(-1) + return { + sessionId: input.snapshot.sessionId, + epoch, + direction: input.direction, + items: input.items, + submissions: input.snapshot.submissions.filter((submission) => + pageItemIds.has(agentJournalSubmissionKey(submission.clientMessageId)) + ), + window: { + oldest: oldest ? { epoch, sequence: oldest.sequence } : null, + newest: newest ? { epoch, sequence: newest.sequence } : null, + nextCursor: input.nextCursor ?? input.fallbackCursor + }, + hasOlder: input.hasOlder, + hasNewer: input.hasNewer + } +} diff --git a/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts b/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts new file mode 100644 index 00000000000..96a3c262c8b --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts @@ -0,0 +1,79 @@ +// Rows appended since a cursor, projected into what a subscriber must apply. +// +// The batch carries each touched item at its CURRENT reduced state rather than +// the raw rows: a client that applies the same batch twice converges, and a +// provider echo that was adopted into a submission slot arrives under the slot +// key instead of appearing as a second copy of the user's own message. + +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import type { AgentJournalSnapshot } from '../../../shared/agent-session-journal-types' +import type { AgentSessionJournalBatch } from '../../../shared/agent-session-wire' +import { findSequenceGap } from '../agent-session-journal/journal-cursor' +import type { JournalRow } from '../agent-session-journal/journal-row-schema' + +export type JournalBatchProjection = + | { ok: true; batch: AgentSessionJournalBatch } + /** A missing sequence means the tail lost a row; the subscriber reloads + * rather than rendering a timeline with a hole in it. */ + | { ok: false; reset: 'journal_gap' } + +export function projectJournalBatch(input: { + rows: readonly JournalRow[] + snapshot: AgentJournalSnapshot + /** Sequence the subscriber has already applied. */ + afterSequence: number +}): JournalBatchProjection { + const gap = findSequenceGap( + input.rows.map((row) => row.seq), + input.afterSequence + 1 + ) + if (gap) { + return { ok: false, reset: 'journal_gap' } + } + const aliases = submissionAliases(input.snapshot) + const touchedItemIds = new Set() + const touchedClientMessageIds = new Set() + for (const row of input.rows) { + if (row.kind === 'item' || row.kind === 'tombstone') { + touchedItemIds.add(aliases.get(row.itemId) ?? row.itemId) + continue + } + if (row.kind === 'submission' || row.kind === 'dispatch') { + touchedClientMessageIds.add(row.clientMessageId) + touchedItemIds.add(agentJournalSubmissionKey(row.clientMessageId)) + } + } + + const live = new Map(input.snapshot.items.map((item) => [item.itemId, item])) + const items = [...touchedItemIds] + .map((itemId) => live.get(itemId)) + .filter((item) => item !== undefined) + .sort((a, b) => a.sequence - b.sequence) + return { + ok: true, + batch: { + cursor: input.snapshot.cursor, + items, + removedItemIds: [...touchedItemIds].filter((itemId) => !live.has(itemId)), + submissions: input.snapshot.submissions.filter((submission) => + touchedClientMessageIds.has(submission.clientMessageId) + ) + } + } +} + +/** + * Provider item id → the submission slot that adopted it, rebuilt from the + * snapshot's own accepted submissions. This mirrors the alias the reducer + * writes on an accepted dispatch; deriving it here keeps the projection a pure + * function of published state instead of reaching into reducer internals. + */ +function submissionAliases(snapshot: AgentJournalSnapshot): Map { + const aliases = new Map() + for (const submission of snapshot.submissions) { + if (submission.dispatchState === 'accepted' && submission.providerItemId) { + aliases.set(submission.providerItemId, agentJournalSubmissionKey(submission.clientMessageId)) + } + } + return aliases +} diff --git a/src/main/native-chat/agent-session-wire/agent-session-journal-recovery.test.ts b/src/main/native-chat/agent-session-wire/agent-session-journal-recovery.test.ts new file mode 100644 index 00000000000..24a985719e1 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-journal-recovery.test.ts @@ -0,0 +1,175 @@ +// Recovery drives the real journal loader against real on-disk damage: a hole +// punched in the log, and a row stamped with a schema this host cannot read. + +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import type { + AgentJournalItemIdentity, + AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import { openAgentSessionJournal } from '../agent-session-journal/journal-store' +import { + openAgentSessionJournalWithRecovery, + providerHistoryId, + recoveryJournalDir +} from './agent-session-journal-recovery' + +const CODEX_SESSION = '019fd532-7c11-7a90-b6de-4e1a2c3d5f60' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: CODEX_SESSION, + workspaceId: 'ws-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: CODEX_SESSION } +} + +const CODEX_LINES = [ + { + type: 'session_meta', + timestamp: '2026-08-05T10:00:00.000Z', + payload: { + id: CODEX_SESSION, + session_id: CODEX_SESSION, + cwd: '/Users/dev/project', + originator: 'codex_cli_rs', + cli_version: '0.146.1' + } + }, + { + type: 'event_msg', + timestamp: '2026-08-05T10:00:02.000Z', + payload: { type: 'user_message', message: 'add a retry', kind: 'plain' } + }, + { + type: 'event_msg', + timestamp: '2026-08-05T10:00:05.000Z', + payload: { type: 'agent_message', message: 'On it.' } + } +] + +let root: string +let journalDir: string +let historyFilePath: string + +function item(ordinal: number): AgentJournalItemIdentity { + return { provider: 'codex', threadId: CODEX_SESSION, turnId: 'turn-1', ordinal } +} + +/** Fills a journal with `count` items and hands back the raw log lines. */ +async function seedJournal(count: number): Promise { + const journal = await openAgentSessionJournal({ identity: IDENTITY, journalDir }) + for (let ordinal = 1; ordinal <= count; ordinal += 1) { + await journal.appendItem( + item(ordinal), + { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: `item-${ordinal}` }] }, + { fence: 1 } + ) + } + const raw = await readFile(join(journalDir, 'log.jsonl'), 'utf-8') + return raw.split('\n').filter((line) => line.trim().length > 0) +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-wire-recovery-')) + journalDir = join(root, 'journal') + historyFilePath = join(root, 'rollout.jsonl') + await writeFile( + historyFilePath, + `${CODEX_LINES.map((line) => JSON.stringify(line)).join('\n')}\n`, + 'utf-8' + ) +}) + +afterEach(async () => { + await rm(root, { recursive: true, force: true }) +}) + +describe('providerHistoryId', () => { + it('uses the provider handle, never the Orca session id', () => { + expect(providerHistoryId({ kind: 'codex', threadId: 'thread-9' })).toBe('thread-9') + expect(providerHistoryId({ kind: 'claude', sessionId: 'sess-9', leafUuid: null })).toBe( + 'sess-9' + ) + }) +}) + +describe('openAgentSessionJournalWithRecovery', () => { + it('opens a healthy journal untouched', async () => { + await seedJournal(2) + const opened = await openAgentSessionJournalWithRecovery({ + identity: IDENTITY, + journalDir, + fence: 1, + historyFilePath + }) + expect(opened.recovery).toBeNull() + expect(opened.journal.snapshot().items).toHaveLength(2) + }) + + it('rebuilds a holed journal in place on a fresh epoch', async () => { + const lines = await seedJournal(3) + const holed = lines.filter((_line, index) => index !== 1) + await writeFile(join(journalDir, 'log.jsonl'), `${holed.join('\n')}\n`, 'utf-8') + + const opened = await openAgentSessionJournalWithRecovery({ + identity: IDENTITY, + journalDir, + fence: 1, + historyFilePath + }) + expect(opened.recovery).toMatchObject({ trigger: 'journal_corrupt', reset: 'epoch_changed' }) + expect(opened.recovery?.imported).toBeGreaterThan(0) + expect(opened.journal.isReadOnly).toBe(false) + // The rebuilt timeline is the only content of its epoch — nothing from the + // damaged prefix survives into it. + const texts = opened.journal.snapshot().items.map((entry) => JSON.stringify(entry.body)) + expect(texts.some((text) => text.includes('item-1'))).toBe(false) + expect(texts.some((text) => text.includes('add a retry'))).toBe(true) + }) + + it('reconstructs a future-schema journal into a schema-scoped sibling, never in place', async () => { + const lines = await seedJournal(1) + await writeFile( + join(journalDir, 'log.jsonl'), + `${lines.join('\n')}\n${JSON.stringify({ v: 99, seq: 2, epoch: 'e', kind: 'item' })}\n`, + 'utf-8' + ) + + const opened = await openAgentSessionJournalWithRecovery({ + identity: IDENTITY, + journalDir, + fence: 1, + historyFilePath + }) + expect(opened.recovery).toMatchObject({ + trigger: 'schema_unreadable', + reset: 'schema_unreadable' + }) + expect(opened.recovery?.imported).toBeGreaterThan(0) + + // The unreadable journal is left exactly as found; a newer host still owns it. + const untouched = await readFile(join(journalDir, 'log.jsonl'), 'utf-8') + expect(untouched).toContain('"v":99') + const sibling = await readFile(join(recoveryJournalDir(journalDir), 'log.jsonl'), 'utf-8') + expect(sibling).toContain('add a retry') + }) + + it('still opens the session when provider history cannot be read', async () => { + const lines = await seedJournal(3) + const holed = lines.filter((_line, index) => index !== 1) + await writeFile(join(journalDir, 'log.jsonl'), `${holed.join('\n')}\n`, 'utf-8') + + const opened = await openAgentSessionJournalWithRecovery({ + identity: IDENTITY, + journalDir, + fence: 1, + historyFilePath: join(root, 'missing.jsonl') + }) + expect(opened.recovery).toMatchObject({ trigger: 'journal_corrupt', imported: 0 }) + expect(opened.recovery?.error).toBeTruthy() + expect(opened.journal.snapshot().items).toHaveLength(0) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/agent-session-journal-recovery.ts b/src/main/native-chat/agent-session-wire/agent-session-journal-recovery.ts new file mode 100644 index 00000000000..14d09730005 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-journal-recovery.ts @@ -0,0 +1,116 @@ +// Journal recovery: rehydrate the timeline from provider history. +// +// Two triggers, and they need different destinations. A journal whose prefix is +// unusable is writable, so it is rebuilt in place on a fresh epoch. A journal +// written by a NEWER schema is not writable by this host at all — rebuilding it +// in place would fork the sequence space a newer host still owns — so the +// reconstruction goes to a schema-scoped sibling directory that is only ever +// written by hosts at this version and is never merged back. + +import type { AgentType } from '../../../shared/agent-status-types' +import { + AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + type AgentJournalResetReason, + type AgentSessionJournalIdentity, + type AgentSessionProviderHandle +} from '../../../shared/agent-session-journal-types' +import { importLegacyTranscriptIntoJournal } from '../agent-session-journal/journal-legacy-import' +import { loadJournal } from '../agent-session-journal/journal-open' +import { + openAgentSessionJournal, + type AgentSessionJournal +} from '../agent-session-journal/journal-store' + +export type AgentSessionJournalRecovery = { + trigger: 'journal_corrupt' | 'schema_unreadable' + /** What subscribers are told; both force a clean snapshot reload. */ + reset: AgentJournalResetReason + epoch: string + imported: number + /** Set when provider history could not be read. The session still opens, on + * an empty fresh epoch — an unreadable transcript is not a reason to refuse + * the user their session. */ + error?: string +} + +export type AgentSessionJournalOpened = { + journal: AgentSessionJournal + recovery: AgentSessionJournalRecovery | null +} + +/** Where a reconstruction lands when the real journal cannot be written. */ +export function recoveryJournalDir(journalDir: string): string { + return `${journalDir}-recovered-v${AGENT_SESSION_JOURNAL_SCHEMA_VERSION}` +} + +/** The provider's own session id, which is what the transcript readers index + * by — never the Orca session id. */ +export function providerHistoryId(handle: AgentSessionProviderHandle): string { + if (handle.kind === 'codex') { + return handle.threadId + } + return handle.kind === 'claude' ? handle.sessionId : handle.value +} + +export async function openAgentSessionJournalWithRecovery(input: { + identity: AgentSessionJournalIdentity + journalDir: string + fence: number + /** Resolve directly to a transcript instead of discovering it by session id. */ + historyFilePath?: string | null +}): Promise { + const probe = await loadJournal(input.journalDir, input.identity.sessionId) + if (probe?.readOnly) { + const journal = await openAgentSessionJournal({ + identity: input.identity, + journalDir: recoveryJournalDir(input.journalDir) + }) + return { + journal, + recovery: await rehydrate({ ...input, journal, trigger: 'schema_unreadable' }) + } + } + const journal = await openAgentSessionJournal({ + identity: input.identity, + journalDir: input.journalDir + }) + if (!probe?.corrupt) { + return { journal, recovery: null } + } + // `open()` already rolled the epoch off the unusable prefix; the import rolls + // once more so the rebuilt timeline is the only content of its epoch. + return { journal, recovery: await rehydrate({ ...input, journal, trigger: 'journal_corrupt' }) } +} + +async function rehydrate(input: { + identity: AgentSessionJournalIdentity + journal: AgentSessionJournal + fence: number + historyFilePath?: string | null + trigger: AgentSessionJournalRecovery['trigger'] +}): Promise { + const reset: AgentJournalResetReason = + input.trigger === 'schema_unreadable' ? 'schema_unreadable' : 'epoch_changed' + const result = await importLegacyTranscriptIntoJournal({ + journal: input.journal, + agent: input.identity.agent satisfies AgentType, + sessionId: providerHistoryId(input.identity.providerHandle), + fence: input.fence, + ...(input.historyFilePath ? { options: { filePath: input.historyFilePath } } : {}) + }) + if (!result.ok) { + return { + trigger: input.trigger, + reset, + epoch: input.journal.epoch, + imported: 0, + error: result.error + } + } + return { + trigger: input.trigger, + reset, + epoch: result.epoch, + imported: result.imported + } +} 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 new file mode 100644 index 00000000000..9b5c9dcc40b --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -0,0 +1,66 @@ +// What the wire needs from a provider adapter. +// +// Phase 2 implements this over the Codex app-server and the Claude Agent SDK; +// nothing here starts, resumes, or talks to a process. The wire owns the +// journal and the lease, so an adapter only has to answer "did the provider +// take this?" — and it answers `unknown` rather than guessing, because the +// journal renders that as delivery unconfirmed instead of as failure. + +import type { + AgentJournalItemIdentity, + AgentJournalMessageItem, + AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import type { AgentSessionProviderHandleLink } from '../../../shared/agent-session-provider-handle' +import type { AgentSessionProcessIdentity } from '../../../shared/agent-session-record' + +/** What a reservation turns into once something is actually running under it: + * the process the host can probe, and the provider handle it was minted with. */ +export type AgentSessionAcquisition = { + process: AgentSessionProcessIdentity + link: AgentSessionProviderHandleLink +} + +export type AgentSessionDispatchOutcome = + /** The provider owns the turn now, under this identity. */ + | { state: 'accepted'; providerIdentity: AgentJournalItemIdentity } + | { state: 'rejected'; reason: string } + /** The call did not settle. Never re-send on the user's behalf. */ + | { state: 'unknown'; reason: string } + +export type StructuredAgentSessionAdapter = { + /** Makes the reservation real. Called once per reservation, with the spawn + * token the lease was reserved under and the fence the handle must be minted + * at — the store rejects a link minted at any other fence. */ + acquire(input: { + identity: AgentSessionJournalIdentity + fence: number + spawnToken: string + }): Promise + dispatch(input: { + sessionId: string + clientMessageId: string + body: AgentJournalMessageItem + fence: number + }): Promise + /** Cancels one turn, not the session: a session-wide interrupt would also kill + * a turn the client never asked to stop. */ + cancelTurn(input: { + sessionId: string + turnId: string + fence: number + }): Promise<{ cancelled: boolean }> + /** Fires the provider callback for an approval or a question. The wire calls + * this only after the durable compare-and-set won, so it runs exactly once. */ + answerPrompt(input: { + sessionId: string + itemId: string + kind: 'approval' | 'question' + optionId: string + fence: number + }): Promise + setOption(input: { sessionId: string; key: string; value: string; fence: number }): Promise + /** Transcript path for journal recovery. Omit to let the existing session-file + * resolver discover it from the provider session id. */ + historyFilePath?(input: { identity: AgentSessionJournalIdentity }): Promise +} 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 new file mode 100644 index 00000000000..1d0292fefaf --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts @@ -0,0 +1,127 @@ +// The attach transition end to end: reserve the lease, make the reservation +// real, open the journal. +// +// Split out of the host so the sequence reads in one place. The host still owns +// the decisions that must not be client-supplied — the spawn token, the claim +// key, the owner probe — and passes them in. + +import type { + AgentSessionAttachResult, + AgentSessionMutationResult +} from '../../../shared/agent-session-wire' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { + admitAttachOrRefuse, + attachJournal, + classifyStoreFailure, + journalIdentityFor, + reserveRequestFor, + type AgentSessionAttachAuthority, + type AgentSessionAttachParams, + type AttachedJournal +} from './structured-agent-session-attach' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' + +export type AttachFlowInput = { + store: AgentSessionRecordStore + adapter: StructuredAgentSessionAdapter + journalRoot: string + authority: AgentSessionAttachAuthority + callerKey: string + params: AgentSessionAttachParams + now: () => number + /** Registers the opened journal and fans out to subscribers before the caller + * sees the result, so no client can send against a session the host has not + * finished publishing. */ + onAttached: (attached: AttachedJournal) => void +} + +export async function performAttach( + input: AttachFlowInput +): Promise> { + const { params, store } = input + const sessionId = params.envelope.sessionId + const admitted = admitAttachOrRefuse(params) + if (!admitted.ok) { + return admitted + } + + let record: AgentSessionRecord + let replayed = false + try { + const reserved = await store.reserveOwner( + reserveRequestFor({ + sessionId, + params, + authority: input.authority, + callerKey: input.callerKey, + fingerprint: admitted.fingerprint, + now: input.now() + }) + ) + record = reserved.record + replayed = reserved.disposition === 'replayed' + if (record.lease.ownerProcess === null) { + record = await acquireOwner(input, record) + } + } catch (error) { + return { + ok: false, + refusal: classifyStoreFailure(error, store.getRecord(sessionId)?.lease.runtimeFence ?? null) + } + } + + const attached = await attachJournal({ + record, + params, + journalRoot: input.journalRoot, + adapter: input.adapter + }) + input.onAttached(attached) + await store.recordOperationOutcome({ + callerKey: input.callerKey, + operationId: params.envelope.clientOperationId, + outcome: { status: 'succeeded', sessionId } + }) + + const fence = record.lease.runtimeFence + return { + ok: true, + replayed, + fence, + cursor: attached.journal.cursor(), + value: { + sessionId, + fence, + snapshot: attached.journal.snapshot(), + unconfirmedClientMessageIds: attached.unconfirmedClientMessageIds + } + } +} + +/** A reservation with no process behind it is only a promise to spawn; the + * adapter makes it real and the store then grants the writer. */ +async function acquireOwner( + input: AttachFlowInput, + record: AgentSessionRecord +): Promise { + const fence = record.lease.runtimeFence + const acquired = await input.adapter.acquire({ + identity: journalIdentityFor(record.sessionId, input.params), + fence, + spawnToken: input.authority.spawnToken + }) + await input.store.commitProcessIdentity({ + sessionId: record.sessionId, + fence, + process: acquired.process, + now: input.now() + }) + return input.store.proveOwner({ + sessionId: record.sessionId, + fence, + link: acquired.link, + now: input.now() + }) +} 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 new file mode 100644 index 00000000000..27fcfd962c7 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts @@ -0,0 +1,188 @@ +// Attach: reserve the session record, then open its journal. +// +// `create` and `ensure` are the same transition with a different starting +// point — a null expected fence means "no session exists yet". Both go through +// the record store's compare-and-swap, which also owns the idempotency row, so +// a retried attach replays instead of reserving a second owner. + +import type { AgentType } from '../../../shared/agent-status-types' +import type { + AgentSessionJournalIdentity, + AgentSessionProviderHandle +} from '../../../shared/agent-session-journal-types' +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import type { AgentSessionHandleProvider } from '../../../shared/agent-session-provider-handle' +import type { + AgentSessionAccountHome, + AgentSessionExecutionLocation, + AgentSessionOwnerRuntimeKind, + AgentSessionRecord +} from '../../../shared/agent-session-record' +import { + AGENT_SESSION_WIRE_REFUSAL_CODES, + type AgentSessionMutationEnvelope, + type AgentSessionWireRefusal, + type AgentSessionWireRefusalCode +} from '../../../shared/agent-session-wire' +import { + agentSessionFingerprintConflict, + computeAgentSessionPayloadFingerprint +} from '../../../shared/agent-session-mutation-envelope' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { journalDirectoryFor } from '../agent-session-journal/journal-paths' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { + openAgentSessionJournalWithRecovery, + type AgentSessionJournalRecovery +} from './agent-session-journal-recovery' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' + +/** + * Everything a client may declare about the session it wants. Deliberately no + * spawn token, claim key, or owner probe: those are host observations, and a + * client that could assert "the previous owner is dead" could steal a live + * session. The host fills them in. + */ +export type AgentSessionAttachParams = { + envelope: AgentSessionMutationEnvelope + location: AgentSessionExecutionLocation + provider: AgentSessionHandleProvider + agent: AgentType + accountHome: AgentSessionAccountHome + runtimeKind: AgentSessionOwnerRuntimeKind + providerHandle: AgentSessionProviderHandle +} + +/** Host-supplied half of the reservation. */ +export type AgentSessionAttachAuthority = { + spawnToken: string + claimKeyId: string + handoffOperationId: string | null + probe: AgentSessionOwnerProbe +} + +/** The fields that define WHICH session this call would attach to. Deliberately + * excludes the spawn token and the probe: those differ between a first attempt + * and its retry, and a retry must replay rather than conflict. */ +export function attachFingerprintFields(params: AgentSessionAttachParams): Record { + return { + location: params.location, + provider: params.provider, + agent: params.agent, + accountHome: params.accountHome, + runtimeKind: params.runtimeKind, + providerHandle: params.providerHandle, + expectedRuntimeFence: params.envelope.expectedRuntimeFence + } +} + +/** Recomputes the fingerprint the client declared and refuses a mismatch before + * anything reaches the store. */ +export function admitAttachOrRefuse( + params: AgentSessionAttachParams +): { ok: true; fingerprint: string } | { ok: false; refusal: AgentSessionWireRefusal } { + const fingerprint = computeAgentSessionPayloadFingerprint({ + method: 'agentSession.attach', + sessionId: params.envelope.sessionId, + fields: attachFingerprintFields(params) + }) + const conflict = agentSessionFingerprintConflict(params.envelope, fingerprint) + return conflict ? { ok: false, refusal: conflict } : { ok: true, fingerprint } +} + +export function journalIdentityFor( + sessionId: string, + params: AgentSessionAttachParams +): AgentSessionJournalIdentity { + return { + sessionId, + workspaceId: params.location.workspaceId, + hostId: params.location.executionHostId, + agent: params.agent, + providerHandle: params.providerHandle + } +} + +export type AttachedJournal = { + journal: AgentSessionJournal + recovery: AgentSessionJournalRecovery | null + /** Submissions the crash boundary settled as `unknown` on this open. */ + unconfirmedClientMessageIds: string[] +} + +/** + * Open the session's journal, recovering it when the stored one is unusable, + * and settle every submission left in flight by a previous process. Orca never + * re-sends those; they surface as delivery unconfirmed. + */ +export async function attachJournal(input: { + record: AgentSessionRecord + params: AgentSessionAttachParams + journalRoot: string + adapter: StructuredAgentSessionAdapter +}): Promise { + const identity = journalIdentityFor(input.record.sessionId, input.params) + const fence = input.record.lease.runtimeFence + const historyFilePath = input.adapter.historyFilePath + ? await input.adapter.historyFilePath({ identity }) + : null + const opened = await openAgentSessionJournalWithRecovery({ + identity, + journalDir: journalDirectoryFor(input.journalRoot, { + workspaceId: identity.workspaceId, + sessionId: identity.sessionId + }), + fence, + historyFilePath + }) + return { + ...opened, + unconfirmedClientMessageIds: await opened.journal.markPendingSubmissionsUnknown(fence) + } +} + +export function reserveRequestFor(input: { + sessionId: string + params: AgentSessionAttachParams + authority: AgentSessionAttachAuthority + callerKey: string + fingerprint: string + now: number +}): Parameters[0] { + const { params, authority } = input + return { + sessionId: input.sessionId, + location: params.location, + provider: params.provider, + accountHome: params.accountHome, + runtimeKind: params.runtimeKind, + expectedFence: params.envelope.expectedRuntimeFence, + spawnToken: authority.spawnToken, + claimKeyId: authority.claimKeyId, + handoffOperationId: authority.handoffOperationId, + probe: authority.probe, + operation: { + callerKey: input.callerKey, + operationId: params.envelope.clientOperationId, + fingerprint: input.fingerprint + }, + now: input.now + } +} + +/** The store signals refusals by throwing the refusal code. Anything not in the + * known set is a defect, not a client error, and is rethrown. */ +export function classifyStoreFailure( + error: unknown, + currentFence: number | null +): AgentSessionWireRefusal { + const code = error instanceof Error ? error.message : String(error) + if (!(AGENT_SESSION_WIRE_REFUSAL_CODES as readonly string[]).includes(code)) { + throw error + } + return { + code: code as AgentSessionWireRefusalCode, + message: `The session store refused this call: ${code}.`, + ...(code === 'agent_session_checkpoint_stale' && currentFence !== null ? { currentFence } : {}) + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts new file mode 100644 index 00000000000..0fd4a23ee38 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts @@ -0,0 +1,603 @@ +// The host against a real record store and a real journal, with only the +// provider adapter stubbed — admission, idempotency, and the journal writes are +// exactly the ones that ship. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import type { + AgentSessionExecutionLocation, + AgentSessionRecord +} from '../../../shared/agent-session-record' +import type { + AgentSessionMutationEnvelope, + AgentSessionSubscribeEvent +} from '../../../shared/agent-session-wire' +import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { journalDirectoryFor } from '../agent-session-journal/journal-paths' +import { openAgentSessionJournal } from '../agent-session-journal/journal-store' +import type { + AgentSessionDispatchOutcome, + StructuredAgentSessionAdapter +} from './structured-agent-session-adapter' +import { attachFingerprintFields } from './structured-agent-session-attach' +import type { AgentSessionAttachParams } from './structured-agent-session-attach' +import { StructuredAgentSessionHost } from './structured-agent-session-host' + +const NOW = 1_800_000_000_000 +const SESSION = 'session-alpha' +const THREAD = '019fd532-7c11-7a90-b6de-4e1a2c3d5f60' + +const LOCATION: AgentSessionExecutionLocation = { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'workspace-1', + workspaceKind: 'git-worktree' +} + +const CALLER = { callerKey: 'client-1' } + +let operations = 0 + +/** `<13-digit ms>-<32 hex>`, the only shape the durable ledger accepts. */ +function operationId(): string { + operations += 1 + return `${NOW}-${operations.toString(16).padStart(32, '0')}` +} + +function message(text: string): AgentJournalMessageItem { + return { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] } +} + +function envelope( + method: string, + fields: Record, + overrides: Partial = {} +): AgentSessionMutationEnvelope { + return { + sessionId: SESSION, + clientOperationId: operationId(), + // A mutation names the fence it believes it is writing under; only attach + // may leave it null. + expectedRuntimeFence: store.getRecord(SESSION)?.lease.runtimeFence ?? 1, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method, + sessionId: SESSION, + fields + }), + ...overrides + } +} + +function attachParams(overrides: Partial = {}): AgentSessionAttachParams { + const params: AgentSessionAttachParams = { + envelope: { + sessionId: SESSION, + clientOperationId: operationId(), + expectedRuntimeFence: null, + payloadFingerprint: '0'.repeat(64) + }, + location: LOCATION, + provider: 'codex', + agent: 'codex', + accountHome: { variable: 'CODEX_HOME', path: '/home/dev/.codex' }, + runtimeKind: 'native', + providerHandle: { kind: 'codex', threadId: THREAD }, + ...overrides + } + return { + ...params, + envelope: { + ...params.envelope, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.attach', + sessionId: params.envelope.sessionId, + fields: attachFingerprintFields(params) + }) + } + } +} + +/** `ensure`: the same transition as create, from a fence the client already holds. */ +function ensureParams(fence: number): AgentSessionAttachParams { + return attachParams({ + envelope: { + sessionId: SESSION, + clientOperationId: operationId(), + expectedRuntimeFence: fence, + payloadFingerprint: '0'.repeat(64) + } + }) +} + +let root: string +let store: AgentSessionRecordStore +let host: StructuredAgentSessionHost +let dispatch: Mock +let cancelTurn: Mock +let answerPrompt: Mock +let setOption: Mock +let ordinal = 0 + +function accepted(): AgentSessionDispatchOutcome { + ordinal += 1 + return { + state: 'accepted', + providerIdentity: { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal } + } +} + +function adapter(): StructuredAgentSessionAdapter { + return { + acquire: async ({ fence }) => ({ + process: { + hostId: 'local', + pid: 4242, + processStartTimeMs: 1_700_000_000_000, + spawnToken: store.getRecord(SESSION)?.lease.reservedSpawnToken ?? 'spawn-a' + }, + link: { + linkId: `link-${fence}`, + handle: { provider: 'codex', threadId: THREAD }, + // A restarted host re-proves the thread it inherited; only the first + // owner of a session may claim to have created it. + origin: store.getRecord(SESSION)?.providerHandleChain.length ? 'resumed' : 'created', + mintedAtFence: fence, + observedAt: NOW + } + }), + dispatch, + cancelTurn, + answerPrompt, + setOption + } +} + +async function attach(): Promise { + const result = await host.attach(CALLER, attachParams()) + expect(result.ok).toBe(true) + return store.getRecord(SESSION) +} + +/** Puts a pending approval in the journal BEFORE attach, which is the only way + * 1d can stage one: the adapter that would emit it is phase 2's. */ +async function seedApproval(optionId = 'allow'): Promise<{ itemId: string; revision: number }> { + const identity = { provider: 'codex' as const, threadId: THREAD, turnId: 'turn-1', ordinal: 99 } + const journalDir = journalDirectoryFor(root, { workspaceId: 'workspace-1', sessionId: SESSION }) + const journal = await openAgentSessionJournal({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: THREAD } + }, + journalDir + }) + const appended = await journal.appendItem( + identity, + { + kind: 'approval', + title: 'Run the command?', + detail: null, + options: [{ id: optionId, label: 'Allow' }], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + }, + { fence: 1 } + ) + return { itemId: appended.itemId, revision: appended.revision } +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-wire-host-')) + operations = 0 + ordinal = 0 + dispatch = vi.fn(async () => accepted()) + cancelTurn = vi.fn(async () => ({ cancelled: true })) + answerPrompt = vi.fn(async () => undefined) + setOption = vi.fn(async () => undefined) + store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' }) + host = new StructuredAgentSessionHost({ + store, + adapter: adapter(), + journalRoot: root, + claimKeyId: 'key-1', + mintSpawnToken: () => 'spawn-a', + now: () => NOW + }) +}) + +afterEach(async () => { + await rm(root, { recursive: true, force: true }) +}) + +describe('attach', () => { + it('reserves the lease, spawns through the adapter, and opens the journal', async () => { + const result = await host.attach(CALLER, attachParams()) + expect(result).toMatchObject({ ok: true, replayed: false }) + expect(host.hasSession(SESSION)).toBe(true) + const record = store.getRecord(SESSION) + expect(record?.lease.ownerProcess?.pid).toBe(4242) + expect(record?.lease.handoffStage).toBeNull() + }) + + it('refuses a payload the client fingerprinted wrong', async () => { + const params = attachParams() + const result = await host.attach(CALLER, { + ...params, + envelope: { ...params.envelope, payloadFingerprint: 'a'.repeat(64) } + }) + expect(result).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_conflict' } + }) + }) + + it('refuses a second create against a live session', async () => { + await attach() + expect(await host.attach(CALLER, attachParams())).toMatchObject({ ok: false }) + }) + + it('replays a retried attach instead of reserving a second owner', async () => { + const params = attachParams() + await host.attach(CALLER, params) + const retry = await host.attach(CALLER, params) + expect(retry).toMatchObject({ ok: true, replayed: true }) + }) +}) + +describe('send', () => { + it('writes the submission before dispatching and resolves it accepted', async () => { + await attach() + const body = message('add a retry') + const result = await host.send(CALLER, { + envelope: envelope('agentSession.send', { body }), + body + }) + if (!result.ok) { + throw new Error(`expected a send, got ${result.refusal.code}`) + } + expect(result.value.submission.dispatchState).toBe('accepted') + expect(dispatch).toHaveBeenCalledTimes(1) + const page = host.history({ sessionId: SESSION, direction: 'tail' }) + expect(page.ok && page.page.items).toHaveLength(1) + }) + + it('settles a thrown dispatch as unknown, never as a rejection', async () => { + await attach() + dispatch.mockRejectedValueOnce(new Error('socket closed')) + const body = message('add a retry') + const result = await host.send(CALLER, { + envelope: envelope('agentSession.send', { body }), + body + }) + expect(result).toMatchObject({ ok: true, value: { submission: { dispatchState: 'unknown' } } }) + }) + + it('replays a retried send from the journal without dispatching twice', async () => { + await attach() + const body = message('add a retry') + const params = { envelope: envelope('agentSession.send', { body }), body } + await host.send(CALLER, params) + const retry = await host.send(CALLER, params) + expect(retry).toMatchObject({ ok: true, replayed: true }) + expect(dispatch).toHaveBeenCalledTimes(1) + }) + + it('refuses a stale fence and hands back the current one', async () => { + const record = await attach() + const body = message('add a retry') + const result = await host.send(CALLER, { + envelope: envelope( + 'agentSession.send', + { body }, + { expectedRuntimeFence: (record?.lease.runtimeFence ?? 1) + 5 } + ), + body + }) + expect(result).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_checkpoint_stale', currentFence: record?.lease.runtimeFence } + }) + }) + + it('does not let a refused call leave a ledger row that replays past the fence', async () => { + const record = await attach() + const body = message('add a retry') + const params = { + envelope: envelope( + 'agentSession.send', + { body }, + { expectedRuntimeFence: (record?.lease.runtimeFence ?? 1) + 5 } + ), + body + } + await host.send(CALLER, params) + expect(await host.send(CALLER, params)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_checkpoint_stale' } + }) + expect(dispatch).not.toHaveBeenCalled() + }) + + it('refuses any mutation against a session this host has not attached', async () => { + const body = message('add a retry') + expect( + await host.send(CALLER, { envelope: envelope('agentSession.send', { body }), body }) + ).toMatchObject({ ok: false, refusal: { code: 'agent_session_ownership_unknown' } }) + }) +}) + +describe('cancel', () => { + it('records the outcome as a status item keyed by the operation id', async () => { + await attach() + const result = await host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', { turnId: 'turn-1' }), + turnId: 'turn-1' + }) + expect(result).toMatchObject({ ok: true, value: { cancelled: true } }) + const page = host.history({ sessionId: SESSION, direction: 'tail' }) + expect(page.ok && page.page.items[0]?.body).toMatchObject({ kind: 'status' }) + }) + + it('reports an unconfirmed cancellation rather than failing the call', async () => { + await attach() + cancelTurn.mockRejectedValueOnce(new Error('no answer')) + const result = await host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', { turnId: 'turn-1' }), + turnId: 'turn-1' + }) + expect(result).toMatchObject({ ok: true, value: { cancelled: false } }) + }) + + it('never interrupts twice on a replay', async () => { + await attach() + const params = { + envelope: envelope('agentSession.cancel', { turnId: 'turn-1' }), + turnId: 'turn-1' + } + await host.cancel(CALLER, params) + expect(await host.cancel(CALLER, params)).toMatchObject({ + ok: true, + replayed: true, + value: { cancelled: false } + }) + expect(cancelTurn).toHaveBeenCalledTimes(1) + }) +}) + +describe('respondToPrompt', () => { + it('commits the answer before the provider callback', async () => { + const prompt = await seedApproval() + await attach() + const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' } + const result = await host.respondToPrompt(CALLER, { + envelope: envelope('agentSession.respondTo:approval', fields), + kind: 'approval', + ...fields + }) + expect(result).toMatchObject({ + ok: true, + value: { resolution: { state: 'resolved', selectedOptionId: 'allow' } } + }) + expect(answerPrompt).toHaveBeenCalledTimes(1) + }) + + it('refuses a second answer to one prompt and says which answer won', async () => { + const prompt = await seedApproval() + await attach() + const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' } + await host.respondToPrompt(CALLER, { + envelope: envelope('agentSession.respondTo:approval', fields), + kind: 'approval', + ...fields + }) + const loser = await host.respondToPrompt( + { callerKey: 'client-2' }, + { + envelope: envelope('agentSession.respondTo:approval', fields), + kind: 'approval', + ...fields + } + ) + expect(loser).toMatchObject({ + ok: false, + refusal: { + code: 'agent_session_item_revision_stale', + resolution: { selectedOptionId: 'allow' } + } + }) + expect(answerPrompt).toHaveBeenCalledTimes(1) + }) + + it('refuses an option the prompt does not offer', async () => { + const prompt = await seedApproval() + await attach() + const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'deny' } + expect( + await host.respondToPrompt(CALLER, { + envelope: envelope('agentSession.respondTo:approval', fields), + kind: 'approval', + ...fields + }) + ).toMatchObject({ ok: false, refusal: { code: 'agent_session_operation_invalid' } }) + expect(answerPrompt).not.toHaveBeenCalled() + }) + + it('keeps the answer and reports it undelivered when the provider callback throws', async () => { + const prompt = await seedApproval() + await attach() + answerPrompt.mockRejectedValueOnce(new Error('pipe closed')) + const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId: 'allow' } + const result = await host.respondToPrompt(CALLER, { + envelope: envelope('agentSession.respondTo:approval', fields), + kind: 'approval', + ...fields + }) + expect(result.ok).toBe(true) + const page = host.history({ sessionId: SESSION, direction: 'tail' }) + const statusId = agentJournalItemKey({ + provider: 'orca', + clientMessageId: `${prompt.itemId}#delivery` + }) + expect(page.ok && page.page.items.some((entry) => entry.itemId === statusId)).toBe(true) + }) +}) + +describe('setOption', () => { + it('goes to the provider and writes nothing to the journal', async () => { + await attach() + const fields = { key: 'model', value: 'gpt-5' } + const result = await host.setOption(CALLER, { + envelope: envelope('agentSession.setOption', fields), + ...fields + }) + expect(result).toMatchObject({ ok: true, value: fields }) + expect(setOption).toHaveBeenCalledTimes(1) + const page = host.history({ sessionId: SESSION, direction: 'tail' }) + expect(page.ok && page.page.items).toHaveLength(0) + }) +}) + +describe('restart', () => { + /** A restarted process: the same directories, a new store and a new host over + * them. Every lease loads unreconciled, so this is the state that decides + * whether a persisted session is reachable at all. */ + async function reboot( + probeOwner: (record: AgentSessionRecord) => Promise + ) { + store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' }) + host = new StructuredAgentSessionHost({ + store, + adapter: adapter(), + journalRoot: root, + claimKeyId: 'key-1', + mintSpawnToken: () => 'spawn-b', + probeOwner, + now: () => NOW + }) + } + + /** The refusal a restarted host owes a client holding the dead generation's + * fence: stale, with the live fence attached so the retry can succeed. */ + async function staleFenceFrom(held: number): Promise { + const refused = await host.attach(CALLER, ensureParams(held)) + if (refused.ok) { + throw new Error('a fence from the previous host generation was accepted') + } + expect(refused.refusal.code).toBe('agent_session_checkpoint_stale') + const current = refused.refusal.currentFence + expect(current).toBeGreaterThan(held) + return current ?? 0 + } + + it('adjudicates the leases it loaded before deciding who may write', async () => { + const before = await attach() + const held = before?.lease.runtimeFence ?? 0 + await reboot(async () => ({ outcome: 'pid-absent' })) + + const reattached = await host.attach(CALLER, ensureParams(await staleFenceFrom(held))) + expect(reattached).toMatchObject({ ok: true }) + expect(store.getRecord(SESSION)?.lease.unreconciled).toBe(false) + expect(store.getRecord(SESSION)?.lease.ownerProcess?.pid).toBe(4242) + }) + + it("keeps a session whose owner cannot be probed out of a live writer's hands", async () => { + await attach() + const held = store.getRecord(SESSION)?.lease.runtimeFence ?? 0 + await reboot(async () => ({ outcome: 'indeterminate', reason: 'no probe on this host' })) + + // Nothing proved the previous process dead, so the lease routes to manual + // recovery rather than letting a second writer in behind it. + expect(await host.attach(CALLER, ensureParams(held))).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_ownership_unknown' } + }) + }) + + it('does not remember a failed adjudication as done', async () => { + const before = await attach() + const held = before?.lease.runtimeFence ?? 0 + const probe = vi + .fn<(record: AgentSessionRecord) => Promise>() + .mockRejectedValueOnce(new Error('probe exploded')) + .mockResolvedValue({ outcome: 'pid-absent' }) + await reboot(probe) + + await expect(host.attach(CALLER, ensureParams(held))).rejects.toThrow('probe exploded') + // One unlucky startup must not strand the session for this host's lifetime. + const reattached = await host.attach(CALLER, ensureParams(await staleFenceFrom(held))) + expect(reattached).toMatchObject({ ok: true }) + expect(probe).toHaveBeenCalledTimes(2) + }) +}) + +describe('subscribe', () => { + it('opens with a snapshot and then streams cursor-qualified batches', async () => { + await attach() + const events: AgentSessionSubscribeEvent[] = [] + const dispose = host.subscribe({ + id: 'sub-1', + sessionId: SESSION, + emit: (event) => events.push(event) + }) + const body = message('add a retry') + await host.send(CALLER, { envelope: envelope('agentSession.send', { body }), body }) + + expect(events[0]?.type).toBe('snapshot') + const batches = events.filter((event) => event.type === 'batch') + expect(batches.length).toBeGreaterThan(0) + const last = batches.at(-1) + expect(last?.type === 'batch' && last.batch.cursor.sequence).toBeGreaterThan(0) + + dispose() + expect(events.at(-1)?.type).toBe('end') + }) + + it('resumes from a client cursor with only the rows it missed', async () => { + await attach() + const body = message('add a retry') + const first = await host.send(CALLER, { + envelope: envelope('agentSession.send', { body }), + body + }) + if (!first.ok) { + throw new Error(`expected a send, got ${first.refusal.code}`) + } + + const events: AgentSessionSubscribeEvent[] = [] + host.subscribe({ + id: 'sub-2', + sessionId: SESSION, + emit: (event) => events.push(event), + cursor: first.cursor + }) + // Nothing new since that cursor: a resume is silent, not a redundant replay. + expect(events).toHaveLength(0) + + const second = message('and a timeout') + await host.send(CALLER, { + envelope: envelope('agentSession.send', { body: second }), + body: second + }) + expect(events.some((event) => event.type === 'batch')).toBe(true) + expect(events.some((event) => event.type === 'snapshot')).toBe(false) + }) + + it('resets a subscriber whose epoch is gone', async () => { + await attach() + const events: AgentSessionSubscribeEvent[] = [] + host.subscribe({ + id: 'sub-3', + sessionId: SESSION, + emit: (event) => events.push(event), + cursor: { epoch: 'epoch-from-a-previous-life', sequence: 3 } + }) + expect(events[0]).toMatchObject({ type: 'reset', reset: 'epoch_changed' }) + }) +}) 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 new file mode 100644 index 00000000000..4f77a503d7e --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -0,0 +1,360 @@ +// The structured agent-session host: one place where the lease, the journal, +// and the provider adapter meet. +// +// Every mutating call takes the same route — recompute the fingerprint, admit +// through the durable operation ledger, check the lease, then act — so no +// method can grow its own admission rules. Calls against one session are +// serialized; the journal's own queue orders writes, but admission has to be +// single-file too or two clients could both read one prompt as pending. + +import { randomUUID } from 'node:crypto' +import type { + AgentJournalCursor, + AgentJournalMessageItem +} from '../../../shared/agent-session-journal-types' +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { + admitAgentSessionMutation, + agentSessionFingerprintConflict, + computeAgentSessionPayloadFingerprint +} from '../../../shared/agent-session-mutation-envelope' +import type { + AgentSessionAttachResult, + AgentSessionCancelResult, + AgentSessionHistoryRequest, + AgentSessionHistoryResult, + AgentSessionMutationEnvelope, + AgentSessionMutationResult, + AgentSessionOptionResult, + AgentSessionPromptResult, + AgentSessionSendResult, + AgentSessionWireRefusal +} from '../../../shared/agent-session-wire' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { readAgentSessionHistory } from './agent-session-history-page' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import type { AgentSessionAttachParams } from './structured-agent-session-attach' +import { performAttach } from './structured-agent-session-attach-flow' +import { createRestartReconciler } from './structured-agent-session-restart-reconcile' +import { + cancelPlan, + promptPlan, + sendPlan, + setOptionPlan, + type MutationPlan +} from './structured-agent-session-mutation-plans' +import { + AgentSessionSubscribers, + type AgentSessionSubscriberEmit +} from './structured-agent-session-subscribers' +import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' + +export type StructuredAgentSessionCaller = { + /** Stable per-client identity; scopes the operation ledger and records who + * answered a prompt. */ + callerKey: string +} + +export type StructuredAgentSessionHostDeps = { + store: AgentSessionRecordStore + adapter: StructuredAgentSessionAdapter + journalRoot: string + /** Key id the host's claims are minted under. */ + claimKeyId: string + probeOwner?: (record: AgentSessionRecord) => Promise + mintSpawnToken?: () => string + now?: () => number +} + +type SessionState = { journal: AgentSessionJournal; params: AgentSessionAttachParams } + +function refuse(refusal: AgentSessionWireRefusal): { ok: false; refusal: AgentSessionWireRefusal } { + return { ok: false, refusal } +} + +const NO_SESSION: AgentSessionWireRefusal = { + code: 'agent_session_ownership_unknown', + message: 'This host holds no attached session by that id.' +} + +export class StructuredAgentSessionHost { + private readonly sessions = new Map() + private readonly subscribers = new AgentSessionSubscribers() + private readonly chains = new Map>() + private readonly reconcileLeases: (sessionId: string) => Promise + + constructor(private readonly deps: StructuredAgentSessionHostDeps) { + this.reconcileLeases = createRestartReconciler({ + store: deps.store, + probe: (record) => this.probeRecord(record), + now: () => this.now() + }) + } + + private now(): number { + return this.deps.now?.() ?? Date.now() + } + + hasSession(sessionId: string): boolean { + return this.sessions.has(sessionId) + } + + /** Per-session single file. A rejected task must not poison the chain. */ + private serialize(sessionId: string, task: () => Promise): Promise { + const prior = this.chains.get(sessionId) ?? Promise.resolve() + const next = prior.then(task, task) + this.chains.set( + sessionId, + next.catch(() => undefined) + ) + return next + } + + /** + * `create` and `ensure` in one: a null expected fence means the session must + * not exist yet, any other value is a compare-and-swap against the lease. + */ + attach( + caller: StructuredAgentSessionCaller, + params: AgentSessionAttachParams + ): Promise> { + return this.serialize(params.envelope.sessionId, async () => { + const sessionId = params.envelope.sessionId + const unreconciled = await this.reconcileLeases(sessionId) + if (unreconciled) { + return refuse(unreconciled) + } + return performAttach({ + store: this.deps.store, + adapter: this.deps.adapter, + journalRoot: this.deps.journalRoot, + authority: { + spawnToken: this.deps.mintSpawnToken?.() ?? randomUUID(), + claimKeyId: this.deps.claimKeyId, + handoffOperationId: params.envelope.clientOperationId, + probe: await this.probeOwner(sessionId) + }, + callerKey: caller.callerKey, + params, + now: () => this.now(), + onAttached: (attached) => { + this.sessions.set(sessionId, { journal: attached.journal, params }) + if (attached.recovery) { + this.subscribers.reset(sessionId, attached.journal, attached.recovery.reset) + } else { + this.subscribers.publish(sessionId, attached.journal) + } + } + }) + }) + } + + /** + * Liveness of whatever currently owns the session. Never client-asserted: a + * client that could claim the owner is gone could evict a live one. Without a + * host prober the answer is `indeterminate`, which routes to manual recovery + * rather than a silent takeover. + */ + private probeOwner(sessionId: string): Promise { + const record = this.deps.store.getRecord(sessionId) + if (!record || record.lease.ownerProcess === null) { + return Promise.resolve({ outcome: 'reservation-unused' }) + } + return this.probeRecord(record) + } + + private probeRecord(record: AgentSessionRecord): Promise { + return ( + this.deps.probeOwner?.(record) ?? + Promise.resolve({ + outcome: 'indeterminate', + reason: 'This host cannot probe structured session owners.' + }) + ) + } + + send( + caller: StructuredAgentSessionCaller, + params: { envelope: AgentSessionMutationEnvelope; body: AgentJournalMessageItem } + ): Promise> { + return this.mutate(caller, params.envelope, sendPlan(params)) + } + + cancel( + caller: StructuredAgentSessionCaller, + params: { envelope: AgentSessionMutationEnvelope; turnId: string } + ): Promise> { + return this.mutate(caller, params.envelope, cancelPlan(params)) + } + + respondToPrompt( + caller: StructuredAgentSessionCaller, + params: { + envelope: AgentSessionMutationEnvelope + kind: 'approval' | 'question' + itemId: string + expectedRevision: number + optionId: string + } + ): Promise> { + return this.mutate(caller, params.envelope, promptPlan(params)) + } + + setOption( + caller: StructuredAgentSessionCaller, + params: { envelope: AgentSessionMutationEnvelope; key: string; value: string } + ): Promise> { + return this.mutate(caller, params.envelope, setOptionPlan(params)) + } + + private mutate( + caller: StructuredAgentSessionCaller, + envelope: AgentSessionMutationEnvelope, + plan: MutationPlan + ): Promise> { + return this.serialize(envelope.sessionId, () => this.runMutation(caller, envelope, plan)) + } + + private async runMutation( + caller: StructuredAgentSessionCaller, + envelope: AgentSessionMutationEnvelope, + plan: MutationPlan + ): Promise> { + const session = this.sessions.get(envelope.sessionId) + const record = this.deps.store.getRecord(envelope.sessionId) + if (!session || !record) { + return refuse(NO_SESSION) + } + const hostFingerprint = computeAgentSessionPayloadFingerprint({ + method: plan.method, + sessionId: envelope.sessionId, + fields: plan.fields + }) + const conflict = agentSessionFingerprintConflict(envelope, hostFingerprint) + if (conflict) { + return refuse(conflict) + } + const admission = admitAgentSessionMutation({ + envelope, + hostFingerprint, + ledger: await this.deps.store.admitOperation({ + callerKey: caller.callerKey, + operationId: envelope.clientOperationId, + fingerprint: hostFingerprint, + now: this.now() + }), + lease: record.lease + }) + if (admission.decision === 'refused') { + return refuse(admission.refusal) + } + + const fence = record.lease.runtimeFence + const ctx = this.contextFor(caller, session, envelope.sessionId, fence) + if (admission.decision === 'replay') { + const recorded = plan.replay(ctx) + if (recorded) { + return { ok: true, replayed: true, fence, cursor: ctx.journal.cursor(), value: recorded } + } + // Nothing durable landed, so this id is about to run for the first time. + // A refused call leaves its ledger row behind, and replaying past the + // lease and the fence would let a resend act under an owner that has since + // changed — so a first run pays the full admission price either way. + const rerun = admitAgentSessionMutation({ + envelope, + hostFingerprint, + ledger: { decision: 'admit', row: admission.row }, + lease: record.lease + }) + if (rerun.decision === 'refused') { + return refuse(rerun.refusal) + } + } + + const outcome = await this.finish(caller, envelope, plan, ctx) + return outcome.ok + ? { ok: true, replayed: false, fence, cursor: ctx.journal.cursor(), value: outcome.value } + : refuse(outcome.refusal) + } + + private contextFor( + caller: StructuredAgentSessionCaller, + session: SessionState, + sessionId: string, + fence: number + ): AgentSessionTurnContext { + return { + sessionId, + journal: session.journal, + fence, + adapter: this.deps.adapter, + resolvedBy: caller.callerKey, + publish: () => this.subscribers.publish(sessionId, session.journal), + now: () => this.now() + } + } + + /** Runs the effect and settles its ledger row, including on a throw — a row + * left `pending` would replay as unresolvable forever. */ + private async finish( + caller: StructuredAgentSessionCaller, + envelope: AgentSessionMutationEnvelope, + plan: MutationPlan, + ctx: AgentSessionTurnContext + ): Promise> { + const settle = ( + outcome: Parameters[0]['outcome'] + ) => + this.deps.store.recordOperationOutcome({ + callerKey: caller.callerKey, + operationId: envelope.clientOperationId, + outcome + }) + try { + const outcome = await plan.run(ctx) + await settle( + outcome.ok + ? { status: 'succeeded', sessionId: envelope.sessionId } + : { status: 'failed', code: outcome.refusal.code } + ) + return outcome + } catch (error) { + await settle({ status: 'unknown' }) + throw error + } + } + + /** + * Paged read. Throws `agent_session_ownership_unknown` for a session this host + * has not attached — a reset would tell the client to reload something that + * does not exist here. + */ + history(request: AgentSessionHistoryRequest): AgentSessionHistoryResult { + return readAgentSessionHistory(this.requireSession(request.sessionId).journal, request) + } + + subscribe(input: { + id: string + sessionId: string + emit: AgentSessionSubscriberEmit + cursor?: AgentJournalCursor + }): () => void { + const session = this.requireSession(input.sessionId) + const fence = this.deps.store.getRecord(input.sessionId)?.lease.runtimeFence ?? 0 + return this.subscribers.open({ ...input, journal: session.journal, fence }) + } + + unsubscribe(sessionId: string, id: string): void { + this.subscribers.close(sessionId, id) + } + + private requireSession(sessionId: string): SessionState { + const session = this.sessions.get(sessionId) + if (!session) { + throw new Error(NO_SESSION.code) + } + return session + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts new file mode 100644 index 00000000000..0f101a6ad70 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts @@ -0,0 +1,113 @@ +// One plan per mutating method: what it fingerprints, what it does, and how its +// answer is rebuilt on a replay. +// +// The replay half matters more than it looks. The ledger records only that an +// operation happened, so the durable answer has to come back out of the journal. +// A plan that cannot find its effect returns null, and the call runs for real — +// which is exactly right when the crash landed before the journal write. + +import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import type { + AgentSessionCancelResult, + AgentSessionMutationEnvelope, + AgentSessionOptionResult, + AgentSessionPromptResult, + AgentSessionSendResult +} from '../../../shared/agent-session-wire' +import { + performCancel, + performPrompt, + performSend, + performSetOption, + type AgentSessionTurnContext, + type TurnOutcome +} from './structured-agent-session-turns' + +export type MutationPlan = { + method: string + fields: Record + run: (ctx: AgentSessionTurnContext) => Promise> + replay: (ctx: AgentSessionTurnContext) => TValue | null +} + +export function sendPlan(params: { + envelope: AgentSessionMutationEnvelope + body: AgentJournalMessageItem +}): MutationPlan { + // The operation id IS the client message id: one send, one durable row, one + // key the client reconciles its optimistic bubble against. + const clientMessageId = params.envelope.clientOperationId + return { + method: 'agentSession.send', + fields: { body: params.body }, + run: (ctx) => + performSend(ctx, { + clientMessageId, + payloadFingerprint: params.envelope.payloadFingerprint, + body: params.body + }), + replay: (ctx) => { + const submission = ctx.journal + .submissions() + .find((entry) => entry.clientMessageId === clientMessageId) + return submission ? { clientMessageId, submission } : null + } + } +} + +export function cancelPlan(params: { + envelope: AgentSessionMutationEnvelope + turnId: string +}): MutationPlan { + return { + method: 'agentSession.cancel', + fields: { turnId: params.turnId }, + run: (ctx) => + performCancel(ctx, { + clientOperationId: params.envelope.clientOperationId, + turnId: params.turnId + }), + // Interrupting twice would kill a turn the client never asked to stop, so a + // replay reports the turn as already handled instead. + replay: () => ({ turnId: params.turnId, cancelled: false }) + } +} + +export function promptPlan(params: { + kind: 'approval' | 'question' + itemId: string + expectedRevision: number + optionId: string +}): MutationPlan { + return { + method: `agentSession.respondTo:${params.kind}`, + fields: { + itemId: params.itemId, + expectedRevision: params.expectedRevision, + optionId: params.optionId + }, + run: (ctx) => performPrompt(ctx, params), + replay: (ctx) => { + const item = ctx.journal.snapshot().items.find((entry) => entry.itemId === params.itemId) + const body = item?.body + if (!item || !body || (body.kind !== 'approval' && body.kind !== 'question')) { + return null + } + return body.resolution.state === 'pending' + ? null + : { itemId: item.itemId, revision: item.revision, resolution: body.resolution } + } + } +} + +export function setOptionPlan(params: { + key: string + value: string +}): MutationPlan { + return { + method: 'agentSession.setOption', + fields: { key: params.key, value: params.value }, + run: (ctx) => performSetOption(ctx, params), + replay: () => ({ key: params.key, value: params.value }) + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-registry.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-registry.ts new file mode 100644 index 00000000000..b7a26bb8e3e --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-registry.ts @@ -0,0 +1,18 @@ +// Where the RPC layer finds the host. +// +// The runtime service is already far past its size budget, so structured +// sessions hang off a module-level slot instead of another field on it — the +// same shape the native-chat RPC methods use to reach their own collaborators. +// Tests install a host with a stub adapter and clear it on teardown. + +import type { StructuredAgentSessionHost } from './structured-agent-session-host' + +let host: StructuredAgentSessionHost | null = null + +export function setStructuredAgentSessionHost(next: StructuredAgentSessionHost | null): void { + host = next +} + +export function getStructuredAgentSessionHost(): StructuredAgentSessionHost | null { + return host +} 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 new file mode 100644 index 00000000000..472fe0140f9 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts @@ -0,0 +1,43 @@ +// Every lease loads from disk unreconciled: the process that wrote it may still +// be alive, so nothing the store persisted grants a writer until this host has +// adjudicated it. Without this an attach after a restart is refused forever with +// `execution_owner_reconciling`, and the session becomes unreachable. + +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { classifyStoreFailure } from './structured-agent-session-attach' + +/** Adjudicates the loaded leases once per host, on the first attach. Answers + * with the refusal that attach owes its caller, or null once the leases are + * settled. */ +export function createRestartReconciler(deps: { + store: AgentSessionRecordStore + probe: (record: AgentSessionRecord) => Promise + now: () => number +}): (sessionId: string) => Promise { + let pending: Promise | null = null + return async (sessionId) => { + if (!pending) { + const run = deps.store + .reconcileOnRestart({ probe: deps.probe, now: deps.now() }) + .then(() => undefined) + // A failed adjudication must not be remembered as done, or one unlucky + // startup would strand every persisted session for this host's lifetime. + pending = run.catch((error: unknown) => { + pending = null + throw error + }) + } + try { + await pending + return null + } catch (error) { + return classifyStoreFailure( + error, + deps.store.getRecord(sessionId)?.lease.runtimeFence ?? null + ) + } + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts new file mode 100644 index 00000000000..8c56efb4c18 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts @@ -0,0 +1,130 @@ +// Per-subscriber cursors over one session's journal. +// +// Each subscriber advances independently: a client that connected two epochs +// ago gets a reset while a caught-up one gets a batch from the same publish. +// Nothing raw reaches a subscriber — every event carries reducer output. + +import type { + AgentJournalCursor, + AgentJournalResetReason +} from '../../../shared/agent-session-journal-types' +import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { projectJournalBatch } from './agent-session-journal-batch' + +export type AgentSessionSubscriberEmit = (event: AgentSessionSubscribeEvent) => void + +type Subscriber = { + id: string + sessionId: string + emit: AgentSessionSubscriberEmit + cursor: AgentJournalCursor +} + +export class AgentSessionSubscribers { + private readonly bySession = new Map>() + + /** Opens the stream with a snapshot or, when the client's cursor still + * resolves, with the rows it missed. Returns the disposer. */ + open(input: { + id: string + sessionId: string + journal: AgentSessionJournal + fence: number + emit: AgentSessionSubscriberEmit + cursor?: AgentJournalCursor + }): () => void { + const snapshot = input.journal.snapshot() + const subscriber: Subscriber = { + id: input.id, + sessionId: input.sessionId, + emit: input.emit, + cursor: input.cursor ?? { epoch: snapshot.cursor.epoch, sequence: 0 } + } + const session = this.bySession.get(input.sessionId) ?? new Map() + session.set(input.id, subscriber) + this.bySession.set(input.sessionId, session) + + if (input.cursor) { + this.deliver(subscriber, input.journal) + } else { + subscriber.emit({ + type: 'snapshot', + sessionId: input.sessionId, + snapshot, + fence: input.fence + }) + subscriber.cursor = snapshot.cursor + } + return () => this.close(input.sessionId, input.id) + } + + close(sessionId: string, id: string): void { + const session = this.bySession.get(sessionId) + const subscriber = session?.get(id) + if (!session || !subscriber) { + return + } + session.delete(id) + if (session.size === 0) { + this.bySession.delete(sessionId) + } + subscriber.emit({ type: 'end' }) + } + + /** Fan out whatever each subscriber has not yet seen. */ + publish(sessionId: string, journal: AgentSessionJournal): void { + for (const subscriber of this.subscribers(sessionId)) { + this.deliver(subscriber, journal) + } + } + + /** Force every subscriber back to a clean snapshot — recovery, epoch + * rollover, an unreadable schema. */ + reset(sessionId: string, journal: AgentSessionJournal, reason: AgentJournalResetReason): void { + const snapshot = journal.snapshot() + for (const subscriber of this.subscribers(sessionId)) { + subscriber.emit({ type: 'reset', sessionId, reset: reason, snapshot }) + subscriber.cursor = snapshot.cursor + } + } + + private subscribers(sessionId: string): Subscriber[] { + return [...(this.bySession.get(sessionId)?.values() ?? [])] + } + + private deliver(subscriber: Subscriber, journal: AgentSessionJournal): void { + const since = journal.readSince(subscriber.cursor) + const snapshot = journal.snapshot() + if (!since.ok) { + subscriber.emit({ + type: 'reset', + sessionId: subscriber.sessionId, + reset: since.reset, + snapshot + }) + subscriber.cursor = snapshot.cursor + return + } + if (since.rows.length === 0) { + return + } + const projected = projectJournalBatch({ + rows: since.rows, + snapshot, + afterSequence: subscriber.cursor.sequence + }) + if (!projected.ok) { + subscriber.emit({ + type: 'reset', + sessionId: subscriber.sessionId, + reset: projected.reset, + snapshot + }) + subscriber.cursor = snapshot.cursor + return + } + subscriber.emit({ type: 'batch', sessionId: subscriber.sessionId, batch: projected.batch }) + subscriber.cursor = projected.batch.cursor + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts new file mode 100644 index 00000000000..c4515e104f7 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts @@ -0,0 +1,246 @@ +// The effects behind send / cancel / respond / setOption. +// +// Admission (lease, fence, idempotency) has already passed by the time anything +// here runs; these functions own only the journal writes and the adapter call, +// in that order. Journal first is deliberate: a crash between the two leaves a +// row the next attach settles as `unknown`, whereas the reverse would lose a +// turn the provider already accepted. + +import type { + AgentJournalItemBody, + AgentJournalMessageItem, + AgentJournalResolution +} from '../../../shared/agent-session-journal-types' +import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import type { + AgentSessionCancelResult, + AgentSessionOptionResult, + AgentSessionPromptResult, + AgentSessionSendResult, + AgentSessionWireRefusal +} from '../../../shared/agent-session-wire' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import type { + AgentSessionDispatchOutcome, + StructuredAgentSessionAdapter +} from './structured-agent-session-adapter' + +export type AgentSessionTurnContext = { + sessionId: string + journal: AgentSessionJournal + fence: number + adapter: StructuredAgentSessionAdapter + /** Opaque client identity recorded as the resolver of a prompt. */ + resolvedBy: string + publish: () => void + now: () => number +} + +export type TurnOutcome = + | { ok: true; value: TValue } + | { ok: false; refusal: AgentSessionWireRefusal } + +function invalid(message: string): { ok: false; refusal: AgentSessionWireRefusal } { + return { ok: false, refusal: { code: 'agent_session_operation_invalid', message } } +} + +/** A thrown adapter error is indistinguishable from a lost reply, so it settles + * as `unknown` rather than as a rejection. */ +async function dispatchSafely( + ctx: AgentSessionTurnContext, + clientMessageId: string, + body: AgentJournalMessageItem +): Promise { + try { + return await ctx.adapter.dispatch({ + sessionId: ctx.sessionId, + clientMessageId, + body, + fence: ctx.fence + }) + } catch (error) { + return { state: 'unknown', reason: error instanceof Error ? error.message : String(error) } + } +} + +async function appendStatus( + ctx: AgentSessionTurnContext, + clientMessageId: string, + text: string +): Promise { + await ctx.journal.appendItem( + { provider: 'orca', clientMessageId }, + { kind: 'status', text }, + { fence: ctx.fence } + ) + ctx.publish() +} + +export async function performSend( + ctx: AgentSessionTurnContext, + input: { clientMessageId: string; payloadFingerprint: string; body: AgentJournalMessageItem } +): Promise> { + await ctx.journal.appendSubmission({ ...input, fence: ctx.fence }) + ctx.publish() + + const outcome = await dispatchSafely(ctx, input.clientMessageId, input.body) + await ctx.journal.resolveDispatch( + outcome.state === 'accepted' + ? { + clientMessageId: input.clientMessageId, + state: 'accepted', + providerIdentity: outcome.providerIdentity, + fence: ctx.fence + } + : { + clientMessageId: input.clientMessageId, + state: outcome.state, + reason: outcome.reason, + fence: ctx.fence + } + ) + ctx.publish() + + const submission = ctx.journal + .submissions() + .find((entry) => entry.clientMessageId === input.clientMessageId) + if (!submission) { + throw new Error('agent_session_submission_lost') + } + return { ok: true, value: { clientMessageId: input.clientMessageId, submission } } +} + +export async function performCancel( + ctx: AgentSessionTurnContext, + input: { clientOperationId: string; turnId: string } +): Promise> { + let cancelled = false + let note = `Cancelled turn ${input.turnId}.` + try { + cancelled = ( + await ctx.adapter.cancelTurn({ + sessionId: ctx.sessionId, + turnId: input.turnId, + fence: ctx.fence + }) + ).cancelled + if (!cancelled) { + note = `The provider had already finished turn ${input.turnId}.` + } + } catch (error) { + note = `Cancellation of turn ${input.turnId} was not confirmed: ${ + error instanceof Error ? error.message : String(error) + }` + } + // Keyed by the operation id so a replayed cancel upserts one item, not two. + await appendStatus(ctx, input.clientOperationId, note) + return { ok: true, value: { turnId: input.turnId, cancelled } } +} + +function promptBodyOf( + body: AgentJournalItemBody +): { options: readonly { id: string }[]; resolution: AgentJournalResolution } | null { + return body.kind === 'approval' || body.kind === 'question' ? body : null +} + +/** + * Durable compare-and-set on (itemId, revision) plus the pending state. The + * journal write commits before the provider callback fires, so two clients + * answering one prompt produce exactly one callback and the loser is told which + * answer won. + */ +export async function performPrompt( + ctx: AgentSessionTurnContext, + input: { + itemId: string + expectedRevision: number + optionId: string + kind: 'approval' | 'question' + } +): Promise> { + const item = ctx.journal.snapshot().items.find((entry) => entry.itemId === input.itemId) + if (!item) { + return invalid(`No item ${input.itemId} in session ${ctx.sessionId}.`) + } + const prompt = promptBodyOf(item.body) + if (!prompt || item.body.kind !== input.kind) { + return invalid(`Item ${input.itemId} is not a pending ${input.kind}.`) + } + if (item.revision !== input.expectedRevision) { + return { + ok: false, + refusal: { + code: 'agent_session_item_revision_stale', + message: `Item ${input.itemId} has moved on.`, + currentRevision: item.revision, + resolution: prompt.resolution + } + } + } + if (prompt.resolution.state !== 'pending') { + return { + ok: false, + refusal: { + code: 'agent_session_already_resolved', + message: `Item ${input.itemId} was already ${prompt.resolution.state}.`, + currentRevision: item.revision, + resolution: prompt.resolution + } + } + } + if (!prompt.options.some((option) => option.id === input.optionId)) { + return invalid(`Option ${input.optionId} is not offered by item ${input.itemId}.`) + } + const identity = parseAgentJournalItemKey(input.itemId) + if (!identity) { + return invalid(`Item id ${input.itemId} is not a well-formed item key.`) + } + + const resolution: AgentJournalResolution = { + state: 'resolved', + selectedOptionId: input.optionId, + resolvedBy: ctx.resolvedBy, + resolvedAt: ctx.now() + } + const appended = await ctx.journal.appendItem( + identity, + { ...item.body, resolution }, + { + fence: ctx.fence + } + ) + ctx.publish() + + try { + await ctx.adapter.answerPrompt({ + sessionId: ctx.sessionId, + itemId: input.itemId, + kind: input.kind, + optionId: input.optionId, + fence: ctx.fence + }) + } catch (error) { + // The answer is committed and will not be offered again; say so rather than + // reopening the prompt and risking a second callback. + await appendStatus( + ctx, + `${input.itemId}#delivery`, + `Your answer was recorded but the agent did not confirm it: ${ + error instanceof Error ? error.message : String(error) + }` + ) + } + return { + ok: true, + value: { itemId: appended.itemId, revision: appended.revision, resolution } + } +} + +/** Options live on the provider, not in the journal, so this writes nothing. */ +export async function performSetOption( + ctx: AgentSessionTurnContext, + input: { key: string; value: string } +): Promise> { + await ctx.adapter.setOption({ sessionId: ctx.sessionId, ...input, fence: ctx.fence }) + return { ok: true, value: input } +} diff --git a/src/main/runtime/agent-session-operation-admission.ts b/src/main/runtime/agent-session-operation-admission.ts new file mode 100644 index 00000000000..520e75a03c3 --- /dev/null +++ b/src/main/runtime/agent-session-operation-admission.ts @@ -0,0 +1,33 @@ +// Ledger admission for mutations that are not reservations — send, cancel, an +// approval answer. Split from the store so the store keeps only the transaction. + +import { + agentSessionOperationKey, + evaluateAgentSessionOperation, + pruneAgentSessionOperationRows, + type AgentSessionOperationDecision, + type AgentSessionOperationRow +} from '../../shared/agent-session-operation-ledger' + +export type AgentSessionOperationAdmission = { + callerKey: string + operationId: string + fingerprint: string + now: number +} + +type OperationRows = Map + +/** Prune, evaluate, and (on admit) place the row. The caller runs this inside one + * transaction, so two concurrent copies of an operation id cannot both admit. */ +export function admitAgentSessionOperationRow( + rows: OperationRows, + args: AgentSessionOperationAdmission +): { rows: OperationRows; decision: AgentSessionOperationDecision } { + const pruned = pruneAgentSessionOperationRows(rows, args.now) + const decision = evaluateAgentSessionOperation({ rows: pruned, ...args }) + if (decision.decision === 'admit') { + pruned.set(agentSessionOperationKey(args.callerKey, args.operationId), decision.row) + } + return { rows: pruned, decision } +} diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index 3f70fa44a73..15fcef3c85e 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -12,9 +12,14 @@ import { pruneAgentSessionOperationRows, agentSessionOperationKey, + type AgentSessionOperationDecision, type AgentSessionOperationOutcome, type AgentSessionOperationRow } from '../../shared/agent-session-operation-ledger' +import { + admitAgentSessionOperationRow, + type AgentSessionOperationAdmission +} from './agent-session-operation-admission' import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication' import { classifyObservedAgentSessionSpawnToken } from '../../shared/agent-session-lease-adjudication' import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' @@ -276,6 +281,17 @@ export class AgentSessionRecordStore { }) } + /** Admits one non-reservation mutation through the durable ledger. */ + async admitOperation( + args: AgentSessionOperationAdmission + ): Promise { + return this.transact(() => { + const admitted = admitAgentSessionOperationRow(this.state.operations, args) + this.state.operations = admitted.rows + return admitted.decision + }) + } + async recordOperationOutcome(args: { callerKey: string operationId: string