diff --git a/src/main/native-chat/agent-session-journal/database/journal-domain-schema-001-outline.ts b/src/main/native-chat/agent-session-journal/database/journal-domain-schema-001-outline.ts index 0cd04c60a5f..ed8e21885b6 100644 --- a/src/main/native-chat/agent-session-journal/database/journal-domain-schema-001-outline.ts +++ b/src/main/native-chat/agent-session-journal/database/journal-domain-schema-001-outline.ts @@ -1,33 +1,11 @@ export const JOURNAL_DOMAIN_SCHEMA_001_OUTLINE = ` -CREATE TABLE journal_outline_members ( +CREATE TABLE journal_outline_cache ( session_id TEXT NOT NULL, epoch TEXT NOT NULL, - item_id TEXT NOT NULL, + source_sequence INTEGER NOT NULL, source_revision INTEGER NOT NULL, - sequence INTEGER NOT NULL, - sequence_index INTEGER NOT NULL, - ordinal INTEGER NOT NULL, - structure_json TEXT NOT NULL, - turn_key TEXT, - reply TEXT, - entry_json TEXT, - short_json TEXT, - empty_json TEXT, - entry_bytes INTEGER NOT NULL, - short_bytes INTEGER NOT NULL, - empty_bytes INTEGER NOT NULL, - PRIMARY KEY(session_id,epoch,item_id) -) WITHOUT ROWID; -CREATE INDEX journal_outline_entries ON journal_outline_members(session_id,epoch,sequence,sequence_index,ordinal) - WHERE entry_json IS NOT NULL; -CREATE INDEX journal_outline_reply_candidates ON journal_outline_members(session_id,epoch,turn_key,sequence DESC,sequence_index DESC,ordinal DESC) - WHERE reply IS NOT NULL; -CREATE TABLE journal_outline_replies ( - session_id TEXT NOT NULL, - epoch TEXT NOT NULL, - turn_key TEXT NOT NULL, - reply TEXT NOT NULL, - reply_bytes INTEGER NOT NULL, - PRIMARY KEY(session_id,epoch,turn_key) + budget_bytes INTEGER NOT NULL, + outline_json TEXT NOT NULL, + PRIMARY KEY(session_id,epoch) ) WITHOUT ROWID; ` diff --git a/src/main/native-chat/agent-session-journal/database/journal-sql-database.ts b/src/main/native-chat/agent-session-journal/database/journal-sql-database.ts index f1cf842a8d3..6e9ef055f55 100644 --- a/src/main/native-chat/agent-session-journal/database/journal-sql-database.ts +++ b/src/main/native-chat/agent-session-journal/database/journal-sql-database.ts @@ -37,7 +37,16 @@ export class JournalSqlDatabase { this.readers = new JournalSqlReaderPool( journalDomainDatabasePath(stateDirectory), journalDatabasePath(stateDirectory), - options.readerPath + options.readerPath, + (cache) => { + if (!this.readOnly && !this.closed && !this.activity.isSessionActive(cache.sessionId)) { + void this.writer.maintain({ kind: 'outline-cache', cache }).catch((error: unknown) => { + if (!this.closed) { + options.onMaintenanceError(error, cache.sessionId) + } + }) + } + } ) this.conversions = new JournalSqlConversions(writer, this.readers, this.activity.waitForIdle) this.maintenance = new JournalSqlBackgroundMaintenance( diff --git a/src/main/native-chat/agent-session-journal/journal-domain-consumer-reads.test.ts b/src/main/native-chat/agent-session-journal/journal-domain-consumer-reads.test.ts index ad4e10e1cc4..e14e56b94bc 100644 --- a/src/main/native-chat/agent-session-journal/journal-domain-consumer-reads.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-domain-consumer-reads.test.ts @@ -1,3 +1,4 @@ +import { saveJournalDomainOutlineCache } from './journal-domain-outline' import { afterEach, expect, it } from 'vitest' import { domainLifecycleFixture, @@ -439,3 +440,87 @@ it('observes a consumed automation card through its first root turn and only tha error: null }) }) + +it('reuses a saved outline after reopen and refuses a cache result from an older source', async () => { + const f = domainLifecycleFixture(cleanup) + await f.run({ + kind: 'item', + identity: identity('prompt'), + body: body('Original prompt'), + options + }) + const read = () => { + const result = readJournalSqlPoint(f.db, { + sessionId: 'chat', + operation: { kind: 'conversation-outline', budgetBytes: 100_000 } + }) + if (result.kind !== 'conversation-outline') { + throw new Error('outline unavailable') + } + return result + } + const cold = read() + if (!cold.cache) { + throw new Error('cold outline unavailable') + } + expect(saveJournalDomainOutlineCache(f.db, cold.cache)).toBe(true) + const source = f.db + .prepare("SELECT outline_source_json FROM journal_items WHERE item_id='orca:prompt'") + .get()!.outline_source_json + f.db + .prepare( + "UPDATE journal_items SET outline_source_json='unreadable source' WHERE item_id='orca:prompt'" + ) + .run() + f.reopen() + expect(read().outline.entries).toEqual([ + { itemId: 'orca:prompt', sequence: 1, preview: 'Original prompt', imageCount: 0 } + ]) + f.db + .prepare("UPDATE journal_items SET outline_source_json=? WHERE item_id='orca:prompt'") + .run(source) + await f.run({ kind: 'item', identity: identity('later'), body: body('Changed prompt'), options }) + expect(saveJournalDomainOutlineCache(f.db, cold.cache)).toBe(false) + const changed = read() + expect(changed.outline.entries).toEqual([ + { itemId: 'orca:prompt', sequence: 1, preview: 'Original prompt', imageCount: 0 }, + { itemId: 'orca:later', sequence: 2, preview: 'Changed prompt', imageCount: 0 } + ]) + await f.run({ + kind: 'item', + identity: identity('turn'), + options, + body: { + kind: 'turn', + turnId: 'turn', + state: 'running', + startedAt: 10, + userItemId: 'orca:prompt' + } + }) + const running = read() + if (!running.cache) { + throw new Error('running outline unavailable') + } + expect(saveJournalDomainOutlineCache(f.db, running.cache)).toBe(false) + await f.run({ + kind: 'item', + identity: identity('turn'), + options, + body: { + kind: 'turn', + turnId: 'turn', + state: 'completed', + startedAt: 10, + completedAt: 20, + userItemId: 'orca:prompt' + } + }) + const idle = read() + if (!idle.cache) { + throw new Error('idle outline unavailable') + } + expect(saveJournalDomainOutlineCache(f.db, idle.cache)).toBe(true) + f.reopen() + expect(read().outline).toEqual(idle.outline) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-domain-import-discard.ts b/src/main/native-chat/agent-session-journal/journal-domain-import-discard.ts index 9cf67510acd..f53f766bf2d 100644 --- a/src/main/native-chat/agent-session-journal/journal-domain-import-discard.ts +++ b/src/main/native-chat/agent-session-journal/journal-domain-import-discard.ts @@ -16,8 +16,7 @@ export function discardJournalDomainImportPage( } let remaining = 512 for (const [table, key] of [ - ['journal_outline_members', 'epoch,item_id'], - ['journal_outline_replies', 'epoch,turn_key'], + ['journal_outline_cache', 'epoch'], ['journal_item_subagent_names', 'epoch,subagent_id,item_id'], ['journal_items', 'epoch,item_id'], ['journal_item_tombstones', 'epoch,item_id'], diff --git a/src/main/native-chat/agent-session-journal/journal-domain-import-rows.ts b/src/main/native-chat/agent-session-journal/journal-domain-import-rows.ts index 8af7d4690ae..30ad82cfc7b 100644 --- a/src/main/native-chat/agent-session-journal/journal-domain-import-rows.ts +++ b/src/main/native-chat/agent-session-journal/journal-domain-import-rows.ts @@ -8,7 +8,6 @@ import type Database from '../../sqlite/sync-database' import { insertJournalDomainQueueRow, journalDomainQueueRow } from './journal-domain-queue-rows' import type { JournalDomainImportHeader, JournalDomainImportPage } from './journal-domain-import' import { writeJournalDomainItemNames } from './journal-domain-item-names' -import { rebuildJournalDomainOutline } from './journal-domain-outline-write' export type JournalDomainImportState = | { state: 'converting'; epoch: null } @@ -253,6 +252,5 @@ export function finishJournalDomainImport( if (state.state !== 'ready') { throw new Error('journal domain import did not publish its chat head') } - rebuildJournalDomainOutline(db, header) return state.epoch } diff --git a/src/main/native-chat/agent-session-journal/journal-domain-maintenance.ts b/src/main/native-chat/agent-session-journal/journal-domain-maintenance.ts index 674a5299401..7a891e0be45 100644 --- a/src/main/native-chat/agent-session-journal/journal-domain-maintenance.ts +++ b/src/main/native-chat/agent-session-journal/journal-domain-maintenance.ts @@ -25,12 +25,7 @@ const RETIRED_TABLES: readonly { name: string; key: string; payload?: readonly s 'outline_source_json' ] }, - { - name: 'journal_outline_members', - key: 'item_id', - payload: ['structure_json', 'entry_json', 'short_json', 'empty_json', 'reply'] - }, - { name: 'journal_outline_replies', key: 'turn_key', payload: ['reply'] }, + { name: 'journal_outline_cache', key: '', payload: ['outline_json'] }, { name: 'journal_item_subagent_names', key: 'subagent_id,item_id' }, { name: 'journal_item_tombstones', key: 'item_id' }, { name: 'journal_removed_turns', key: 'turn_id' }, @@ -63,7 +58,7 @@ export function pruneJournalDomainRetiredPage( let deleted = 0 let bytes = 0 for (const { name, key, payload } of RETIRED_TABLES) { - const columns = ['session_id', 'epoch', ...key.split(',')] + const columns = ['session_id', 'epoch', ...(key ? key.split(',') : [])] const sizes = [...columns, ...(payload ?? [])] .map((column) => `COALESCE(octet_length(${column}),0)`) .join('+') diff --git a/src/main/native-chat/agent-session-journal/journal-domain-outline-source.ts b/src/main/native-chat/agent-session-journal/journal-domain-outline-source.ts index e90645a00df..bfb4a190d10 100644 --- a/src/main/native-chat/agent-session-journal/journal-domain-outline-source.ts +++ b/src/main/native-chat/agent-session-journal/journal-domain-outline-source.ts @@ -14,8 +14,6 @@ import { readAgentJournalTurn } from '../../../shared/agent-session/lifecycle/ag import { isAgentSessionContextClear } from '../../../shared/agent-session/agent-session-context-clear' import { projectStructuredItemToNativeChat } from '../../../shared/agent-session/view/structured-agent-session-projection' import { nativeChatToolFoldKind } from '../../../shared/native-chat/tools/native-chat-tool-fold' -import { isNoiseMessage } from '../../../shared/native-chat/native-chat-noise' -import { nativeChatRowRendersContent } from '../../../shared/native-chat/transcript/native-chat-row-content' import type { NativeChatBlock, NativeChatMessage @@ -100,26 +98,6 @@ export function journalDomainOutlineSource(item: AgentJournalRenderItem): string } satisfies JournalOutlineSource) } -/** Source position and membership only; body revisions and prose do not regroup a turn. */ -export function journalOutlineStructure(source: JournalOutlineSource): string { - const { revision: _revision, ...item } = source.item - const { blocks: _blocks, ...message } = source.message ?? { blocks: [] } - return JSON.stringify({ - item, - message, - fold: source.message && nativeChatToolFoldKind(source.message), - noise: source.message && isNoiseMessage(source.message), - renders: source.message && nativeChatRowRendersContent(source.message.blocks), - tools: source.message?.blocks.flatMap((block) => - block.type === 'tool-call' || block.type === 'tool-result' ? [[block.type, block.callId]] : [] - ) - }) -} - -export function journalOutlineMessage(source: JournalOutlineSource): NativeChatMessage | null { - return source.message -} - export function journalDomainOutlineSubmission(submission: AgentJournalSubmission): string { const { clientMessageId, diff --git a/src/main/native-chat/agent-session-journal/journal-domain-outline-write.ts b/src/main/native-chat/agent-session-journal/journal-domain-outline-write.ts deleted file mode 100644 index 28e73915646..00000000000 --- a/src/main/native-chat/agent-session-journal/journal-domain-outline-write.ts +++ /dev/null @@ -1,205 +0,0 @@ -import type Database from '../../sqlite/sync-database' -import type { SqliteRow } from '../../sqlite/sqlite-statement' -import { - projectAgentSessionConversationOutline, - truncateOutlinePreview, - type AgentSessionConversationOutlineEntry -} from '../../../shared/agent-session/agent-session-conversation-outline' -import type { AgentJournalSubmission } from '../../../shared/agent-session/journal/agent-session-journal-types' -import { - journalOutlineMessage, - journalOutlineStructure, - type JournalOutlineSource -} from './journal-domain-outline-source' - -type Target = { sessionId: string; epoch: string } -type SourceRow = { source: JournalOutlineSource; ordinal: number; structure: string } - -function writeMember( - db: Database.Database, - target: Target, - row: SourceRow, - turnKey: string | undefined, - entry: AgentSessionConversationOutlineEntry | undefined -): void { - const json = entry ? JSON.stringify(entry) : null - const short = entry - ? JSON.stringify({ ...entry, preview: truncateOutlinePreview(entry.preview, 60) }) - : null - const empty = entry ? JSON.stringify({ ...entry, preview: '' }) : null - db.prepare(`INSERT INTO journal_outline_members(session_id,epoch,item_id,source_revision, - sequence,sequence_index,ordinal,structure_json,turn_key,reply,entry_json,short_json,empty_json, - entry_bytes,short_bytes,empty_bytes) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) - ON CONFLICT(session_id,epoch,item_id) DO UPDATE SET - source_revision=excluded.source_revision,sequence=excluded.sequence,sequence_index=excluded.sequence_index, - ordinal=excluded.ordinal,structure_json=excluded.structure_json,turn_key=excluded.turn_key, - reply=excluded.reply,entry_json=excluded.entry_json,short_json=excluded.short_json,empty_json=excluded.empty_json, - entry_bytes=excluded.entry_bytes,short_bytes=excluded.short_bytes,empty_bytes=excluded.empty_bytes - WHERE source_revision<>excluded.source_revision OR structure_json<>excluded.structure_json - OR turn_key IS NOT excluded.turn_key OR reply IS NOT excluded.reply OR entry_json IS NOT excluded.entry_json`).run( - target.sessionId, - target.epoch, - row.source.item.itemId, - row.source.item.revision, - row.source.item.sequence, - row.source.item.sequenceIndex ?? 0, - row.ordinal, - row.structure, - turnKey ?? null, - row.source.reply, - json, - short, - empty, - json === null ? 0 : Buffer.byteLength(json, 'utf8') + 1, - short === null ? 0 : Buffer.byteLength(short, 'utf8') + 1, - empty === null ? 0 : Buffer.byteLength(empty, 'utf8') + 1 - ) -} - -function refreshReply(db: Database.Database, target: Target, key: string): void { - const prefix = [target.sessionId, target.epoch, key] as const - const row = db - .prepare(`SELECT reply FROM journal_outline_members - INDEXED BY journal_outline_reply_candidates WHERE session_id=? AND epoch=? AND turn_key=? AND reply IS NOT NULL - ORDER BY sequence DESC,sequence_index DESC,ordinal DESC LIMIT 1`) - .get(...prefix) - if (!row) { - db.prepare( - 'DELETE FROM journal_outline_replies WHERE session_id=? AND epoch=? AND turn_key=?' - ).run(...prefix) - return - } - const reply = String(row.reply) - db.prepare(`INSERT INTO journal_outline_replies(session_id,epoch,turn_key,reply,reply_bytes) VALUES(?,?,?,?,?) - ON CONFLICT(session_id,epoch,turn_key) DO UPDATE SET reply=excluded.reply,reply_bytes=excluded.reply_bytes - WHERE reply<>excluded.reply`).run( - ...prefix, - reply, - Buffer.byteLength(JSON.stringify({ reply }), 'utf8') - 1 - ) -} - -function sourceRow(row: SqliteRow): SourceRow { - const source: JournalOutlineSource = JSON.parse(String(row.outline_source_json)) - return { source, ordinal: Number(row.ordinal), structure: journalOutlineStructure(source) } -} - -/** Membership changes can refer backward through aliases/Stop; reproject compact facts, never originals. */ -export function rebuildJournalDomainOutline(db: Database.Database, target: Target): void { - const prefix = [target.sessionId, target.epoch] as const - const sources = db - .prepare(`SELECT ordinal,outline_source_json FROM journal_items - INDEXED BY journal_items_visible_position WHERE session_id=? AND epoch=? AND hidden=0 - AND outline_source_json IS NOT NULL ORDER BY sequence,sequence_index,ordinal`) - .all(...prefix) - .map(sourceRow) - const byId = new Map(sources.map((row) => [row.source.item.itemId, row])) - const submissions = db - .prepare(`SELECT outline_json FROM journal_submission_attempts - INDEXED BY journal_submission_ordinal WHERE session_id=? AND epoch=? AND current=1 ORDER BY ordinal`) - .all(...prefix) - .map((row): AgentJournalSubmission => JSON.parse(String(row.outline_json))) - const turnKeys = new Map() - const entries = projectAgentSessionConversationOutline( - sources.map((row) => row.source.item), - submissions, - (messages, keys) => { - messages.forEach((message, index) => { - const key = keys[index] - if (key !== undefined) { - turnKeys.set(message.id, key) - } - }) - return new Map() - }, - { - projectItems: (items) => - items.flatMap((item) => { - const source = byId.get(item.itemId)?.source - const message = source && journalOutlineMessage(source) - return message ? [message] : [] - }), - userPreview: (message) => byId.get(message.id)!.source.preview! - } - ) - const entriesById = new Map(entries.map((entry) => [entry.itemId, entry])) - const old = db - .prepare('SELECT item_id,turn_key FROM journal_outline_members WHERE session_id=? AND epoch=?') - .all(...prefix) - const affected = new Set() - const remove = db.prepare( - 'DELETE FROM journal_outline_members WHERE session_id=? AND epoch=? AND item_id=?' - ) - for (const row of old) { - if (row.turn_key !== null) { - affected.add(String(row.turn_key)) - } - if (!byId.has(String(row.item_id))) { - remove.run(...prefix, String(row.item_id)) - } - } - for (const source of sources) { - const key = turnKeys.get(source.source.item.itemId) - if (key !== undefined) { - affected.add(key) - } - writeMember(db, target, source, key, entriesById.get(source.source.item.itemId)) - } - for (const key of affected) { - refreshReply(db, target, key) - } -} - -/** Stable prose revisions replace one preview/candidate; only structural facts regroup membership. */ -export function refreshJournalDomainOutline( - db: Database.Database, - target: Target, - itemIds: readonly string[], - submissionsChanged: boolean -): void { - if (submissionsChanged) { - rebuildJournalDomainOutline(db, target) - return - } - const prefix = [target.sessionId, target.epoch] as const - const current = db.prepare(`SELECT ordinal,outline_source_json FROM journal_items - WHERE session_id=? AND epoch=? AND item_id=? AND hidden=0 AND outline_source_json IS NOT NULL`) - const member = db.prepare(`SELECT structure_json,turn_key,entry_json FROM journal_outline_members - WHERE session_id=? AND epoch=? AND item_id=?`) - const updates: { row: SourceRow; old: SqliteRow }[] = [] - for (const id of new Set(itemIds)) { - const saved = current.get(...prefix, id) - const old = member.get(...prefix, id) - if (!saved && !old) { - continue - } - const row = saved && sourceRow(saved) - if (!row || !old || row.structure !== old.structure_json) { - rebuildJournalDomainOutline(db, target) - return - } - updates.push({ row, old }) - } - const affected = new Set() - for (const { row, old } of updates) { - const key = old.turn_key === null ? undefined : String(old.turn_key) - const preview = row.source.preview - const entry = - old.entry_json === null || preview === null - ? undefined - : { - itemId: row.source.item.itemId, - sequence: row.source.item.sequence, - preview: preview.text, - imageCount: preview.imageCount, - ...(key === undefined || key === row.source.item.itemId ? {} : { turnKey: key }) - } - writeMember(db, target, row, key, entry) - if (key !== undefined) { - affected.add(key) - } - } - for (const key of affected) { - refreshReply(db, target, key) - } -} diff --git a/src/main/native-chat/agent-session-journal/journal-domain-outline.ts b/src/main/native-chat/agent-session-journal/journal-domain-outline.ts index 74b9075c9f4..742c90b5580 100644 --- a/src/main/native-chat/agent-session-journal/journal-domain-outline.ts +++ b/src/main/native-chat/agent-session-journal/journal-domain-outline.ts @@ -1,60 +1,126 @@ -import type { AgentSessionConversationOutlineEntry } from '../../../shared/agent-session/agent-session-conversation-outline' +import { + projectAgentSessionConversationOutline, + type AgentSessionConversationOutline +} from '../../../shared/agent-session/agent-session-conversation-outline' +import type { AgentJournalSubmission } from '../../../shared/agent-session/journal/agent-session-journal-types' +import type Database from '../../sqlite/sync-database' +import { boundJournalConversationOutline } from './journal-outline-budget' +import type { JournalOutlineSource } from './journal-domain-outline-source' import { journalDomainContextHead } from './journal-sql-continuation-context' import type { JournalSqlScope } from './journal-sql-scope' -/** Choose the global degradation before reading the admitted recent suffix. */ +export type JournalOutlineCache = { + sessionId: string + epoch: string + sequence: number + revision: number + budgetBytes: number + json: string +} + +/** Cold reads reconstruct compact facts; ordinary writes never regroup the chat. */ export function readJournalDomainOutline(scope: JournalSqlScope, budgetBytes: number) { const { db, prefix } = scope - const totals = db - .prepare(`SELECT COUNT(*) AS count, - COALESCE(SUM(member.entry_bytes + COALESCE(reply.reply_bytes,0)),0) AS with_reply, - COALESCE(SUM(member.entry_bytes),0) AS full, - COALESCE(SUM(member.short_bytes),0) AS short, - COALESCE(SUM(member.empty_bytes),0) AS empty - FROM journal_outline_members member INDEXED BY journal_outline_entries - LEFT JOIN journal_outline_replies reply ON reply.session_id=member.session_id - AND reply.epoch=member.epoch AND reply.turn_key=member.turn_key - WHERE member.session_id=? AND member.epoch=? AND member.entry_json IS NOT NULL`) + const head = db + .prepare( + 'SELECT last_sequence,submissions_revision FROM journal_chat_state WHERE session_id=? AND epoch=?' + ) .get(...prefix)! - const replies = Number(totals.with_reply) <= budgetBytes - const variant = - replies || Number(totals.full) <= budgetBytes - ? 'entry' - : Number(totals.short) <= budgetBytes - ? 'short' - : 'empty' - const json = variant === 'entry' ? 'entry_json' : `${variant}_json` - const bytes = variant === 'entry' ? 'entry_bytes' : `${variant}_bytes` - const entries: AgentSessionConversationOutlineEntry[] = [] - let total = 0 - for (const row of db - .prepare(`SELECT member.${json} AS entry_json, - member.${bytes} + ${replies ? 'COALESCE(reply.reply_bytes,0)' : '0'} AS bytes, - ${replies ? 'reply.reply' : 'NULL'} AS reply - FROM journal_outline_members member INDEXED BY journal_outline_entries - ${ - replies - ? `LEFT JOIN journal_outline_replies reply ON reply.session_id=member.session_id - AND reply.epoch=member.epoch AND reply.turn_key=member.turn_key` - : '' + const sequence = Number(head.last_sequence) + const revision = Number(head.submissions_revision) + const stored = db + .prepare(`SELECT outline_json FROM journal_outline_cache + WHERE session_id=? AND epoch=? AND source_sequence=? AND source_revision=? AND budget_bytes=?`) + .get(...prefix, sequence, revision, budgetBytes) + if (stored) { + try { + return { outline: scope.decode(stored.outline_json) } + } catch { + // A damaged cache can be reconstructed from the current compact source. } - WHERE member.session_id=? AND member.epoch=? AND member.entry_json IS NOT NULL - ORDER BY member.sequence DESC,member.sequence_index DESC,member.ordinal DESC`) - .iterate(...prefix)) { - const next = Number(row.bytes) - if (total + next > budgetBytes) { - break - } - total += next - scope.examined(row) - const entry = scope.decode(row.entry_json) - entries.push(row.reply === null ? entry : { ...entry, reply: String(row.reply) }) } - entries.reverse() - return { - sessionId: scope.sessionId, - cursor: journalDomainContextHead(db, scope), + const sources = db + .prepare(`SELECT outline_source_json FROM journal_items + INDEXED BY journal_items_visible_position WHERE session_id=? AND epoch=? AND hidden=0 + AND outline_source_json IS NOT NULL ORDER BY sequence,sequence_index,ordinal`) + .all(...prefix) + .map((row) => scope.decode(row.outline_source_json)) + const byId = new Map(sources.map((source) => [source.item.itemId, source])) + const submissions = db + .prepare(`SELECT outline_json FROM journal_submission_attempts + INDEXED BY journal_submission_ordinal WHERE session_id=? AND epoch=? AND current=1 ORDER BY ordinal`) + .all(...prefix) + .map((row) => scope.decode(row.outline_json)) + const entries = projectAgentSessionConversationOutline( + sources.map((source) => source.item), + submissions, + (messages, keys) => { + const replies = new Map() + for (let index = messages.length - 1; index >= 0; index--) { + const key = keys[index] + const reply = byId.get(messages[index].id)?.reply + if (key !== undefined && reply && !replies.has(key)) { + replies.set(key, reply) + } + } + return replies + }, + { + projectItems: (items) => + items.flatMap((item) => { + const message = byId.get(item.itemId)?.message + return message ? [message] : [] + }), + userPreview: (message) => byId.get(message.id)!.preview! + } + ) + const outline = boundJournalConversationOutline( + { sessionId: scope.sessionId, cursor: journalDomainContextHead(db, scope) }, entries, - omittedEntries: Number(totals.count) - entries.length + budgetBytes + ) + return { + outline, + cache: { + sessionId: scope.sessionId, + epoch: scope.epoch, + sequence, + revision, + budgetBytes, + json: JSON.stringify(outline) + } satisfies JournalOutlineCache } } + +/** Cache admission is optional and rechecks the reader's exact source inside the sole writer. */ +export function saveJournalDomainOutlineCache( + db: Database.Database, + cache: JournalOutlineCache +): boolean { + const prefix = [cache.sessionId, cache.epoch] as const + if ( + db + .prepare(`SELECT 1 FROM journal_items INDEXED BY journal_items_unsettled + WHERE session_id=? AND epoch=? AND hidden=0 + AND (lifecycle_state IN ('running','pending') OR live_work=1) LIMIT 1`) + .get(...prefix) || + db + .prepare(`SELECT 1 FROM journal_submission_attempts INDEXED BY journal_submission_sending + WHERE session_id=? AND epoch=? AND current=1 AND dispatch_state='pending' LIMIT 1`) + .get(...prefix) + ) { + return false + } + return ( + Number( + db + .prepare(`INSERT INTO journal_outline_cache + (session_id,epoch,source_sequence,source_revision,budget_bytes,outline_json) + SELECT session_id,epoch,last_sequence,submissions_revision,?,? FROM journal_chat_state + WHERE session_id=? AND epoch=? AND last_sequence=? AND submissions_revision=? + ON CONFLICT(session_id,epoch) DO UPDATE SET source_sequence=excluded.source_sequence, + source_revision=excluded.source_revision,budget_bytes=excluded.budget_bytes,outline_json=excluded.outline_json`) + .run(cache.budgetBytes, cache.json, ...prefix, cache.sequence, cache.revision).changes + ) > 0 + ) +} diff --git a/src/main/native-chat/agent-session-journal/journal-sql-activity.ts b/src/main/native-chat/agent-session-journal/journal-sql-activity.ts index ff8c4d0ca63..0ddb5af353c 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-activity.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-activity.ts @@ -13,6 +13,10 @@ export class JournalSqlActivity { } } + isSessionActive(sessionId: string): boolean { + return this.sessions.has(sessionId) + } + hold(): () => void { this.foreground++ let held = true diff --git a/src/main/native-chat/agent-session-journal/journal-sql-event-tail.ts b/src/main/native-chat/agent-session-journal/journal-sql-event-tail.ts index e436bcb124b..4045d60d9e7 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-event-tail.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-event-tail.ts @@ -2,7 +2,6 @@ import { appendJournalRequestEvent } from './journal-writer-request-events' import type Database from '../../sqlite/sync-database' import type { JournalDomainEventChanges } from './journal-domain-schema' import { invalidateJournalDomainQueue } from './journal-domain-queue-revision' -import { refreshJournalDomainOutline } from './journal-domain-outline-write' export const JOURNAL_EVENT_HISTORY_MS = 24 * 60 * 60 * 1000 const MAX_EVENT_SEQUENCES = 4096 @@ -18,12 +17,6 @@ export function appendJournalSqlEvent( activity?: { fence: number; observedAt: number } ): void { const prefix = [target.sessionId, target.epoch] as const - refreshJournalDomainOutline( - db, - target, - [...changes.items.map((item) => item.itemId), ...(changes.removedItemIds ?? [])], - changes.submissions.length > 0 - ) const json = JSON.stringify(changes) const oversized = Buffer.byteLength(json, 'utf8') > MAX_EVENT_BYTES if (changes.cards.length > 0 || changes.submissions.length > 0) { diff --git a/src/main/native-chat/agent-session-journal/journal-sql-maintenance.ts b/src/main/native-chat/agent-session-journal/journal-sql-maintenance.ts index fbf887daae1..4847596aa72 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-maintenance.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-maintenance.ts @@ -1,3 +1,4 @@ +import { saveJournalDomainOutlineCache, type JournalOutlineCache } from './journal-domain-outline' import type Database from '../../sqlite/sync-database' import type { JournalDomainImportHeader, JournalDomainImportPage } from './journal-domain-import' import { @@ -21,6 +22,7 @@ import { } from './journal-domain-import-discard' export type JournalSqlMaintenance = + | { kind: 'outline-cache'; cache: JournalOutlineCache } | { kind: 'projection-state'; sessionId: string } | { kind: 'import-discard'; sessionId: string } | { @@ -48,6 +50,8 @@ export function applyJournalSqlMaintenance( operation: JournalSqlMaintenance ): JournalSqlMaintenanceResult { switch (operation.kind) { + case 'outline-cache': + return saveJournalDomainOutlineCache(db, operation.cache) case 'projection-state': return journalDomainImportState(db, operation.sessionId) case 'import-discard': diff --git a/src/main/native-chat/agent-session-journal/journal-sql-point-read-types.ts b/src/main/native-chat/agent-session-journal/journal-sql-point-read-types.ts index e771f9c63ae..c58440892a0 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-point-read-types.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-point-read-types.ts @@ -1,3 +1,4 @@ +import type { JournalOutlineCache } from './journal-domain-outline' import type { JournalRequestFailureRead } from './journal-request-failure' import type { JournalRewindCut, JournalRewindSelection } from './journal-domain-rewind-cut' import type { AgentSessionConversationOutline } from '../../../shared/agent-session/agent-session-conversation-outline' @@ -86,7 +87,11 @@ export type JournalSqlPointReadResult = | { kind: 'rewind-cut'; cut: JournalRewindCut | null } | { kind: 'fork-boundary'; boundary: JournalForkBoundary } | { kind: 'saved-content'; content: JournalSavedContent } - | { kind: 'conversation-outline'; outline: AgentSessionConversationOutline } + | { + kind: 'conversation-outline' + outline: AgentSessionConversationOutline + cache?: JournalOutlineCache + } | JournalRequestFailureResult | { kind: 'rewind-target'; target: JournalRewindTarget } | { kind: 'history-source-items'; items: AgentJournalRenderItem[] } diff --git a/src/main/native-chat/agent-session-journal/journal-sql-point-read.ts b/src/main/native-chat/agent-session-journal/journal-sql-point-read.ts index 7718c21aee1..bbe6e846dbd 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-point-read.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-point-read.ts @@ -62,7 +62,7 @@ function read( case 'conversation-outline': return { kind: operation.kind, - outline: readJournalDomainOutline(scope, operation.budgetBytes) + ...readJournalDomainOutline(scope, operation.budgetBytes) } case 'request-failure': case 'running-request': diff --git a/src/main/native-chat/agent-session-journal/journal-sql-reader-pool.ts b/src/main/native-chat/agent-session-journal/journal-sql-reader-pool.ts index 54bdcce84ef..bf12cdd58d2 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-reader-pool.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-reader-pool.ts @@ -1,3 +1,4 @@ +import type { JournalOutlineCache } from './journal-domain-outline' import type { Worker } from 'node:worker_threads' import { spawnWorkerThread } from '../../worker-threads/worker-thread-spawn' import { @@ -34,7 +35,8 @@ export class JournalSqlReaderPool { private readonly workerPath = resolveWorkerThreadEntryPath( currentWorkerEntryLayout(__dirname), 'journal-sql-reader-entry.js' - ) + ), + private readonly cacheOutline?: (cache: JournalOutlineCache) => void ) {} async domainHistory(request: JournalDomainHistoryRequest): Promise { @@ -73,6 +75,11 @@ export class JournalSqlReaderPool { if (response.type !== 'point') { throw new Error('journal reader returned no point result') } + if (response.result.kind === 'conversation-outline' && response.result.cache) { + const { cache, ...result } = response.result + this.cacheOutline?.(cache) + return result + } return response.result }