diff --git a/src/main/native-chat/agent-session-journal/command-receipt-schema.test.ts b/src/main/native-chat/agent-session-journal/command-receipt-schema.test.ts new file mode 100644 index 00000000000..6c5fbcfa452 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/command-receipt-schema.test.ts @@ -0,0 +1,174 @@ +import { mkdtemp, readFile, rm, stat } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import Database from '../../sqlite/sync-database' +import { journalPragmaNumber, openJournalDatabase, runJournalTransaction } from './journal-database' +import { + createAgentSessionRecordTablesSql, + createJournalTablesSql, + JOURNAL_DB_SCHEMA_VERSION +} from './journal-database-schema' +import { journalDatabasePath } from './journal-host-database' +import { + closeTestJournalHostDatabases, + openTestJournalHostDatabase +} from './journal-host-database-test-support' +import { insertCommandReceiptIfAbsent, readCommandReceipt } from './command-receipt-table' +import { commandReceiptScope } from './command-receipt-schema' +import { + commandReceiptFixture, + writeCommandReceiptTestRecord +} from './command-receipt-test-support' +import { AgentSessionJournalError } from './journal-write-guards' +import { buildCommandReceiptTransaction } from './command-receipt-transaction' + +let root: string +let dbPath: string +const scope = commandReceiptScope('caller-1', 'global') + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-command-receipt-schema-')) + dbPath = journalDatabasePath(root) +}) + +afterEach(async () => { + closeTestJournalHostDatabases() + await rm(root, { recursive: true, force: true }) +}) + +function hasReceiptTable(db: Database.Database): boolean { + return Boolean( + db + .prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?") + .get('agent_session_command_receipts') + ) +} + +describe('command receipt schema at writable open', () => { + it('creates the operation/caller primary key and session index on a fresh host database', () => { + const database = openTestJournalHostDatabase(root) + expect(hasReceiptTable(database.db)).toBe(true) + expect(journalPragmaNumber(database.db, 'user_version')).toBe(JOURNAL_DB_SCHEMA_VERSION) + expect(database.db.prepare('PRAGMA table_info(agent_session_command_receipts)').all()).toEqual([ + expect.objectContaining({ name: 'operation_id', pk: 1 }), + expect.objectContaining({ name: 'session_id', pk: 0 }), + expect.objectContaining({ name: 'caller_key', pk: 2 }), + expect.objectContaining({ name: 'method', pk: 0 }), + expect.objectContaining({ name: 'fingerprint', pk: 0 }), + expect.objectContaining({ name: 'status', pk: 0 }), + expect.objectContaining({ name: 'result_json', pk: 0 }), + expect.objectContaining({ name: 'rejection_json', pk: 0 }), + expect.objectContaining({ name: 'accepted_at', pk: 0 }) + ]) + const indexes = database.db.prepare('PRAGMA index_list(agent_session_command_receipts)').all() + expect(indexes).toHaveLength(2) + expect(indexes).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + name: 'sqlite_autoindex_agent_session_command_receipts_1', + origin: 'pk' + }), + expect.objectContaining({ name: 'agent_session_command_receipts_session', origin: 'c' }) + ]) + ) + expect( + database.db.prepare('PRAGMA index_info(agent_session_command_receipts_session)').all() + ).toEqual([expect.objectContaining({ name: 'session_id' })]) + expect( + database.db.prepare('PRAGMA foreign_key_list(agent_session_command_receipts)').all() + ).toEqual([ + expect.objectContaining({ + table: 'agent_session_records', + from: 'session_id', + to: 'session_id', + on_delete: 'CASCADE' + }) + ]) + }) + + it.each([3, 4])('adds the table to an older-shaped version %i database', (version) => { + const seeded = new Database(dbPath) + seeded.exec(createJournalTablesSql()) + if (version === 4) { + seeded.exec(createAgentSessionRecordTablesSql()) + writeCommandReceiptTestRecord(seeded) + } + seeded.pragma(`user_version = ${version}`) + expect(hasReceiptTable(seeded)).toBe(false) + seeded.close() + + for (let open = 0; open < 2; open += 1) { + const { db, readOnly } = openJournalDatabase(dbPath) + try { + expect(readOnly).toBe(false) + expect(hasReceiptTable(db)).toBe(true) + expect(journalPragmaNumber(db, 'user_version')).toBe(4) + if (version === 4) { + expect(db.prepare('SELECT session_id FROM agent_session_records').all()).toEqual([ + { session_id: 'session-1' } + ]) + } + } finally { + db.close() + } + } + }) +}) + +describe('command receipts on a newer database', () => { + it('reads receipts, refuses writes with the journal guard and leaves the file byte-identical', async () => { + const seeded = openJournalDatabase(dbPath).db + const receipt = commandReceiptFixture() + runJournalTransaction(seeded, (db) => { + writeCommandReceiptTestRecord(db) + insertCommandReceiptIfAbsent(db, scope, receipt) + }) + seeded.pragma(`user_version = ${JOURNAL_DB_SCHEMA_VERSION + 1}`) + seeded.close() + const before = await readFile(dbPath) + + const { db, readOnly } = openJournalDatabase(dbPath) + try { + expect(readOnly).toBe(true) + expect(readCommandReceipt(db, scope, receipt.operationId)).toEqual({ + verdict: 'readable', + receipt + }) + const write = () => insertCommandReceiptIfAbsent(db, scope, receipt) + expect(write).toThrow(AgentSessionJournalError) + expect(write).toThrow(expect.objectContaining({ code: 'journal_read_only' })) + expect(() => buildCommandReceiptTransaction(scope, receipt).write(db)).toThrow( + expect.objectContaining({ code: 'journal_read_only' }) + ) + } finally { + db.close() + } + + expect(await readFile(dbPath)).toEqual(before) + expect((await stat(`${dbPath}-wal`).catch(() => null))?.size ?? 0).toBe(0) + }) + + it('does not create a missing table or mistake unavailable proof for an absent command', async () => { + const seeded = new Database(dbPath) + seeded.exec(createJournalTablesSql()) + seeded.exec(createAgentSessionRecordTablesSql()) + seeded.pragma(`user_version = ${JOURNAL_DB_SCHEMA_VERSION + 1}`) + seeded.close() + const before = await readFile(dbPath) + + const { db } = openJournalDatabase(dbPath) + try { + const receipt = commandReceiptFixture() + expect(hasReceiptTable(db)).toBe(false) + expect(readCommandReceipt(db, scope, receipt.operationId)).toEqual({ + verdict: 'unreadable', + scope, + operationId: receipt.operationId + }) + } finally { + db.close() + } + expect(await readFile(dbPath)).toEqual(before) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/command-receipt-schema.ts b/src/main/native-chat/agent-session-journal/command-receipt-schema.ts new file mode 100644 index 00000000000..dfadddbb063 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/command-receipt-schema.ts @@ -0,0 +1,96 @@ +import { isDeepStrictEqual } from 'node:util' +import { z } from 'zod' +import type Database from '../../sqlite/sync-database' +import { readAgentSessionRefusalReference } from '../../../shared/agent-session-wire-refusals' + +export const commandReceiptScopeSchema = z.discriminatedUnion('kind', [ + z.strictObject({ kind: z.literal('global') }), + z.strictObject({ kind: z.literal('caller'), callerKey: z.string().min(1) }) +]) + +export type CommandReceiptScope = z.infer + +/** Sends look across callers; other methods look up only the authenticated caller's row. */ +export function commandReceiptScope( + callerKey: string, + operationIdScope?: 'global' +): CommandReceiptScope { + return operationIdScope === 'global' ? { kind: 'global' } : { kind: 'caller', callerKey } +} + +/** The pointer locates the result; if its row is gone, the command stays spent: read what remains, else unknown, never rerun. */ +export const commandReceiptResultSchema = z.discriminatedUnion('kind', [ + z.strictObject({ + kind: z.literal('journal-row'), + epoch: z.string().min(1), + sequence: z.int().positive() + }), + z.strictObject({ + kind: z.literal('no-op'), + outcome: z.discriminatedUnion('kind', [ + z.strictObject({ + kind: z.literal('cancel'), + cancelled: z.literal(false), + turnId: z.string().optional() + }), + z.strictObject({ kind: z.literal('queue-resume'), resumed: z.literal(false) }) + ]) + }) +]) + +export type CommandReceiptResult = z.infer + +const rejectionSchema = z + .strictObject({ reference: z.unknown(), message: z.string().optional() }) + .transform((value, ctx) => { + const reference = readAgentSessionRefusalReference(value.reference) + if (!reference || !isDeepStrictEqual(reference, value.reference)) { + ctx.addIssue({ code: 'custom', message: 'unreadable command refusal' }) + return z.NEVER + } + return { reference, ...(value.message !== undefined ? { message: value.message } : {}) } + }) + +const receiptIdentity = { + operationId: z.string().min(1), + sessionId: z.string().min(1), + callerKey: z.string().min(1), + method: z.string().min(1), + fingerprint: z.string().min(1), + acceptedAt: z.int().nonnegative() +} + +export const commandReceiptSchema = z.discriminatedUnion('status', [ + z.strictObject({ + ...receiptIdentity, + status: z.literal('accepted'), + result: commandReceiptResultSchema + }), + z.strictObject({ + ...receiptIdentity, + status: z.literal('rejected'), + rejection: rejectionSchema + }) +]) + +export type CommandReceipt = z.infer + +/** Additive and inert until cutover; a version bump here would prevent older builds writing. */ +export function ensureCommandReceiptsTable(db: Database.Database): void { + db.exec(` +CREATE TABLE IF NOT EXISTS agent_session_command_receipts ( + operation_id TEXT NOT NULL, + session_id TEXT NOT NULL REFERENCES agent_session_records(session_id) ON DELETE CASCADE, + caller_key TEXT NOT NULL, + method TEXT NOT NULL, + fingerprint TEXT NOT NULL, + status TEXT NOT NULL, + result_json TEXT, + rejection_json TEXT, + accepted_at INTEGER NOT NULL, + PRIMARY KEY (operation_id, caller_key) +); +CREATE INDEX IF NOT EXISTS agent_session_command_receipts_session + ON agent_session_command_receipts (session_id); +`) +} diff --git a/src/main/native-chat/agent-session-journal/command-receipt-table.test.ts b/src/main/native-chat/agent-session-journal/command-receipt-table.test.ts new file mode 100644 index 00000000000..5fb26b71e1d --- /dev/null +++ b/src/main/native-chat/agent-session-journal/command-receipt-table.test.ts @@ -0,0 +1,348 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { JournalHostDatabase } from './journal-host-database' +import { + closeTestJournalHostDatabases, + openTestJournalHostDatabase +} from './journal-host-database-test-support' +import { + commandReceiptScope, + type CommandReceipt, + type CommandReceiptScope +} from './command-receipt-schema' +import { insertCommandReceiptIfAbsent, readCommandReceipt } from './command-receipt-table' +import { + commandReceiptFixture, + writeCommandReceiptTestRecord +} from './command-receipt-test-support' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' + +let root: string +let database: JournalHostDatabase +const globalScope = commandReceiptScope('caller-1', 'global') + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-command-receipt-table-')) + database = openTestJournalHostDatabase(root) + database.transaction((db) => writeCommandReceiptTestRecord(db)) +}) + +afterEach(async () => { + closeTestJournalHostDatabases() + await rm(root, { recursive: true, force: true }) +}) + +function insert(receipt: CommandReceipt, scope: CommandReceiptScope = globalScope) { + return database.transaction((db) => insertCommandReceiptIfAbsent(db, scope, receipt)) +} + +function read(receipt: CommandReceipt, scope: CommandReceiptScope = globalScope) { + return readCommandReceipt(database.db, scope, receipt.operationId) +} + +function cancelReceipt(callerKey: string, turnId = 'turn-1') { + return commandReceiptFixture({ + callerKey, + method: 'agentSession.cancel', + fingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.cancel', + sessionId: 'session-1', + fields: { turnId } + }), + result: { kind: 'no-op', outcome: { kind: 'cancel', cancelled: false, turnId } } + }) +} + +describe('command receipt identity', () => { + it('inserts a send once and replays the original receipt across callers with the same fingerprint', () => { + const receipt = commandReceiptFixture() + expect(read(receipt)).toEqual({ verdict: 'absent' }) + expect(insert(receipt)).toEqual({ inserted: true }) + expect(insert({ ...receipt, acceptedAt: 2000, callerKey: 'caller-2' })).toEqual({ + inserted: false, + reason: 'duplicate', + existing: { verdict: 'readable', receipt } + }) + expect(read(receipt)).toEqual({ verdict: 'readable', receipt }) + }) + + it('reports a changed canonical hash, including a different method or chat, as conflict', () => { + const receipt = commandReceiptFixture() + insert(receipt) + database.transaction((db) => writeCommandReceiptTestRecord(db, 'session-2')) + for (const target of [ + { method: 'agentSession.cancel', sessionId: 'session-1' }, + { method: 'agentSession.send', sessionId: 'session-2' } + ]) { + const changed = commandReceiptFixture({ + ...target, + fingerprint: computeAgentSessionPayloadFingerprint({ ...target, fields: {} }) + }) + expect(insert(changed)).toEqual({ + inserted: false, + reason: 'conflict', + existing: { verdict: 'readable', receipt } + }) + } + expect(read(receipt)).toEqual({ verdict: 'readable', receipt }) + }) + + it('conflicts when the same caller sends and then stops with the same id', () => { + const send = commandReceiptFixture() + const stop = cancelReceipt(send.callerKey) + const scope = commandReceiptScope(stop.callerKey) + expect(insert(send)).toEqual({ inserted: true }) + expect(read(stop, scope)).toEqual({ verdict: 'readable', receipt: send }) + expect(insert(stop, scope)).toEqual({ + inserted: false, + reason: 'conflict', + existing: { verdict: 'readable', receipt: send } + }) + }) + + it('conflicts when one caller stops and another sends with the same id', () => { + const stop = cancelReceipt('caller-1') + expect(insert(stop, commandReceiptScope(stop.callerKey))).toEqual({ inserted: true }) + expect(insert(commandReceiptFixture({ callerKey: 'caller-2' }))).toEqual({ + inserted: false, + reason: 'conflict', + existing: { verdict: 'readable', receipt: stop } + }) + }) + + it('allows separate callers to stop with the same id', () => { + for (const callerKey of ['caller-1', 'caller-2']) { + const receipt = cancelReceipt(callerKey) + const scope = commandReceiptScope(callerKey) + expect(insert(receipt, scope)).toEqual({ inserted: true }) + expect(read(receipt, scope)).toEqual({ verdict: 'readable', receipt }) + } + }) + + it('allows another caller to stop with an id already used by a send', () => { + const send = commandReceiptFixture() + const stop = cancelReceipt('caller-2') + const scope = commandReceiptScope(stop.callerKey) + expect(insert(send)).toEqual({ inserted: true }) + expect(read(stop, scope)).toEqual({ verdict: 'absent' }) + expect(insert(stop, scope)).toEqual({ inserted: true }) + expect(read(stop, scope)).toEqual({ verdict: 'readable', receipt: stop }) + }) + + it('treats a caller key literally named global as an ordinary caller', () => { + const stop = cancelReceipt('global') + const scope = commandReceiptScope(stop.callerKey) + expect(insert(stop, scope)).toEqual({ inserted: true }) + expect(insert(cancelReceipt('caller-2'), commandReceiptScope('caller-2'))).toEqual({ + inserted: true + }) + expect(read(stop, scope)).toEqual({ verdict: 'readable', receipt: stop }) + expect(insert(commandReceiptFixture())).toMatchObject({ inserted: false, reason: 'conflict' }) + }) + + it.each([500, 1000])( + 'orders global candidates by insertion and checks every fingerprint (%i)', + (acceptedAt) => { + const first = cancelReceipt('caller-b', 'turn-b') + const later = { ...cancelReceipt('caller-a', 'turn-a'), acceptedAt } + expect(insert(first, commandReceiptScope(first.callerKey))).toEqual({ inserted: true }) + expect(insert(later, commandReceiptScope(later.callerKey))).toEqual({ inserted: true }) + expect(read(later)).toEqual({ verdict: 'readable', receipt: first }) + expect(insert({ ...later, callerKey: 'caller-c' })).toEqual({ + inserted: false, + reason: 'duplicate', + existing: { verdict: 'readable', receipt: later } + }) + expect(insert(commandReceiptFixture())).toEqual({ + inserted: false, + reason: 'conflict', + existing: { verdict: 'readable', receipt: first } + }) + } + ) + + it('refuses a caller that does not match its namespace', () => { + const receipt = commandReceiptFixture() + expect(() => insert(receipt, commandReceiptScope('caller-2'))).toThrow( + 'command caller does not match its scope' + ) + expect(read(receipt)).toEqual({ verdict: 'absent' }) + }) + + it('requires the chat record to exist and does not leave an orphan receipt', () => { + const receipt = commandReceiptFixture({ sessionId: 'missing-chat' }) + expect(() => insert(receipt)).toThrow(/FOREIGN KEY/i) + expect(read(receipt)).toEqual({ verdict: 'absent' }) + }) + + it('requires a receipt to be written inside an open transaction', () => { + const receipt = commandReceiptFixture() + expect(() => insertCommandReceiptIfAbsent(database.db, globalScope, receipt)).toThrow( + /inside an open transaction/ + ) + expect(read(receipt)).toEqual({ verdict: 'absent' }) + }) + + it('throws and rolls back an unexpected key collision after an absent lookup', () => { + database.db + .exec(`CREATE TRIGGER collide_command_receipt BEFORE INSERT ON agent_session_command_receipts + BEGIN + INSERT INTO agent_session_command_receipts + (operation_id, session_id, caller_key, method, fingerprint, status, result_json, rejection_json, accepted_at) + VALUES (NEW.operation_id, NEW.session_id, NEW.caller_key, NEW.method, NEW.fingerprint, + NEW.status, NEW.result_json, NEW.rejection_json, NEW.accepted_at); + END`) + const receipt = commandReceiptFixture() + expect(() => insert(receipt)).toThrow(/UNIQUE constraint/i) + expect(read(receipt)).toEqual({ verdict: 'absent' }) + expect(database.db.isTransaction).toBe(false) + }) +}) + +describe('command receipt outcomes and lifetime', () => { + it.each([ + { kind: 'cancel', cancelled: false, turnId: 'turn-1' }, + { kind: 'cancel', cancelled: false }, + { kind: 'queue-resume', resumed: false } + ] as const)('reads the small no-op outcome $kind', (outcome) => { + const receipt = commandReceiptFixture({ result: { kind: 'no-op', outcome } }) + insert(receipt) + expect(read(receipt)).toEqual({ verdict: 'readable', receipt }) + }) + + it('retains a refused outcome with its code, message and details', () => { + const { result: _result, ...identity } = commandReceiptFixture() + const receipt: CommandReceipt = { + ...identity, + status: 'rejected', + rejection: { + reference: { code: 'agent_session_operation_invalid', details: { reason: 'promptGone' } }, + message: 'The prompt is no longer available.' + } + } + insert(receipt) + expect(read(receipt)).toEqual({ verdict: 'readable', receipt }) + expect(insert(commandReceiptFixture())).toMatchObject({ inserted: false, reason: 'duplicate' }) + }) + + it('preserves receipts across record upserts and cascades only on record deletion', () => { + const receipt = commandReceiptFixture() + insert(receipt) + database.transaction((db) => writeCommandReceiptTestRecord(db, 'session-1', '{"updated":true}')) + expect(read(receipt)).toEqual({ verdict: 'readable', receipt }) + database.transaction((db) => writeCommandReceiptTestRecord(db, 'session-1', null)) + expect(read(receipt)).toEqual({ verdict: 'absent' }) + }) +}) + +describe('unreadable command receipts', () => { + it.each([ + ['result_json', '{'], + ['result_json', '{"kind":"journal-row","epoch":"e1","sequence":0}'], + ['result_json', '{"kind":"future-result"}'], + ['result_json', null], + ['status', 'pending'], + ['method', ''], + ['fingerprint', ''], + ['accepted_at', -1], + ['caller_key', ''], + ['rejection_json', '{}'] + ])('preserves the key when %s holds malformed data %s', (column, value) => { + const receipt = commandReceiptFixture() + insert(receipt) + database.db.prepare(`UPDATE agent_session_command_receipts SET ${column} = ?`).run(value) + const existing = { + verdict: 'unreadable', + scope: globalScope, + operationId: receipt.operationId + } + expect(read(receipt)).toEqual(existing) + expect(insert(receipt)).toEqual({ inserted: false, reason: 'unreadable', existing }) + }) + + it('treats a malformed rejection as unreadable rather than granting another execution', () => { + const receipt = commandReceiptFixture() + insert(receipt) + database.db + .prepare(`UPDATE agent_session_command_receipts SET status = 'rejected', result_json = NULL, + rejection_json = ? WHERE caller_key = ? AND operation_id = ?`) + .run( + JSON.stringify({ + reference: { code: 'agent_session_operation_invalid', details: { reason: 'madeUp' } } + }), + receipt.callerKey, + receipt.operationId + ) + expect(read(receipt)).toMatchObject({ verdict: 'unreadable' }) + expect(insert(receipt)).toMatchObject({ inserted: false, reason: 'unreadable' }) + }) + + it('refuses a global lookup if any candidate is unreadable, even after a matching receipt', () => { + const readable = cancelReceipt('caller-a') + const unreadable = cancelReceipt('caller-b') + insert(readable, commandReceiptScope(readable.callerKey)) + insert(unreadable, commandReceiptScope(unreadable.callerKey)) + database.db + .prepare("UPDATE agent_session_command_receipts SET result_json = '{' WHERE caller_key = ?") + .run(unreadable.callerKey) + expect(read(readable, commandReceiptScope(readable.callerKey))).toEqual({ + verdict: 'readable', + receipt: readable + }) + const existing = { + verdict: 'unreadable', + scope: globalScope, + operationId: readable.operationId + } + expect(read(readable)).toEqual(existing) + expect(insert({ ...readable, callerKey: 'caller-c' })).toEqual({ + inserted: false, + reason: 'unreadable', + existing + }) + }) + + it('reads an oversized accepted_at integer as unreadable', () => { + const receipt = commandReceiptFixture() + insert(receipt) + database.db.exec('UPDATE agent_session_command_receipts SET accepted_at = 9007199254740993') + expect(read(receipt)).toEqual({ + verdict: 'unreadable', + scope: globalScope, + operationId: receipt.operationId + }) + expect(insert(receipt)).toMatchObject({ inserted: false, reason: 'unreadable' }) + }) + + it('reads a table missing an expected column as unreadable', () => { + database.db.exec( + 'ALTER TABLE agent_session_command_receipts RENAME COLUMN result_json TO future_result_json' + ) + const receipt = commandReceiptFixture() + for (const scope of [globalScope, commandReceiptScope(receipt.callerKey)]) { + expect(read(receipt, scope)).toEqual({ + verdict: 'unreadable', + scope, + operationId: receipt.operationId + }) + expect(insert(receipt, scope)).toMatchObject({ inserted: false, reason: 'unreadable' }) + } + }) + + it.each([ + [5, 'database is locked'], + [10, 'disk I/O error'] + ] as const)('propagates operational SQLite error %i', (errcode, message) => { + const failure = Object.assign(new Error(message), { code: 'ERR_SQLITE_ERROR', errcode }) + const prepare = vi.spyOn(database.db, 'prepare').mockImplementationOnce(() => { + throw failure + }) + try { + expect(() => read(commandReceiptFixture())).toThrow(failure) + } finally { + prepare.mockRestore() + } + }) +}) diff --git a/src/main/native-chat/agent-session-journal/command-receipt-table.ts b/src/main/native-chat/agent-session-journal/command-receipt-table.ts new file mode 100644 index 00000000000..3b8c2d38bc5 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/command-receipt-table.ts @@ -0,0 +1,170 @@ +import type Database from '../../sqlite/sync-database' +import { journalPragmaNumber } from './journal-database' +import { JOURNAL_DB_SCHEMA_VERSION } from './journal-database-schema' +import { assertJournalWritable } from './journal-write-guards' +import { + commandReceiptSchema, + commandReceiptScopeSchema, + type CommandReceipt, + type CommandReceiptScope +} from './command-receipt-schema' + +export type CommandReceiptRead = + | { verdict: 'absent' } + | { verdict: 'readable'; receipt: CommandReceipt } + | { verdict: 'unreadable'; scope: CommandReceiptScope; operationId: string } + +export type ExistingCommandReceipt = Exclude + +export type CommandReceiptInsert = + | { inserted: true } + | { + inserted: false + reason: 'duplicate' | 'conflict' + existing: Extract + } + | { + inserted: false + reason: 'unreadable' + existing: Extract + } + +const SELECT_RECEIPT = `SELECT session_id, caller_key, method, fingerprint, status, + result_json, rejection_json, accepted_at FROM agent_session_command_receipts` + +type CommandReceiptCandidates = + | { verdict: 'readable'; receipts: CommandReceipt[] } + | Extract + +function parseReceiptJson(value: unknown): unknown { + return typeof value === 'string' ? JSON.parse(value) : undefined +} + +function readCommandReceiptCandidates( + db: Database.Database, + scope: CommandReceiptScope, + operationId: string +): CommandReceiptCandidates { + const unreadable: Extract = { + verdict: 'unreadable', + scope, + operationId + } + try { + // A newer read-only database may not expose this table; that cannot authorize a retry. + if ( + !db + .prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?") + .get('agent_session_command_receipts') + ) { + return unreadable + } + const rows = + scope.kind === 'caller' + ? db + .prepare(`${SELECT_RECEIPT} WHERE caller_key = ? AND operation_id = ?`) + .all(scope.callerKey, operationId) + : db.prepare(`${SELECT_RECEIPT} WHERE operation_id = ? ORDER BY rowid`).all(operationId) + const receipts: CommandReceipt[] = [] + for (const row of rows) { + if ( + (row.status === 'accepted' && row.rejection_json !== null) || + (row.status === 'rejected' && row.result_json !== null) + ) { + return unreadable + } + const parsed = commandReceiptSchema.safeParse({ + operationId, + sessionId: row.session_id, + callerKey: row.caller_key, + method: row.method, + fingerprint: row.fingerprint, + status: row.status, + acceptedAt: row.accepted_at, + ...(row.status === 'accepted' + ? { result: parseReceiptJson(row.result_json) } + : { rejection: parseReceiptJson(row.rejection_json) }) + }) + if (!parsed.success) { + return unreadable + } + receipts.push(parsed.data) + } + return { verdict: 'readable', receipts } + } catch (error) { + if ( + error instanceof RangeError || + error instanceof SyntaxError || + (error instanceof Error && + 'code' in error && + error.code === 'ERR_SQLITE_ERROR' && + /^no such (?:column|table):/.test(error.message)) + ) { + return unreadable + } + throw error + } +} + +export function readCommandReceipt( + db: Database.Database, + scope: CommandReceiptScope, + operationId: string +): CommandReceiptRead { + const candidates = readCommandReceiptCandidates(db, scope, operationId) + if (candidates.verdict === 'unreadable') { + return candidates + } + const receipt = candidates.receipts[0] + return receipt ? { verdict: 'readable', receipt } : { verdict: 'absent' } +} + +/** The caller owns the effect's transaction; this function never reserves or commits on its own. */ +export function insertCommandReceiptIfAbsent( + db: Database.Database, + scope: CommandReceiptScope, + receipt: CommandReceipt +): CommandReceiptInsert { + assertJournalWritable( + journalPragmaNumber(db, 'user_version') > JOURNAL_DB_SCHEMA_VERSION, + receipt.sessionId + ) + if (!db.isTransaction) { + throw new Error('a command receipt must be written inside an open transaction') + } + const written = commandReceiptSchema.parse(receipt) + const claim = commandReceiptScopeSchema.parse(scope) + if (claim.kind === 'caller' && claim.callerKey !== written.callerKey) { + throw new Error('command caller does not match its scope') + } + const candidates = readCommandReceiptCandidates(db, claim, written.operationId) + if (candidates.verdict === 'unreadable') { + return { inserted: false, reason: 'unreadable', existing: candidates } + } + const duplicate = candidates.receipts.find( + (candidate) => candidate.fingerprint === written.fingerprint + ) + const existing = duplicate ?? candidates.receipts[0] + if (existing) { + return { + inserted: false, + existing: { verdict: 'readable', receipt: existing }, + reason: duplicate ? 'duplicate' : 'conflict' + } + } + db.prepare(`INSERT INTO agent_session_command_receipts + (operation_id, session_id, caller_key, method, fingerprint, status, + result_json, rejection_json, accepted_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`).run( + written.operationId, + written.sessionId, + written.callerKey, + written.method, + written.fingerprint, + written.status, + written.status === 'accepted' ? JSON.stringify(written.result) : null, + written.status === 'rejected' ? JSON.stringify(written.rejection) : null, + written.acceptedAt + ) + return { inserted: true } +} diff --git a/src/main/native-chat/agent-session-journal/command-receipt-test-support.ts b/src/main/native-chat/agent-session-journal/command-receipt-test-support.ts new file mode 100644 index 00000000000..b26b8e1f45c --- /dev/null +++ b/src/main/native-chat/agent-session-journal/command-receipt-test-support.ts @@ -0,0 +1,40 @@ +import type Database from '../../sqlite/sync-database' +import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' +import { writeAgentSessionStoreRows } from '../../runtime/agent-session-record-rows' +import type { CommandReceipt } from './command-receipt-schema' + +type AcceptedReceipt = Extract + +export function commandReceiptFixture(overrides: Partial = {}): AcceptedReceipt { + return { + operationId: 'operation-1', + sessionId: 'session-1', + callerKey: 'caller-1', + method: 'agentSession.send', + fingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.send', + sessionId: 'session-1', + fields: { body: { text: 'hello' } } + }), + acceptedAt: 1000, + status: 'accepted', + result: { kind: 'journal-row', epoch: 'epoch-1', sequence: 1 }, + ...overrides + } +} + +export function writeCommandReceiptTestRecord( + db: Database.Database, + sessionId = 'session-1', + recordJson: string | null = '{}' +): void { + writeAgentSessionStoreRows(db, { + records: { + upsert: recordJson === null ? [] : [[sessionId, recordJson]], + remove: recordJson === null ? [sessionId] : [] + }, + operations: { upsert: [], remove: [] }, + retiredClaimKeys: null, + sessionTabs: null + }) +} diff --git a/src/main/native-chat/agent-session-journal/command-receipt-transaction.test.ts b/src/main/native-chat/agent-session-journal/command-receipt-transaction.test.ts new file mode 100644 index 00000000000..7c98aa5c716 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/command-receipt-transaction.test.ts @@ -0,0 +1,188 @@ +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 { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' +import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import type { JournalHostDatabase } from './journal-host-database' +import { + closeTestJournalHostDatabases, + openTestJournalHostDatabase +} from './journal-host-database-test-support' +import { replaceJournalEpoch } from './journal-epoch-replacement' +import { runJournalTransaction } from './journal-database' +import { + deleteJournalEpochRows, + insertJournalRow, + publishJournalSessionEpoch +} from './journal-row-table' +import type { JournalRow } from './journal-row-schema' +import { JournalRowWriter } from './journal-row-writer' +import { JournalWriteQueue } from './journal-write-queue' +import { insertCommandReceiptIfAbsent, readCommandReceipt } from './command-receipt-table' +import { commandReceiptScope, type CommandReceiptResult } from './command-receipt-schema' +import { + commandReceiptFixture, + writeCommandReceiptTestRecord +} from './command-receipt-test-support' +import { + buildCommandReceiptTransaction, + CommandReceiptExistsError +} from './command-receipt-transaction' + +let root: string +let database: JournalHostDatabase +const receipt = commandReceiptFixture() +const scope = commandReceiptScope(receipt.callerKey, 'global') +const identity = { + sessionId: 'session-1', + workspaceId: 'workspace-1', + hostId: 'host-1', + agent: 'codex' as const, + providerHandle: codexProviderHandle('thread-1') +} + +function epochRow(seq: number, ts: number): JournalRow { + return { + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + epoch: 'epoch-1', + seq, + ts, + fence: 0, + kind: 'epoch', + reason: 'session_created', + providerHandle: { kind: 'codex', threadId: 'thread-1' } + } +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-command-receipt-transaction-')) + database = openTestJournalHostDatabase(root) + database.transaction((db) => { + writeCommandReceiptTestRecord(db) + publishJournalSessionEpoch(db, identity, 'epoch-1') + }) +}) + +afterEach(async () => { + closeTestJournalHostDatabases() + await rm(root, { recursive: true, force: true }) +}) + +function storedRows() { + return database.db.prepare('SELECT seq FROM journal_rows ORDER BY seq').all() +} + +function readReceipt() { + return readCommandReceipt(database.db, scope, receipt.operationId) +} + +describe('command receipt transaction hook', () => { + it('rolls back the effect and receipt together if the enclosing transaction throws', () => { + const hook = buildCommandReceiptTransaction(scope, receipt) + expect(() => + runJournalTransaction(database.db, (db) => { + insertJournalRow(db, identity.sessionId, epochRow(1, 1000)) + hook.write(db) + throw new Error('effect failed') + }) + ).toThrow('effect failed') + expect(storedRows()).toEqual([]) + expect(readReceipt()).toEqual({ verdict: 'absent' }) + expect(database.db.isTransaction).toBe(false) + }) + + it('commits through the existing row writer with a pointer to the row it assigned', async () => { + const queue = new JournalWriteQueue(identity.sessionId) + const committed: JournalRow[] = [] + let result: CommandReceiptResult = { kind: 'journal-row', epoch: 'epoch-0', sequence: 99 } + const writer = new JournalRowWriter({ + sessionId: identity.sessionId, + now: () => 1000, + serialize: (run) => queue.serialize(run), + database: () => database, + readOnly: () => database.readOnly, + highestFence: () => 0, + nextSequence: () => 7, + commit: (row) => committed.push(row) + }) + await writer.enqueue( + epochRow, + (_db, row) => { + result = { kind: 'journal-row', epoch: row.epoch, sequence: row.seq } + }, + buildCommandReceiptTransaction(scope, () => ({ ...receipt, result })) + ) + expect(committed).toHaveLength(1) + expect(storedRows()).toEqual([{ seq: 7 }]) + expect(readReceipt()).toEqual({ + verdict: 'readable', + receipt: { ...receipt, result: { kind: 'journal-row', epoch: 'epoch-1', sequence: 7 } } + }) + }) + + it.each(['duplicate', 'conflict', 'unreadable'] as const)( + 'throws a typed %s verdict and rolls back the second effect', + (reason) => { + database.transaction((db) => insertCommandReceiptIfAbsent(db, scope, receipt)) + if (reason === 'unreadable') { + database.db.prepare("UPDATE agent_session_command_receipts SET result_json = '{'").run() + } + const hook = buildCommandReceiptTransaction(scope, { + ...receipt, + fingerprint: reason === 'conflict' ? 'changed' : receipt.fingerprint + }) + const write = () => + database.transaction((db) => { + insertJournalRow(db, identity.sessionId, epochRow(1, 1000)) + hook.write(db) + }) + expect(write).toThrow(CommandReceiptExistsError) + expect(write).toThrow( + expect.objectContaining({ result: expect.objectContaining({ reason }) }) + ) + expect(storedRows()).toEqual([]) + expect(readReceipt().verdict).toBe(reason === 'unreadable' ? 'unreadable' : 'readable') + } + ) +}) + +describe('command receipts across epoch replacement', () => { + it('survives deletion of its referenced journal epoch as retained spent proof', () => { + database.transaction((db) => { + insertJournalRow(db, identity.sessionId, epochRow(1, 1000)) + insertCommandReceiptIfAbsent(db, scope, receipt) + deleteJournalEpochRows(db, identity.sessionId, 'epoch-1') + }) + expect(storedRows()).toEqual([]) + expect(readReceipt()).toEqual({ verdict: 'readable', receipt }) + }) + + it('survives the actual replacement transaction without transferring or copying results', () => { + database.transaction((db) => { + insertJournalRow(db, identity.sessionId, epochRow(1, 1000)) + insertCommandReceiptIfAbsent(db, scope, receipt) + }) + replaceJournalEpoch({ + database, + identity, + reason: 'legacy_import', + fence: 0, + items: [], + queuePause: { lifted: false, liveStop: null, reopened: false }, + now: () => 2000, + mintEpoch: () => 'epoch-2', + onPublished: () => undefined + }) + expect(database.db.prepare('SELECT epoch FROM journal_rows').all()).toEqual([ + { epoch: 'epoch-2' } + ]) + expect(readReceipt()).toEqual({ verdict: 'readable', receipt }) + expect( + database.transaction((db) => insertCommandReceiptIfAbsent(db, scope, receipt)) + ).toMatchObject({ + inserted: false, + reason: 'duplicate' + }) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/command-receipt-transaction.ts b/src/main/native-chat/agent-session-journal/command-receipt-transaction.ts new file mode 100644 index 00000000000..c4b3330cc28 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/command-receipt-transaction.ts @@ -0,0 +1,32 @@ +import type { CommandReceipt, CommandReceiptScope } from './command-receipt-schema' +import { insertCommandReceiptIfAbsent, type CommandReceiptInsert } from './command-receipt-table' +import type { JournalOperationReceipt } from './journal-row-writer' + +type CommandReceiptNotInserted = Extract + +/** The enclosing effect rolls back; its caller replays, refuses conflict, or answers unknown. */ +export class CommandReceiptExistsError extends Error { + constructor(readonly result: CommandReceiptNotInserted) { + super(`command receipt already exists: ${result.reason}`) + this.name = 'CommandReceiptExistsError' + } +} + +export function buildCommandReceiptTransaction( + scope: CommandReceiptScope, + receipt: CommandReceipt | (() => CommandReceipt) +): JournalOperationReceipt { + return { + write: (db) => { + const result = insertCommandReceiptIfAbsent( + db, + scope, + typeof receipt === 'function' ? receipt() : receipt + ) + if (!result.inserted) { + throw new CommandReceiptExistsError(result) + } + }, + committed: () => undefined + } +} diff --git a/src/main/native-chat/agent-session-journal/journal-database.ts b/src/main/native-chat/agent-session-journal/journal-database.ts index ba961541a7e..0dee9537b3f 100644 --- a/src/main/native-chat/agent-session-journal/journal-database.ts +++ b/src/main/native-chat/agent-session-journal/journal-database.ts @@ -14,6 +14,7 @@ import { } from './journal-database-schema' import { JournalUnreleasedSchemaError } from './journal-open-failure' import { ensureQueuedMessagesTable } from './queued-message-schema' +import { ensureCommandReceiptsTable } from './command-receipt-schema' import { ensureAgentSessionAttachmentClaimTables } from '../agent-session-attachments/agent-session-attachment-claims' export const JOURNAL_BUSY_TIMEOUT_MS = 5000 @@ -81,6 +82,7 @@ export function openJournalDatabase(dbPath: string): OpenJournalDatabase { // already at the current version, and this table must exist at EVERY // writable open with no `user_version` bump (see `ensureQueuedMessagesTable`). ensureQueuedMessagesTable(probe) + ensureCommandReceiptsTable(probe) ensureAgentSessionAttachmentClaimTables(probe) hardenSqliteDatabaseFiles(dbPath) transferred = true