From fc36d58bbb697e10ade0ecbd5df22e53bd2e7cf6 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Sun, 9 Aug 2026 13:35:19 -0700 Subject: [PATCH] Add the structured agent-session host One place where the lease, the journal, and the provider adapter meet: every mutating call recomputes the fingerprint, admits through the durable operation ledger, checks the lease, then acts, so no method can grow its own admission rules. Journal recovery rehydrates an unreadable journal from provider history into a fresh epoch and tells subscribers which reset they are seeing. Attach adjudicates the leases it loaded from disk before granting any writer. Without that step every lease stays unreconciled for the life of the process and a restarted host refuses every persisted session forever. --- .../agent-session-history-page.test.ts | 229 +++++++ .../agent-session-history-page.ts | 143 +++++ .../agent-session-journal-batch.ts | 79 +++ .../agent-session-journal-recovery.test.ts | 175 +++++ .../agent-session-journal-recovery.ts | 116 ++++ .../structured-agent-session-adapter.ts | 66 ++ .../structured-agent-session-attach-flow.ts | 127 ++++ .../structured-agent-session-attach.ts | 188 ++++++ .../structured-agent-session-host.test.ts | 603 ++++++++++++++++++ .../structured-agent-session-host.ts | 360 +++++++++++ ...structured-agent-session-mutation-plans.ts | 113 ++++ .../structured-agent-session-registry.ts | 18 + ...uctured-agent-session-restart-reconcile.ts | 43 ++ .../structured-agent-session-subscribers.ts | 130 ++++ .../structured-agent-session-turns.ts | 246 +++++++ .../agent-session-operation-admission.ts | 33 + .../runtime/agent-session-record-store.ts | 16 + 17 files changed, 2685 insertions(+) create mode 100644 src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts create mode 100644 src/main/native-chat/agent-session-wire/agent-session-history-page.ts create mode 100644 src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts create mode 100644 src/main/native-chat/agent-session-wire/agent-session-journal-recovery.test.ts create mode 100644 src/main/native-chat/agent-session-wire/agent-session-journal-recovery.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-host.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-registry.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts create mode 100644 src/main/runtime/agent-session-operation-admission.ts 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