From b98a2544bffe1ac617088fd677b2af8145c6d29c Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 12 Nov 2026 10:14:19 -0800 Subject: [PATCH] perf(native-chat): reuse narrow SQLite work reads Read canonical current work and queue eligibility in one snapshot, and reuse validated work through the existing journal observation entry. Preserve full status publication, live authority validation, optional queue failures and final writer claims. Frozen700 matches the reviewed 27-file source manifest and measured tree 6423b1cfbb45c6e8d7804174f8e4501a214b33b7. Cold and changed-head reads meet the one-point budget; unchanged publication frames improve. Disclose report698 strict-rule misses accepted by the coordinator as noise: the approximately 4us warm in-memory hit has paired p50 ratio 1.047; one writer-ack cell (empty / 2 subscribers / changed) is +5.5% at p50, 0.259 to 0.274 ms, while completion is approximately twice as fast. --- .../journal-domain-input-eligibility.ts | 90 ++++---- .../journal-domain-input-populations.ts | 4 +- .../journal-domain-queue-pauses.ts | 36 ++-- .../journal-domain-status-reads.test.ts | 105 +++++++++- .../journal-domain-submission-consume.ts | 21 +- .../journal-sql-accepted-inputs.ts | 9 + .../journal-sql-current-work.ts | 50 +++++ .../journal-sql-point-read.ts | 10 +- .../journal-sql-queue-read.ts | 51 ++++- .../journal-sql-status-context.ts | 192 ++++++++++++++---- .../journal-sql-status-head.ts | 30 +++ .../journal-sql-status-items.ts | 28 +++ .../journal-sql-store.test.ts | 82 +------- .../journal-sql-work-answer.ts | 20 +- .../journal-sql-work-items.ts | 79 +++++++ .../journal-sql-work-read.ts | 104 +++++----- .../journal-work-read.ts | 15 ++ ...ctured-agent-session-queued-publication.ts | 58 +++--- .../structured-agent-session-observed-work.ts | 4 +- ...agent-session-status-journal-projection.ts | 102 ++++++++-- .../structured-agent-session-status-reader.ts | 42 ++-- ...tured-agent-session-status-reentry.test.ts | 109 ++++++++++ .../agent-session-subscriber-frame-fields.ts | 3 +- ...d-agent-session-background-task-channel.ts | 5 +- ...tructured-agent-session-client-delivery.ts | 25 +-- .../structured-agent-session-subscribers.ts | 11 + ...agent-session-chat-wait-work-state.test.ts | 22 +- 27 files changed, 989 insertions(+), 318 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-sql-current-work.ts create mode 100644 src/main/native-chat/agent-session-journal/journal-sql-status-head.ts create mode 100644 src/main/native-chat/agent-session-journal/journal-sql-work-items.ts diff --git a/src/main/native-chat/agent-session-journal/journal-domain-input-eligibility.ts b/src/main/native-chat/agent-session-journal/journal-domain-input-eligibility.ts index 6cbaef782bd..63679a96c68 100644 --- a/src/main/native-chat/agent-session-journal/journal-domain-input-eligibility.ts +++ b/src/main/native-chat/agent-session-journal/journal-domain-input-eligibility.ts @@ -1,12 +1,8 @@ import { decodePersistedAgentSessionRecord } from '../../../shared/agent-session/lifecycle/agent-session-record-stored-form' import { isReadableAgentSessionStoreRecord } from '../../runtime/agent-session/records/agent-session-store-row-rules' -import type { JournalCurrentWorkAuthority } from './journal-observer-read' +import type { JournalCurrentWorkAuthority, JournalObserverResult } from './journal-observer-read' import { JournalSqlStatusContext } from './journal-sql-status-context' -import { readJournalSqlCurrentWork } from './journal-sql-work-read' -import { - journalDomainFirstQueuedInput, - journalDomainEnteredInputs -} from './journal-domain-input-populations' +import { readJournalSqlCurrentWork } from './journal-sql-current-work' import { nextJournalDomainQueueCard, type JournalDomainQueuePick @@ -17,47 +13,68 @@ import type { JournalSqlQueueFacts } from './journal-sql-queued-messages' /** Send, automatic claim and publication derive the same barriers from current SQL rows. */ export function journalDomainInputEligibility( scope: JournalSqlScope, - authority: JournalCurrentWorkAuthority + authority: JournalCurrentWorkAuthority, + context = new JournalSqlStatusContext(scope), + work = readJournalSqlCurrentWork(context, authority) ) { - const row = scope.db - .prepare('SELECT record_json FROM agent_session_records WHERE session_id=?') - .get(scope.sessionId) - const saved = row && scope.decode(row.record_json) - if (row && !isReadableAgentSessionStoreRecord(scope.sessionId, saved)) { - throw new Error('input admission requires a readable session record') - } - const record = isReadableAgentSessionStoreRecord(scope.sessionId, saved) - ? decodePersistedAgentSessionRecord(saved).record - : null - const blocked = - !record || record.rewind?.phase === 'prepared' || record.rewind?.phase === 'provider-succeeded' - const context = new JournalSqlStatusContext(scope) - const work = readJournalSqlCurrentWork(context, authority) + const blocked = journalDomainInputRecordBlocked(scope, context) const prompts = [...context.select({ kind: 'pending-prompts' })].flatMap(({ item }) => { const displayed = work.isCurrentItem(item.itemId) ? context.displayed(item.itemId) : null return displayed ? [displayed] : [] }) - const pending = journalDomainFirstQueuedInput(scope) - const entered = journalDomainEnteredInputs(scope) return { blocked, context, work, prompts, - pending, - handover: [...entered].some((send) => work.owesSend(send)), + pending: context.firstQueuedInput(), + handover: [...context.enteredSubmissions()].some((send) => work.owesSend(send)), activeTurnId: work.activeTurn()?.turnId } } +function journalDomainInputRecordBlocked(scope: JournalSqlScope, context: JournalSqlStatusContext) { + const row = + context.inputHead ?? + scope.db + .prepare('SELECT record_json FROM agent_session_records WHERE session_id=?') + .get(scope.sessionId) + const saved = row?.record_json == null ? null : scope.decode(row.record_json) + if (row?.record_json != null && !isReadableAgentSessionStoreRecord(scope.sessionId, saved)) { + throw new Error('input admission requires a readable session record') + } + const record = isReadableAgentSessionStoreRecord(scope.sessionId, saved) + ? decodePersistedAgentSessionRecord(saved).record + : null + return ( + !record || record.rewind?.phase === 'prepared' || record.rewind?.phase === 'provider-succeeded' + ) +} + +/** A validated work answer already owns prompt, handover and active-turn policy. */ +export function journalDomainResolvedInputEligibility( + context: JournalSqlStatusContext, + work: JournalObserverResult['work'] +) { + return { + blocked: journalDomainInputRecordBlocked(context.scope, context), + context, + prompts: work.actionablePrompts, + pending: context.firstQueuedInput(), + handover: work.owedEnteredSubmissions.length > 0, + activeTurnId: work.activeTurn?.turnId + } +} + export function selectJournalEligibleQueueCard( scope: JournalSqlScope, authority: JournalCurrentWorkAuthority, now: number, facts: JournalSqlQueueFacts, - allowSteer: boolean + allowSteer: boolean, + resolved?: Omit, 'work'> ): JournalDomainQueuePick & { steer: boolean; command: boolean } { - const state = journalDomainInputEligibility(scope, authority) + const state = resolved ?? journalDomainInputEligibility(scope, authority) const none = { kind: 'selected' as const, card: null, steer: false, command: false } const steer = state.activeTurnId !== undefined if ( @@ -70,13 +87,18 @@ export function selectJournalEligibleQueueCard( return none } return { - ...nextJournalDomainQueueCard(scope.db, scope, { - reopenFloor: facts.reopenFloor, - now, - steer, - automatic: true, - mail: facts.mail - }), + ...nextJournalDomainQueueCard( + scope.db, + scope, + { + reopenFloor: facts.reopenFloor, + now, + steer, + automatic: true, + mail: facts.mail + }, + state.context.inputHead + ), steer, command: state.activeTurnId?.startsWith('compact:') === true } diff --git a/src/main/native-chat/agent-session-journal/journal-domain-input-populations.ts b/src/main/native-chat/agent-session-journal/journal-domain-input-populations.ts index 8711a53c744..ccfc55484e7 100644 --- a/src/main/native-chat/agent-session-journal/journal-domain-input-populations.ts +++ b/src/main/native-chat/agent-session-journal/journal-domain-input-populations.ts @@ -27,14 +27,14 @@ export function* journalDomainUnsettledInputs(scope: JournalSqlScope, pendingOnl let after: [number, string] = [-1, ''] for (;;) { const rows = scope.db - .prepare(`SELECT message_id,ordinal FROM journal_submission_attempts + .prepare(`SELECT message_id,ordinal,payload_json FROM journal_submission_attempts WHERE session_id=? AND epoch=? AND current=1 AND dispatch_state IN ('pending','unknown') ${pendingOnly ? "AND dispatch_state='pending'" : ''} AND (ordinal,message_id)>(?,?) ORDER BY ordinal,message_id LIMIT ?`) .all(...scope.prefix, ...after, PAGE) for (const row of rows) { scope.examined(row) - const input = journalDomainSubmission(scope, String(row.message_id)) + const input = scope.decode(row.payload_json) if (input) { yield input } diff --git a/src/main/native-chat/agent-session-journal/journal-domain-queue-pauses.ts b/src/main/native-chat/agent-session-journal/journal-domain-queue-pauses.ts index e351e943504..55f568f6b68 100644 --- a/src/main/native-chat/agent-session-journal/journal-domain-queue-pauses.ts +++ b/src/main/native-chat/agent-session-journal/journal-domain-queue-pauses.ts @@ -5,7 +5,8 @@ import { createQueuePauseContext, type QueuePauseContext } from './queued-messages/queued-message-pause' -import { readJournalSqlAcceptedInputs } from './journal-sql-accepted-inputs' +import { JOURNAL_SQL_ACCEPTED_INPUT_COLUMNS } from './journal-sql-accepted-inputs' +import type { SqliteRow } from '../../sqlite/sqlite-statement' import { readJournalDomainQueueFailure } from './journal-domain-queue-failures' import { invalidateJournalDomainQueue } from './journal-domain-queue-revision' @@ -16,15 +17,19 @@ export function readJournalDomainQueuePauseContext( db: Database.Database, target: Target, reopenFloor: AgentJournalCursor | null, - messageIds: readonly string[] + messageIds: readonly string[], + snapshot?: { head: SqliteRow; clearMembers: ReadonlySet } ): QueuePauseContext { const prefix = [target.sessionId, target.epoch] as const - const head = db - .prepare(`SELECT queue_latest_stop_json, queue_manual_stop_json, + const head = + snapshot?.head ?? + db + .prepare(`SELECT queue_latest_stop_json, queue_manual_stop_json, queue_resumed_sequence, queue_reopened_sequence, queue_clear_sequence, - queue_clear_operation_id, queue_clear_lifted FROM journal_chat_state + queue_clear_operation_id, queue_clear_lifted, ${JOURNAL_SQL_ACCEPTED_INPUT_COLUMNS} + FROM journal_chat_state head WHERE session_id=? AND epoch=?`) - .get(...prefix) + .get(...prefix) if (!head) { throw new Error('journal queue pause targets a retired chat epoch') } @@ -47,15 +52,20 @@ export function readJournalDomainQueuePauseContext( } }) } - const clearMembers = new Set( - db - .prepare(`SELECT member.message_id FROM json_each(?) wanted + const clearMembers = + snapshot?.clearMembers ?? + new Set( + db + .prepare(`SELECT member.message_id FROM json_each(?) wanted CROSS JOIN journal_queue_clear_members member WHERE member.session_id=? AND member.epoch=? AND member.message_id=wanted.value`) - .all(JSON.stringify(messageIds), ...prefix) - .map((row) => String(row.message_id)) - ) - const accepted = readJournalSqlAcceptedInputs(db, prefix) + .all(JSON.stringify(messageIds), ...prefix) + .map((row) => String(row.message_id)) + ) + const accepted = { + latestAcceptedTurnSequence: Number(head.latest_accepted ?? 0), + latestPersonAcceptedSequence: Number(head.latest_person_accepted ?? 0) + } return createQueuePauseContext( { epoch: target.epoch, diff --git a/src/main/native-chat/agent-session-journal/journal-domain-status-reads.test.ts b/src/main/native-chat/agent-session-journal/journal-domain-status-reads.test.ts index 8896ff33879..c2ac11a3cad 100644 --- a/src/main/native-chat/agent-session-journal/journal-domain-status-reads.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-domain-status-reads.test.ts @@ -1,4 +1,5 @@ -import { afterEach, expect, it } from 'vitest' +import { readJournalSqlPoint } from './journal-sql-point-read' +import { afterEach, expect, it, vi } from 'vitest' import Database from '../../sqlite/sync-database' import type { AgentJournalMessageItem } from '../../../shared/agent-session/journal/agent-session-journal-types' import { agentSessionFailureWords } from '../../../shared/agent-session/failures/agent-session-failure-words' @@ -90,6 +91,108 @@ it('answers opening and owed phase membership in one work snapshot and honors a db.exec(`UPDATE journal_items SET hidden=1,turn_marker=1,context_clear=0,root=1, sequence=99 WHERE item_id='orca:send'`) expect(read(1)).toEqual({ kind: 'opening-send', opening: false }) + const sql = vi.spyOn(db, 'prepare') + const input = readJournalSqlPoint(db, { + sessionId: 'chat', + operation: { + kind: 'work', + input: { + authority, + operation: { kind: 'input', now: 100, facts: { reopenFloor: null }, allowSteer: false } + } + } + }) + expect(input).toMatchObject({ + kind: 'work', + answer: { result: { kind: 'input', state: { working: true }, queue: { card: null } } } + }) + const seeks = [ + ...new Set( + sql.mock.calls + .map(([query]) => query) + .filter( + (query) => + query.includes('raw_boundary AS') || + query.includes('AS work_floor') || + query.includes('payload_json FROM journal_submission_attempts') + ) + ) + ] + sql.mockRestore() + const plans = seeks.map((query) => + db + .prepare(`EXPLAIN QUERY PLAN ${query}`) + .all() + .map((row) => String(row.detail)) + .join('\n') + ) + expect(plans.join('\n')).toContain('journal_items_raw_turn_marker') + expect(plans.join('\n')).toContain('journal_items_turn_marker') + expect(plans.join('\n')).toContain('journal_items_status_pending') + expect(plans.join('\n')).toContain('journal_items_context_clear') + expect(plans.join('\n')).toContain('journal_items_raw_context_clear') + expect(plans.join('\n')).toContain('journal_submission_unsettled') + for (const plan of plans) { + expect(plan).not.toMatch(/SCAN (?:journal_items|item|source)\b|TEMP B-TREE/) + } + const resolved = readJournalSqlWork(context(), { authority, operation: { kind: 'state' } }) + const resolvedQueue = () => + readJournalSqlPoint(db, { + sessionId: 'chat', + operation: { + kind: 'queue', + operation: { + kind: 'eligible-input-card', + authority, + now: 100, + facts: { reopenFloor: null }, + allowSteer: false, + resolved + } + } + }) + expect(resolvedQueue()).toMatchObject({ + kind: 'queue', + result: { + kind: 'eligible-input-card', + card: null, + work: { result: { kind: 'input', state: { working: true }, queue: { card: null } } } + } + }) + db.prepare( + "INSERT INTO agent_session_records(session_id,record_json) VALUES ('chat','unreadable')" + ).run() + expect( + readJournalSqlPoint(db, { + sessionId: 'chat', + operation: { + kind: 'work', + input: { + authority, + operation: { kind: 'input', now: 100, facts: { reopenFloor: null }, allowSteer: false } + } + } + }) + ).toMatchObject({ + kind: 'work', + answer: { + result: { + state: { working: true }, + queue: { failure: { message: expect.stringContaining('JSON') } } + } + } + }) + expect(resolvedQueue()).toMatchObject({ + kind: 'queue', + result: { + work: { + result: { + state: { working: true }, + queue: { failure: { message: expect.stringContaining('JSON') } } + } + } + } + }) }) const cleanup: Database.Database[] = [] afterEach(() => { diff --git a/src/main/native-chat/agent-session-journal/journal-domain-submission-consume.ts b/src/main/native-chat/agent-session-journal/journal-domain-submission-consume.ts index 8cdf9d16d53..18e538cd10e 100644 --- a/src/main/native-chat/agent-session-journal/journal-domain-submission-consume.ts +++ b/src/main/native-chat/agent-session-journal/journal-domain-submission-consume.ts @@ -1,3 +1,4 @@ +import type { SqliteRow } from '../../sqlite/sqlite-statement' import type Database from '../../sqlite/sync-database' import { journalMailSenderKey, type JournalMailCustody } from './journal-writer-host-facts' import type { AgentMessageSource } from '../../../shared/agent-session/agent-session-message-source' @@ -40,7 +41,8 @@ export type JournalDomainQueuePick = export function nextJournalDomainQueueCard( db: Database.Database, target: JournalDomainSubmissionTarget, - selection: JournalDomainQueueSelection + selection: JournalDomainQueueSelection, + pauseHead?: SqliteRow ): JournalDomainQueuePick { const { steer, automatic } = selection let after: { position: number; messageId: string } | undefined @@ -52,6 +54,13 @@ export function nextJournalDomainQueueCard( WHERE lease.session_id=card.session_id AND lease.epoch=card.epoch AND lease.message_id=card.message_id AND lease.card_revision=card.revision AND lease.expires_at>?) AS edit_held + ${ + pauseHead + ? `,EXISTS(SELECT 1 FROM journal_queue_clear_members member + WHERE member.session_id=card.session_id AND member.epoch=card.epoch + AND member.message_id=card.message_id) AS clear_member` + : '' + } FROM journal_queue_cards AS card WHERE card.session_id=? AND card.epoch=? AND card.state IN ('waiting','returned') ${after ? 'AND (card.position,card.message_id)>(?,?)' : ''} @@ -69,7 +78,15 @@ export function nextJournalDomainQueueCard( db, target, selection.reopenFloor, - rows.map((row) => String(row.message_id)) + rows.map((row) => String(row.message_id)), + pauseHead + ? { + head: pauseHead, + clearMembers: new Set( + rows.filter((row) => row.clear_member).map((row) => String(row.message_id)) + ) + } + : undefined ) for (const row of rows) { if (automatic && row.edit_held) { diff --git a/src/main/native-chat/agent-session-journal/journal-sql-accepted-inputs.ts b/src/main/native-chat/agent-session-journal/journal-sql-accepted-inputs.ts index 719b9683def..52efcd38d7b 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-accepted-inputs.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-accepted-inputs.ts @@ -26,3 +26,12 @@ export function readJournalSqlAcceptedInputs( latestPersonAcceptedSequence: Number(person?.accepted_sequence ?? 0) } } + +export const JOURNAL_SQL_ACCEPTED_INPUT_COLUMNS = ` + (SELECT accepted_sequence FROM journal_submission_attempts + WHERE session_id=head.session_id AND epoch=head.epoch AND dispatch_state='accepted' + AND accepted_sequence IS NOT NULL ORDER BY accepted_sequence DESC LIMIT 1) AS latest_accepted, + (SELECT accepted_sequence FROM journal_submission_attempts + WHERE session_id=head.session_id AND epoch=head.epoch AND dispatch_state='accepted' + AND accepted_sequence IS NOT NULL AND admitted_personal=1 + ORDER BY accepted_sequence DESC LIMIT 1) AS latest_person_accepted` diff --git a/src/main/native-chat/agent-session-journal/journal-sql-current-work.ts b/src/main/native-chat/agent-session-journal/journal-sql-current-work.ts new file mode 100644 index 00000000000..1fe57ce1855 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-sql-current-work.ts @@ -0,0 +1,50 @@ +import { + StructuredAgentSessionWorkPolicy, + structuredAgentSessionLiveFence, + structuredAgentSessionOwnerUnverifiable +} from '../agent-session-wire/observations/structured-agent-session-work-policy' +import { journalWroteBeforeOpen } from './replay/journal-cursor' +import type { JournalSqlStatusContext } from './journal-sql-status-context' +import type { JournalCurrentWorkAuthority } from './journal-observer-read' + +export function readJournalSqlCurrentWork( + context: JournalSqlStatusContext, + authority: JournalCurrentWorkAuthority +) { + const handingOver = new Set(authority.handingOver) + const continuing = new Set(authority.continuingUnadmitted) + const finalizing = new Set(authority.finalizing) + const held = + authority.child?.heldSends === undefined ? undefined : new Set(authority.child.heldSends) + const evidence = { + record: authority.record, + ended: authority.ended, + replaced: authority.replaced, + child: + authority.child === null + ? null + : authority.child === undefined + ? undefined + : { fence: authority.child.fence } + } + return new StructuredAgentSessionWorkPolicy( + { + item: context.displayed, + itemFence: context.itemFence, + runningTurn: context.runningTurn, + recoverableSubmissions: context.recoverableSubmissions, + wroteBeforeOpen: (sequence) => + journalWroteBeforeOpen({ epoch: context.scope.epoch }, authority.openedThrough, sequence), + continuingInput: (id) => + continuing.has(id) && context.submission(id)?.dispatchState === 'pending', + finalizingInput: (id) => finalizing.has(id), + handingOver: (id) => handingOver.has(id) + }, + structuredAgentSessionLiveFence(evidence), + authority.operationalRevision, + authority.ended, + authority.replaced, + structuredAgentSessionOwnerUnverifiable(evidence), + () => held + ) +} 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 b70d07c540d..4fc364b5625 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 @@ -94,7 +94,15 @@ function read( case 'work': return { kind: 'work', - answer: readJournalSqlWork(new JournalSqlStatusContext(scope), operation.input) + answer: readJournalSqlWork( + new JournalSqlStatusContext( + scope, + operation.input.operation.kind === 'state' || operation.input.operation.kind === 'input' + ? operation.input.operation.kind + : undefined + ), + operation.input + ) } case 'observer': return { diff --git a/src/main/native-chat/agent-session-journal/journal-sql-queue-read.ts b/src/main/native-chat/agent-session-journal/journal-sql-queue-read.ts index b65ee8b07df..415f528764f 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-queue-read.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-queue-read.ts @@ -1,5 +1,13 @@ import type { JournalCurrentWorkAuthority } from './journal-observer-read' -import { selectJournalEligibleQueueCard } from './journal-domain-input-eligibility' +import { + selectJournalEligibleQueueCard, + journalDomainResolvedInputEligibility +} from './journal-domain-input-eligibility' +import { JournalSqlStatusContext } from './journal-sql-status-context' +import { readJournalSqlWork, readJournalSqlWorkInput } from './journal-sql-work-read' +import { sameJournalCursor } from './replay/journal-cursor' +import { copyJournalCurrentWorkAuthority } from './journal-observer-read' +import type { JournalSqlWorkResult, JournalWorkRead } from './journal-work-read' import type { JournalMailCustody } from './journal-writer-host-facts' import { readJournalDomainQueuePauseContext } from './journal-domain-queue-pauses' import type { AgentJournalCursor } from '../../../shared/agent-session/journal/agent-session-journal-identity' @@ -29,6 +37,7 @@ export type JournalSqlQueueRead = { facts: JournalSqlQueueFacts } & ( authority: JournalCurrentWorkAuthority allowSteer: boolean now: number + resolved?: JournalSqlWorkResult } | { kind: 'list' } | { kind: 'get'; messageId: string } @@ -42,7 +51,7 @@ export type JournalSqlQueueRead = { facts: JournalSqlQueueFacts } & ( ) export type JournalSqlQueueReadResult = - | { kind: 'eligible-input-card'; card: QueuedMessageRow | null } + | { kind: 'eligible-input-card'; card: QueuedMessageRow | null; work?: JournalSqlWorkResult } | { kind: 'list'; cards: QueuedMessageRow[] } | { kind: 'get'; card: QueuedMessageRow | null } | { kind: 'revision'; revision: string } @@ -64,6 +73,44 @@ export function readJournalSqlQueue( const { facts } = operation switch (operation.kind) { case 'eligible-input-card': { + if (operation.resolved) { + const context = new JournalSqlStatusContext(read, 'input') + const prior = operation.resolved + const cursor = { epoch: context.epoch, sequence: context.sequence } + const physicalCursor = { epoch: read.epoch, sequence: context.sequence } + const input: Extract = { + kind: 'input', + now: operation.now, + facts, + allowSteer: operation.allowSteer + } + const state = prior.result.kind === 'state' ? prior.result.state : undefined + const reusable = + state && + sameJournalCursor(prior.cursor, cursor) && + sameJournalCursor(prior.physicalCursor, physicalCursor) && + JSON.stringify(copyJournalCurrentWorkAuthority(prior.authority)) === + JSON.stringify(operation.authority) + const work = + reusable && state + ? { + cursor, + physicalCursor, + authority: operation.authority, + result: readJournalSqlWorkInput(context, operation.authority, input, state, () => + journalDomainResolvedInputEligibility(context, state) + ) + } + : readJournalSqlWork(context, { authority: operation.authority, operation: input }) + if (work.result.kind !== 'input') { + throw new Error('journal queue work answer changed kind') + } + return { + kind: operation.kind, + card: 'card' in work.result.queue ? work.result.queue.card : null, + work + } + } const selected = selectJournalEligibleQueueCard( read, operation.authority, diff --git a/src/main/native-chat/agent-session-journal/journal-sql-status-context.ts b/src/main/native-chat/agent-session-journal/journal-sql-status-context.ts index 671112da65e..4b0fc40ed1e 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-status-context.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-status-context.ts @@ -1,3 +1,14 @@ +import { readJournalSqlStatusHead } from './journal-sql-status-head' +import type { SqliteRow } from '../../sqlite/sqlite-statement' +import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session/requests/agent-session-queued-submission' +import { + readJournalSqlWorkItems, + hydrateJournalSqlWorkDependencies +} from './journal-sql-work-items' +import { + journalDomainFirstQueuedInput, + journalDomainUnsettledInputs +} from './journal-domain-input-populations' import type { AgentJournalRenderItem, AgentJournalSubmission @@ -8,14 +19,10 @@ import { journalContextEpoch } from './context-epochs/journal-context-epoch' import type { JournalSqlScope } from './journal-sql-scope' import { AGENT_SESSION_ITEM_WIRE_VERSION } from '../agent-session-wire/publication/history/agent-session-item-presentation-contract' import type { JournalStatusItem, JournalStatusItemSelection } from './journal-status-items' -import { sqlJournalStatusItems } from './journal-sql-status-items' +import { sqlJournalStatusItems, sqlJournalStatusItem } from './journal-sql-status-items' import { sqlFailedStatusItems, sqlHasStatusRequestItem } from './journal-sql-status-request-items' import { sqlStatusRelatedInputIds } from './journal-sql-status-inputs' -import { - readJournalDomainStopStates, - resolveJournalDomainDisplay, - journalDomainDisplayTurnState -} from './journal-domain-item-display' +import { readJournalDomainStopStates } from './journal-domain-item-display' import type { JournalDomainSubmissionStoredFacts } from './journal-domain-submission-records' import { journalDomainInputForItem, @@ -32,7 +39,12 @@ export class JournalSqlStatusContext { private readonly selected = new Map() private readonly items = new Map< string, - | { item: AgentJournalRenderItem; displayJson: unknown; hidden: boolean; fence: number } + | { + item: AgentJournalRenderItem + displayed: AgentJournalRenderItem + hidden: boolean + fence: number + } | undefined >() private readonly submissions = new Map() @@ -45,17 +57,26 @@ export class JournalSqlStatusContext { readonly submissionCount: number readonly stop: JournalLatestStop | null - constructor(readonly scope: JournalSqlScope) { - const row = scope.db - .prepare(`SELECT context_boundary,last_sequence,last_activity_at,queue_latest_stop_json, - (SELECT COALESCE(MAX(ordinal),-1)+1 FROM journal_submission_attempts - WHERE session_id=? AND epoch=?) AS submission_count - FROM journal_chat_state WHERE session_id=? AND epoch=?`) - .get(...scope.prefix, ...scope.prefix) + private unsettled: AgentJournalSubmission[] | undefined + private workFloor: number | undefined + + readonly inputHead: SqliteRow | undefined + private rawWorkFloor: number | undefined + + constructor( + readonly scope: JournalSqlScope, + work?: 'state' | 'input' + ) { + const row = readJournalSqlStatusHead(scope, work) if (!row) { throw new Error('journal status read targets a retired chat epoch') } scope.examined(row) + this.inputHead = work === 'input' ? row : undefined + if (work) { + this.workFloor = Number(row.work_floor ?? 0) + this.rawWorkFloor = Number(row.raw_work_floor ?? 0) + } this.epoch = journalContextEpoch(scope.epoch, Number(row.context_boundary)) this.sequence = Number(row.last_sequence) this.lastActivityAt = Number(row.last_activity_at) @@ -74,46 +95,122 @@ export class JournalSqlStatusContext { FROM journal_items WHERE session_id=? AND epoch=? AND item_id=?`) .get(...this.scope.prefix, id) if (row) { - this.scope.examined(row) - const source = this.scope.decode(row.payload_json) - const states = - row.wire_version === AGENT_SESSION_ITEM_WIRE_VERSION - ? undefined - : readJournalDomainStopStates(this.scope.db, this.scope, [id]) - const displayed = resolveJournalDomainDisplay( - source, - row.wire_version, - row.wire_json, - states && journalDomainDisplayTurnState(source, states) - ) - row.display_json = displayed === source ? row.payload_json : JSON.stringify(displayed) + this.remember(row, id) + } else { + this.items.set(id, undefined) } - - this.items.set( - id, - row - ? { - item: this.scope.decode(row.payload_json), - displayJson: row.display_json === row.payload_json ? null : row.display_json, - hidden: Boolean(row.hidden), - fence: Number(row.fence) - } - : undefined - ) } return this.items.get(id) } + private remember( + row: SqliteRow, + id: string, + states?: ReturnType + ) { + if (this.items.has(id)) { + return + } + this.items.set(id, sqlJournalStatusItem(this.scope, row, id, states)) + } + + private rememberAll(rows: SqliteRow[]) { + const stale = rows.filter( + (row) => row.payload_json != null && row.wire_version !== AGENT_SESSION_ITEM_WIRE_VERSION + ) + const states = stale.length + ? readJournalDomainStopStates( + this.scope.db, + this.scope, + stale.map((row) => String(row.item_id)) + ) + : undefined + for (const row of rows) { + const id = String(row.item_id) + if (row.payload_json === null) { + this.items.set(id, undefined) + } else { + this.remember(row, id, states) + } + } + } + + /** Work-only populations and their reciprocal turn/opener facts share this snapshot. */ + prepareWork() { + const selections = new Map([ + ['raw', []], + ['boundary', []], + ['pending-prompts', []] + ]) + const rows = readJournalSqlWorkItems(this.scope) + this.rememberAll(rows) + for (const row of rows) { + const id = String(row.item_id) + selections.get(String(row.selection))?.push({ + item: this.items.get(id)!.item, + position: { + sequence: Number(row.sequence), + sequenceIndex: Number(row.sequence_index), + ordinal: Number(row.ordinal) + } + }) + } + for (const [key, items] of selections) { + this.selected.set( + JSON.stringify(key === 'raw' ? { kind: 'boundary', raw: true } : { kind: key }), + items + ) + } + const sources = rows.flatMap((row) => { + const current = this.items.get(String(row.item_id)) + return current && !current.hidden ? [current.displayed] : [] + }) + hydrateJournalSqlWorkDependencies( + this.scope, + sources, + (id) => { + const current = this.items.get(id) + return current && !current.hidden + ? current.displayed + : this.items.has(id) + ? null + : undefined + }, + (rows) => this.rememberAll(rows) + ) + this.unsettled = [...journalDomainUnsettledInputs(this.scope)] + this.recoverable = this.unsettled.filter( + (send) => send.dispatchState === 'pending' || !send.recovered + ) + for (const send of this.unsettled) { + this.submissions.set(send.clientMessageId, send) + } + } + + commandFloor = () => + this.workFloor ?? + this.select({ kind: 'clear' })[Symbol.iterator]().next().value?.item.sequence ?? + 0 + + enteredSubmissions = () => + (this.unsettled ?? [...journalDomainUnsettledInputs(this.scope)]).filter( + (send) => !isQueuedAgentJournalSubmission(send) + ) + sendingSubmissions = () => + (this.unsettled ?? [...journalDomainUnsettledInputs(this.scope)]).filter( + (send) => send.dispatchState === 'pending' + ) + firstQueuedInput = () => + this.unsettled + ? this.sendingSubmissions().find(isQueuedAgentJournalSubmission) + : journalDomainFirstQueuedInput(this.scope) + item = (id: string) => this.current(id)?.item /** Semantic display uses the writer's saved Stop rewrite without shortening ordinary items. */ displayed = (id: string): AgentJournalRenderItem | null => { const row = this.current(id) - return row && !row.hidden - ? row.displayJson === null - ? row.item - : this.scope.decode(row.displayJson) - : null + return row && !row.hidden ? row.displayed : null } submission = (id: string) => { @@ -146,6 +243,10 @@ export class JournalSqlStatusContext { steerInput = (itemId: string) => journalDomainLatestSteer(this.scope, itemId) select = (selection: JournalStatusItemSelection): Iterable => { + const prepared = this.selected.get(JSON.stringify(selection)) + if (prepared) { + return prepared + } if (!['clear', 'boundary', 'prompt', 'assistant-exists'].includes(selection.kind)) { return sqlJournalStatusItems(this.scope, this.item, selection) } @@ -163,6 +264,9 @@ export class JournalSqlStatusContext { sqlStatusRelatedInputIds(this.scope, start, floor) contextFloor = (): number => { + if (this.rawWorkFloor !== undefined) { + return this.rawWorkFloor + } const row = this.scope.db .prepare(`SELECT sequence FROM journal_items WHERE session_id=? AND epoch=? AND context_clear=1 ORDER BY sequence DESC LIMIT 1`) diff --git a/src/main/native-chat/agent-session-journal/journal-sql-status-head.ts b/src/main/native-chat/agent-session-journal/journal-sql-status-head.ts new file mode 100644 index 00000000000..ca9310494b4 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-sql-status-head.ts @@ -0,0 +1,30 @@ +import { JOURNAL_SQL_ACCEPTED_INPUT_COLUMNS } from './journal-sql-accepted-inputs' +import type { JournalSqlScope } from './journal-sql-scope' + +export function readJournalSqlStatusHead(scope: JournalSqlScope, work?: 'state' | 'input') { + return scope.db + .prepare(`SELECT context_boundary,last_sequence,last_activity_at,queue_latest_stop_json, + (SELECT COALESCE(MAX(ordinal),-1)+1 FROM journal_submission_attempts + WHERE session_id=? AND epoch=?) AS submission_count + ${ + work + ? `,(SELECT sequence FROM journal_items INDEXED BY journal_items_context_clear + WHERE session_id=head.session_id AND epoch=head.epoch AND hidden=0 AND context_clear=1 + ORDER BY sequence DESC LIMIT 1) AS work_floor, + (SELECT sequence FROM journal_items INDEXED BY journal_items_raw_context_clear + WHERE session_id=head.session_id AND epoch=head.epoch AND context_clear=1 + ORDER BY sequence DESC LIMIT 1) AS raw_work_floor` + : '' + } + ${ + work === 'input' + ? `,(SELECT record_json FROM agent_session_records + WHERE session_id=head.session_id) AS record_json, + head.queue_manual_stop_json,head.queue_resumed_sequence,head.queue_reopened_sequence, + head.queue_clear_sequence,head.queue_clear_operation_id,head.queue_clear_lifted, + ${JOURNAL_SQL_ACCEPTED_INPUT_COLUMNS}` + : '' + } + FROM journal_chat_state head WHERE session_id=? AND epoch=?`) + .get(...scope.prefix, ...scope.prefix) +} diff --git a/src/main/native-chat/agent-session-journal/journal-sql-status-items.ts b/src/main/native-chat/agent-session-journal/journal-sql-status-items.ts index 4f4ea878a43..e202e0caec1 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-status-items.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-status-items.ts @@ -1,3 +1,10 @@ +import type { SqliteRow } from '../../sqlite/sqlite-statement' +import { AGENT_SESSION_ITEM_WIRE_VERSION } from '../agent-session-wire/publication/history/agent-session-item-presentation-contract' +import { + readJournalDomainStopStates, + resolveJournalDomainDisplay, + journalDomainDisplayTurnState +} from './journal-domain-item-display' import type { AgentJournalRenderItem } from '../../../shared/agent-session/journal/agent-session-journal-types' import type { JournalSqlScope } from './journal-sql-scope' import type { JournalStatusItem, JournalStatusItemSelection } from './journal-status-items' @@ -114,3 +121,24 @@ export function* sqlJournalStatusItems( after = columns.map((column) => Number(last[column])) } } + +export function sqlJournalStatusItem( + scope: JournalSqlScope, + row: SqliteRow, + id: string, + states?: ReturnType +) { + scope.examined(row) + const item = scope.decode(row.payload_json) + const stopStates = + row.wire_version === AGENT_SESSION_ITEM_WIRE_VERSION + ? undefined + : (states ?? readJournalDomainStopStates(scope.db, scope, [id])) + const displayed = resolveJournalDomainDisplay( + item, + row.wire_version, + row.wire_json, + stopStates && journalDomainDisplayTurnState(item, stopStates) + ) + return { item, displayed, hidden: Boolean(row.hidden), fence: Number(row.fence) } +} diff --git a/src/main/native-chat/agent-session-journal/journal-sql-store.test.ts b/src/main/native-chat/agent-session-journal/journal-sql-store.test.ts index 771440c87a8..63dd47b1494 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-store.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-store.test.ts @@ -1,10 +1,9 @@ import { AgentSessionSubscribers } from '../agent-session-wire/publication/subscribers/structured-agent-session-subscribers' import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session/agent-session-wire' import Database from '../../sqlite/sync-database' -import { StructuredAgentSessionJournalProjections } from '../agent-session-wire/observations/structured-agent-session-status-journal-projection' import { StructuredAgentSessionStatusFeed } from '../agent-session-wire/observations/structured-agent-session-status-feed' import type { StatusFeedSession } from '../agent-session-wire/observations/structured-agent-session-status-feed-deps' -import type { JournalWorkAuthority, JournalCurrentWorkAuthority } from './journal-observer-read' +import type { JournalCurrentWorkAuthority } from './journal-observer-read' import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' @@ -300,85 +299,6 @@ it('preserves a corrupt database on worker startup refusal', async () => { expect(await readFile(path)).toEqual(bytes) }) -it('discards row-free authority changes before publishing an asynchronous observer answer', async () => { - const { journal } = await fixture() - const projections = new StructuredAgentSessionJournalProjections() - let revision = 0 - const held: string[] = [] - const capture = () => { - const authority: JournalWorkAuthority = { - handle: journal.observationHandle(), - record: { - currentFence: 1, - ownerObservation: { claimStatus: 'live', unreconciled: false, handoffStage: null } - }, - child: { fence: 1, heldSends: held }, - handingOver: [], - continuingUnadmitted: [], - finalizing: [], - openedThrough: journal.openingCursor(), - reopenFloor: null, - operationalRevision: revision, - stopRevision: 0, - stop: null - } - return { authority, contextId: null } - } - const reached = Promise.withResolvers() - const release = Promise.withResolvers() - const read = journal.reads.observer.bind(journal.reads) - const observer = vi.spyOn(journal.reads, 'observer').mockImplementation(async (input) => { - const result = await read(input) - reached.resolve() - await release.promise - return result - }) - const invalidated: string[] = [] - const observed: unknown[] = [] - const pending = Promise.all( - ['first', 'second'].map((reader) => - projections.readSql( - journal, - capture, - () => invalidated.push(reader), - (projection) => observed.push(projection) - ) - ) - ) - await reached.promise - revision = 1 - held.push('new-live-owner') - release.resolve() - expect(await pending).toEqual([null, null]) - expect(invalidated).toEqual(['first', 'second']) - expect(observed).toEqual([]) - observer.mockRestore() - const current = await projections.readSql(journal, capture, () => {}) - expect(current).toMatchObject({ - workRevision: 1, - authority: { child: { heldSends: ['new-live-owner'] } } - }) - held.push('later-owner') - expect(current?.authority?.child?.heldSends).toEqual(['new-live-owner']) - expect(await projections.readSql(journal, capture, () => {})).toMatchObject({ - authority: { child: { heldSends: ['later-owner', 'new-live-owner'] } } - }) - revision = 2 - const failed = Promise.withResolvers>>() - const failure = new Error('observer read failed') - vi.spyOn(journal.reads, 'observer').mockImplementationOnce(() => failed.promise) - const failures = Promise.allSettled([ - projections.readSql(journal, capture, () => {}), - projections.readSql(journal, capture, () => {}) - ]) - failed.reject(failure) - expect(await failures).toEqual([ - { status: 'rejected', reason: failure }, - { status: 'rejected', reason: failure } - ]) - expect(await projections.readSql(journal, capture, () => {})).toMatchObject({ workRevision: 2 }) -}) - it('rechecks row-free owner death before answering an attempt-end action read', async () => { const { journal } = await fixture() await journal.appendItem( diff --git a/src/main/native-chat/agent-session-journal/journal-sql-work-answer.ts b/src/main/native-chat/agent-session-journal/journal-sql-work-answer.ts index 4a086c1c27c..57ea8aa7308 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-work-answer.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-work-answer.ts @@ -43,7 +43,7 @@ export async function readJournalSqlWorkAnswer JournalCurrentWorkAuthority, operation: Extract ): Promise> { - const requested = structuredClone(operation) + const requested: JournalWorkRead = structuredClone(operation) for (;;) { if (journal.isClosed) { throw new AgentSessionJournalError( @@ -52,7 +52,23 @@ export async function readJournalSqlWorkAnswer