From ccc32a137fcd98eeccb789a1408f48b5841b8348 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 8 Oct 2026 12:21:50 -0700 Subject: [PATCH 1/5] refactor(native-chat): read the chat queue by row headers instead of loading every message Every queue decision (next card, pauses, consume recheck, settlement, retention, head placement, admission sum) loaded and parsed every stored row of every state and cached the whole array per chat. Read indexed, body-free headers instead and stop at the first decision; bodies are read only for the card being handed off, explicit card actions, receipts, /clear's carry and the existing full-list publication. No behaviour change. --- .../journal-queued-message-reads.test.ts | 148 ++++++ .../journal-queued-messages.ts | 98 ++-- .../journal-unsent-send-hold.ts | 53 ++- .../queued-message-delivered-echo.test.ts | 18 +- .../queued-message-delivered-echo.ts | 40 +- .../queued-message-headers.test.ts | 426 ++++++++++++++++++ .../queued-message-headers.ts | 152 +++++++ .../queued-message-pause.ts | 31 +- .../queued-message-retention.ts | 7 +- .../queued-message-schema.ts | 6 + .../queued-message-settlement.ts | 78 ++-- .../queued-message-stored-row.ts | 42 +- .../queued-message-table.ts | 47 +- ...tructured-agent-session-queued-messages.ts | 30 +- ...ctured-agent-session-queued-publication.ts | 4 +- ...ed-agent-session-queued-published-bytes.ts | 6 +- ...ured-agent-session-queued-stop-row.test.ts | 2 +- .../structured-agent-session-queued-stop.ts | 12 +- 18 files changed, 1012 insertions(+), 188 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-queued-message-reads.test.ts create mode 100644 src/main/native-chat/agent-session-journal/queued-message-headers.test.ts create mode 100644 src/main/native-chat/agent-session-journal/queued-message-headers.ts diff --git a/src/main/native-chat/agent-session-journal/journal-queued-message-reads.test.ts b/src/main/native-chat/agent-session-journal/journal-queued-message-reads.test.ts new file mode 100644 index 000000000000..01f7b380b87a --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-queued-message-reads.test.ts @@ -0,0 +1,148 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import { claudeProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import { readQueuePublication } from '../agent-session-wire/structured-agent-session-queued-publication' +import { + createTrackedJournalOpener, + openTestJournalHostDatabase +} from './journal-host-database-test-support' +import { JournalQueuedMessages } from './journal-queued-messages' +import type { AgentSessionJournal } from './journal-store' + +const SESSION = 'queue-read-integration' +const journals = createTrackedJournalOpener() +let root: string +let journal: AgentSessionJournal + +function message(text: string): AgentJournalMessageItem { + return { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] } +} + +function queue(messageId: string) { + return journal.queuedMessages.insert({ + messageId, + body: message(messageId), + fingerprint: `fp-${messageId}`, + hostInstance: 'host' + }) +} + +function publication() { + return readQueuePublication(journal, () => ({ record: null, fence: 0 })) +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-queue-reads-')) + journal = await journals.open({ + identity: { + sessionId: SESSION, + workspaceId: 'workspace', + hostId: 'host', + agent: 'claude', + providerHandle: claudeProviderHandle('native', null) + }, + stateDirectory: root, + mintEpoch: () => 'epoch' + }) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +it('journal gates and the transactional consume recheck never select or parse draft bodies', async () => { + await queue('head') + await queue('tail') + const db = openTestJournalHostDatabase(root).db + const prepare = vi.spyOn(db, 'prepare') + const parse = vi.spyOn(JSON, 'parse') + expect(journal.queuedMessages.nextSendable()?.messageId).toBe('head') + expect(journal.queuedMessages.pauses()).toEqual([]) + expect(journal.queuedMessages.awaitReopenMark()).toBe(true) + expect(journal.queuedMessages.settlementOwed()).toBe(false) + expect(journal.queuedMessages.deliveredByEchoOwed()).toBe(false) + db.exec('BEGIN IMMEDIATE') + try { + journal.queuedMessages.consumeInTransaction(db, { + messageId: 'head', + expect: 'waiting', + settledByOp: null, + consumedAs: 'handoff', + yieldsToPause: true + }) + expect(journal.queuedMessages.nextSendable()?.messageId).toBe('tail') + } finally { + db.exec('ROLLBACK') + journal.queuedMessages.invalidate() + } + expect(journal.queuedMessages.nextSendable()?.messageId).toBe('head') + expect(parse).not.toHaveBeenCalled() + for (const [sql] of prepare.mock.calls.filter(([sql]) => sql.startsWith('SELECT'))) { + expect(sql.split('FROM')[0]).not.toMatch(/body_json|returned_rejection/) + } +}) + +it('v1 publication parses only unsettled bodies, once per queue revision, with unchanged output', async () => { + await queue('withdrawn') + await journal.queuedMessages.withdraw({ messageIds: ['withdrawn'], settledByOp: 'op' }) + await queue('waiting') + await queue('returned') + const db = openTestJournalHostDatabase(root).db + db.prepare( + "UPDATE queued_messages SET state = 'returned', returned_reason = 'refused' WHERE message_id = 'returned'" + ).run() + const list = vi.spyOn(JournalQueuedMessages.prototype, 'list') + const parse = vi.spyOn(JSON, 'parse') + const first = publication() + expect(first.queuedMessages).toEqual([ + { messageId: 'waiting', position: 2, body: message('waiting'), state: 'waiting' }, + { + messageId: 'returned', + position: 3, + body: message('returned'), + state: 'returned', + returnedReason: 'refused' + } + ]) + expect(publication()).toBe(first) + expect(list).toHaveBeenCalledTimes(1) + expect(parse).toHaveBeenCalledTimes(2) + parse.mockRestore() + await queue('new') + expect(publication().queuedMessages.map((row) => row.messageId)).toEqual([ + 'waiting', + 'returned', + 'new' + ]) + expect(list).toHaveBeenCalledTimes(2) +}) + +it('a publication read inside a rolled-back consume cannot survive invalidation', async () => { + await queue('head') + const db = openTestJournalHostDatabase(root).db + const first = publication() + const revision = journal.queuedMessages.revision() + db.exec('BEGIN IMMEDIATE') + try { + journal.queuedMessages.consumeInTransaction(db, { + messageId: 'head', + expect: 'waiting', + consumedAs: 'handoff', + settledByOp: null, + yieldsToPause: true + }) + expect(publication().queuedMessages).toEqual([]) + expect(journal.queuedMessages.list()).toEqual([]) + } finally { + db.exec('ROLLBACK') + journal.queuedMessages.invalidate() + } + expect(journal.queuedMessages.revision()).toBeGreaterThan(revision) + expect(publication()).toEqual(first) + expect(journal.queuedMessages.list().map((row) => row.messageId)).toEqual(['head']) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts index 5b0e2580c45e..9e35f65542ce 100644 --- a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts +++ b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts @@ -31,11 +31,20 @@ import { getQueuedMessage, insertQueuedMessage, listQueuedMessages, + QueuedMessageNotConsumableError, queuedMessagesSettledByOp, withdrawQueuedMessages, type QueuedMessageHoldReason, type QueuedMessageRow } from './queued-message-table' +import { + hasReadableQueuedMessage, + getQueuedMessageHeader, + queuedMessageAwaitingReopen, + queuedMessageHeaders, + unsettledQueuedMessageBodyBytes, + type QueuedMessageHeaderSelection +} from './queued-message-headers' import { draftsDeliveredByAppliedEcho } from './queued-message-delivered-echo' import { moveQueuedMessages, type QueuedMessagePositionMove } from './queued-message-positions' import { pruneQueuedMessages, retainedSubmissionVerdict } from './queued-message-retention' @@ -73,7 +82,6 @@ export type JournalQueuedMessagesDeps = { export class JournalQueuedMessages { /** Bumped on every draft-table write, so publication memos recompute only when they must. */ private changeRevision = 0 - private listed: { revision: number; rows: readonly QueuedMessageRow[] } | null = null constructor(private readonly deps: JournalQueuedMessagesDeps) {} @@ -90,16 +98,21 @@ export class JournalQueuedMessages { this.changeRevision++ } - /** Cached per revision: the drain re-checks on every journal publish, so an - * unchanged table must cost no SQL read or body parse on token streams. */ + /** Full bodies only for v1 publication and /clear's carry, filtered in SQL. */ list(): readonly QueuedMessageRow[] { - if (this.listed?.revision !== this.changeRevision) { - this.listed = { - revision: this.changeRevision, - rows: listQueuedMessages(this.deps.database().db, this.deps.sessionId) - } - } - return this.listed.rows + return listQueuedMessages(this.deps.database().db, this.deps.sessionId) + } + + headers(selection?: QueuedMessageHeaderSelection) { + return queuedMessageHeaders(this.deps.database().db, this.deps.sessionId, selection) + } + + publishedBodyBytes(): number { + return unsettledQueuedMessageBodyBytes(this.deps.database().db, this.deps.sessionId) + } + + nextSendable() { + return nextSendableQueuedCard(this.pauses(), this.headers('unsettled')) } get(messageId: string): QueuedMessageRow | null { @@ -168,20 +181,24 @@ export class JournalQueuedMessages { /** The queue's pauses in force, derived from the fold and the cards (`queued-message-pause.ts`). */ pauses(): DerivedQueuePause[] { - return this.derivePauses(this.list()) + const state = this.deps.state() + return deriveQueuePauses({ + epoch: state.epoch, + marks: state.queuePauseMarks, + latestAcceptedTurnSequence: state.latestAcceptedTurnSequence, + cards: this.headers('waiting'), + reopenFloor: this.deps.reopenFloor() + }) } /** A card waits, or is mid-hand-off and may come back to waiting: a chat that stops running * marks it (`AgentSessionJournal.markQueueReopen`). */ awaitReopenMark(): boolean { - const { submissions } = this.deps.state() - return this.list().some((row) => { - if (row.state === 'waiting') { - return true - } - const handOff = row.consumedAs ? submissions.get(row.consumedAs)?.dispatchState : undefined - return row.state === 'dispatched' && (handOff === 'pending' || handOff === 'unknown') - }) + return queuedMessageAwaitingReopen( + this.deps.database().db, + this.deps.sessionId, + this.deps.state().submissions + ) } /** The person's Stop still pausing the queue, if any (`journalUserStopInForce`). */ @@ -190,17 +207,6 @@ export class JournalQueuedMessages { return journalUserStopInForce(state.queuePauseMarks, state.latestAcceptedTurnSequence) } - private derivePauses(cards: readonly QueuedMessageRow[]): DerivedQueuePause[] { - const state = this.deps.state() - return deriveQueuePauses({ - epoch: state.epoch, - marks: state.queuePauseMarks, - latestAcceptedTurnSequence: state.latestAcceptedTurnSequence, - cards, - reopenFloor: this.deps.reopenFloor() - }) - } - /** Inside the caller's journal-row transaction (`journal-unsent-send-hold.ts`): one kept send * becomes a card, and the cards ahead of the queue take the positions given. False when a card * by that id already exists, which then stands. */ @@ -267,7 +273,7 @@ export class JournalQueuedMessages { this.changeRevision += settleQueuedMessagesForRow(db, { sessionId: this.deps.sessionId, state: this.deps.state(), - drafts: () => this.list(), + drafts: () => (messageId) => getQueuedMessageHeader(db, this.deps.sessionId, messageId), row, now: this.deps.now() }) @@ -288,9 +294,7 @@ export class JournalQueuedMessages { // Judged again here, by the drain's own rule. Today a Stop cannot land between the drain's // pick and this claim (both run on the session's serialized lane, held across the send), so // this guards any pause-relevant row written off that lane from overtaking a held card. - const cards = listQueuedMessages(db, this.deps.sessionId) - const pauses = this.derivePauses(cards) - if (nextSendableQueuedCard(pauses, cards)?.messageId !== input.messageId) { + if (this.nextSendable()?.messageId !== input.messageId) { throw new QueuedMessageNotConsumableError(input.messageId, input.expect) } } @@ -307,13 +311,20 @@ export class JournalQueuedMessages { /** A skipped live settlement the journal already decided (`queued-message-settlement.ts`). */ settlementOwed(): boolean { - return queuedMessageSettlementOwed(this.list(), this.deps.state().submissions) + return queuedMessageSettlementOwed( + this.deps.database().db, + this.deps.sessionId, + this.deps.state().submissions + ) } /** Waiting drafts a skipped echo hook left unwithdrawn; reads every item, so only the drain * step asks, right before a draft would send. */ deliveredByEchoOwed(): boolean { - return draftsDeliveredByAppliedEcho(this.deps.state(), this.list()).length > 0 + const delivered = draftsDeliveredByAppliedEcho(this.deps.state(), (messageId) => + getQueuedMessageHeader(this.deps.database().db, this.deps.sessionId, messageId) + ) + return delivered.next().done === false } /** Applies owed settlements now, so a skipped live transition heals without a reopen. */ @@ -348,7 +359,10 @@ export class JournalQueuedMessages { */ repairAndPrune(): Promise { // No draft, no work, and no write. - if (this.deps.readOnly() || this.list().length === 0) { + if ( + this.deps.readOnly() || + !hasReadableQueuedMessage(this.deps.database().db, this.deps.sessionId) + ) { return Promise.resolve() } const { sessionId } = this.deps @@ -370,12 +384,4 @@ export class JournalQueuedMessages { } } -export class QueuedMessageNotConsumableError extends Error { - constructor( - readonly messageId: string, - readonly expected: 'waiting' | 'returned' - ) { - super(`queued message ${messageId} is no longer ${expected}`) - this.name = 'QueuedMessageNotConsumableError' - } -} +export { QueuedMessageNotConsumableError } diff --git a/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.ts b/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.ts index fd5a774c7bf2..2031e73c5c7e 100644 --- a/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.ts +++ b/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.ts @@ -196,40 +196,59 @@ function headOfQueuePositions( journal: AgentSessionJournal, batch: readonly { submission: AgentJournalSubmission; body: AgentJournalMessageItem | null }[] ): { placed: QueuedMessagePositionMove[]; earlier: QueuedMessagePositionMove[] } { - const cards = journal.queuedMessages.list() const keptIds = new Set( journal.submissions().flatMap((entry) => entry.keptAsQueuedMessageId ?? []) ) const { epoch } = journal.cursor() // Kept by an earlier settlement: its rejection names it, or, past an epoch that dropped that // row, it was queued in an earlier epoch, before anything this one accepted. - const earlier = cards - .filter( - (card) => - card.state === 'waiting' && - (keptIds.has(card.messageId) || (card.queuedAt !== null && card.queuedAt.epoch !== epoch)) - ) - .map((card) => card.messageId) + const earlier: string[] = [] + const clientSends = new Set( + batch + .filter(({ submission }) => submission.origin === 'client') + .map(({ submission }) => submission.clientMessageId) + ) + const consumedClientSends = new Set() + for (const card of journal.queuedMessages.headers()) { + if ( + card.state === 'waiting' && + (keptIds.has(card.messageId) || (card.queuedAt !== null && card.queuedAt.epoch !== epoch)) + ) { + earlier.push(card.messageId) + } + if (card.consumedAs !== null && clientSends.has(card.consumedAs)) { + consumedClientSends.add(card.consumedAs) + } + } + const earlierIds = new Set(earlier) const placed: QueuedMessagePositionMove[] = [] for (const { submission, body } of batch) { const { clientMessageId } = submission - if (body && !earlier.includes(clientMessageId)) { + if (body && !earlierIds.has(clientMessageId)) { placed.push({ messageId: clientMessageId, position: 0 }) } else if ( // Kept by its settlement (`rejectedDraftSettlement`): a Send the person asked for. submission.origin === 'client' && - cards.some((card) => card.consumedAs === clientMessageId) + consumedClientSends.has(clientMessageId) ) { placed.push({ consumedAs: clientMessageId, position: 0 }) } } - const inHead = (card: (typeof cards)[number]): boolean => - earlier.includes(card.messageId) || - placed.some((move) => - 'messageId' in move ? move.messageId === card.messageId : move.consumedAs === card.consumedAs - ) - const others = cards.filter((card) => !inHead(card)).map((card) => card.position) - const anchor = others.length > 0 ? Math.min(...others) : 1 + const placedIds = new Set(placed.flatMap((move) => ('messageId' in move ? [move.messageId] : []))) + const placedSubmissions = new Set( + placed.flatMap((move) => ('consumedAs' in move ? [move.consumedAs] : [])) + ) + let anchor = 1 + for (const card of journal.queuedMessages.headers()) { + if ( + !earlierIds.has(card.messageId) && + !placedIds.has(card.messageId) && + (card.consumedAs === null || !placedSubmissions.has(card.consumedAs)) + ) { + anchor = card.position + break + } + } const first = anchor - earlier.length - placed.length return { earlier: earlier.map((messageId, index) => ({ messageId, position: first + index })), diff --git a/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.test.ts b/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.test.ts index e6905e848460..aaf011da3ccf 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.test.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.test.ts @@ -16,7 +16,10 @@ import { agentSessionFailureWords } from '../../../shared/agent-session-failure- import { agentSessionSendBodyFingerprint } from '../../../shared/structured-agent-session-send-mutation' import { JournalQueuedMessages } from './journal-queued-messages' import type { AgentSessionJournal } from './journal-store' -import { createTrackedJournalOpener } from './journal-host-database-test-support' +import { + createTrackedJournalOpener, + openTestJournalHostDatabase +} from './journal-host-database-test-support' import { claudeProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' const IDENTITY: AgentSessionJournalIdentity = { @@ -122,6 +125,19 @@ afterEach(async () => { }) describe("a waiting draft whose 'never delivered' claim an echo disproves", () => { + it('ignores an unreadable waiting draft in both the live hook and owed-echo repair', async () => { + const journal = await withdrawnDraft('did it land?') + const db = openTestJournalHostDatabase(root).db + db.prepare("UPDATE queued_messages SET body_json = '{' WHERE message_id = 'draft-1'").run() + await echo(journal, 'echo-1', 'did it land?') + expect(journal.queuedMessages.get('draft-1')).toBeNull() + expect(journal.queuedMessages.deliveredByEchoOwed()).toBe(false) + await journal.queuedMessages.settleOwed() + expect( + db.prepare("SELECT state FROM queued_messages WHERE message_id = 'draft-1'").get() + ).toEqual({ state: 'waiting' }) + }) + it('is withdrawn when the provider echoes the message it was withdrawn from', async () => { const journal = await withdrawnDraft('did it land?') await echo(journal, 'echo-1', 'did it land?') diff --git a/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.ts b/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.ts index 7d739a3f6224..ec4f0cde2449 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.ts @@ -11,14 +11,16 @@ import { type JournalReducerState } from './journal-reducer' import type { JournalRow } from './journal-row-schema' -import type { QueuedMessageRow } from './queued-message-table' +import type { QueuedMessageHeader } from './queued-message-table' + +export type QueuedMessageEchoDraftReader = (messageId: string) => QueuedMessageHeader | null /** The waiting draft this appended row proves was delivered, or null. Called * before the row applies, for every row, so a row that is no new provider * echo of a user message returns before anything else is read. */ export function draftDeliveredByEcho( state: JournalReducerState, - drafts: () => readonly QueuedMessageRow[], + drafts: () => QueuedMessageEchoDraftReader, row: JournalRow ): string | null { const echoes = appendedItems(row).filter( @@ -48,13 +50,13 @@ export function draftDeliveredByEcho( * was folded into its submission's item. Reads every item, so callers run it * only where a draft is about to send, never per streamed row. */ -export function draftsDeliveredByAppliedEcho( +export function* draftsDeliveredByAppliedEcho( state: JournalReducerState, - drafts: readonly QueuedMessageRow[] -): string[] { + drafts: QueuedMessageEchoDraftReader +): IterableIterator { const spent = spentWaitingDrafts(state, drafts) if (spent.length === 0) { - return [] + return } const delivered = new Set() for (const item of state.items.values()) { @@ -62,22 +64,25 @@ export function draftsDeliveredByAppliedEcho( continue } for (const candidate of spent) { - if (echoProvesDelivered(state, candidate, item.body, item.sequence)) { + if ( + !delivered.has(candidate.draft.messageId) && + echoProvesDelivered(state, candidate, item.body, item.sequence) + ) { delivered.add(candidate.draft.messageId) + yield candidate.draft.messageId } } } - return [...delivered] } -type SpentDraft = { draft: QueuedMessageRow; since: number } +type SpentDraft = { draft: QueuedMessageHeader; since: number } /** Waiting drafts some hand-off of which was handed over, then rejected as never delivered; * `since` is the earliest such hand-off's row. A hand-off rejected before hand-over is * provably unwritten: an echo matching it is some other message, and must not delete the card. */ function spentWaitingDrafts( state: JournalReducerState, - drafts: readonly QueuedMessageRow[] + drafts: QueuedMessageEchoDraftReader ): SpentDraft[] { const since = new Map() for (const submission of state.submissions.values()) { @@ -91,10 +96,17 @@ function spentWaitingDrafts( since.set(submission.queuedMessageId, Math.min(earliest ?? sequence, sequence)) } } - return drafts.flatMap((draft) => { - const from = since.get(draft.messageId) - return draft.state === 'waiting' && from !== undefined ? [{ draft, since: from }] : [] - }) + if (since.size === 0) { + return [] + } + const spent: SpentDraft[] = [] + for (const [messageId, from] of since) { + const draft = drafts(messageId) + if (draft?.state === 'waiting') { + spent.push({ draft, since: from }) + } + } + return spent.sort((a, b) => a.draft.position - b.draft.position) } /** The one predicate both paths share: an unclaimed echo appended after a disproved hand-off, diff --git a/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts b/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts new file mode 100644 index 000000000000..50e5f5521896 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts @@ -0,0 +1,426 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { + AgentJournalMessageItem, + AgentJournalSubmission +} from '../../../shared/agent-session-journal-types' +import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import Database from '../../sqlite/sync-database' +import type { SqliteBindings } from '../../sqlite/sqlite-statement' +import { + dispatchedQueuedMessageHeader, + expiredDispatchedQueuedMessageHeaders, + getQueuedMessageHeader, + hasReadableQueuedMessage, + hasWaitingQueuedMessage, + queuedMessageAwaitingReopen, + queuedMessageHeaders, + unsettledQueuedMessageBodyBytes +} from './queued-message-headers' +import { + deriveQueuePauses, + nextSendableQueuedCard, + resumableQueuePause +} from './queued-message-pause' +import { ensureQueuedMessagesTable } from './queued-message-schema' +import { + insertQueuedMessage, + listQueuedMessages, + type QueuedMessageState +} from './queued-message-table' +import { createJournalReducerState } from './journal-reducer' +import { queuedMessageSettlementOwed, settleOwedQueuedMessages } from './queued-message-settlement' +import { pruneQueuedMessages } from './queued-message-retention' + +const SESSION = 'queue-reads' +const EPOCH = 'epoch' +let db: Database.Database + +function message(text: string): AgentJournalMessageItem { + return { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] } +} + +function insert(messageId: string, position?: number, body = message(messageId)) { + return insertQueuedMessage(db, { + sessionId: SESSION, + messageId, + position, + body, + fingerprint: `fp-${messageId}`, + hostInstance: 'host', + queuedAt: { epoch: EPOCH, sequence: 12 }, + now: 1 + }) +} + +function transition( + messageId: string, + state: QueuedMessageState, + consumedAs: string | null = null, + settledAt: number | null = null +) { + db.prepare( + 'UPDATE queued_messages SET state = ?, consumed_as = ?, settled_at = ? WHERE message_id = ?' + ).run(state, consumedAs, settledAt, messageId) +} + +function submission( + clientMessageId: string, + dispatchState: AgentJournalSubmission['dispatchState'] +): AgentJournalSubmission { + return { + clientMessageId, + dispatchState, + fence: 0, + payloadFingerprint: 'fp', + providerItemId: null, + reason: null, + submittedAt: 1, + resolvedAt: null + } +} + +function seedTombstones() { + db.prepare( + `WITH RECURSIVE positions(n) AS (VALUES(1) UNION ALL SELECT n + 1 FROM positions WHERE n < 3000) + INSERT INTO queued_messages + (session_id, message_id, position, body_json, fingerprint, created_at, host_instance, state, settled_at) + SELECT ?, 'tombstone-' || n, -n, ?, 'fp', 1, 'host', 'withdrawn', 900 FROM positions` + ).run(SESSION, JSON.stringify(message('old text'))) +} + +function queryPlan(sql: string | undefined, bindings: SqliteBindings): string { + if (!sql) { + throw new Error('the query was not prepared') + } + return db + .prepare(`EXPLAIN QUERY PLAN ${sql}`) + .all(...bindings) + .map((row) => row.detail) + .join('\n') +} + +beforeEach(() => { + db = new Database(':memory:') + ensureQueuedMessagesTable(db) +}) + +afterEach(() => { + vi.restoreAllMocks() + db.close() +}) + +describe('body-free queue reads', () => { + it('selects headers without body or rejection JSON, and never calls JSON.parse', () => { + insert('head') + insert('tail', undefined, message('a large body 💬'.repeat(1000))) + transition('tail', 'returned', 'handoff') + db.prepare('UPDATE queued_messages SET returned_rejection = ? WHERE message_id = ?').run( + '{', + 'tail' + ) + const prepare = vi.spyOn(db, 'prepare') + const parse = vi.spyOn(JSON, 'parse') + const headers = [...queuedMessageHeaders(db, SESSION, 'unsettled')] + expect(headers.map((row) => row.messageId)).toEqual(['head', 'tail']) + expect(headers.every((row) => !('body' in row) && !('returnedRejection' in row))).toBe(true) + expect(getQueuedMessageHeader(db, SESSION, 'head')).toEqual(headers[0]) + expect(hasWaitingQueuedMessage(db, SESSION)).toBe(true) + expect(hasReadableQueuedMessage(db, SESSION)).toBe(true) + expect(parse).not.toHaveBeenCalled() + for (const [sql] of prepare.mock.calls) { + expect(sql.split('FROM')[0]).not.toMatch(/body_json|returned_rejection/) + } + }) + + it.each(['waiting', 'returned'] as const)( + 'stops the next-card reader at a decisive %s head', + (state) => { + insert('head') + insert('tail') + transition('head', state) + const visited: string[] = [] + function* counted() { + for (const header of queuedMessageHeaders(db, SESSION, 'unsettled')) { + visited.push(header.messageId) + yield header + } + } + expect(nextSendableQueuedCard([], counted())?.messageId ?? null).toBe( + state === 'waiting' ? 'head' : null + ) + expect(visited).toEqual(['head']) + // Early return releases the SQLite reader before the next transaction. + db.exec('BEGIN IMMEDIATE') + db.exec('COMMIT') + } + ) + + it('keeps SQL UTF-8 byte sums equal to JSON.stringify(body), including escaping and emoji', () => { + const bodies = [ + message('é漢字🦦'), + message('"quoted"\\path\n\t\u0000'), + message('\ud800'), + { + ...message('blocks'), + blocks: [ + { type: 'text' as const, text: '😀' }, + { type: 'text' as const, text: '\r\n' } + ] + } + ] + for (const [index, body] of bodies.entries()) { + insert(`body-${index}`, undefined, body) + expect( + db + .prepare('SELECT body_json FROM queued_messages WHERE message_id = ?') + .get(`body-${index}`)?.body_json + ).toBe(JSON.stringify(body)) + } + transition('body-1', 'returned') + insert('dispatched') + transition('dispatched', 'dispatched', 'handoff', 1) + insert('withdrawn') + transition('withdrawn', 'withdrawn', null, 1) + insert('corrupt') + db.prepare('UPDATE queued_messages SET body_json = ? WHERE message_id = ?').run('{', 'corrupt') + const expected = bodies.reduce( + (bytes, body) => bytes + Buffer.byteLength(JSON.stringify(body), 'utf8'), + 0 + ) + const parse = vi.spyOn(JSON, 'parse') + expect(unsettledQueuedMessageBodyBytes(db, SESSION)).toBe(expected) + expect(unsettledQueuedMessageBodyBytes(db, 'other-session')).toBe(0) + expect(parse).not.toHaveBeenCalled() + }) + + it.each([ + { name: 'head', position: 0, carried: null, hold: null, state: 'returned', body: '{' }, + { name: 'middle', position: 2, carried: null, hold: null, state: 'returned', body: '{' }, + { name: 'carried', position: 0, carried: 'source', hold: null, state: 'waiting', body: '{' }, + { + name: 'held', + position: 0, + carried: 'source', + hold: 'send_failed', + state: 'waiting', + body: '{' + }, + { + name: 'unknown state', + position: 0, + carried: 'source', + hold: null, + state: 'future', + body: '{}' + } + ])('an unreadable $name row is never shown, selected, or a pause/reopen obligation', (bad) => { + insert('first', 1) + insert('last', 3) + insert('bad', bad.position) + db.prepare( + `UPDATE queued_messages SET body_json = ?, state = ?, carried_from = ?, hold_reason = ?, queued_sequence = 0 + WHERE message_id = 'bad'` + ).run(bad.body, bad.state, bad.carried, bad.hold) + const state = createJournalReducerState(SESSION, EPOCH) + state.queuePauseMarks.reopenedSequence = 5 + const cards = [...queuedMessageHeaders(db, SESSION, 'unsettled')] + const pauses = deriveQueuePauses({ + epoch: EPOCH, + marks: state.queuePauseMarks, + latestAcceptedTurnSequence: 0, + cards: queuedMessageHeaders(db, SESSION, 'waiting'), + reopenFloor: null + }) + expect(cards.map((row) => row.messageId)).toEqual(['first', 'last']) + expect(listQueuedMessages(db, SESSION).map((row) => row.messageId)).toEqual(['first', 'last']) + expect(pauses).toEqual([]) + expect(nextSendableQueuedCard(pauses, cards)?.messageId).toBe('first') + db.prepare("DELETE FROM queued_messages WHERE message_id != 'bad'").run() + expect(hasWaitingQueuedMessage(db, SESSION)).toBe(false) + expect(queuedMessageAwaitingReopen(db, SESSION, state.submissions)).toBe(false) + expect(getQueuedMessageHeader(db, SESSION, 'bad')).toBeNull() + }) + + it('preserves FIFO and pause decisions over a mixed queue using only headers', () => { + insert('held') + insert('before-stop') + insert('returned') + insert('carried') + insert('after-reopen') + insert('dispatched') + transition('returned', 'returned') + transition('dispatched', 'dispatched', 'handoff', 1) + db.prepare( + "UPDATE queued_messages SET hold_reason = 'send_failed' WHERE message_id = 'held'" + ).run() + db.prepare( + "UPDATE queued_messages SET queued_sequence = 2 WHERE message_id = 'before-stop'" + ).run() + db.prepare( + "UPDATE queued_messages SET carried_from = 'old-session' WHERE message_id = 'carried'" + ).run() + const state = createJournalReducerState(SESSION, EPOCH) + state.queuePauseMarks.latestStop = { sequence: 5, event: { reason: 'user-stop', at: 1 } } + state.queuePauseMarks.reopenedSequence = 8 + const input = { + epoch: EPOCH, + marks: state.queuePauseMarks, + latestAcceptedTurnSequence: 0, + reopenFloor: null + } + const bodies = listQueuedMessages(db, SESSION) + const fromBodies = deriveQueuePauses({ ...input, cards: bodies }) + const fromHeaders = deriveQueuePauses({ + ...input, + cards: queuedMessageHeaders(db, SESSION, 'waiting') + }) + expect(fromHeaders.map((pause) => pause.reason)).toEqual(['stopped', 'cleared', 'restarted']) + expect(fromHeaders).toEqual(fromBodies) + expect( + nextSendableQueuedCard(fromHeaders, queuedMessageHeaders(db, SESSION, 'unsettled')) + ).toBeNull() + expect( + resumableQueuePause(fromHeaders, queuedMessageHeaders(db, SESSION, 'unsettled'))?.reason + ).toBe('stopped') + expect( + nextSendableQueuedCard([], queuedMessageHeaders(db, SESSION, 'unsettled'))?.messageId + ).toBe('before-stop') + transition('before-stop', 'withdrawn', null, 1) + expect(nextSendableQueuedCard([], queuedMessageHeaders(db, SESSION, 'unsettled'))).toBeNull() + transition('returned', 'withdrawn', null, 1) + expect( + nextSendableQueuedCard([], queuedMessageHeaders(db, SESSION, 'unsettled'))?.messageId + ).toBe('carried') + }) + + it('stops pause derivation once its carried and pre-reopen existence questions are answered', () => { + const card = insert('carried') + const state = createJournalReducerState(SESSION, EPOCH) + state.queuePauseMarks.reopenedSequence = 20 + function* cards() { + yield { ...card, carriedFrom: 'source' } + throw new Error('the rest of the queue must not be read') + } + expect( + deriveQueuePauses({ + epoch: EPOCH, + marks: state.queuePauseMarks, + latestAcceptedTurnSequence: 0, + cards: cards(), + reopenFloor: null + }).map((pause) => pause.reason) + ).toEqual(['cleared', 'restarted']) + }) +}) + +describe('indexed bookkeeping', () => { + it('settles only the rejected current consume links among thousands of tombstones', () => { + seedTombstones() + insert('owed') + transition('owed', 'dispatched', 'rejected', 1) + insert('accepted') + transition('accepted', 'dispatched', 'accepted', 1) + insert('corrupt') + transition('corrupt', 'dispatched', 'corrupt-ref', 1) + db.prepare("UPDATE queued_messages SET body_json = '{' WHERE message_id = 'corrupt'").run() + const state = createJournalReducerState(SESSION, EPOCH) + state.submissions.set('rejected', { + ...submission('rejected', 'rejected'), + rejection: agentSessionFailureFact('providerRejected') + }) + state.submissions.set('accepted', submission('accepted', 'accepted')) + state.submissions.set('corrupt-ref', submission('corrupt-ref', 'rejected')) + const prepare = vi.spyOn(db, 'prepare') + const parse = vi.spyOn(JSON, 'parse') + expect(queuedMessageSettlementOwed(db, SESSION, state.submissions)).toBe(true) + expect(settleOwedQueuedMessages(db, { sessionId: SESSION, state, now: 10 })).toBe(1) + expect(queuedMessageSettlementOwed(db, SESSION, state.submissions)).toBe(false) + expect(parse).not.toHaveBeenCalled() + expect( + prepare.mock.calls + .filter(([sql]) => sql.startsWith('SELECT')) + .every(([sql]) => sql.includes('consumed_as = ?')) + ).toBe(true) + expect(getQueuedMessageHeader(db, SESSION, 'owed')?.state).toBe('returned') + }) + + it('asks retention verdicts only for expired dispatched rows and keeps pending/refused ones', () => { + seedTombstones() + for (const id of [ + 'accepted', + 'pending', + 'rejected', + 'absent', + 'fresh', + 'corrupt', + 'same-time' + ]) { + insert(id) + transition(id, 'dispatched', `ref-${id}`, id === 'fresh' ? 901 : 1) + } + db.prepare("UPDATE queued_messages SET body_json = '{' WHERE message_id = 'corrupt'").run() + const verdict = vi.fn((ref: string) => { + if (ref === 'ref-pending') { + return 'pending' as const + } + if (ref === 'ref-rejected') { + return 'rejected' as const + } + return 'terminal-not-refused' as const + }) + const prepare = vi.spyOn(db, 'prepare') + const parse = vi.spyOn(JSON, 'parse') + expect( + pruneQueuedMessages(db, { + sessionId: SESSION, + now: 1000, + replayWindowMs: 100, + submissionVerdict: verdict + }) + ).toBe(3) + expect(verdict.mock.calls.map(([ref]) => ref)).toEqual([ + 'ref-absent', + 'ref-accepted', + 'ref-pending', + 'ref-rejected', + 'ref-same-time' + ]) + expect(parse).not.toHaveBeenCalled() + for (const [sql] of prepare.mock.calls.filter(([sql]) => sql.startsWith('SELECT'))) { + expect(sql).toContain("state = 'dispatched' AND settled_at < ?") + expect(sql).toContain('LIMIT 1') + expect(sql.split('FROM')[0]).not.toContain('body_json') + } + expect( + [...expiredDispatchedQueuedMessageHeaders(db, SESSION, 900)].map((row) => row.messageId) + ).toEqual(['pending', 'rejected']) + }) + + it('creates downgrade-safe indexes idempotently and uses them for the matching queries', () => { + ensureQueuedMessagesTable(db) + seedTombstones() + insert('waiting') + insert('dispatched') + transition('dispatched', 'dispatched', 'handoff', 1) + const prepare = vi.spyOn(db, 'prepare') + nextSendableQueuedCard([], queuedMessageHeaders(db, SESSION, 'unsettled')) + dispatchedQueuedMessageHeader(db, SESSION, 'handoff') + const expired = expiredDispatchedQueuedMessageHeaders(db, SESSION, 900) + const expiredRows = [...expired] + expect(expiredRows).toHaveLength(1) + hasWaitingQueuedMessage(db, SESSION) + const queries = prepare.mock.calls.map(([sql]) => sql) + prepare.mockRestore() + expect(queryPlan(queries[0], [SESSION])).toContain('queued_messages_unsettled_position') + expect(queryPlan(queries[1], [SESSION, 'handoff'])).toContain('queued_messages_consumed_as') + expect(queryPlan(queries[2], [SESSION, 900])).toContain('queued_messages_state_settled') + expect(queryPlan(queries[3], [SESSION, 900, 1, 'dispatched'])).toContain( + 'queued_messages_state_settled' + ) + expect(queryPlan(queries[4], [SESSION])).toContain('queued_messages_state_settled') + expect( + queryPlan('SELECT MAX(position) FROM queued_messages WHERE session_id = ?', [SESSION]) + ).toContain('queued_messages_position') + expect(db.pragma('user_version', { simple: true })).toBe(0) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/queued-message-headers.ts b/src/main/native-chat/agent-session-journal/queued-message-headers.ts new file mode 100644 index 000000000000..30f12efb4d5d --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-headers.ts @@ -0,0 +1,152 @@ +// Body-free reads of the existing queue rows, never cached across a transaction. + +import type Database from '../../sqlite/sync-database' +import type { SqliteBindings } from '../../sqlite/sqlite-statement' +import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' +import { readStoredQueuedMessageHeader } from './queued-message-stored-row' +import type { QueuedMessageHeader } from './queued-message-table' + +const HEADER_COLUMNS = + 'session_id, message_id, position, fingerprint, created_at, host_instance, state, hold_reason, returned_reason, settled_at, settled_by_op, consumed_as, carried_from, queued_epoch, queued_sequence' +const READABLE = + "state IN ('waiting', 'returned', 'dispatched', 'withdrawn') AND json_valid(body_json)" + +export type QueuedMessageHeaderSelection = 'all' | 'unsettled' | 'waiting' + +const SELECTIONS = { + all: '1', + unsettled: "state IN ('waiting', 'returned')", + waiting: "state = 'waiting'" +} as const + +export function* queuedMessageHeaders( + db: Database.Database, + sessionId: string, + selection: QueuedMessageHeaderSelection = 'all' +): IterableIterator { + for (const row of db + .prepare( + `SELECT ${HEADER_COLUMNS} FROM queued_messages + WHERE session_id = ? AND ${SELECTIONS[selection]} AND ${READABLE} ORDER BY position ASC` + ) + .iterate(sessionId)) { + const header = readStoredQueuedMessageHeader(row) + if (header) { + yield header + } + } +} + +export function hasWaitingQueuedMessage(db: Database.Database, sessionId: string): boolean { + return ( + db + .prepare( + `SELECT 1 FROM queued_messages + WHERE session_id = ? AND state = 'waiting' AND json_valid(body_json) LIMIT 1` + ) + .get(sessionId) !== undefined + ) +} + +export function hasReadableQueuedMessage(db: Database.Database, sessionId: string): boolean { + return ( + db + .prepare(`SELECT 1 FROM queued_messages WHERE session_id = ? AND ${READABLE} LIMIT 1`) + .get(sessionId) !== undefined + ) +} + +export function dispatchedQueuedMessageHeader( + db: Database.Database, + sessionId: string, + consumedAs: string +): QueuedMessageHeader | null { + const row = db + .prepare( + `SELECT ${HEADER_COLUMNS} FROM queued_messages + WHERE session_id = ? AND consumed_as = ? AND consumed_as IS NOT NULL + AND state = 'dispatched' AND json_valid(body_json)` + ) + .get(sessionId, consumedAs) + return row ? readStoredQueuedMessageHeader(row) : null +} + +export function getQueuedMessageHeader( + db: Database.Database, + sessionId: string, + messageId: string +): QueuedMessageHeader | null { + const row = db + .prepare( + `SELECT ${HEADER_COLUMNS} FROM queued_messages + WHERE session_id = ? AND message_id = ? AND ${READABLE}` + ) + .get(sessionId, messageId) + return row ? readStoredQueuedMessageHeader(row) : null +} + +export function queuedMessageAwaitingReopen( + db: Database.Database, + sessionId: string, + submissions: ReadonlyMap +): boolean { + if (hasWaitingQueuedMessage(db, sessionId)) { + return true + } + for (const submission of submissions.values()) { + if ( + (submission.dispatchState === 'pending' || submission.dispatchState === 'unknown') && + dispatchedQueuedMessageHeader(db, sessionId, submission.clientMessageId) + ) { + return true + } + } + return false +} + +/** Only dispatched rows old enough for retention need the submission verdict. */ +export function* expiredDispatchedQueuedMessageHeaders( + db: Database.Database, + sessionId: string, + cutoff: number +): IterableIterator { + let settledAt: number | null = null + let messageId: string | null = null + // Release each reader before deletion changes the index it traversed. + for (;;) { + const bindings: SqliteBindings = [sessionId, cutoff] + if (settledAt !== null && messageId !== null) { + bindings.push(settledAt, messageId) + } + const row = db + .prepare( + `SELECT ${HEADER_COLUMNS} FROM queued_messages + WHERE session_id = ? AND state = 'dispatched' AND settled_at < ? + AND json_valid(body_json) + ${settledAt === null ? '' : 'AND (settled_at, message_id) > (?, ?)'} + ORDER BY settled_at, message_id LIMIT 1` + ) + .get(...bindings) + if (!row) { + return + } + const header = readStoredQueuedMessageHeader(row) + if (!header || header.settledAt === null) { + return + } + settledAt = header.settledAt + messageId = header.messageId + yield header + } +} + +/** Inserts store JSON.stringify(body), so BLOB length is the existing UTF-8 admission measure. */ +export function unsettledQueuedMessageBodyBytes(db: Database.Database, sessionId: string): number { + const row = db + .prepare( + `SELECT COALESCE(SUM(length(CAST(body_json AS BLOB))), 0) AS bytes FROM queued_messages + WHERE session_id = ? AND state IN ('waiting', 'returned') AND json_valid(body_json)` + ) + .get(sessionId) + return typeof row?.bytes === 'number' ? row.bytes : 0 +} diff --git a/src/main/native-chat/agent-session-journal/queued-message-pause.ts b/src/main/native-chat/agent-session-journal/queued-message-pause.ts index a7a652d81b7c..8d9415c6561c 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-pause.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-pause.ts @@ -139,7 +139,7 @@ export function deriveQueuePauses(input: { epoch: string marks: JournalQueuePauseMarks latestAcceptedTurnSequence: number - cards: readonly QueueCard[] + cards: Iterable /** Where the reopen's pause begins when this handle could not mark it; null otherwise. */ reopenFloor: AgentJournalCursor | null }): DerivedQueuePause[] { @@ -149,16 +149,25 @@ export function deriveQueuePauses(input: { if (stop) { pauses.push({ reason: 'stopped', since: { epoch, sequence: stop.sequence } }) } - const waiting = input.cards.filter((card) => card.state === 'waiting') - const carried = waiting.filter((card) => card.carriedFrom !== null) - if (carried.length > 0 && latestAcceptedTurnSequence === 0 && marks.resumedSequence === 0) { + const needsCarried = latestAcceptedTurnSequence === 0 && marks.resumedSequence === 0 + const reopened = reopenPause(input) + let carried = false + let beforeReopen = false + if (needsCarried || reopened) { + for (const card of input.cards) { + if (card.state === 'waiting') { + carried ||= card.carriedFrom !== null + beforeReopen ||= !!reopened && card.holdReason === null && queuedBefore(reopened, card) + } + if ((!needsCarried || carried) && (!reopened || beforeReopen)) { + break + } + } + } + if (needsCarried && carried) { pauses.push({ reason: 'cleared', since: null }) } - const reopened = reopenPause(input) - if ( - reopened && - waiting.some((card) => card.holdReason === null && queuedBefore(reopened, card)) - ) { + if (reopened && beforeReopen) { pauses.push(reopened) } return pauses @@ -222,7 +231,7 @@ export function queuePauseHolding( * consume both read this. */ export function nextSendableQueuedCard( pauses: readonly DerivedQueuePause[], - cards: readonly T[] + cards: Iterable ): T | null { for (const card of cards) { if (card.state === 'returned' || queuePauseHolding(pauses, card)) { @@ -240,7 +249,7 @@ export function nextSendableQueuedCard( * never offers a Resume that sends nothing. */ export function resumableQueuePause( pauses: readonly DerivedQueuePause[], - cards: readonly QueueCard[] + cards: Iterable ): DerivedQueuePause | null { for (const card of cards) { if (card.state === 'returned') { diff --git a/src/main/native-chat/agent-session-journal/queued-message-retention.ts b/src/main/native-chat/agent-session-journal/queued-message-retention.ts index 46717268ad10..d835377727e6 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-retention.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-retention.ts @@ -3,7 +3,7 @@ import type Database from '../../sqlite/sync-database' import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' -import { listQueuedMessages } from './queued-message-table' +import { expiredDispatchedQueuedMessageHeaders } from './queued-message-headers' /** What the loaded journal says about a dispatched draft's consumed submission. */ export type QueuedMessageSubmissionVerdict = @@ -39,10 +39,7 @@ export function pruneQueuedMessages( ) .run(input.sessionId, cutoff) let pruned = Number(tombstones.changes ?? 0) - for (const row of listQueuedMessages(db, input.sessionId)) { - if (row.state !== 'dispatched' || row.settledAt === null || row.settledAt >= cutoff) { - continue - } + for (const row of expiredDispatchedQueuedMessageHeaders(db, input.sessionId, cutoff)) { // A dispatched row always names its hand-off; one that does not has nothing to wait for. const verdict = row.consumedAs === null ? 'absent' : input.submissionVerdict(row.consumedAs) if (verdict === 'terminal-not-refused' || verdict === 'absent') { diff --git a/src/main/native-chat/agent-session-journal/queued-message-schema.ts b/src/main/native-chat/agent-session-journal/queued-message-schema.ts index 3f9d0122fda2..b343b2f0b508 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-schema.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-schema.ts @@ -67,5 +67,11 @@ CREATE TABLE IF NOT EXISTS queued_messages ( db.exec(` CREATE UNIQUE INDEX IF NOT EXISTS queued_messages_consumed_as ON queued_messages (session_id, consumed_as) WHERE consumed_as IS NOT NULL; +CREATE INDEX IF NOT EXISTS queued_messages_position + ON queued_messages (session_id, position); +CREATE INDEX IF NOT EXISTS queued_messages_unsettled_position + ON queued_messages (session_id, position) WHERE state IN ('waiting', 'returned'); +CREATE INDEX IF NOT EXISTS queued_messages_state_settled + ON queued_messages (session_id, state, settled_at, message_id); `) } diff --git a/src/main/native-chat/agent-session-journal/queued-message-settlement.ts b/src/main/native-chat/agent-session-journal/queued-message-settlement.ts index 9c30bf4311dd..b481898e2b25 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-settlement.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-settlement.ts @@ -12,27 +12,41 @@ import { } from './journal-dispatch-settlement' import type { JournalReducerState } from './journal-reducer' import type { JournalRow } from './journal-row-schema' -import { draftDeliveredByEcho, draftsDeliveredByAppliedEcho } from './queued-message-delivered-echo' import { - listQueuedMessages, + draftDeliveredByEcho, + draftsDeliveredByAppliedEcho, + type QueuedMessageEchoDraftReader +} from './queued-message-delivered-echo' +import { settleRejectedQueuedMessage, - withdrawQueuedMessages, - type QueuedMessageRow + withdrawQueuedMessageInTransaction } from './queued-message-table' +import { dispatchedQueuedMessageHeader, getQueuedMessageHeader } from './queued-message-headers' type Submissions = ReadonlyMap /** Some dispatched draft still waits on a settlement the journal already decided. */ export function queuedMessageSettlementOwed( - rows: readonly QueuedMessageRow[], + db: Database.Database, + sessionId: string, submissions: Submissions ): boolean { - return rows.some( - (row) => - row.state === 'dispatched' && - row.consumedAs !== null && - consumedSubmissionWasRejected(submissions.get(row.consumedAs)) - ) + return owedQueuedMessageSettlements(db, sessionId, submissions).next().done === false +} + +function* owedQueuedMessageSettlements( + db: Database.Database, + sessionId: string, + submissions: Submissions +): IterableIterator { + for (const submission of submissions.values()) { + if ( + consumedSubmissionWasRejected(submission) && + dispatchedQueuedMessageHeader(db, sessionId, submission.clientMessageId) + ) { + yield submission + } + } } /** Applies each owed settlement, and withdraws each waiting draft an applied echo proves @@ -43,36 +57,25 @@ export function settleOwedQueuedMessages( ): number { const { submissions } = input.state let settled = 0 - for (const row of listQueuedMessages(db, input.sessionId)) { - const consumedRef = row.consumedAs - const submission = consumedRef === null ? undefined : submissions.get(consumedRef) - if ( - row.state !== 'dispatched' || - consumedRef === null || - !consumedSubmissionWasRejected(submission) - ) { - continue - } + for (const submission of owedQueuedMessageSettlements(db, input.sessionId, submissions)) { const changed = settleRejectedQueuedMessage(db, { sessionId: input.sessionId, - consumedRef, - reason: submission?.reason ?? null, - rejection: submission?.rejection, + consumedRef: submission.clientMessageId, + reason: submission.reason ?? null, + rejection: submission.rejection, now: input.now }) settled += changed ? 1 : 0 } - const delivered = draftsDeliveredByAppliedEcho( - input.state, - listQueuedMessages(db, input.sessionId) - ) - if (delivered.length > 0) { - settled += withdrawQueuedMessages(db, { + for (const messageId of draftsDeliveredByAppliedEcho(input.state, (messageId) => + getQueuedMessageHeader(db, input.sessionId, messageId) + )) { + settled += withdrawQueuedMessageInTransaction(db, { sessionId: input.sessionId, - messageIds: delivered, + messageId, settledByOp: null, now: input.now - }).length + }) } return settled } @@ -92,9 +95,8 @@ export function settleQueuedMessagesForRow( input: { sessionId: string state: JournalReducerState - /** Read only once the row holds an unclaimed echo: a list read inside the append's - * transaction must not be cached under state a rollback could undo. */ - drafts: () => readonly QueuedMessageRow[] + /** Read only for an unclaimed echo, without caching a transaction's provisional state. */ + drafts: () => QueuedMessageEchoDraftReader row: JournalRow now: number } @@ -104,12 +106,12 @@ export function settleQueuedMessagesForRow( const delivered = draftDeliveredByEcho(input.state, input.drafts, row) if (delivered !== null) { // Its first send reached the agent after all; sending it again would repeat it. - changed += withdrawQueuedMessages(db, { + changed += withdrawQueuedMessageInTransaction(db, { sessionId: input.sessionId, - messageIds: [delivered], + messageId: delivered, settledByOp: null, now: input.now - }).length + }) } if (row.kind !== 'dispatch' || row.state !== 'rejected') { return changed diff --git a/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts b/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts index 51233b66abb0..c1599d2e6f8b 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts @@ -4,23 +4,44 @@ import type { UnreadAgentSessionFailureFact } from '../../../shared/agent-session-failure' import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' import { readStoredRejectionFact } from './journal-dispatch-reducer' -import type { QueuedMessageRow } from './queued-message-table' +import type { SqliteRow } from '../../sqlite/sqlite-statement' +import type { QueuedMessageHeader, QueuedMessageRow } from './queued-message-table' /** A `queued_messages` row as this file's SELECTs return it; null when it cannot be read back. */ -export function readStoredQueuedMessageRow(row: unknown): QueuedMessageRow | null { - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: rows come from this file's own SELECTs, which name exactly these columns; better-sqlite3 types them as unknown. +export function readStoredQueuedMessageRow(row: SqliteRow): QueuedMessageRow | null { + const header = readStoredQueuedMessageHeader(row) + if (!header || typeof row.body_json !== 'string') { + return null + } + let body: AgentJournalMessageItem + try { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: body_json is written only by insertQueuedMessage from a schema-validated AgentJournalMessageItem. + body = JSON.parse(row.body_json) as AgentJournalMessageItem + } catch { + return null + } + return { + ...header, + body, + returnedRejection: storedRejection( + typeof row.returned_rejection === 'string' ? row.returned_rejection : null + ) + } +} + +/** Header queries exclude unreadable bodies in SQL without bringing their text into JS. */ +export function readStoredQueuedMessageHeader(row: SqliteRow): QueuedMessageHeader | null { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the queue repository SELECTs name these columns with the table's declared string, number, and nullable types. const record = row as { session_id: string message_id: string position: number - body_json: string fingerprint: string created_at: number host_instance: string state: string hold_reason: string | null returned_reason: string | null - returned_rejection: string | null settled_at: number | null settled_by_op: string | null consumed_as: string | null @@ -28,15 +49,6 @@ export function readStoredQueuedMessageRow(row: unknown): QueuedMessageRow | nul queued_epoch: string | null queued_sequence: number | null } - let body: AgentJournalMessageItem - try { - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: body_json is written only by insertQueuedMessage from a schema-validated AgentJournalMessageItem. - body = JSON.parse(record.body_json) as AgentJournalMessageItem - } catch { - // Our own writer stringified it; an unreadable body is corruption, and a - // row we cannot re-materialize must not masquerade as an empty message. - return null - } const state = record.state if ( state !== 'waiting' && @@ -50,14 +62,12 @@ export function readStoredQueuedMessageRow(row: unknown): QueuedMessageRow | nul sessionId: record.session_id, messageId: record.message_id, position: record.position, - body, fingerprint: record.fingerprint, createdAt: record.created_at, hostInstance: record.host_instance, state, holdReason: storedHoldReason(record.hold_reason), returnedReason: record.returned_reason, - returnedRejection: storedRejection(record.returned_rejection), settledAt: record.settled_at, settledByOp: record.settled_by_op, consumedAs: record.consumed_as, diff --git a/src/main/native-chat/agent-session-journal/queued-message-table.ts b/src/main/native-chat/agent-session-journal/queued-message-table.ts index 852ebc5d7ddf..4b618f002247 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-table.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-table.ts @@ -19,6 +19,16 @@ import { readStoredQueuedMessageRow } from './queued-message-stored-row' export type QueuedMessageState = 'waiting' | 'dispatched' | 'returned' | 'withdrawn' +export class QueuedMessageNotConsumableError extends Error { + constructor( + readonly messageId: string, + readonly expected: 'waiting' | 'returned' + ) { + super(`queued message ${messageId} is no longer ${expected}`) + this.name = 'QueuedMessageNotConsumableError' + } +} + /** Why ONE waiting draft is held from auto-sending: its conversion failed (`send_failed`). Stored * on the row, so it survives handle eviction and restart; a wire marker (it publishes as * `pausedReason`). A Stop, a /clear or a reopen pauses the queue instead. A reader treats an @@ -33,11 +43,10 @@ export function isUnsettledQueuedMessage(row: Pick): return row.state === 'waiting' || row.state === 'returned' } -export type QueuedMessageRow = { +export type QueuedMessageHeader = { sessionId: string messageId: string position: number - body: AgentJournalMessageItem fingerprint: string createdAt: number hostInstance: string @@ -47,7 +56,6 @@ export type QueuedMessageRow = { holdReason: string | null /** A returned card's refusal, mirroring its submission's `reason` and `rejection` pair. */ returnedReason: string | null - returnedRejection: UnreadAgentSessionFailureFact | null settledAt: number | null /** The operation ledger's caller-scoped key, making settled rows mutation receipts. */ settledByOp: string | null @@ -63,6 +71,11 @@ export type QueuedMessageRow = { queuedAt: AgentJournalCursor | null } +export type QueuedMessageRow = QueuedMessageHeader & { + body: AgentJournalMessageItem + returnedRejection: UnreadAgentSessionFailureFact | null +} + const COLUMNS = 'session_id, message_id, position, body_json, fingerprint, created_at, host_instance, state, hold_reason, returned_reason, returned_rejection, settled_at, settled_by_op, consumed_as, carried_from, queued_epoch, queued_sequence' @@ -125,7 +138,10 @@ export function insertQueuedMessage( export function listQueuedMessages(db: Database.Database, sessionId: string): QueuedMessageRow[] { return db - .prepare(`SELECT ${COLUMNS} FROM queued_messages WHERE session_id = ? ORDER BY position ASC`) + .prepare( + `SELECT ${COLUMNS} FROM queued_messages + WHERE session_id = ? AND state IN ('waiting', 'returned') ORDER BY position ASC` + ) .all(sessionId) .flatMap((row) => readStoredQueuedMessageRow(row) ?? []) } @@ -206,11 +222,7 @@ export function withdrawQueuedMessages( if (!row || !isUnsettledQueuedMessage(row)) { continue } - db.prepare( - `UPDATE queued_messages - SET state = 'withdrawn', hold_reason = NULL, settled_at = ?, settled_by_op = ? - WHERE session_id = ? AND message_id = ? AND state IN ('waiting', 'returned')` - ).run(input.now, input.settledByOp, input.sessionId, messageId) + withdrawQueuedMessageInTransaction(db, { ...input, messageId }) withdrawn.push({ ...row, state: 'withdrawn', @@ -222,6 +234,23 @@ export function withdrawQueuedMessages( return withdrawn } +/** Echo settlement needs only the transition count, never the draft's text. */ +export function withdrawQueuedMessageInTransaction( + db: Database.Database, + input: { sessionId: string; messageId: string; settledByOp: string | null; now: number } +): number { + return Number( + db + .prepare( + `UPDATE queued_messages + SET state = 'withdrawn', hold_reason = NULL, settled_at = ?, settled_by_op = ? + WHERE session_id = ? AND message_id = ? AND state IN ('waiting', 'returned') + AND json_valid(body_json)` + ) + .run(input.now, input.settledByOp, input.sessionId, input.messageId).changes + ) +} + /** * dispatched → returned, or back to waiting (`rejectedDraftSettlement`), * matched on the draft's CURRENT hand-off (`consumed_as`), so a re-send refused diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts index 23a8af3582a6..e349a455415e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts @@ -23,13 +23,12 @@ import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/struct import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import type { AgentSessionTurnContext } from './structured-agent-session-turns' import { QueuedMessageNotConsumableError } from '../agent-session-journal/journal-queued-messages' -import type { QueuedMessageRow } from '../agent-session-journal/queued-message-table' +import type { + QueuedMessageHeader, + QueuedMessageRow +} from '../agent-session-journal/queued-message-table' import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' -import { - structuredAgentSessionHostInstance, - structuredQueuePauses -} from './structured-agent-session-queued-pause' -import { nextSendableQueuedCard } from '../agent-session-journal/queued-message-pause' +import { structuredAgentSessionHostInstance } from './structured-agent-session-queued-pause' import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' import { agentSessionAttachmentExpiredRefusal } from './structured-agent-session-turns' import { isAgentSessionAttachmentExpiredError } from '../agent-session-attachments/agent-session-attachment-claims' @@ -60,13 +59,8 @@ export function pendingPromptExists(journal: Pick -): QueuedMessageRow | null { - const rows = journal.queuedMessages.list() - // Nothing waiting costs no pause derivation: this runs on every journal publish. - if (!rows.some((row) => row.state === 'waiting')) { - return null - } - return nextSendableQueuedCard(structuredQueuePauses(journal), rows) +): QueuedMessageHeader | null { + return journal.queuedMessages.nextSendable() } /** @@ -124,7 +118,7 @@ export function nextStructuredQueuedMessage(input: { journal: AgentSessionJournal record: AgentSessionRecord | null fence: number -}): QueuedMessageRow | null { +}): QueuedMessageHeader | null { const next = oldestActionableQueuedMessage(input.journal) const { journal, fence } = input // The gate's cheap `working` first: publication asks on every streamed frame, and the gate's @@ -340,8 +334,12 @@ export class StructuredAgentSessionQueuedMessageDrain { const fence = this.deps.conversationFence(sessionId) // Whatever clears a hold publishes or commits, which re-derives this step. const record = this.deps.getRecord(sessionId) - const next = nextStructuredQueuedMessage({ journal, record, fence }) - if (this.disposed || !next) { + const header = nextStructuredQueuedMessage({ journal, record, fence }) + if (this.disposed || !header) { + return + } + const next = journal.queuedMessages.get(header.messageId) + if (!next || next.state !== 'waiting') { return } // Always a fresh id: the submission names its draft by `queuedMessageId`, never by id equality. diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts index d48d9bf27c39..d970940be949 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts @@ -129,7 +129,9 @@ export function readQueuePublication( stop.epoch !== pause.since.epoch || stop.sequence < pause.since.sequence)) ) - const resumable = silent ? null : resumableQueuePause(pauses, journal.queuedMessages.list()) + const resumable = silent + ? null + : resumableQueuePause(pauses, journal.queuedMessages.headers('unsettled')) // The submissions are read only while a pause would show, never while the queue runs freely. const pause = resumable && !queuePauseLiftOnItsWay(resumable, journal.submissions()) ? resumable : null diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-published-bytes.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-published-bytes.ts index 4e1222b83429..5a38c7fe9ab0 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-published-bytes.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-published-bytes.ts @@ -8,7 +8,6 @@ import { REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES } from '../../../shared/remote-r import { MAX_PROMPT_BYTES } from '../../../shared/rpc-contract/structured-agent-session-params' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { AGENT_SESSION_HISTORY_MAX_PAGE_BYTES } from './agent-session-history-page-bounds' -import { unsettledQueuedMessages } from './structured-agent-session-queued-stop' /** A frame bound, not a queue limit: every unsettled card's body rides each hydrating frame and * history answer beside a full page, and a frame past the outbound cap closes the remote session. */ @@ -26,10 +25,7 @@ export function queuedMessagesPublishedBytesRefusal( body: AgentJournalMessageItem, fromPerson: boolean ): AgentSessionWireRefusal | null { - const bytes = unsettledQueuedMessages(journal).reduce( - (sum, row) => sum + publishedBodyBytes(row.body), - publishedBodyBytes(body) - ) + const bytes = journal.queuedMessages.publishedBodyBytes() + publishedBodyBytes(body) const limit = fromPerson ? QUEUED_MESSAGES_PUBLISHED_MAX_BYTES : QUEUED_MESSAGES_PUBLISHED_MAX_BYTES - QUEUED_MESSAGES_PERSON_RESERVE_BYTES diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop-row.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop-row.test.ts index 2b0aae62f487..129e8db21e88 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop-row.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop-row.test.ts @@ -278,7 +278,7 @@ describe("a /clear's carried cards", () => { clientMessageId: sent.value.clientMessageId, providerIdentity: { provider: 'codex', threadId: 'thread-1', turnId: 'turn-hi', ordinal: 0 } }) - expect(journal(replacementId).queuedMessages.list()[0]?.messageId).toBe(carried) + expect([...journal(replacementId).queuedMessages.headers()][0]?.messageId).toBe(carried) expect(structuredQueuePauses(journal(replacementId))).toEqual([]) }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts index fea528147494..3d16379c3fe1 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts @@ -8,10 +8,7 @@ import type { JournalStopEvent } from '../agent-session-journal/journal-row-schema' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' -import { - isUnsettledQueuedMessage, - type QueuedMessageRow -} from '../agent-session-journal/queued-message-table' +import type { QueuedMessageRow } from '../agent-session-journal/queued-message-table' import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' import { isMainAgentWorking } from './structured-agent-session-turns-cancel' import { @@ -19,10 +16,9 @@ import { structuredAgentSessionStoppedTurnId } from './structured-agent-session-turn-stop-notes' -/** The one unsettled-card predicate /clear's carry and the published-bytes bound share: - * waiting or returned. Pending/unknown/accepted deliveries stay outside it. */ -export function unsettledQueuedMessages(journal: AgentSessionJournal): QueuedMessageRow[] { - return journal.queuedMessages.list().filter(isUnsettledQueuedMessage) +/** /clear copies only waiting and returned bodies, filtered in the repository. */ +export function unsettledQueuedMessages(journal: AgentSessionJournal): readonly QueuedMessageRow[] { + return journal.queuedMessages.list() } /** From b64ca94375016ac80c6b0c489ec0d65e1fae4061 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 8 Oct 2026 13:48:49 -0700 Subject: [PATCH 2/5] perf(native-chat): keep queue reads off streamed frames and body bytes Check whether the agent is working before reading the queue, skip the resumable-pause walk when nothing is paused, and probe settlement only for sends that took a card. One readable-row definition backs the partial indexes, header reads and body reads, so a corrupt row can never be picked and then fail to load. --- .../journal-queued-message-reads.test.ts | 9 + .../journal-queued-messages.ts | 12 +- .../queued-message-headers.test.ts | 16 +- .../queued-message-headers.ts | 31 +-- .../queued-message-pause.ts | 3 + .../queued-message-read-plans.test.ts | 204 ++++++++++++++++++ .../queued-message-readability.test.ts | 81 +++++++ .../queued-message-readability.ts | 7 + .../queued-message-schema.ts | 12 +- .../queued-message-settlement.ts | 1 + .../queued-message-store.test.ts | 2 + .../queued-message-stored-row.ts | 2 +- .../queued-message-table.ts | 16 +- ...tructured-agent-session-queued-messages.ts | 21 +- ...nt-session-queued-read-regressions.test.ts | 109 ++++++++++ 15 files changed, 485 insertions(+), 41 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/queued-message-read-plans.test.ts create mode 100644 src/main/native-chat/agent-session-journal/queued-message-readability.test.ts create mode 100644 src/main/native-chat/agent-session-journal/queued-message-readability.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-queued-read-regressions.test.ts diff --git a/src/main/native-chat/agent-session-journal/journal-queued-message-reads.test.ts b/src/main/native-chat/agent-session-journal/journal-queued-message-reads.test.ts index 01f7b380b87a..8fd49dfd17c8 100644 --- a/src/main/native-chat/agent-session-journal/journal-queued-message-reads.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-queued-message-reads.test.ts @@ -146,3 +146,12 @@ it('a publication read inside a rolled-back consume cannot survive invalidation' expect(publication()).toEqual(first) expect(journal.queuedMessages.list().map((row) => row.messageId)).toEqual(['head']) }) + +it('does not derive pauses when no readable card is waiting', async () => { + await queue('returned') + const db = openTestJournalHostDatabase(root).db + db.prepare("UPDATE queued_messages SET state = 'returned'").run() + const pauses = vi.spyOn(journal.queuedMessages, 'pauses') + expect(journal.queuedMessages.nextSendable()).toBeNull() + expect(pauses).not.toHaveBeenCalled() +}) diff --git a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts index 9e35f65542ce..438f27e55c55 100644 --- a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts +++ b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts @@ -39,6 +39,7 @@ import { } from './queued-message-table' import { hasReadableQueuedMessage, + hasWaitingQueuedMessage, getQueuedMessageHeader, queuedMessageAwaitingReopen, queuedMessageHeaders, @@ -112,7 +113,9 @@ export class JournalQueuedMessages { } nextSendable() { - return nextSendableQueuedCard(this.pauses(), this.headers('unsettled')) + return hasWaitingQueuedMessage(this.deps.database().db, this.deps.sessionId) + ? nextSendableQueuedCard(this.pauses(), this.headers('unsettled')) + : null } get(messageId: string): QueuedMessageRow | null { @@ -359,13 +362,10 @@ export class JournalQueuedMessages { */ repairAndPrune(): Promise { // No draft, no work, and no write. - if ( - this.deps.readOnly() || - !hasReadableQueuedMessage(this.deps.database().db, this.deps.sessionId) - ) { + const { sessionId } = this.deps + if (this.deps.readOnly() || !hasReadableQueuedMessage(this.deps.database().db, sessionId)) { return Promise.resolve() } - const { sessionId } = this.deps return this.transact( (db) => { const [now, state] = [this.deps.now(), this.deps.state()] diff --git a/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts b/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts index 50e5f5521896..b360178697d0 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts @@ -326,10 +326,14 @@ describe('indexed bookkeeping', () => { const state = createJournalReducerState(SESSION, EPOCH) state.submissions.set('rejected', { ...submission('rejected', 'rejected'), + queuedMessageId: 'owed', rejection: agentSessionFailureFact('providerRejected') }) state.submissions.set('accepted', submission('accepted', 'accepted')) - state.submissions.set('corrupt-ref', submission('corrupt-ref', 'rejected')) + state.submissions.set('corrupt-ref', { + ...submission('corrupt-ref', 'rejected'), + queuedMessageId: 'corrupt' + }) const prepare = vi.spyOn(db, 'prepare') const parse = vi.spyOn(JSON, 'parse') expect(queuedMessageSettlementOwed(db, SESSION, state.submissions)).toBe(true) @@ -411,13 +415,17 @@ describe('indexed bookkeeping', () => { hasWaitingQueuedMessage(db, SESSION) const queries = prepare.mock.calls.map(([sql]) => sql) prepare.mockRestore() - expect(queryPlan(queries[0], [SESSION])).toContain('queued_messages_unsettled_position') - expect(queryPlan(queries[1], [SESSION, 'handoff'])).toContain('queued_messages_consumed_as') + expect(queryPlan(queries[0], [SESSION])).toContain( + 'queued_messages_readable_unsettled_position' + ) + expect(queryPlan(queries[1], [SESSION, 'handoff'])).toContain('queued_messages_consume_state') expect(queryPlan(queries[2], [SESSION, 900])).toContain('queued_messages_state_settled') expect(queryPlan(queries[3], [SESSION, 900, 1, 'dispatched'])).toContain( 'queued_messages_state_settled' ) - expect(queryPlan(queries[4], [SESSION])).toContain('queued_messages_state_settled') + expect(queryPlan(queries[4], [SESSION])).toContain( + 'queued_messages_readable_unsettled_position' + ) expect( queryPlan('SELECT MAX(position) FROM queued_messages WHERE session_id = ?', [SESSION]) ).toContain('queued_messages_position') diff --git a/src/main/native-chat/agent-session-journal/queued-message-headers.ts b/src/main/native-chat/agent-session-journal/queued-message-headers.ts index 30f12efb4d5d..b40b543340a3 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-headers.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-headers.ts @@ -5,18 +5,20 @@ import type { SqliteBindings } from '../../sqlite/sqlite-statement' import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' import { readStoredQueuedMessageHeader } from './queued-message-stored-row' import type { QueuedMessageHeader } from './queued-message-table' +import { + READABLE_QUEUED_MESSAGE, + READABLE_UNSETTLED_QUEUED_MESSAGE +} from './queued-message-readability' const HEADER_COLUMNS = 'session_id, message_id, position, fingerprint, created_at, host_instance, state, hold_reason, returned_reason, settled_at, settled_by_op, consumed_as, carried_from, queued_epoch, queued_sequence' -const READABLE = - "state IN ('waiting', 'returned', 'dispatched', 'withdrawn') AND json_valid(body_json)" export type QueuedMessageHeaderSelection = 'all' | 'unsettled' | 'waiting' const SELECTIONS = { - all: '1', - unsettled: "state IN ('waiting', 'returned')", - waiting: "state = 'waiting'" + all: READABLE_QUEUED_MESSAGE, + unsettled: READABLE_UNSETTLED_QUEUED_MESSAGE, + waiting: `${READABLE_UNSETTLED_QUEUED_MESSAGE} AND state = 'waiting'` } as const export function* queuedMessageHeaders( @@ -24,10 +26,11 @@ export function* queuedMessageHeaders( sessionId: string, selection: QueuedMessageHeaderSelection = 'all' ): IterableIterator { + // Close each walk before reusing its cached statement; for-of also closes on early return. for (const row of db .prepare( `SELECT ${HEADER_COLUMNS} FROM queued_messages - WHERE session_id = ? AND ${SELECTIONS[selection]} AND ${READABLE} ORDER BY position ASC` + WHERE session_id = ? AND ${SELECTIONS[selection]} ORDER BY position ASC` ) .iterate(sessionId)) { const header = readStoredQueuedMessageHeader(row) @@ -42,7 +45,7 @@ export function hasWaitingQueuedMessage(db: Database.Database, sessionId: string db .prepare( `SELECT 1 FROM queued_messages - WHERE session_id = ? AND state = 'waiting' AND json_valid(body_json) LIMIT 1` + WHERE session_id = ? AND ${SELECTIONS.waiting} ORDER BY position ASC LIMIT 1` ) .get(sessionId) !== undefined ) @@ -51,7 +54,10 @@ export function hasWaitingQueuedMessage(db: Database.Database, sessionId: string export function hasReadableQueuedMessage(db: Database.Database, sessionId: string): boolean { return ( db - .prepare(`SELECT 1 FROM queued_messages WHERE session_id = ? AND ${READABLE} LIMIT 1`) + .prepare( + `SELECT 1 FROM queued_messages WHERE session_id = ? AND ${READABLE_QUEUED_MESSAGE} + ORDER BY position ASC LIMIT 1` + ) .get(sessionId) !== undefined ) } @@ -65,7 +71,7 @@ export function dispatchedQueuedMessageHeader( .prepare( `SELECT ${HEADER_COLUMNS} FROM queued_messages WHERE session_id = ? AND consumed_as = ? AND consumed_as IS NOT NULL - AND state = 'dispatched' AND json_valid(body_json)` + AND state = 'dispatched' AND ${READABLE_QUEUED_MESSAGE}` ) .get(sessionId, consumedAs) return row ? readStoredQueuedMessageHeader(row) : null @@ -79,7 +85,7 @@ export function getQueuedMessageHeader( const row = db .prepare( `SELECT ${HEADER_COLUMNS} FROM queued_messages - WHERE session_id = ? AND message_id = ? AND ${READABLE}` + WHERE session_id = ? AND message_id = ? AND ${READABLE_QUEUED_MESSAGE}` ) .get(sessionId, messageId) return row ? readStoredQueuedMessageHeader(row) : null @@ -95,6 +101,7 @@ export function queuedMessageAwaitingReopen( } for (const submission of submissions.values()) { if ( + submission.queuedMessageId !== undefined && (submission.dispatchState === 'pending' || submission.dispatchState === 'unknown') && dispatchedQueuedMessageHeader(db, sessionId, submission.clientMessageId) ) { @@ -122,7 +129,7 @@ export function* expiredDispatchedQueuedMessageHeaders( .prepare( `SELECT ${HEADER_COLUMNS} FROM queued_messages WHERE session_id = ? AND state = 'dispatched' AND settled_at < ? - AND json_valid(body_json) + AND ${READABLE_QUEUED_MESSAGE} ${settledAt === null ? '' : 'AND (settled_at, message_id) > (?, ?)'} ORDER BY settled_at, message_id LIMIT 1` ) @@ -145,7 +152,7 @@ export function unsettledQueuedMessageBodyBytes(db: Database.Database, sessionId const row = db .prepare( `SELECT COALESCE(SUM(length(CAST(body_json AS BLOB))), 0) AS bytes FROM queued_messages - WHERE session_id = ? AND state IN ('waiting', 'returned') AND json_valid(body_json)` + WHERE session_id = ? AND ${READABLE_UNSETTLED_QUEUED_MESSAGE}` ) .get(sessionId) return typeof row?.bytes === 'number' ? row.bytes : 0 diff --git a/src/main/native-chat/agent-session-journal/queued-message-pause.ts b/src/main/native-chat/agent-session-journal/queued-message-pause.ts index 8d9415c6561c..db80cdb5bac2 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-pause.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-pause.ts @@ -251,6 +251,9 @@ export function resumableQueuePause( pauses: readonly DerivedQueuePause[], cards: Iterable ): DerivedQueuePause | null { + if (pauses.length === 0) { + return null + } for (const card of cards) { if (card.state === 'returned') { return null diff --git a/src/main/native-chat/agent-session-journal/queued-message-read-plans.test.ts b/src/main/native-chat/agent-session-journal/queued-message-read-plans.test.ts new file mode 100644 index 000000000000..134789235806 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-read-plans.test.ts @@ -0,0 +1,204 @@ +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import Database from '../../sqlite/sync-database' +import type { SqliteBindings } from '../../sqlite/sqlite-statement' +import { + dispatchedQueuedMessageHeader, + expiredDispatchedQueuedMessageHeaders, + getQueuedMessageHeader, + hasReadableQueuedMessage, + hasWaitingQueuedMessage, + queuedMessageAwaitingReopen, + queuedMessageHeaders, + unsettledQueuedMessageBodyBytes +} from './queued-message-headers' +import { ensureQueuedMessagesTable } from './queued-message-schema' +import { createJournalReducerState } from './journal-reducer' +import { queuedMessageSettlementOwed } from './queued-message-settlement' +import { + getQueuedMessage, + insertQueuedMessage, + listQueuedMessages, + queuedMessagesSettledByOp, + withdrawQueuedMessageInTransaction +} from './queued-message-table' +import { resumableQueuePause } from './queued-message-pause' + +let db: Database.Database +const SESSION = 'plans' + +beforeEach(() => { + db = new Database(':memory:') + ensureQueuedMessagesTable(db) + insertQueuedMessage(db, { + sessionId: SESSION, + messageId: 'head', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'head' }] }, + fingerprint: 'fp', + hostInstance: 'host', + queuedAt: { epoch: 'epoch', sequence: 1 }, + now: 1 + }) +}) + +afterEach(() => { + vi.restoreAllMocks() + db.close() +}) + +function expectPlan(read: () => unknown, bindings: SqliteBindings, index: string) { + const prepare = vi.spyOn(db, 'prepare') + read() + const sql = prepare.mock.calls[0]?.[0] + prepare.mockRestore() + if (!sql) { + throw new Error('expected a prepared query') + } + const plan = db + .prepare(`EXPLAIN QUERY PLAN ${sql}`) + .all(...bindings) + .map((row) => row.detail) + .join('\n') + expect(plan).toContain(index) + expect(plan).not.toMatch(/SCAN queued_messages|USE TEMP B-TREE/) + if ( + index.startsWith('queued_messages_readable_') && + !sql.split('FROM')[0].includes('body_json') + ) { + expect( + db + .prepare(`EXPLAIN ${sql}`) + .all(...bindings) + .map((row) => row.p4) + .join('\n') + ).not.toMatch(/json_valid|instr\(|typeof\(/) + } +} + +it('keeps all ordered and existence header reads on readable partial indexes', () => { + for (const selection of ['all', 'waiting', 'unsettled'] as const) { + expectPlan( + () => [...queuedMessageHeaders(db, SESSION, selection)], + [SESSION], + selection === 'all' + ? 'queued_messages_readable_position' + : 'queued_messages_readable_unsettled_position' + ) + } + expectPlan( + () => hasWaitingQueuedMessage(db, SESSION), + [SESSION], + 'queued_messages_readable_unsettled_position' + ) + expectPlan( + () => hasReadableQueuedMessage(db, SESSION), + [SESSION], + 'queued_messages_readable_position' + ) + expectPlan( + () => unsettledQueuedMessageBodyBytes(db, SESSION), + [SESSION], + 'queued_messages_state_settled' + ) + expectPlan( + () => listQueuedMessages(db, SESSION), + [SESSION], + 'queued_messages_readable_unsettled_position' + ) +}) + +it('keeps point, consume, retention, withdrawal and receipt queries indexed', () => { + expectPlan( + () => getQueuedMessageHeader(db, SESSION, 'head'), + [SESSION, 'head'], + 'sqlite_autoindex_queued_messages_1' + ) + expectPlan( + () => getQueuedMessage(db, SESSION, 'head'), + [SESSION, 'head'], + 'sqlite_autoindex_queued_messages_1' + ) + expectPlan( + () => queuedMessagesSettledByOp(db, SESSION, 'op'), + [SESSION, 'op'], + 'queued_messages_position' + ) + expectPlan( + () => + withdrawQueuedMessageInTransaction(db, { + sessionId: SESSION, + messageId: 'head', + settledByOp: 'op', + now: 2 + }), + [2, 'op', SESSION, 'head'], + 'sqlite_autoindex_queued_messages_1' + ) + db.prepare("UPDATE queued_messages SET state = 'dispatched', consumed_as = 'handoff'").run() + expectPlan( + () => dispatchedQueuedMessageHeader(db, SESSION, 'handoff'), + [SESSION, 'handoff'], + 'queued_messages_consume_state' + ) + expectPlan( + () => [...expiredDispatchedQueuedMessageHeaders(db, SESSION, 3)], + [SESSION, 3], + 'queued_messages_state_settled' + ) +}) + +it('never probes direct submissions for settlement or reopen', () => { + db.prepare("UPDATE queued_messages SET state = 'withdrawn'").run() + const state = createJournalReducerState(SESSION, 'epoch') + for (const dispatchState of ['rejected', 'pending', 'unknown'] as const) { + state.submissions.set(dispatchState, { + clientMessageId: dispatchState, + dispatchState, + fence: 0, + payloadFingerprint: 'fp', + providerItemId: null, + reason: null, + submittedAt: 1, + resolvedAt: null + }) + } + const prepare = vi.spyOn(db, 'prepare') + expect(queuedMessageSettlementOwed(db, SESSION, state.submissions)).toBe(false) + expect(prepare).not.toHaveBeenCalled() + expect(queuedMessageAwaitingReopen(db, SESSION, state.submissions)).toBe(false) + expect(prepare).toHaveBeenCalledTimes(1) + expect(prepare.mock.calls[0]?.[0]).not.toContain('consumed_as') +}) + +it('reads a consume link fresh after provisional state changes roll back', () => { + db.prepare("UPDATE queued_messages SET state = 'returned', consumed_as = 'handoff'").run() + const state = createJournalReducerState(SESSION, 'epoch') + state.submissions.set('handoff', { + clientMessageId: 'handoff', + queuedMessageId: 'head', + dispatchState: 'rejected', + fence: 0, + payloadFingerprint: 'fp', + providerItemId: null, + reason: null, + submittedAt: 1, + resolvedAt: null + }) + expect(queuedMessageSettlementOwed(db, SESSION, state.submissions)).toBe(false) + db.exec('BEGIN') + try { + db.prepare("UPDATE queued_messages SET state = 'dispatched'").run() + expect(queuedMessageSettlementOwed(db, SESSION, state.submissions)).toBe(true) + } finally { + db.exec('ROLLBACK') + } + expect(queuedMessageSettlementOwed(db, SESSION, state.submissions)).toBe(false) +}) + +it('does not open the card iterator when there is no resumable pause', () => { + const cards = { + [Symbol.iterator]() { + throw new Error('no pause can hold a card') + } + } + expect(resumableQueuePause([], cards)).toBeNull() +}) diff --git a/src/main/native-chat/agent-session-journal/queued-message-readability.test.ts b/src/main/native-chat/agent-session-journal/queued-message-readability.test.ts new file mode 100644 index 000000000000..1ff408b2864b --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-readability.test.ts @@ -0,0 +1,81 @@ +import { afterEach, beforeEach, expect, it } from 'vitest' +import Database from '../../sqlite/sync-database' +import { ensureQueuedMessagesTable } from './queued-message-schema' +import { + getQueuedMessageHeader, + hasReadableQueuedMessage, + hasWaitingQueuedMessage, + queuedMessageHeaders, + unsettledQueuedMessageBodyBytes +} from './queued-message-headers' +import { + getQueuedMessage, + insertQueuedMessage, + listQueuedMessages, + queuedMessagesSettledByOp, + withdrawQueuedMessageInTransaction +} from './queued-message-table' + +let db: Database.Database +const SESSION = 'readability' +const body = { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'card' }] } as const + +beforeEach(() => { + db = new Database(':memory:') + ensureQueuedMessagesTable(db) + for (const messageId of ['a', 'b']) { + insertQueuedMessage(db, { + sessionId: SESSION, + messageId, + body: { ...body, blocks: [...body.blocks] }, + fingerprint: 'fp', + hostInstance: 'host', + queuedAt: { epoch: 'epoch', sequence: 1 }, + now: 1 + }) + } +}) + +afterEach(() => db.close()) + +it.each([ + { name: 'NUL suffix', stored: `${JSON.stringify(body)}\u0000garbage` }, + { name: 'BLOB JSON', stored: Buffer.from(JSON.stringify(body)) }, + { name: 'invalid JSON', stored: '{' }, + { name: 'SQLite nesting limit', stored: `${'['.repeat(1001)}0${']'.repeat(1001)}` } +])('uses the same readable predicate for a $name body everywhere', ({ stored }) => { + db.prepare( + "UPDATE queued_messages SET body_json = ?, settled_by_op = 'op' WHERE message_id = 'a'" + ).run(stored) + expect(getQueuedMessage(db, SESSION, 'a')).toBeNull() + expect(getQueuedMessageHeader(db, SESSION, 'a')).toBeNull() + expect(listQueuedMessages(db, SESSION).map((row) => row.messageId)).toEqual(['b']) + expect(queuedMessagesSettledByOp(db, SESSION, 'op')).toEqual([]) + for (const selection of ['all', 'waiting', 'unsettled'] as const) { + expect([...queuedMessageHeaders(db, SESSION, selection)].map((row) => row.messageId)).toEqual([ + 'b' + ]) + } + expect(unsettledQueuedMessageBodyBytes(db, SESSION)).toBe(Buffer.byteLength(JSON.stringify(body))) + expect( + withdrawQueuedMessageInTransaction(db, { + sessionId: SESSION, + messageId: 'a', + settledByOp: 'op', + now: 2 + }) + ).toBe(0) + db.prepare("DELETE FROM queued_messages WHERE message_id = 'b'").run() + expect(hasWaitingQueuedMessage(db, SESSION)).toBe(false) + expect(hasReadableQueuedMessage(db, SESSION)).toBe(false) +}) + +it('retains valid escaped NUL text in both header and body reads', () => { + const text = 'contains\u0000a NUL' + const valid = { ...body, blocks: [{ type: 'text', text }] } + db.prepare("UPDATE queued_messages SET body_json = ? WHERE message_id = 'a'").run( + JSON.stringify(valid) + ) + expect(getQueuedMessage(db, SESSION, 'a')?.body).toEqual(valid) + expect(getQueuedMessageHeader(db, SESSION, 'a')?.messageId).toBe('a') +}) diff --git a/src/main/native-chat/agent-session-journal/queued-message-readability.ts b/src/main/native-chat/agent-session-journal/queued-message-readability.ts new file mode 100644 index 000000000000..59a3f8ed8772 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-readability.ts @@ -0,0 +1,7 @@ +// SQLite accepts JSON text in BLOBs and ignores NUL suffixes that JSON.parse rejects. +export const READABLE_QUEUED_MESSAGE_BODY = + "typeof(body_json) = 'text' AND instr(body_json, char(0)) = 0 AND json_valid(body_json)" + +export const READABLE_QUEUED_MESSAGE = `state IN ('waiting', 'returned', 'dispatched', 'withdrawn') AND ${READABLE_QUEUED_MESSAGE_BODY}` + +export const READABLE_UNSETTLED_QUEUED_MESSAGE = `state IN ('waiting', 'returned') AND ${READABLE_QUEUED_MESSAGE}` diff --git a/src/main/native-chat/agent-session-journal/queued-message-schema.ts b/src/main/native-chat/agent-session-journal/queued-message-schema.ts index b343b2f0b508..4dc24121654f 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-schema.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-schema.ts @@ -1,6 +1,10 @@ // The draft table's shape, created and healed at every writable open. import type Database from '../../sqlite/sync-database' +import { + READABLE_QUEUED_MESSAGE, + READABLE_UNSETTLED_QUEUED_MESSAGE +} from './queued-message-readability' /** Columns a later build added, so an older draft table can gain them in place. */ const NULLABLE_COLUMNS: readonly (readonly [name: string, type: string])[] = [ @@ -69,8 +73,12 @@ CREATE UNIQUE INDEX IF NOT EXISTS queued_messages_consumed_as ON queued_messages (session_id, consumed_as) WHERE consumed_as IS NOT NULL; CREATE INDEX IF NOT EXISTS queued_messages_position ON queued_messages (session_id, position); -CREATE INDEX IF NOT EXISTS queued_messages_unsettled_position - ON queued_messages (session_id, position) WHERE state IN ('waiting', 'returned'); +CREATE INDEX IF NOT EXISTS queued_messages_readable_position + ON queued_messages (session_id, position) WHERE ${READABLE_QUEUED_MESSAGE}; +CREATE INDEX IF NOT EXISTS queued_messages_readable_unsettled_position + ON queued_messages (session_id, position) WHERE ${READABLE_UNSETTLED_QUEUED_MESSAGE}; +CREATE INDEX IF NOT EXISTS queued_messages_consume_state + ON queued_messages (session_id, consumed_as, state) WHERE consumed_as IS NOT NULL; CREATE INDEX IF NOT EXISTS queued_messages_state_settled ON queued_messages (session_id, state, settled_at, message_id); `) diff --git a/src/main/native-chat/agent-session-journal/queued-message-settlement.ts b/src/main/native-chat/agent-session-journal/queued-message-settlement.ts index b481898e2b25..a511bab4aac5 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-settlement.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-settlement.ts @@ -41,6 +41,7 @@ function* owedQueuedMessageSettlements( ): IterableIterator { for (const submission of submissions.values()) { if ( + submission.queuedMessageId !== undefined && consumedSubmissionWasRejected(submission) && dispatchedQueuedMessageHeader(db, sessionId, submission.clientMessageId) ) { diff --git a/src/main/native-chat/agent-session-journal/queued-message-store.test.ts b/src/main/native-chat/agent-session-journal/queued-message-store.test.ts index 02e07bc53ee6..764b9893e658 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-store.test.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-store.test.ts @@ -717,6 +717,8 @@ describe('open-time repair and retention', () => { await journal.close() clock += QUEUED_MESSAGE_REPLAY_WINDOW_MS + 1_000 journal = await open() + expect(journal.queuedMessages.get('accepted-1')).toBeNull() + expect(journal.queuedMessages.get('withdrawn-1')).toBeNull() expect(journal.queuedMessages.list().map((row) => [row.messageId, row.state])).toEqual([ ['waiting-1', 'waiting'], ['returned-1', 'returned'] diff --git a/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts b/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts index c1599d2e6f8b..36711e68c61d 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts @@ -10,7 +10,7 @@ import type { QueuedMessageHeader, QueuedMessageRow } from './queued-message-tab /** A `queued_messages` row as this file's SELECTs return it; null when it cannot be read back. */ export function readStoredQueuedMessageRow(row: SqliteRow): QueuedMessageRow | null { const header = readStoredQueuedMessageHeader(row) - if (!header || typeof row.body_json !== 'string') { + if (!header || row.readable_body !== 1 || typeof row.body_json !== 'string') { return null } let body: AgentJournalMessageItem diff --git a/src/main/native-chat/agent-session-journal/queued-message-table.ts b/src/main/native-chat/agent-session-journal/queued-message-table.ts index 4b618f002247..cad8f37a6c1d 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-table.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-table.ts @@ -16,6 +16,10 @@ import type { import type { QUEUED_MESSAGE_PAUSED_SEND_FAILED } from '../../../shared/agent-session-queued-message-wire' import { rejectedDraftSettlement } from './journal-dispatch-settlement' import { readStoredQueuedMessageRow } from './queued-message-stored-row' +import { + READABLE_QUEUED_MESSAGE_BODY, + READABLE_UNSETTLED_QUEUED_MESSAGE +} from './queued-message-readability' export type QueuedMessageState = 'waiting' | 'dispatched' | 'returned' | 'withdrawn' @@ -78,6 +82,7 @@ export type QueuedMessageRow = QueuedMessageHeader & { const COLUMNS = 'session_id, message_id, position, body_json, fingerprint, created_at, host_instance, state, hold_reason, returned_reason, returned_rejection, settled_at, settled_by_op, consumed_as, carried_from, queued_epoch, queued_sequence' +const READ_COLUMNS = `${COLUMNS}, (${READABLE_QUEUED_MESSAGE_BODY}) AS readable_body` export function insertQueuedMessage( db: Database.Database, @@ -139,8 +144,8 @@ export function insertQueuedMessage( export function listQueuedMessages(db: Database.Database, sessionId: string): QueuedMessageRow[] { return db .prepare( - `SELECT ${COLUMNS} FROM queued_messages - WHERE session_id = ? AND state IN ('waiting', 'returned') ORDER BY position ASC` + `SELECT ${READ_COLUMNS} FROM queued_messages + WHERE session_id = ? AND ${READABLE_UNSETTLED_QUEUED_MESSAGE} ORDER BY position ASC` ) .all(sessionId) .flatMap((row) => readStoredQueuedMessageRow(row) ?? []) @@ -152,7 +157,7 @@ export function getQueuedMessage( messageId: string ): QueuedMessageRow | null { const row = db - .prepare(`SELECT ${COLUMNS} FROM queued_messages WHERE session_id = ? AND message_id = ?`) + .prepare(`SELECT ${READ_COLUMNS} FROM queued_messages WHERE session_id = ? AND message_id = ?`) .get(sessionId, messageId) return row === undefined ? null : readStoredQueuedMessageRow(row) } @@ -244,8 +249,7 @@ export function withdrawQueuedMessageInTransaction( .prepare( `UPDATE queued_messages SET state = 'withdrawn', hold_reason = NULL, settled_at = ?, settled_by_op = ? - WHERE session_id = ? AND message_id = ? AND state IN ('waiting', 'returned') - AND json_valid(body_json)` + WHERE session_id = ? AND message_id = ? AND ${READABLE_UNSETTLED_QUEUED_MESSAGE}` ) .run(input.now, input.settledByOp, input.sessionId, input.messageId).changes ) @@ -303,7 +307,7 @@ export function queuedMessagesSettledByOp( ): QueuedMessageRow[] { return db .prepare( - `SELECT ${COLUMNS} FROM queued_messages + `SELECT ${READ_COLUMNS} FROM queued_messages WHERE session_id = ? AND settled_by_op = ? ORDER BY position ASC` ) .all(sessionId, settledByOp) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts index e349a455415e..f42faa278495 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts @@ -119,16 +119,17 @@ export function nextStructuredQueuedMessage(input: { record: AgentSessionRecord | null fence: number }): QueuedMessageHeader | null { - const next = oldestActionableQueuedMessage(input.journal) const { journal, fence } = input - // The gate's cheap `working` first: publication asks on every streamed frame, and the gate's - // prompt check walks the whole fold. + // Token-frame publication checks working before reading the queue or walking prompts. if ( - next === null || isStructuredAgentSessionMainAgentWorking(journal.activeTurnId(), journal.submissions(), fence) ) { return null } + const next = oldestActionableQueuedMessage(journal) + if (next === null) { + return null + } return structuredQueueHold(input) === null ? next : null } @@ -283,12 +284,12 @@ export class StructuredAgentSessionQueuedMessageDrain { try { if ( !journal.queuedMessages.settlementOwed() && - (oldestActionableQueuedMessage(journal) === null || - isStructuredAgentSessionMainAgentWorking( - journal.activeTurnId(), - journal.submissions(), - this.deps.conversationFence(sessionId) - )) + (isStructuredAgentSessionMainAgentWorking( + journal.activeTurnId(), + journal.submissions(), + this.deps.conversationFence(sessionId) + ) || + oldestActionableQueuedMessage(journal) === null) ) { return } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-read-regressions.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-read-regressions.test.ts new file mode 100644 index 000000000000..926b0391650d --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-read-regressions.test.ts @@ -0,0 +1,109 @@ +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import { + createQueuedMessageTestRig, + eventually, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_SESSION as SESSION, + hostTestMessage +} from './structured-agent-session-host-test-data' +import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' +import { insertQueuedMessage } from '../agent-session-journal/queued-message-table' +import { readQueuePublication } from './structured-agent-session-queued-publication' +import { nextStructuredQueuedMessage } from './structured-agent-session-queued-messages' + +let rig: QueuedMessageTestRig + +beforeEach(async () => { + rig = await createQueuedMessageTestRig() +}) + +afterEach(async () => { + vi.restoreAllMocks() + await rig.dispose() +}) + +function openJournal() { + const journal = rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal + if (!journal) { + throw new Error('expected the conversation open') + } + return journal +} + +it.each([0, 5, 2000])( + 'uses zero queue SQL on a streamed frame with %i waiting cards', + async (n) => { + const working = await rig.workingSend() + await rig.settleAccepted(working, 'accepted') + const journal = openJournal() + vi.spyOn(journal, 'activeTurnId').mockReturnValue('streaming') + const db = openTestJournalHostDatabase(rig.root).db + db.exec('BEGIN') + try { + for (let i = 0; i < n; i++) { + insertQueuedMessage(db, { + sessionId: SESSION, + messageId: `card-${i}`, + body: hostTestMessage('x'.repeat(400)), + fingerprint: `fp-${i}`, + hostInstance: 'host', + queuedAt: { epoch: 'epoch', sequence: 1 }, + now: 1 + }) + } + db.exec('COMMIT') + } catch (error) { + db.exec('ROLLBACK') + throw error + } + journal.queuedMessages.invalidate() + const gate = () => ({ record: rig.store.getRecord(SESSION), fence: 1 }) + const first = readQueuePublication(journal, gate) + const prepare = vi.spyOn(db, 'prepare') + const pick = vi.spyOn(journal.queuedMessages, 'nextSendable') + for (let frame = 0; frame < 3; frame++) { + rig.host['queued'].drain.schedule(SESSION) + expect(nextStructuredQueuedMessage({ journal, ...gate() })).toBeNull() + expect(readQueuePublication(journal, gate)).toBe(first) + } + expect(pick).not.toHaveBeenCalled() + expect(prepare.mock.calls.filter(([sql]) => sql.includes('queued_messages'))).toEqual([]) + } +) + +it('still settles owed bookkeeping while an agent works', async () => { + await rig.workingSend() + const journal = openJournal() + vi.spyOn(journal.queuedMessages, 'settlementOwed').mockReturnValue(true) + const settle = vi.spyOn(journal.queuedMessages, 'settleOwed').mockResolvedValue(undefined) + const pick = vi.spyOn(journal.queuedMessages, 'nextSendable') + rig.host['queued'].drain.schedule(SESSION) + await eventually(() => expect(settle).toHaveBeenCalled()) + expect(pick).not.toHaveBeenCalled() +}) + +it.each(['NUL suffix', 'BLOB JSON'])('sends card b past a corrupt %s head', async (kind) => { + const working = await rig.workingSend() + const a = await rig.send('a', 'queue-if-active').result + const b = await rig.send('b', 'queue-if-active').result + if (!a.ok || !b.ok || !('queued' in a.value) || !('queued' in b.value)) { + throw new Error('expected both cards queued') + } + const aId = a.value.queued.messageId + const bId = b.value.queued.messageId + const json = JSON.stringify(hostTestMessage('a')) + const stored = kind === 'NUL suffix' ? `${json}\u0000garbage` : Buffer.from(json) + openTestJournalHostDatabase(rig.root) + .db.prepare('UPDATE queued_messages SET body_json = ? WHERE message_id = ?') + .run(stored, aId) + const journal = openJournal() + journal.queuedMessages.invalidate() + expect(journal.queuedMessages.get(aId)).toBeNull() + expect(journal.queuedMessages.nextSendable()?.messageId).toBe(bId) + await rig.settleAccepted(working, 'done') + await eventually(async () => expect(await rig.handoff(bId)).toBeDefined()) + expect(await rig.handoff(aId)).toBeUndefined() + expect((await rig.handoff(bId))?.queuedMessageId).toBe(bId) +}) From be7c141444cf3039221d397519553b039aad3a97 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 8 Oct 2026 14:05:26 -0700 Subject: [PATCH 3/5] perf(native-chat): validate queued bodies only where reads need it Drop the all-state readable index and the consume-state index (they cost every write and served rare or unproven reads) and the extra readable_body column on body reads; JSON.parse already rejects the corrupt shapes. Receipts read exactly as before. --- .../queued-message-headers.test.ts | 2 +- .../queued-message-read-plans.test.ts | 17 +++++----------- .../queued-message-readability.test.ts | 20 ++++++++++++++++--- .../queued-message-schema.ts | 10 ++-------- .../queued-message-stored-row.ts | 2 +- .../queued-message-table.ts | 12 ++++------- 6 files changed, 30 insertions(+), 33 deletions(-) diff --git a/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts b/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts index b360178697d0..03573eca8978 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts @@ -418,7 +418,7 @@ describe('indexed bookkeeping', () => { expect(queryPlan(queries[0], [SESSION])).toContain( 'queued_messages_readable_unsettled_position' ) - expect(queryPlan(queries[1], [SESSION, 'handoff'])).toContain('queued_messages_consume_state') + expect(queryPlan(queries[1], [SESSION, 'handoff'])).toContain('queued_messages_consumed_as') expect(queryPlan(queries[2], [SESSION, 900])).toContain('queued_messages_state_settled') expect(queryPlan(queries[3], [SESSION, 900, 1, 'dispatched'])).toContain( 'queued_messages_state_settled' diff --git a/src/main/native-chat/agent-session-journal/queued-message-read-plans.test.ts b/src/main/native-chat/agent-session-journal/queued-message-read-plans.test.ts index 134789235806..86f313954d02 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-read-plans.test.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-read-plans.test.ts @@ -60,10 +60,7 @@ function expectPlan(read: () => unknown, bindings: SqliteBindings, index: string .join('\n') expect(plan).toContain(index) expect(plan).not.toMatch(/SCAN queued_messages|USE TEMP B-TREE/) - if ( - index.startsWith('queued_messages_readable_') && - !sql.split('FROM')[0].includes('body_json') - ) { + if (index === 'queued_messages_readable_unsettled_position') { expect( db .prepare(`EXPLAIN ${sql}`) @@ -74,13 +71,13 @@ function expectPlan(read: () => unknown, bindings: SqliteBindings, index: string } } -it('keeps all ordered and existence header reads on readable partial indexes', () => { +it('keeps unsettled reads on the readable index and rare all-state reads on the position index', () => { for (const selection of ['all', 'waiting', 'unsettled'] as const) { expectPlan( () => [...queuedMessageHeaders(db, SESSION, selection)], [SESSION], selection === 'all' - ? 'queued_messages_readable_position' + ? 'queued_messages_position' : 'queued_messages_readable_unsettled_position' ) } @@ -89,11 +86,7 @@ it('keeps all ordered and existence header reads on readable partial indexes', ( [SESSION], 'queued_messages_readable_unsettled_position' ) - expectPlan( - () => hasReadableQueuedMessage(db, SESSION), - [SESSION], - 'queued_messages_readable_position' - ) + expectPlan(() => hasReadableQueuedMessage(db, SESSION), [SESSION], 'queued_messages_position') expectPlan( () => unsettledQueuedMessageBodyBytes(db, SESSION), [SESSION], @@ -137,7 +130,7 @@ it('keeps point, consume, retention, withdrawal and receipt queries indexed', () expectPlan( () => dispatchedQueuedMessageHeader(db, SESSION, 'handoff'), [SESSION, 'handoff'], - 'queued_messages_consume_state' + 'queued_messages_consumed_as' ) expectPlan( () => [...expiredDispatchedQueuedMessageHeaders(db, SESSION, 3)], diff --git a/src/main/native-chat/agent-session-journal/queued-message-readability.test.ts b/src/main/native-chat/agent-session-journal/queued-message-readability.test.ts index 1ff408b2864b..af22e201b7a5 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-readability.test.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-readability.test.ts @@ -41,9 +41,8 @@ afterEach(() => db.close()) it.each([ { name: 'NUL suffix', stored: `${JSON.stringify(body)}\u0000garbage` }, { name: 'BLOB JSON', stored: Buffer.from(JSON.stringify(body)) }, - { name: 'invalid JSON', stored: '{' }, - { name: 'SQLite nesting limit', stored: `${'['.repeat(1001)}0${']'.repeat(1001)}` } -])('uses the same readable predicate for a $name body everywhere', ({ stored }) => { + { name: 'invalid JSON', stored: '{' } +])('excludes a $name body from headers and body reads', ({ stored }) => { db.prepare( "UPDATE queued_messages SET body_json = ?, settled_by_op = 'op' WHERE message_id = 'a'" ).run(stored) @@ -70,6 +69,21 @@ it.each([ expect(hasReadableQueuedMessage(db, SESSION)).toBe(false) }) +it('filters bodies beyond the SQLite nesting limit from the unsettled list', () => { + db.prepare("UPDATE queued_messages SET body_json = ? WHERE message_id = 'a'").run( + `${'['.repeat(1001)}0${']'.repeat(1001)}` + ) + expect(listQueuedMessages(db, SESSION).map((row) => row.messageId)).toEqual(['b']) +}) + +it('reads valid bodies and receipts without a computed readability column', () => { + db.prepare("UPDATE queued_messages SET settled_by_op = 'op' WHERE message_id = 'a'").run() + expect(getQueuedMessage(db, SESSION, 'a')?.body).toEqual(body) + expect(queuedMessagesSettledByOp(db, SESSION, 'op')).toEqual([ + expect.objectContaining({ messageId: 'a', body }) + ]) +}) + it('retains valid escaped NUL text in both header and body reads', () => { const text = 'contains\u0000a NUL' const valid = { ...body, blocks: [{ type: 'text', text }] } diff --git a/src/main/native-chat/agent-session-journal/queued-message-schema.ts b/src/main/native-chat/agent-session-journal/queued-message-schema.ts index 4dc24121654f..6038237af93b 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-schema.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-schema.ts @@ -1,10 +1,7 @@ // The draft table's shape, created and healed at every writable open. import type Database from '../../sqlite/sync-database' -import { - READABLE_QUEUED_MESSAGE, - READABLE_UNSETTLED_QUEUED_MESSAGE -} from './queued-message-readability' +import { READABLE_UNSETTLED_QUEUED_MESSAGE } from './queued-message-readability' /** Columns a later build added, so an older draft table can gain them in place. */ const NULLABLE_COLUMNS: readonly (readonly [name: string, type: string])[] = [ @@ -73,12 +70,9 @@ CREATE UNIQUE INDEX IF NOT EXISTS queued_messages_consumed_as ON queued_messages (session_id, consumed_as) WHERE consumed_as IS NOT NULL; CREATE INDEX IF NOT EXISTS queued_messages_position ON queued_messages (session_id, position); -CREATE INDEX IF NOT EXISTS queued_messages_readable_position - ON queued_messages (session_id, position) WHERE ${READABLE_QUEUED_MESSAGE}; +-- Keep WHERE textually identical to the query predicate; changing it requires renaming this index. CREATE INDEX IF NOT EXISTS queued_messages_readable_unsettled_position ON queued_messages (session_id, position) WHERE ${READABLE_UNSETTLED_QUEUED_MESSAGE}; -CREATE INDEX IF NOT EXISTS queued_messages_consume_state - ON queued_messages (session_id, consumed_as, state) WHERE consumed_as IS NOT NULL; CREATE INDEX IF NOT EXISTS queued_messages_state_settled ON queued_messages (session_id, state, settled_at, message_id); `) diff --git a/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts b/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts index 36711e68c61d..c1599d2e6f8b 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-stored-row.ts @@ -10,7 +10,7 @@ import type { QueuedMessageHeader, QueuedMessageRow } from './queued-message-tab /** A `queued_messages` row as this file's SELECTs return it; null when it cannot be read back. */ export function readStoredQueuedMessageRow(row: SqliteRow): QueuedMessageRow | null { const header = readStoredQueuedMessageHeader(row) - if (!header || row.readable_body !== 1 || typeof row.body_json !== 'string') { + if (!header || typeof row.body_json !== 'string') { return null } let body: AgentJournalMessageItem diff --git a/src/main/native-chat/agent-session-journal/queued-message-table.ts b/src/main/native-chat/agent-session-journal/queued-message-table.ts index cad8f37a6c1d..a9d4d7cd3141 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-table.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-table.ts @@ -16,10 +16,7 @@ import type { import type { QUEUED_MESSAGE_PAUSED_SEND_FAILED } from '../../../shared/agent-session-queued-message-wire' import { rejectedDraftSettlement } from './journal-dispatch-settlement' import { readStoredQueuedMessageRow } from './queued-message-stored-row' -import { - READABLE_QUEUED_MESSAGE_BODY, - READABLE_UNSETTLED_QUEUED_MESSAGE -} from './queued-message-readability' +import { READABLE_UNSETTLED_QUEUED_MESSAGE } from './queued-message-readability' export type QueuedMessageState = 'waiting' | 'dispatched' | 'returned' | 'withdrawn' @@ -82,7 +79,6 @@ export type QueuedMessageRow = QueuedMessageHeader & { const COLUMNS = 'session_id, message_id, position, body_json, fingerprint, created_at, host_instance, state, hold_reason, returned_reason, returned_rejection, settled_at, settled_by_op, consumed_as, carried_from, queued_epoch, queued_sequence' -const READ_COLUMNS = `${COLUMNS}, (${READABLE_QUEUED_MESSAGE_BODY}) AS readable_body` export function insertQueuedMessage( db: Database.Database, @@ -144,7 +140,7 @@ export function insertQueuedMessage( export function listQueuedMessages(db: Database.Database, sessionId: string): QueuedMessageRow[] { return db .prepare( - `SELECT ${READ_COLUMNS} FROM queued_messages + `SELECT ${COLUMNS} FROM queued_messages WHERE session_id = ? AND ${READABLE_UNSETTLED_QUEUED_MESSAGE} ORDER BY position ASC` ) .all(sessionId) @@ -157,7 +153,7 @@ export function getQueuedMessage( messageId: string ): QueuedMessageRow | null { const row = db - .prepare(`SELECT ${READ_COLUMNS} FROM queued_messages WHERE session_id = ? AND message_id = ?`) + .prepare(`SELECT ${COLUMNS} FROM queued_messages WHERE session_id = ? AND message_id = ?`) .get(sessionId, messageId) return row === undefined ? null : readStoredQueuedMessageRow(row) } @@ -307,7 +303,7 @@ export function queuedMessagesSettledByOp( ): QueuedMessageRow[] { return db .prepare( - `SELECT ${READ_COLUMNS} FROM queued_messages + `SELECT ${COLUMNS} FROM queued_messages WHERE session_id = ? AND settled_by_op = ? ORDER BY position ASC` ) .all(sessionId, settledByOp) From bcc33e0b82ba9006937cd34e25a37138fbf20e03 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 8 Oct 2026 14:26:36 -0700 Subject: [PATCH 4/5] fix(native-chat): never let queue index builds make the journal unopenable The new queue indexes only speed up reads. Build them best-effort after the required schema, so a damaged row elsewhere or a full disk logs once instead of refusing every chat. --- .../agent-session-journal/journal-database.ts | 2 +- ...urnal-queued-message-index-failure.test.ts | 197 ++++++++++++++++++ .../queued-message-schema.ts | 11 + 3 files changed, 209 insertions(+), 1 deletion(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-queued-message-index-failure.test.ts diff --git a/src/main/native-chat/agent-session-journal/journal-database.ts b/src/main/native-chat/agent-session-journal/journal-database.ts index ba961541a7e9..5ae18bbe46fa 100644 --- a/src/main/native-chat/agent-session-journal/journal-database.ts +++ b/src/main/native-chat/agent-session-journal/journal-database.ts @@ -77,11 +77,11 @@ export function openJournalDatabase(dbPath: string): OpenJournalDatabase { } configureJournalPragmas(probe, stored) migrateJournalSchema(probe, stored) + ensureAgentSessionAttachmentClaimTables(probe) // Outside `migrateJournalSchema` on purpose: its early return skips a db // already at the current version, and this table must exist at EVERY // writable open with no `user_version` bump (see `ensureQueuedMessagesTable`). ensureQueuedMessagesTable(probe) - ensureAgentSessionAttachmentClaimTables(probe) hardenSqliteDatabaseFiles(dbPath) transferred = true return { db: probe, readOnly: false } diff --git a/src/main/native-chat/agent-session-journal/journal-queued-message-index-failure.test.ts b/src/main/native-chat/agent-session-journal/journal-queued-message-index-failure.test.ts new file mode 100644 index 000000000000..37702db49217 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-queued-message-index-failure.test.ts @@ -0,0 +1,197 @@ +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import Database from '../../sqlite/sync-database' +import { journalPragmaNumber, openJournalDatabase } from './journal-database' +import { JOURNAL_DB_SCHEMA_VERSION } from './journal-database-schema' +import { journalDatabasePath } from './journal-host-database' +import { hasWaitingQueuedMessage, queuedMessageHeaders } from './queued-message-headers' +import { getQueuedMessage, insertQueuedMessage, listQueuedMessages } from './queued-message-table' + +const OPTIONAL_INDEXES = [ + 'queued_messages_position', + 'queued_messages_readable_unsettled_position', + 'queued_messages_state_settled' +] +let root: string +let dbPath: string + +function message(text: string): AgentJournalMessageItem { + return { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] } +} + +function insert(db: Database.Database, sessionId: string, messageId: string, text = messageId) { + return insertQueuedMessage(db, { + sessionId, + messageId, + body: message(text), + fingerprint: `fp-${messageId}`, + hostInstance: 'host', + queuedAt: { epoch: 'epoch', sequence: 1 }, + now: 1 + }) +} + +function indexes(db: Database.Database) { + return db + .prepare("SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = 'queued_messages'") + .all() + .map((row) => row.name) +} + +function tables(db: Database.Database) { + return db + .prepare("SELECT name FROM sqlite_master WHERE type = 'table'") + .all() + .map((row) => row.name) +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-queue-index-failure-')) + dbPath = journalDatabasePath(root) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await rm(root, { recursive: true, force: true }) +}) + +it('opens a healthy chat when building indexes reaches another chat’s damaged overflow page', async () => { + const seed = openJournalDatabase(dbPath).db + let victim: number + let pageSize: number + try { + seed.pragma('journal_mode = DELETE') + for (const name of OPTIONAL_INDEXES) { + seed.exec(`DROP INDEX ${name}`) + } + insert(seed, 'healthy', 'a1') + insert(seed, 'damaged', 'b1', 'y'.repeat(20_000)) + const pages = seed + .prepare( + "SELECT pageno FROM dbstat WHERE name = 'queued_messages' AND pagetype = 'overflow' ORDER BY pageno" + ) + .all() + const page = pages[1]?.pageno + if (typeof page !== 'number') { + throw new Error('the large queued body must have a second overflow page') + } + victim = page + pageSize = journalPragmaNumber(seed, 'page_size') + } finally { + seed.close() + } + const bytes = await readFile(dbPath) + bytes.fill(0xff, (victim - 1) * pageSize, victim * pageSize) + await writeFile(dbPath, bytes) + + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + for (let attempt = 1; attempt <= 2; attempt++) { + const opened = openJournalDatabase(dbPath) + try { + expect(opened.readOnly).toBe(false) + expect(opened.db.isTransaction).toBe(false) + expect(warn).toHaveBeenCalledTimes(attempt) + expect(warn).toHaveBeenLastCalledWith('[journal-open] queued-message indexes skipped:', { + reason: 'journalCorrupt', + error: 'database disk image is malformed' + }) + expect(indexes(opened.db)).toEqual( + expect.arrayContaining(['queued_messages_consumed_as', 'queued_messages_position']) + ) + expect(indexes(opened.db)).not.toContain('queued_messages_readable_unsettled_position') + expect(indexes(opened.db)).not.toContain('queued_messages_state_settled') + expect(hasWaitingQueuedMessage(opened.db, 'healthy')).toBe(true) + expect([...queuedMessageHeaders(opened.db, 'healthy', 'waiting')]).toMatchObject([ + { messageId: 'a1', state: 'waiting' } + ]) + expect(listQueuedMessages(opened.db, 'healthy')).toMatchObject([{ messageId: 'a1' }]) + expect(getQueuedMessage(opened.db, 'healthy', 'a1')?.body).toEqual(message('a1')) + } finally { + opened.db.close() + } + } +}) + +it('keeps required schema and queue reads after SQLITE_FULL, then retries indexes on the next open', () => { + const failure = Object.assign(new Error('database or disk is full'), { + code: 'ERR_SQLITE_ERROR', + errcode: 13 + }) + const original = Database.prototype.exec + const transactionAtBuild: boolean[] = [] + const tablesAtBuild: unknown[][] = [] + const exec = vi.spyOn(Database.prototype, 'exec').mockImplementation(function ( + this: Database.Database, + sql: string + ) { + if (sql.includes('CREATE INDEX IF NOT EXISTS queued_messages_position')) { + transactionAtBuild.push(this.isTransaction) + tablesAtBuild.push(tables(this)) + throw failure + } + return original.call(this, sql) + }) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + const { db } = openJournalDatabase(dbPath) + try { + expect(transactionAtBuild).toEqual([false]) + expect(tablesAtBuild[0]).toEqual( + expect.arrayContaining([ + 'journal_rows', + 'agent_session_records', + 'agent_session_attachment_claims', + 'agent_session_attachment_sweeps', + 'queued_messages' + ]) + ) + expect(journalPragmaNumber(db, 'user_version')).toBe(JOURNAL_DB_SCHEMA_VERSION) + expect(indexes(db)).toContain('queued_messages_consumed_as') + for (const name of OPTIONAL_INDEXES) { + expect(indexes(db)).not.toContain(name) + } + insert(db, 'healthy', 'unreadable') + db.prepare('UPDATE queued_messages SET body_json = ? WHERE message_id = ?').run( + `${JSON.stringify(message('unreadable'))}\0suffix`, + 'unreadable' + ) + insert(db, 'healthy', 'a1') + expect(hasWaitingQueuedMessage(db, 'healthy')).toBe(true) + expect([...queuedMessageHeaders(db, 'healthy', 'waiting')]).toMatchObject([{ messageId: 'a1' }]) + expect(listQueuedMessages(db, 'healthy')).toMatchObject([{ messageId: 'a1' }]) + expect(warn).toHaveBeenCalledExactlyOnceWith('[journal-open] queued-message indexes skipped:', { + reason: 'journalUnavailable', + error: failure.message + }) + } finally { + db.close() + exec.mockRestore() + } + const reopened = openJournalDatabase(dbPath).db + try { + expect(indexes(reopened)).toEqual(expect.arrayContaining(OPTIONAL_INDEXES)) + expect(listQueuedMessages(reopened, 'healthy')).toMatchObject([{ messageId: 'a1' }]) + expect(warn).toHaveBeenCalledTimes(1) + } finally { + reopened.close() + } +}) + +it('still refuses an open when the required unique consumed_as constraint cannot be created', () => { + const failure = new Error('required unique index failed') + const original = Database.prototype.exec + vi.spyOn(Database.prototype, 'exec').mockImplementation(function ( + this: Database.Database, + sql: string + ) { + if (sql.includes('CREATE UNIQUE INDEX IF NOT EXISTS queued_messages_consumed_as')) { + throw failure + } + return original.call(this, sql) + }) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + expect(() => openJournalDatabase(dbPath)).toThrow(failure) + expect(warn).not.toHaveBeenCalled() +}) diff --git a/src/main/native-chat/agent-session-journal/queued-message-schema.ts b/src/main/native-chat/agent-session-journal/queued-message-schema.ts index 6038237af93b..afa535fb1586 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-schema.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-schema.ts @@ -1,6 +1,7 @@ // The draft table's shape, created and healed at every writable open. import type Database from '../../sqlite/sync-database' +import { classifyJournalOpenFailure } from './journal-open-failure' import { READABLE_UNSETTLED_QUEUED_MESSAGE } from './queued-message-readability' /** Columns a later build added, so an older draft table can gain them in place. */ @@ -68,6 +69,10 @@ CREATE TABLE IF NOT EXISTS queued_messages ( db.exec(` CREATE UNIQUE INDEX IF NOT EXISTS queued_messages_consumed_as ON queued_messages (session_id, consumed_as) WHERE consumed_as IS NOT NULL; +`) + // Autocommit keeps optional index failures from rolling back the required schema. + try { + db.exec(` CREATE INDEX IF NOT EXISTS queued_messages_position ON queued_messages (session_id, position); -- Keep WHERE textually identical to the query predicate; changing it requires renaming this index. @@ -76,4 +81,10 @@ CREATE INDEX IF NOT EXISTS queued_messages_readable_unsettled_position CREATE INDEX IF NOT EXISTS queued_messages_state_settled ON queued_messages (session_id, state, settled_at, message_id); `) + } catch (error) { + console.warn('[journal-open] queued-message indexes skipped:', { + reason: classifyJournalOpenFailure(error), + error: error instanceof Error ? error.message : String(error) + }) + } } From 0ff27c73afeb64bb4b64f2319a22689617b8d03c Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:02:05 -0700 Subject: [PATCH 5/5] test(native-chat): run the queued-read regression test in the Node runtime project --- config/scripts/vitest-node-runtime-files.mjs | 1 + 1 file changed, 1 insertion(+) diff --git a/config/scripts/vitest-node-runtime-files.mjs b/config/scripts/vitest-node-runtime-files.mjs index 1233564b961a..45c0e1cc38f5 100644 --- a/config/scripts/vitest-node-runtime-files.mjs +++ b/config/scripts/vitest-node-runtime-files.mjs @@ -65,6 +65,7 @@ export const NODE_RUNTIME_INCLUDE = [ 'src/renderer/src/components/right-sidebar/parent-pr-checks-projection-selector.test.ts', 'src/renderer/src/store/github/cache-persistence.test.ts', 'src/main/native-chat/agent-session-journal/**/*.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-queued-read-regressions.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-status-feed.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-conversation-name.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-status-first-input-identity.test.ts',