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', 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/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..8fd49dfd17c8 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-queued-message-reads.test.ts @@ -0,0 +1,157 @@ +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']) +}) + +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 3337727689cd..8bb29de37d32 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 @@ -9,10 +9,6 @@ import type { AgentJournalCursor, AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' -import { - AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS, - AGENT_SESSION_OPERATION_FUTURE_SKEW_MS -} from '../../../shared/agent-session-host-authority' import type { JournalHostDatabase } from './journal-host-database' import type { JournalReducerState } from './journal-reducer' import type { JournalRow } from './journal-row-schema' @@ -36,9 +32,18 @@ import { type QueuedMessageHoldReason, type QueuedMessageRow } from './queued-message-table' +import { + hasReadableQueuedMessage, + hasWaitingQueuedMessage, + 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' +import { repairAndPruneQueuedMessages } from './queued-message-retention' import { queuedMessageSettlementOwed, settleOwedQueuedMessages, @@ -50,9 +55,7 @@ import type { JournalWriteBody, JournalWriteResult } from './journal-write-queue import { QueuedMessageNotConsumableError } from './queued-message-consume-error' export { QueuedMessageNotConsumableError } from './queued-message-consume-error' -/** Tombstones must outlive the window in which their operation id could still be admitted as new. */ -export const QUEUED_MESSAGE_REPLAY_WINDOW_MS = - AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS +export { QUEUED_MESSAGE_REPLAY_WINDOW_MS } from './queued-message-retention' export type JournalQueuedMessagesDeps = { sessionId: string @@ -75,7 +78,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) {} @@ -92,16 +94,23 @@ 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 command withdrawal, 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 hasWaitingQueuedMessage(this.deps.database().db, this.deps.sessionId) + ? nextSendableQueuedCard(this.pauses(), this.headers('unsettled')) + : null } get(messageId: string): QueuedMessageRow | null { @@ -165,20 +174,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`). */ @@ -187,17 +200,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. */ @@ -279,7 +281,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() }) @@ -300,9 +302,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) } } @@ -319,13 +319,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. */ @@ -360,23 +367,17 @@ export class JournalQueuedMessages { */ repairAndPrune(): Promise { // No draft, no work, and no write. - if (this.deps.readOnly() || this.list().length === 0) { + 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()] - return ( - settleOwedQueuedMessages(db, { sessionId, state, now }) + - pruneQueuedMessages(db, { - sessionId, - now, - replayWindowMs: QUEUED_MESSAGE_REPLAY_WINDOW_MS, - submissionVerdict: retainedSubmissionVerdict(state.submissions) - }) - ) - }, + (db) => + repairAndPruneQueuedMessages(db, { + sessionId, + state: this.deps.state(), + now: this.deps.now() + }), (changed) => changed > 0 ).then(() => undefined) } 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-command-owed-settlement.test.ts b/src/main/native-chat/agent-session-journal/queued-command-owed-settlement.test.ts new file mode 100644 index 000000000000..dda41a805976 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-command-owed-settlement.test.ts @@ -0,0 +1,110 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { structuredAgentSessionCommandTurnIdentity } from '../../../shared/structured-agent-session-command-turn-identity' +import Database from '../../sqlite/sync-database' +import { createJournalReducerState, type JournalReducerState } from './journal-reducer' +import { ensureQueuedMessagesTable } from './queued-message-schema' +import { queuedMessageSettlementOwed, settleOwedQueuedMessages } from './queued-message-settlement' +import { + consumeQueuedMessageInTransaction, + getQueuedMessage, + insertQueuedMessage +} from './queued-message-table' + +const SESSION = 'owed-command' +const SUBMISSION = 'submission' +let db: Database.Database + +beforeEach(() => { + db = new Database(':memory:') + ensureQueuedMessagesTable(db) +}) + +afterEach(() => db.close()) + +function refusedCard(command: boolean) { + insertQueuedMessage(db, { + sessionId: SESSION, + messageId: 'card', + body: { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: command ? '/compact' : 'plain message' }], + ...(command ? { command: { name: 'compact' } } : {}) + }, + fingerprint: 'fp', + hostInstance: 'host', + queuedAt: { epoch: 'epoch', sequence: 1 }, + now: 1 + }) + db.exec('BEGIN IMMEDIATE') + expect( + consumeQueuedMessageInTransaction(db, { + sessionId: SESSION, + messageId: 'card', + expect: 'waiting', + consumedAs: SUBMISSION, + settledByOp: null, + now: 2 + }) + ).toBe(true) + db.exec('COMMIT') + const state = createJournalReducerState(SESSION, 'epoch') + state.submissions.set(SUBMISSION, { + clientMessageId: SUBMISSION, + queuedMessageId: 'card', + dispatchState: 'rejected', + fence: 0, + payloadFingerprint: 'fp', + providerItemId: null, + reason: null, + rejection: agentSessionFailureFact('providerRejected'), + submittedAt: 1, + resolvedAt: 2 + }) + expect(queuedMessageSettlementOwed(db, SESSION, state.submissions)).toBe(true) + return state +} + +function reportCommandTurn(state: JournalReducerState) { + const itemId = agentJournalItemKey(structuredAgentSessionCommandTurnIdentity(SUBMISSION)) + state.items.set(itemId, { + itemId, + revision: 0, + sequence: 2, + observedAt: 2, + body: { + kind: 'turn', + turnId: `compact:${SUBMISSION}`, + state: 'completed', + outcome: 'failure', + completedAt: 2 + } + }) +} + +describe('owed settlement of refused command cards', () => { + it('withdraws a refused command card whose command turn already reports the refusal', () => { + const state = refusedCard(true) + reportCommandTurn(state) + expect(settleOwedQueuedMessages(db, { sessionId: SESSION, state, now: 10 })).toBe(1) + expect(getQueuedMessage(db, SESSION, 'card')?.state).toBe('withdrawn') + expect(queuedMessageSettlementOwed(db, SESSION, state.submissions)).toBe(false) + }) + + it('returns a refused command card when its command turn is absent', () => { + const state = refusedCard(true) + expect(settleOwedQueuedMessages(db, { sessionId: SESSION, state, now: 10 })).toBe(1) + expect(getQueuedMessage(db, SESSION, 'card')?.state).toBe('returned') + expect(queuedMessageSettlementOwed(db, SESSION, state.submissions)).toBe(false) + }) + + it('returns a plain message even when its submission has a command-turn key', () => { + const state = refusedCard(false) + reportCommandTurn(state) + expect(settleOwedQueuedMessages(db, { sessionId: SESSION, state, now: 10 })).toBe(1) + expect(getQueuedMessage(db, SESSION, 'card')?.state).toBe('returned') + expect(queuedMessageSettlementOwed(db, SESSION, state.submissions)).toBe(false) + }) +}) 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..cd57f7567740 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-headers.test.ts @@ -0,0 +1,433 @@ +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, hold: null, state: 'returned', body: '{' }, + { name: 'middle', position: 2, hold: null, state: 'returned', body: '{' }, + { name: 'cleared', position: 0, hold: null, state: 'waiting', body: '{' }, + { name: 'held', position: 0, hold: 'send_failed', state: 'waiting', body: '{' }, + { name: 'unknown state', position: 0, 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 = ?, hold_reason = ?, queued_sequence = 0 + WHERE message_id = 'bad'` + ).run(bad.body, bad.state, bad.hold) + const state = createJournalReducerState(SESSION, EPOCH) + state.queuePauseMarks.reopenedSequence = 5 + state.queuePauseMarks.cleared = { sequence: 4, operationId: 'clear', messageIds: ['bad'] } + 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() + const state = createJournalReducerState(SESSION, EPOCH) + state.queuePauseMarks.latestStop = { sequence: 5, event: { reason: 'user-stop', at: 1 } } + state.queuePauseMarks.cleared = { sequence: 6, operationId: 'clear', messageIds: ['carried'] } + 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 every cleared card and the pre-reopen card are found', () => { + const first = insert('cleared-1') + const second = insert('cleared-2') + const state = createJournalReducerState(SESSION, EPOCH) + state.queuePauseMarks.reopenedSequence = 20 + state.queuePauseMarks.cleared = { + sequence: 19, + operationId: 'clear', + messageIds: ['cleared-1', 'cleared-2'] + } + function* cards() { + yield first + yield second + 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 + }) + ).toEqual([ + { + reason: 'cleared', + since: { epoch: EPOCH, sequence: 19 }, + messageIds: ['cleared-1', 'cleared-2'] + }, + { reason: 'restarted', since: { epoch: EPOCH, sequence: 20 } } + ]) + }) +}) + +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'), + queuedMessageId: 'owed', + rejection: agentSessionFailureFact('providerRejected') + }) + state.submissions.set('accepted', submission('accepted', 'accepted')) + 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) + 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_readable_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_readable_unsettled_position' + ) + 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..b40b543340a3 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-headers.ts @@ -0,0 +1,159 @@ +// 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' +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' + +export type QueuedMessageHeaderSelection = 'all' | 'unsettled' | 'waiting' + +const SELECTIONS = { + all: READABLE_QUEUED_MESSAGE, + unsettled: READABLE_UNSETTLED_QUEUED_MESSAGE, + waiting: `${READABLE_UNSETTLED_QUEUED_MESSAGE} AND state = 'waiting'` +} as const + +export function* queuedMessageHeaders( + db: Database.Database, + 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]} 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 ${SELECTIONS.waiting} ORDER BY position ASC 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_QUEUED_MESSAGE} + ORDER BY position ASC 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 ${READABLE_QUEUED_MESSAGE}` + ) + .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_QUEUED_MESSAGE}` + ) + .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.queuedMessageId !== undefined && + (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 ${READABLE_QUEUED_MESSAGE} + ${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 ${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 688265f7c278..1e43a40eb9fa 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 @@ -170,7 +170,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[] { @@ -180,26 +180,37 @@ 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 cleared = marks.cleared - if ( - cleared && + const clearHolds = + cleared !== undefined && !cleared.lifted && cleared.sequence > Math.max(latestAcceptedTurnSequence, marks.resumedSequence) - ) { - const membership = new Set(cleared.messageIds) - const messageIds = waiting.flatMap((card) => - card.messageId && membership.has(card.messageId) ? [card.messageId] : [] - ) - if (messageIds.length > 0) { - pauses.push({ reason: 'cleared', since: { epoch, sequence: cleared.sequence }, messageIds }) + const members = clearHolds ? new Set(cleared.messageIds) : null + const reopened = reopenPause(input) + // Every waiting member, in queue order: the pause holds exactly these and a rewind restates them. + const clearedIds: string[] = [] + let beforeReopen = false + if ((members && members.size > 0) || reopened) { + for (const card of input.cards) { + if (card.state === 'waiting') { + if (members && card.messageId && members.has(card.messageId)) { + clearedIds.push(card.messageId) + } + beforeReopen ||= !!reopened && card.holdReason === null && queuedBefore(reopened, card) + } + if ((!members || clearedIds.length === members.size) && (!reopened || beforeReopen)) { + break + } } } - const reopened = reopenPause(input) - if ( - reopened && - waiting.some((card) => card.holdReason === null && queuedBefore(reopened, card)) - ) { + if (cleared && clearedIds.length > 0) { + pauses.push({ + reason: 'cleared', + since: { epoch, sequence: cleared.sequence }, + messageIds: clearedIds + }) + } + if (reopened && beforeReopen) { pauses.push(reopened) } return pauses @@ -261,7 +272,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)) { @@ -279,8 +290,11 @@ export function nextSendableQueuedCard( * never offers a Resume that sends nothing. */ export function resumableQueuePause( pauses: readonly DerivedQueuePause[], - cards: readonly QueueCard[] + 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..86f313954d02 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-read-plans.test.ts @@ -0,0 +1,197 @@ +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 === 'queued_messages_readable_unsettled_position') { + expect( + db + .prepare(`EXPLAIN ${sql}`) + .all(...bindings) + .map((row) => row.p4) + .join('\n') + ).not.toMatch(/json_valid|instr\(|typeof\(/) + } +} + +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_position' + : 'queued_messages_readable_unsettled_position' + ) + } + expectPlan( + () => hasWaitingQueuedMessage(db, SESSION), + [SESSION], + 'queued_messages_readable_unsettled_position' + ) + expectPlan(() => hasReadableQueuedMessage(db, SESSION), [SESSION], 'queued_messages_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_consumed_as' + ) + 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..af22e201b7a5 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-readability.test.ts @@ -0,0 +1,95 @@ +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: '{' } +])('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) + 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('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 }] } + 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-retention.ts b/src/main/native-chat/agent-session-journal/queued-message-retention.ts index 46717268ad10..1d4717ec6ef5 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,34 @@ import type Database from '../../sqlite/sync-database' import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' -import { listQueuedMessages } from './queued-message-table' +import { + AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS, + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS +} from '../../../shared/agent-session-host-authority' +import { expiredDispatchedQueuedMessageHeaders } from './queued-message-headers' +import type { JournalReducerState } from './journal-reducer' +import { settleOwedQueuedMessages } from './queued-message-settlement' + +/** Tombstones must outlive the window in which their operation id could still be admitted as new. */ +export const QUEUED_MESSAGE_REPLAY_WINDOW_MS = + AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS + +/** Open-time repair: owed settlements apply as the live hook would have, then retention runs. */ +export function repairAndPruneQueuedMessages( + db: Database.Database, + input: { sessionId: string; state: JournalReducerState; now: number } +): number { + const { sessionId, state, now } = input + return ( + settleOwedQueuedMessages(db, input) + + pruneQueuedMessages(db, { + sessionId, + now, + replayWindowMs: QUEUED_MESSAGE_REPLAY_WINDOW_MS, + submissionVerdict: retainedSubmissionVerdict(state.submissions) + }) + ) +} /** What the loaded journal says about a dispatched draft's consumed submission. */ export type QueuedMessageSubmissionVerdict = @@ -39,10 +66,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..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,8 @@ // 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. */ const NULLABLE_COLUMNS: readonly (readonly [name: string, type: string])[] = [ @@ -68,4 +70,21 @@ CREATE TABLE IF NOT EXISTS queued_messages ( 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. +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_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) + }) + } } 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 dffe87f3af09..3214f07d30e1 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 @@ -14,27 +14,42 @@ import type { JournalReducerState } from './journal-reducer' import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' import { structuredAgentSessionCommandTurnIdentity } from '../../../shared/structured-agent-session-command-turn-identity' 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 ( + submission.queuedMessageId !== undefined && + consumedSubmissionWasRejected(submission) && + dispatchedQueuedMessageHeader(db, sessionId, submission.clientMessageId) + ) { + yield submission + } + } } /** Applies each owed settlement, and withdraws each waiting draft an applied echo proves @@ -45,37 +60,26 @@ 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, - commandTurnReported: commandTurnReported(input.state, consumedRef), + consumedRef: submission.clientMessageId, + reason: submission.reason ?? null, + rejection: submission.rejection, + commandTurnReported: commandTurnReported(input.state, submission.clientMessageId), 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 } @@ -95,9 +99,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 } @@ -107,12 +110,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-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 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 303ea060c633..1d8c8bdd2000 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 @@ -17,6 +17,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_UNSETTLED_QUEUED_MESSAGE } from './queued-message-readability' export type QueuedMessageState = 'waiting' | 'dispatched' | 'returned' | 'withdrawn' @@ -34,11 +35,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 @@ -48,7 +48,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 +62,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' @@ -124,7 +128,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 ${READABLE_UNSETTLED_QUEUED_MESSAGE} ORDER BY position ASC` + ) .all(sessionId) .flatMap((row) => readStoredQueuedMessageRow(row) ?? []) } @@ -205,11 +212,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', @@ -221,6 +224,22 @@ 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 ${READABLE_UNSETTLED_QUEUED_MESSAGE}` + ) + .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 249396d38b5d..9b5f980c5271 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,17 +118,18 @@ export function nextStructuredQueuedMessage(input: { journal: AgentSessionJournal record: AgentSessionRecord | null fence: number -}): QueuedMessageRow | null { - const next = oldestActionableQueuedMessage(input.journal) +}): QueuedMessageHeader | null { 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 } @@ -289,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 } @@ -340,8 +335,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-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts index 28be32c15759..4c8c2ab2e6c8 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts @@ -70,7 +70,6 @@ export async function withdrawQueuedMessagesForOperation( }) } - /** Draft actions run like any mutation: admitted on the session's lane, the * conversation opened for the write. */ function mutateQueued( 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-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) +}) 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 e6ecad009d4c..604184cf81df 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 @@ -284,7 +284,7 @@ describe("a /clear's carried cards", () => { 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..169208541d6d 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,6 @@ 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 { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' import { isMainAgentWorking } from './structured-agent-session-turns-cancel' import { @@ -19,12 +15,6 @@ 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) -} - /** * Runs a Stop, which calls `tookEffect` where it takes effect: after it issued the withdrawal of * the queued sends, and BEFORE the interrupt or anything that ends the child (the at-start stop, a