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