mirror of
https://github.com/stablyai/orca.git
synced 2026-10-31 09:07:40 +00:00
feat(native-chat): add a chat-owned command receipt store (#26653)
* feat(native-chat): add a chat-owned command receipt store Adds the agent_session_command_receipts table and an insert-if-absent receipt write that runs inside a command's own journal transaction. A receipt records the command's identity, canonical hash, status and a pointer to the row holding its result; it dies with its chat record. Nothing uses it yet: the existing operation ledger stays authoritative. * fix(native-chat): key command receipts like today's ledger ids Rows are keyed by caller and operation id; a send looks across callers and other methods look up only their caller's row, matching the existing ledger. Bad rows and newer table shapes read as unreadable instead of throwing. * fix(native-chat): order receipt lookups by insertion and key by operation id Global lookups now take rows in insertion order, so a read and an insert agree on which receipt answers an id regardless of the host clock. The primary key leads with the operation id, so its own index serves both lookups.
This commit is contained in:
Brennan Benson
@@ -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)
|
||||
})
|
||||
})
|
||||
@@ -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<typeof commandReceiptScopeSchema>
|
||||
|
||||
/** 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<typeof commandReceiptResultSchema>
|
||||
|
||||
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<typeof commandReceiptSchema>
|
||||
|
||||
/** 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);
|
||||
`)
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -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<CommandReceiptRead, { verdict: 'absent' }>
|
||||
|
||||
export type CommandReceiptInsert =
|
||||
| { inserted: true }
|
||||
| {
|
||||
inserted: false
|
||||
reason: 'duplicate' | 'conflict'
|
||||
existing: Extract<CommandReceiptRead, { verdict: 'readable' }>
|
||||
}
|
||||
| {
|
||||
inserted: false
|
||||
reason: 'unreadable'
|
||||
existing: Extract<CommandReceiptRead, { verdict: 'unreadable' }>
|
||||
}
|
||||
|
||||
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<CommandReceiptRead, { verdict: 'unreadable' }>
|
||||
|
||||
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<CommandReceiptRead, { verdict: 'unreadable' }> = {
|
||||
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 }
|
||||
}
|
||||
@@ -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<CommandReceipt, { status: 'accepted' }>
|
||||
|
||||
export function commandReceiptFixture(overrides: Partial<AcceptedReceipt> = {}): 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
|
||||
})
|
||||
}
|
||||
@@ -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'
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -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<CommandReceiptInsert, { inserted: false }>
|
||||
|
||||
/** 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
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user