diff --git a/src/main/native-chat/agent-session-journal/journal-lifecycle-reads.ts b/src/main/native-chat/agent-session-journal/journal-lifecycle-reads.ts index 8cebc00330a..e3a5d93d62c 100644 --- a/src/main/native-chat/agent-session-journal/journal-lifecycle-reads.ts +++ b/src/main/native-chat/agent-session-journal/journal-lifecycle-reads.ts @@ -1,21 +1,7 @@ -import { readWholeAgentSessionFailureFact } from '../../../shared/agent-session/failures/agent-session-failure' -import { isRootAgentJournalItem } from '../../../shared/agent-session/journal/agent-session-journal-producer' -import type { AgentJournalCursor } from '../../../shared/agent-session/journal/agent-session-journal-identity' -import { agentJournalSubmissionKey } from '../../../shared/agent-session/journal/agent-session-journal-item-key' import type { - AgentJournalItemBody, AgentJournalMessageItem, - AgentJournalRenderItem, AgentJournalSubmission } from '../../../shared/agent-session/journal/agent-session-journal-types' -import { latestStructuredAgentSessionPrompt } from '../../../shared/agent-session/requests/structured-agent-session-latest-request' -import { - readJournalInput, - type JournalInputRead, - type JournalInputReadResult -} from './journal-input-reads' -import type { JournalItemLinkageVisitor } from './journal-store-contracts' -import type { JournalReducerState } from './projection/journal-reducer' /** The live epoch with the sends accepted after an offer's cursor, read together. */ export type JournalRestartProgress = { @@ -30,92 +16,3 @@ export type JournalOutgrownChild = { newestTurnCompletedAt: number | undefined nextMessage: AgentJournalMessageItem | null } - -/** One chat snapshot as both backends expose it to these lifecycle reads. */ -export type JournalLifecycleSnapshot = { - epoch: () => string - input: (operation: JournalInputRead) => JournalInputReadResult - /** Presented item body, as `reads.itemBody` answers it. */ - itemBody: (itemId: string) => AgentJournalItemBody | null | undefined - /** Whether a presented root status at this execution fence reports sign-out. */ - rootStatusSignedOutAt: (fence: number) => boolean -} - -/** A root status row reporting sign-out; a subagent's says nothing about the parent. */ -export function journalStatusReportsSignedOut( - body: AgentJournalItemBody, - attribution: Pick -): boolean { - return ( - body.kind === 'status' && - isRootAgentJournalItem(attribution) && - readWholeAgentSessionFailureFact(body.failure)?.kind === 'notSignedIn' - ) -} - -export function readJournalRestartProgress( - snapshot: JournalLifecycleSnapshot, - cursor: AgentJournalCursor -): JournalRestartProgress { - const accepted = snapshot.input({ kind: 'accepted-inputs-after', sequence: cursor.sequence }) - if (accepted.kind !== 'accepted-inputs-after') { - throw new Error('journal accepted inputs read returned a different answer') - } - return { epoch: snapshot.epoch(), acceptedInputs: accepted.submissions } -} - -export function readJournalOutgrownChild( - snapshot: JournalLifecycleSnapshot, - { fence, nextClientMessageId }: JournalOutgrownChildRead -): JournalOutgrownChild { - const answer = snapshot.input({ - kind: 'batch', - reads: [{ kind: 'signed-out-at', fence }, { kind: 'newest-turn' }] - }) - const [rejected, newest] = answer.kind === 'batch' ? answer.results : [] - if (rejected?.kind !== 'signed-out-at' || newest?.kind !== 'newest-turn') { - throw new Error('journal outgrown child read returned a different answer') - } - const body = - nextClientMessageId === undefined - ? null - : snapshot.itemBody(agentJournalSubmissionKey(nextClientMessageId)) - return { - signedOut: rejected.answer || snapshot.rootStatusSignedOutAt(fence), - newestTurnCompletedAt: newest.turn?.completedAt, - nextMessage: body?.kind === 'message' ? body : null - } -} - -type FoldedLifecycleJournal = { - cursor(): AgentJournalCursor - itemsRead(): { items: readonly AgentJournalRenderItem[] } - itemBody(itemId: string): AgentJournalItemBody | null - itemFence(itemId: string): number | undefined - visitItemsWithLinkage(visit: JournalItemLinkageVisitor): void -} - -/** The in-memory chat answers from its fold at call time. */ -export function foldedJournalLifecycleReads( - journal: FoldedLifecycleJournal, - state: () => JournalReducerState -) { - const snapshot: JournalLifecycleSnapshot = { - epoch: () => journal.cursor().epoch, - input: (operation) => readJournalInput(state(), operation), - itemBody: journal.itemBody, - rootStatusSignedOutAt: (fence) => { - let found = false - journal.visitItemsWithLinkage((itemId, _sequence, body, attribution) => { - found ||= - journalStatusReportsSignedOut(body, attribution) && journal.itemFence(itemId) === fence - }) - return found - } - } - return { - latestPrompt: () => latestStructuredAgentSessionPrompt(journal.itemsRead().items), - restartProgress: (cursor: AgentJournalCursor) => readJournalRestartProgress(snapshot, cursor), - outgrownChild: (input: JournalOutgrownChildRead) => readJournalOutgrownChild(snapshot, input) - } -} diff --git a/src/main/native-chat/agent-session-journal/journal-reads.ts b/src/main/native-chat/agent-session-journal/journal-reads.ts index 9197d6f5a7d..d3e51d37950 100644 --- a/src/main/native-chat/agent-session-journal/journal-reads.ts +++ b/src/main/native-chat/agent-session-journal/journal-reads.ts @@ -4,46 +4,33 @@ import type { AgentJournalMessageItem, AgentJournalRenderItem, AgentJournalSubmission, - AgentJournalSnapshot, AgentJournalTurnLifecycle } from '../../../shared/agent-session/journal/agent-session-journal-types' import type { AgentJournalCursor } from '../../../shared/agent-session/journal/agent-session-journal-identity' import type { AgentSessionContextUsage } from '../../../shared/agent-session/agent-session-context-usage' import type { JournalItemLinkageVisitor } from './journal-store-contracts' -import { - readJournalHistoryItems, - readClaudeHistoryPrompts, - type JournalHistoryItem, - type JournalHistoryReadOptions, - type JournalHistoryPrompt +import type { + JournalHistoryItem, + JournalHistoryReadOptions, + JournalHistoryPrompt } from './journal-history-items' import type { JournalReducerState } from './projection/journal-reducer' import type { JournalItemEcho } from './projection/journal-item-echo' import type { JournalThreadGoalSource } from './projection/journal-thread-goal-source' -import { - readClaudeJournalSnapshot, - type ClaudeJournalSnapshot -} from '../../claude/claude-subagent-journaled-identities' -import { readJournalInput, type JournalInputReader } from './journal-input-reads' -import { readJournalUsageLimitCut, readJournalOrcaStopCut } from './journal-status-cuts' +import type { ClaudeJournalSnapshot } from '../../claude/claude-subagent-journaled-identities' +import type { JournalInputReader } from './journal-input-reads' import type { NativeChatUsageLimitCut } from '../../../shared/native-chat/native-chat-usage-limit-cut' -import { readJournalStatusState, type JournalStatusState } from './journal-status-state' +import type { JournalStatusState } from './journal-status-state' import type { NativeChatOrcaStopCut } from '../../../shared/native-chat/native-chat-orca-stop-cut' import type { AgentJournalTurnScope } from '../../../shared/agent-session/journal/agent-session-journal-turn-scope' -import { memoryJournalMap } from './journal-memory-collections' -import { journalInputPresentationMember } from './input-order/journal-input-obligations' -import { agentJournalSubmissionKey } from '../../../shared/agent-session/journal/agent-session-journal-item-key' -import { agentSessionCurrentContextRows } from '../../../shared/agent-session/agent-session-context-clear' -import { - readJournalStatusIndicators, - type JournalStatusIndicatorRead, - type JournalStatusIndicators +import type { + JournalStatusIndicatorRead, + JournalStatusIndicators } from './journal-status-indicators' -import { - foldedJournalLifecycleReads, - type JournalOutgrownChild, - type JournalOutgrownChildRead, - type JournalRestartProgress +import type { + JournalOutgrownChild, + JournalOutgrownChildRead, + JournalRestartProgress } from './journal-lifecycle-reads' /** Which items a visit decodes; a database serves only the named kinds, in sequence order. */ @@ -71,7 +58,6 @@ export type AgentSessionJournalReads = { input: JournalInputReader usageLimitCut(): Promise orcaStopCut(): Promise - // Builder B: context metadata used by options reports. contextFloor(): Promise cursor(): Promise physicalCursor(): Promise @@ -118,181 +104,3 @@ export type AgentSessionJournalReads = { contextUsage(): Promise visitItems(filter: JournalItemFilter, visit: JournalItemLinkageVisitor): Promise } - -/** The synchronous readers today's in-memory journal already answers. */ -type FoldedJournalReads = { - [ - K in Exclude< - keyof AgentSessionJournalReads, - | 'visitItems' - | 'hasItems' - | 'canonicalItemIds' - | 'historyItems' - | 'claudeHistoryPrompts' - | 'namedTurn' - | 'messageOptions' - | 'continuationContext' - | 'latestPrompt' - | 'restartProgress' - | 'outgrownChild' - | 'contextFloor' - | 'input' - | 'usageLimitCut' - | 'orcaStopCut' - | 'statusState' - | 'statusIndicators' - | 'claudeSnapshot' - > - ]: ( - ...args: Parameters - ) => Awaited> -} & { - snapshot(): AgentJournalSnapshot - itemsRead(): { items: readonly AgentJournalRenderItem[] } - visitItemsWithLinkage(visit: JournalItemLinkageVisitor): void - // Builder B: context metadata stays fold-backed until the database reader replaces it. - context: { floor(): AgentJournalCursor | null } -} - -/** Answers at call time, so an awaited answer reflects writes admitted before the call. */ -export function foldedJournalReads( - journal: FoldedJournalReads, - state: () => JournalReducerState -): AgentSessionJournalReads { - const now = - (read: (...args: A) => T) => - (...args: A): Promise => { - try { - return Promise.resolve(read(...args)) - } catch (error) { - return Promise.reject(error) - } - } - const lifecycle = foldedJournalLifecycleReads(journal, state) - return { - statusIndicators: now((input: JournalStatusIndicatorRead) => - readJournalStatusIndicators(state(), input) - ), - statusState: now((fence?: number | null) => readJournalStatusState(state(), fence)), - input: async (operation) => readJournalInput(state(), operation), - usageLimitCut: now(() => readJournalUsageLimitCut(state())), - orcaStopCut: now(() => readJournalOrcaStopCut(state())), - // Builder B: read through the context controller without retaining a copy. - contextFloor: now(() => journal.context.floor()), - cursor: now(journal.cursor), - physicalCursor: now(journal.physicalCursor), - lastActivityAt: now(journal.lastActivityAt), - historyOrigin: now(journal.historyOrigin), - item: now(journal.item), - itemBody: now(journal.itemBody), - itemFence: now(journal.itemFence), - itemProvenance: now(journal.itemProvenance), - hasRecordedItem: now(journal.hasRecordedItem), - hasItems: now(() => { - const current = state() - for (const item of memoryJournalMap(current.items).values()) { - if (journalInputPresentationMember(current, item.itemId)) { - return true - } - } - return false - }), - presentationRemoved: now(journal.presentationRemoved), - canonicalItemId: now(journal.canonicalItemId), - canonicalItemIds: now((itemIds: readonly string[]) => - itemIds.map((itemId) => ({ itemId, canonicalItemId: journal.canonicalItemId(itemId) })) - ), - historyItems: now((itemIds: readonly string[], options?: JournalHistoryReadOptions) => { - if (!itemIds.length) { - return [] - } - const turns = new Map() - if (options?.includeOpeners) { - const wanted = new Set(itemIds.map(journal.canonicalItemId)) - journal.visitItemsWithLinkage((key, _sequence, body) => { - if (body.kind !== 'turn' || !body.userItemId) { - return - } - const user = journal.canonicalItemId(body.userItemId) - if (!wanted.has(user)) { - return - } - const held = turns.get(user) ?? [] - held.push(key) - turns.set(user, held) - }) - } - return readJournalHistoryItems( - { - canonicalItemId: journal.canonicalItemId, - item: journal.item, - presentationRemoved: journal.presentationRemoved, - ...(options?.includeOpeners - ? { turnsOpenedBy: (key: string) => turns.get(key) ?? [] } - : {}) - }, - itemIds - ) - }), - claudeHistoryPrompts: now((sessionId: string, uuids: readonly string[]) => { - if (!uuids.length) { - return [] - } - const users: string[] = [] - journal.visitItemsWithLinkage((key, _sequence, body) => { - if (body.kind === 'message' && body.role === 'user') { - users.push(key) - } - }) - return readClaudeHistoryPrompts(users, sessionId, uuids) - }), - submission: now(journal.submission), - submissionBody: now(journal.submissionBody), - receiptFor: now(journal.receiptFor), - pendingSubmissions: now(journal.pendingSubmissions), - lastProviderActivityAt: now(journal.lastProviderActivityAt), - runningTurn: now(journal.runningTurn), - newestTurn: now(journal.newestTurn), - namedTurn: now((turnId: string) => { - const current = state() - const item = current.queries.latestItemForTurn(current, turnId) - return { - scope: item ? { kind: 'turn' as const, turnItemId: item.itemId } : null, - removed: current.removedTurnIds.has(turnId) - } - }), - messageOptions: now((clientMessageId: string) => { - const body = journal.itemBody(agentJournalSubmissionKey(clientMessageId)) - return { - ordinaryMessage: body?.kind === 'message' && !body.command, - runningTurn: journal.newestTurn()?.state === 'running' - } - }), - continuationContext: now(() => { - const snapshot = journal.snapshot() - const { items, submissions } = agentSessionCurrentContextRows( - snapshot.items, - snapshot.submissions - ) - return { - cursor: snapshot.cursor, - items, - submissions - } - }), - latestPrompt: now(lifecycle.latestPrompt), - restartProgress: now(lifecycle.restartProgress), - outgrownChild: now(lifecycle.outgrownChild), - threadGoalSource: now(journal.threadGoalSource), - claudeSnapshot: now(() => readClaudeJournalSnapshot(journal)), - contextUsage: now(journal.contextUsage), - visitItems: now((filter: JournalItemFilter, visit: JournalItemLinkageVisitor) => { - const kinds = new Set(filter.kinds) - journal.visitItemsWithLinkage((itemId, sequence, body, attribution, item) => { - if (kinds.has(body.kind)) { - visit(itemId, sequence, body, attribution, item) - } - }) - }) - } -} diff --git a/src/main/native-chat/agent-session-journal/journal-sql-message-commit.ts b/src/main/native-chat/agent-session-journal/journal-sql-message-commit.ts index 7c09b0026ba..b17589687d0 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-message-commit.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-message-commit.ts @@ -1,5 +1,4 @@ import type Database from '../../sqlite/sync-database' -import type { JournalSqlBatch, JournalSqlBatchResult } from './journal-sql-batch' import { applyJournalDomainItem, type JournalDomainItemWrite, @@ -19,41 +18,12 @@ type MessageCommand = { savedAt: number } -export type JournalSqlMessageJob = { - command: MessageCommand - bytes: number -} - export type JournalSqlMessagesCommit = Omit & { publicCursor: AgentJournalCursor outcomes: JournalStepOutcome[] failure?: JournalSqlFailure } -/** Sender work; the writer drains the already classified and sized command. */ -export function prepareJournalSqlMessages(command: MessageCommand): JournalSqlMessageJob { - return { - command, - bytes: Buffer.byteLength(JSON.stringify(command), 'utf8') - } -} - -/** Called by the existing writer admission lane; publication completes before the next drain. */ -export async function commitJournalSqlMessages( - db: Database.Database, - batch: JournalSqlBatch, - jobs: readonly JournalSqlMessageJob[], - publish: (commit: JournalSqlMessagesCommit, index: number) => void -): Promise> { - const result = await batch.run(jobs, (job) => applyJournalSqlMessageSteps(db, job.command)) - for (const [index, outcome] of result.outcomes.entries()) { - if (outcome.committed) { - publish(outcome.value, index) - } - } - return result -} - export function applyJournalSqlMessageSteps( db: Database.Database, command: MessageCommand diff --git a/src/main/native-chat/agent-session-journal/journal-sql-message.test-fixture.ts b/src/main/native-chat/agent-session-journal/journal-sql-message.test-fixture.ts index 1c1f188235b..2370a112831 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-message.test-fixture.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-message.test-fixture.ts @@ -15,14 +15,12 @@ import { agentSessionSendBodyFingerprint } from '../../../shared/agent-session/r import { migrateJournalDomainSchema } from './journal-domain-schema' import { writeJournalDomainItemNames } from './journal-domain-item-names' import { JournalSqlBatch } from './journal-sql-batch' -import { - commitJournalSqlMessages, - prepareJournalSqlMessages, - type JournalSqlMessagesCommit -} from './journal-sql-message-commit' +import { JournalSqlCommand } from './journal-sql-command' +import type { JournalWriterCommand } from './journal-writer-command' +import type { JournalWriterCommit } from './journal-writer-changes' import type { JournalSqlMessage } from './journal-sql-message-commit' -export const target = { sessionId: 'chat', epoch: 'epoch' } +export const target = { sessionId: 'chat', epoch: 'epoch', ownerToken: 'owner' } export function message( text: string, clientMessageId = 'assistant', @@ -42,19 +40,39 @@ export function messageFixture(db = new Database(':memory:')) { `INSERT INTO journal_chat_state(workspace_id,session_id,epoch,last_sequence,highest_fence) VALUES('workspace',?,?,10,1)` ).run(target.sessionId, target.epoch) const batch = new JournalSqlBatch(db) - const published: JournalSqlMessagesCommit[] = [] - const commit = (...steps: JournalSqlMessage[]) => - commitJournalSqlMessages( - db, - batch, - [prepareJournalSqlMessages({ target, steps, savedAt: 100 })], - (result) => { - if (db.isTransaction) { - throw new Error('publication preceded commit') - } - published.push(result) - } + const published: JournalWriterCommit[] = [] + const command = ( + steps: readonly JournalSqlMessage[], + destination = target + ): JournalWriterCommand => ({ + target: destination, + operation: { + kind: 'steps', + steps: steps.map((step) => { + const reservedBytes = Buffer.byteLength(JSON.stringify(step)) + 512 + return step.kind === 'text-snapshot' + ? { + kind: 'resolved-item', + resolution: { kind: 'text-snapshot', identity: step.identity, body: step.body }, + options: step.options, + reservedBytes + } + : { ...step, reservedBytes } + }) + } + }) + const commitCommands = async (commands: readonly JournalWriterCommand[], savedAt = 100) => { + const result = await batch.run(commands, (command) => + new JournalSqlCommand(db, '').apply(command, savedAt) ) + for (const outcome of result.outcomes) { + if (outcome.committed) { + published.push(outcome.value) + } + } + return result + } + const commit = (...steps: JournalSqlMessage[]) => commitCommands([command(steps)]) const item = (itemId: string): AgentJournalRenderItem | undefined => { const row = db.prepare('SELECT payload_json FROM journal_items WHERE item_id=?').get(itemId) return row ? JSON.parse(String(row.payload_json)) : undefined @@ -143,5 +161,5 @@ export function messageFixture(db = new Database(':memory:')) { ended: false }) } - return { db, batch, published, commit, item, seedItem, seedSend } + return { db, batch, published, command, commitCommands, commit, item, seedItem, seedSend } } diff --git a/src/main/native-chat/agent-session-journal/journal-sql-message.test.ts b/src/main/native-chat/agent-session-journal/journal-sql-message.test.ts index 396764eb6a1..ebb2d581c8c 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-message.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-sql-message.test.ts @@ -1,12 +1,10 @@ import { afterEach, expect, it } from 'vitest' import type Database from '../../sqlite/sync-database' import { message, messageFixture, target } from './journal-sql-message.test-fixture' -import { commitJournalSqlMessages, prepareJournalSqlMessages } from './journal-sql-message-commit' import type { AgentJournalSubmission } from '../../../shared/agent-session/journal/agent-session-journal-types' import { journalDomainItemPlacement } from './journal-domain-item-placement' import { applyJournalDomainItem } from './journal-domain-item' import { readJournalDomainItems } from './journal-sql-item-reads' -import { writeJournalDomainCurrentItem } from './journal-domain-current-item' const databases: Database.Database[] = [] const fixture = () => { @@ -20,75 +18,28 @@ afterEach(() => { } }) -it('admits revision zero and moves the same item without changing its admission ordinal', () => { - const { db, item } = fixture() - const admitted = { - itemId: 'orca:send', - revision: 0, - sequence: 11, - observedAt: 100, - turnScope: { kind: 'thread' as const }, - body: { kind: 'message' as const, role: 'user' as const, blocks: [] } - } - db.exec('BEGIN') - writeJournalDomainCurrentItem(db, target, { - item: admitted, - ordinal: 11, - hidden: false, - ownerFence: 1 - }) - expect(item('orca:send')).toEqual(admitted) - writeJournalDomainCurrentItem(db, target, { - item: { - ...admitted, - sequence: 20, - observedAt: 200, - turnScope: { kind: 'turn', turnItemId: 'orca:turn' } - }, - previous: { item: admitted, ordinal: 11, hidden: false }, - ordinal: 11, - hidden: false, - ownerFence: 1 - }) - db.exec('COMMIT') - expect(item('orca:send')).toMatchObject({ - revision: 0, - sequence: 20, - observedAt: 200, - turnScope: { kind: 'turn', turnItemId: 'orca:turn' } - }) - expect(db.prepare("SELECT ordinal FROM journal_items WHERE item_id='orca:send'").get()).toEqual({ - ordinal: 11 - }) -}) - it('publishes exact committed snapshots and their final body in order', async () => { - const { db, batch, commit, item, published } = fixture() + const { db, command, commitCommands, commit, item, published } = fixture() db.prepare('UPDATE journal_chat_state SET context_boundary=7').run() expect((await commit(message('first'))).outcomes[0]).toMatchObject({ committed: true, value: { - cursor: { epoch: 'epoch', sequence: 11 }, - publicCursor: { + cursor: { epoch: 'context-v1:2aa4945736094af6247b635644bf6763d06f6ba2c1f8d5a69f5ef000feafd5f1', sequence: 11 } } }) - const result = await commitJournalSqlMessages( - db, - batch, - ['second', 'third'].map((text) => - prepareJournalSqlMessages({ target, steps: [message(text)], savedAt: 200 }) - ), - (result) => published.push(result) + const result = await commitCommands( + ['second', 'third'].map((text) => command([message(text)])), + 200 ) expect(result.outcomes.map((entry) => entry.committed)).toEqual([true, true]) - expect(published.at(-1)?.publicCursor).toEqual({ + expect(published.at(-1)?.cursor).toEqual({ epoch: 'context-v1:2aa4945736094af6247b635644bf6763d06f6ba2c1f8d5a69f5ef000feafd5f1', sequence: 13 }) - expect(published.map((entry) => entry.items[0]?.body)).toEqual([ + expect(published.map((entry) => entry.publication?.items[0]?.body)).toEqual([ { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'first' }] }, { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'second' }] }, { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'third' }] } @@ -99,30 +50,22 @@ it('publishes exact committed snapshots and their final body in order', async () expect((await commit(ended)).outcomes[0]).toMatchObject({ committed: true, value: { - cursor: { epoch: 'epoch', sequence: 14 }, - items: [{ body: { completedAt: 300 } }] + cursor: { + epoch: 'context-v1:2aa4945736094af6247b635644bf6763d06f6ba2c1f8d5a69f5ef000feafd5f1', + sequence: 14 + }, + publication: { items: [{ body: { completedAt: 300 } }] } } }) }) -it('commits ready revisions from different chats atomically before publishing either payload', async () => { - const { db, batch, commit } = fixture() +it('commits ready revisions from different chats atomically', async () => { + const { db, command, commitCommands, commit } = fixture() await commit(message('first', 'assistant', 'item')) db.exec(`INSERT INTO journal_chat_state(workspace_id,session_id,epoch,last_sequence,highest_fence) VALUES('workspace','other','epoch',0,1)`) - const other = { sessionId: 'other', epoch: 'epoch' } - await commitJournalSqlMessages( - db, - batch, - [ - prepareJournalSqlMessages({ - target: other, - steps: [message('other first', 'assistant', 'item')], - savedAt: 100 - }) - ], - () => {} - ) + const other = { sessionId: 'other', epoch: 'epoch', ownerToken: 'other-owner' } + await commitCommands([command([message('other first', 'assistant', 'item')], other)]) db.exec(`PRAGMA foreign_keys=ON; CREATE TABLE published_pair (id TEXT PRIMARY KEY); CREATE TABLE pending_pair (id TEXT REFERENCES published_pair(id) DEFERRABLE INITIALLY DEFERRED); @@ -130,25 +73,33 @@ it('commits ready revisions from different chats atomically before publishing ei BEGIN INSERT INTO pending_pair VALUES ('paired'); END; CREATE TEMP TRIGGER pair_second AFTER UPDATE OF payload_json ON journal_items WHEN NEW.session_id='other' BEGIN INSERT INTO published_pair VALUES ('paired'); END`) - const delivered: string[] = [] - const result = await commitJournalSqlMessages( - db, - batch, - [target, other].map((target) => - prepareJournalSqlMessages({ target, steps: [message('last')], savedAt: 200 }) - ), - (commit, index) => { - expect(db.isTransaction).toBe(false) - expect(db.prepare('SELECT id FROM published_pair').get()).toEqual({ id: 'paired' }) - delivered.push(`${index}:${commit.cursor.sequence}`) - } + const result = await commitCommands( + [target, other].map((target) => command([message('last')], target)), + 200 ) expect(result.outcomes.map((outcome) => outcome.committed)).toEqual([true, true]) - expect(delivered).toEqual(['0:12', '1:2']) + expect(db.isTransaction).toBe(false) + expect(db.prepare('SELECT id FROM published_pair').get()).toEqual({ id: 'paired' }) + expect(result.outcomes).toMatchObject([ + { + committed: true, + value: { + cursor: { epoch: 'epoch', sequence: 12 }, + publication: { items: [{ body: { blocks: [{ type: 'text', text: 'last' }] } }] } + } + }, + { + committed: true, + value: { + cursor: { epoch: 'epoch', sequence: 2 }, + publication: { items: [{ body: { blocks: [{ type: 'text', text: 'last' }] } }] } + } + } + ]) }) it('accepts an optimistic echo once, preserving its local body and moved placement', async () => { - const { db, commit, item, seedSend, seedItem } = fixture() + const { db, commit, commitCommands, item, seedSend, seedItem } = fixture() seedSend() db.exec(`INSERT INTO journal_submission_attempts(session_id,epoch,message_id,attempt_sequence, attempt_index,current,item_id,ordinal,submitted_at,accepted_sequence,fingerprint, @@ -186,13 +137,18 @@ it('accepts an optimistic echo once, preserving its local body and moved placeme committed: true, value: { inputChanged: true, - acceptedInputs: { latestAcceptedTurnSequence: 1, latestPersonAcceptedSequence: 1 }, - cards: [{ messageId: 'card', state: 'withdrawn' }], - receipts: [ - { clientMessageId: 'send', acceptedAt: 100, cursor: { epoch: 'epoch', sequence: 11 } } - ] + publication: { + acceptedInputs: { latestAcceptedTurnSequence: 1, latestPersonAcceptedSequence: 1 }, + cards: [], + receipts: [ + { clientMessageId: 'send', acceptedAt: 100, cursor: { epoch: 'epoch', sequence: 11 } } + ] + } } }) + expect(db.prepare("SELECT state FROM journal_queue_cards WHERE message_id='card'").get()).toEqual( + { state: 'returned' } + ) const accepted: AgentJournalSubmission = JSON.parse( String( db.prepare('SELECT payload_json FROM journal_submission_attempts WHERE current=1').get()! @@ -211,9 +167,6 @@ it('accepts an optimistic echo once, preserving its local body and moved placeme db.prepare('SELECT payload_json FROM journal_submission_attempts WHERE current=0').get()! .payload_json ).toBe(historical) - expect( - db.prepare('SELECT candidate FROM journal_submission_attempts WHERE current=1').get()!.candidate - ).toBe(0) expect(item('orca:send')).toMatchObject({ sequence: 7, observedAt: 70, @@ -222,14 +175,28 @@ it('accepts an optimistic echo once, preserving its local body and moved placeme input.body.blocks = [{ type: 'text', text: 'provider normalized' }] expect((await commit(input)).outcomes[0]).toMatchObject({ committed: true, - value: { receipts: [], inputChanged: false } + value: { publication: { receipts: [] } } }) expect(item('orca:send')?.body).toEqual({ kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'original' }] }) - expect(db.prepare('SELECT COUNT(*) AS count FROM journal_items').get()!.count).toBe(2) + const reported = await commitCommands([ + { + target, + operation: { + kind: 'item', + identity: { provider: 'orca', clientMessageId: 'reported-command' }, + body: { kind: 'turn', turnId: 'command', state: 'completed' }, + options: { fence: 1, turnScope: { kind: 'thread' } } + } + } + ]) + expect(reported.outcomes[0]).toMatchObject({ + committed: true, + value: { publication: { cards: [{ messageId: 'card', state: 'withdrawn' }] } } + }) }) it('refuses a snapshot after the saved turn ends and commits only a successful step prefix', async () => { @@ -258,8 +225,8 @@ it('refuses a snapshot after the saved turn ends and commits only a successful s committed: true, value: { cursor: { epoch: 'epoch', sequence: 12 }, - items: [], - outcomes: [{ kind: 'dropped', reason: 'turn-settled' }] + publication: { items: [] }, + value: { outcomes: [{ kind: 'dropped', reason: 'turn-settled' }] } } }) expect(item('orca:assistant')).toBeUndefined() @@ -277,13 +244,12 @@ it('refuses a snapshot after the saved turn ends and commits only a successful s committed: true, value: { cursor: { epoch: 'epoch', sequence: 13 }, - items: [{ itemId: 'orca:prefix' }], - failure: { message: 'refused' } + publication: { items: [{ itemId: 'orca:prefix' }] }, + value: { failure: { message: 'refused' } } } }) expect(item('orca:refused')).toBeUndefined() expect(item('orca:suffix')).toBeUndefined() - expect(db.prepare('SELECT COUNT(*) AS count FROM journal_events').get()!.count).toBe(3) }) it('stores one growing text body and a metadata-only replay tail', async () => { @@ -354,7 +320,9 @@ it('retains a stream placement while honoring a newer moved row and a settled tu const late = await commit(second) expect(late.outcomes[0]).toMatchObject({ committed: true, - value: { outcomes: [{ kind: 'dropped', reason: 'turn-settled', turnItemId: 'orca:turn' }] } + value: { + value: { outcomes: [{ kind: 'dropped', reason: 'turn-settled', turnItemId: 'orca:turn' }] } + } }) expect(item('orca:assistant')?.revision).toBe(4) }) @@ -379,7 +347,6 @@ it('retains removed turn identity when its item is reinserted', () => { ) db.exec('COMMIT') expect(item('orca:turn')).toMatchObject({ revision: 10 }) - expect(db.prepare('SELECT 1 FROM journal_item_tombstones').get()).toBeUndefined() expect( readJournalDomainItems(db, target, { kind: 'named-turn', turnId: 'removed-turn' }) ).toEqual({