mirror of
https://github.com/stablyai/orca.git
synced 2026-11-15 10:27:33 +00:00
Cache bounded conversation outlines after reader projection
This commit is contained in:
+5
-27
@@ -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;
|
||||
`
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
|
||||
@@ -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'],
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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('+')
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<string, string>()
|
||||
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<string>()
|
||||
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<string>()
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -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<AgentSessionConversationOutline>(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<AgentSessionConversationOutlineEntry>(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<JournalOutlineSource>(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<AgentJournalSubmission>(row.outline_json))
|
||||
const entries = projectAgentSessionConversationOutline(
|
||||
sources.map((source) => source.item),
|
||||
submissions,
|
||||
(messages, keys) => {
|
||||
const replies = new Map<string, string>()
|
||||
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
|
||||
)
|
||||
}
|
||||
|
||||
@@ -13,6 +13,10 @@ export class JournalSqlActivity {
|
||||
}
|
||||
}
|
||||
|
||||
isSessionActive(sessionId: string): boolean {
|
||||
return this.sessions.has(sessionId)
|
||||
}
|
||||
|
||||
hold(): () => void {
|
||||
this.foreground++
|
||||
let held = true
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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':
|
||||
|
||||
@@ -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[] }
|
||||
|
||||
@@ -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':
|
||||
|
||||
@@ -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<AgentSessionHistoryResult> {
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user