diff --git a/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts b/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts index a00b1367ac..7af2ca13db 100644 --- a/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts +++ b/packages/runtime-host/src/__tests__/execution-host-recovery.test.ts @@ -217,6 +217,151 @@ test('startup recovery replays an admitted regenerate with its source lineage', }); }); +test('startup recovery materializes legacy terminal Root sources exactly once', async () => { + await withExecutionRoot(async (fixture) => { + const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts(); + assert.deepEqual( + (await fixture.readSessionUserMessages()).filter((message) => + legacy.sources.some((source) => source.messageId === message.id), + ), + [], + ); + + const firstHost = await fixture.startHost(); + await fixture.stopHost(firstHost); + assert.deepEqual( + (await fixture.readSessionUserMessages()) + .filter((message) => legacy.sources.some((source) => source.messageId === message.id)) + .map(({ id, turnId, ts, text }) => ({ id, turnId, ts, text })), + legacy.sources.map((source) => ({ + id: source.messageId, + turnId: legacy.turnId, + ts: source.admittedAt, + text: source.content.text, + })), + ); + + const secondHost = await fixture.startHost(); + await fixture.stopHost(secondHost); + assert.deepEqual( + (await fixture.readSessionUserMessages()) + .filter((message) => legacy.sources.some((source) => source.messageId === message.id)) + .map(({ id, turnId, ts, text }) => ({ id, turnId, ts, text })), + legacy.sources.map((source) => ({ + id: source.messageId, + turnId: legacy.turnId, + ts: source.admittedAt, + text: source.content.text, + })), + ); + }); +}); + +test('startup recovery replays a legacy Root without a Run before materializing its sources', async () => { + await withExecutionRoot(async (fixture) => { + const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts('missing'); + + const firstHost = await fixture.startHost(); + await fixture.stopHost(firstHost); + const secondHost = await fixture.startHost(); + await fixture.stopHost(secondHost); + + assert.deepEqual( + (await fixture.readSessionUserMessages()) + .filter((message) => legacy.sources.some((source) => source.messageId === message.id)) + .map(({ id, turnId, ts, text }) => ({ id, turnId, ts, text })), + legacy.sources.map((source) => ({ + id: source.messageId, + turnId: legacy.turnId, + ts: source.admittedAt, + text: source.content.text, + })), + ); + const ledger = await fixture.readTurn(legacy.turnId); + assert.equal(ledger.runs.length, 1); + assert.equal(ledger.terminalEvents.length, 1); + }); +}); + +test('startup recovery closes a legacy non-terminal Run before materializing its sources', async () => { + await withExecutionRoot(async (fixture) => { + const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts('created'); + + const firstHost = await fixture.startHost(); + await fixture.stopHost(firstHost); + const secondHost = await fixture.startHost(); + await fixture.stopHost(secondHost); + + assert.deepEqual( + (await fixture.readSessionUserMessages()) + .filter((message) => legacy.sources.some((source) => source.messageId === message.id)) + .map(({ id, turnId, ts, text }) => ({ id, turnId, ts, text })), + legacy.sources.map((source) => ({ + id: source.messageId, + turnId: legacy.turnId, + ts: source.admittedAt, + text: source.content.text, + })), + ); + const ledger = await fixture.readTurn(legacy.turnId); + assert.equal(ledger.runs.length, 1); + assert.equal(ledger.terminalEvents.length, 1); + }); +}); + +test('startup recovery rejects an unproven legacy Root without creating its missing Run', async () => { + await withExecutionRoot(async (fixture) => { + const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts('missing'); + fixture.deleteRootSourceProof(legacy.sources[1].messageId); + + await fixture.expectHostStartupFailure(); + await fixture.assertOwnerAvailable(); + assert.deepEqual(await fixture.readTurnRuns(legacy.turnId), []); + assert.deepEqual( + (await fixture.readSessionUserMessages()).filter((message) => + legacy.sources.some((source) => source.messageId === message.id), + ), + [], + ); + }); +}); + +test('startup recovery rejects an unproven legacy non-terminal Run before closing it', async () => { + await withExecutionRoot(async (fixture) => { + const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts('created'); + fixture.deleteRootSourceProof(legacy.sources[1].messageId); + + await fixture.expectHostStartupFailure(); + await fixture.assertOwnerAvailable(); + const ledger = await fixture.readTurn(legacy.turnId); + assert.equal(ledger.runs.length, 1); + assert.equal(ledger.runs[0]?.status, 'created'); + assert.equal(ledger.terminalEvents.length, 0); + assert.deepEqual( + (await fixture.readSessionUserMessages()).filter((message) => + legacy.sources.some((source) => source.messageId === message.id), + ), + [], + ); + }); +}); + +test('startup recovery rejects a legacy terminal Root source without its durable receipt', async () => { + await withExecutionRoot(async (fixture) => { + const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts(); + fixture.deleteRootSourceProof(legacy.sources[1].messageId); + + await fixture.expectHostStartupFailure(); + await fixture.assertOwnerAvailable(); + assert.deepEqual( + (await fixture.readSessionUserMessages()).filter((message) => + legacy.sources.some((source) => source.messageId === message.id), + ), + [], + ); + }); +}); + test('a fresh quoted Turn preserves durable and Runtime handoff content', async () => { await withExecutionRoot(async (fixture) => { const host = await fixture.startHost(); diff --git a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts index 986e4328d4..47863eb2a2 100644 --- a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts +++ b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts @@ -34,11 +34,13 @@ import { createServer, type Server } from 'node:http'; import { connect, type Socket } from 'node:net'; import { tmpdir } from 'node:os'; import { dirname, join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; import { test } from 'node:test'; import { TOOL_BOUNDARY_PROTOCOL_V1 } from '@maka/core/runtime-event'; import { canonicalToolArgsHash } from '@maka/core/tool-args-identity'; import type { AgentRunHeader } from '@maka/core/agent-run'; import { + aggregateMessageContents, messageContentDigest, normalizeMessageContent, type MessageContent, @@ -65,6 +67,7 @@ import { openInteractiveExecutionStoresForRead, openInteractiveExecutionStoresForWrite, } from '@maka/storage/execution-stores'; +import { OPERATIONAL_STATE_DATABASE_NAME } from '@maka/storage/operational-state-store'; import { openInteractiveRuntimePolicyStoresForWrite } from '@maka/storage/runtime-policy-stores'; import { resolveRootControlNamespace, @@ -729,6 +732,127 @@ export class ExecutionFixture { } } + async seedLegacyRootWithoutSourceTranscripts( + runState: 'missing' | 'created' | 'terminal' = 'terminal', + ): Promise<{ + turnId: string; + runId: string; + sources: readonly [ + { messageId: string; content: MessageContent; admittedAt: number }, + { messageId: string; content: MessageContent; admittedAt: number }, + ]; + }> { + const owner = await tryAcquireInteractiveRootOwner(this.capability); + assert.ok(owner); + if (!owner) throw new Error('Unable to acquire execution root for legacy Root setup'); + let stores: Awaited> | undefined; + try { + stores = await openInteractiveExecutionStoresForWrite(owner.lease); + const turnId = randomUUID(); + const runId = randomUUID(); + const admittedAt = Date.now(); + const followup = { + messageId: randomUUID(), + content: { text: 'legacy follow-up source' }, + admittedAt, + }; + const steering = { + messageId: randomUUID(), + content: { text: 'legacy steering source' }, + admittedAt, + }; + const normalizedInput = aggregateMessageContents([followup.content, steering.content]); + const admission = await stores.agentRunStore.admitRootTurn({ + sessionId: this.sessionId, + turnId, + proposedRunId: runId, + proposedUserMessageId: null, + execution: { + kind: 'external_message', + inputDigest: messageContentDigest(normalizedInput), + }, + previousRootTurnId: null, + normalizedInput, + sourceMessages: [ + { + messageId: followup.messageId, + content: followup.content, + submittedContentDigest: messageContentDigest(followup.content), + placement: 'next_turn', + disposition: 'followup', + }, + { + messageId: steering.messageId, + content: steering.content, + submittedContentDigest: messageContentDigest(steering.content), + placement: 'current_turn', + disposition: 'steering', + }, + ], + admittedAt, + }); + assert.equal(admission.kind, 'admitted'); + const run: AgentRunHeader = { + runId, + invocationId: runId, + sessionId: this.sessionId, + turnId, + status: 'created', + backendKind: 'fake', + llmConnectionSlug: 'fake', + modelId: 'fake-model', + cwd: this.root, + permissionMode: 'ask', + createdAt: admittedAt, + updatedAt: admittedAt, + }; + if (runState !== 'missing') { + await stores.agentRunStore.createRun(run, { durable: true }); + } + if (runState === 'terminal') { + const terminalAt = admittedAt + 1; + const terminal = buildRecoveredTerminalRuntimeEvent({ + id: randomUUID(), + run, + status: 'failed', + ts: terminalAt, + failureClass: 'legacy_terminal', + recoveryReason: 'test_legacy_terminal_root', + }); + await commitTerminalRunWithRuntimeFact({ + runStore: stores.agentRunStore, + runtimeEventStore: stores.runtimeEventStore, + newId: randomUUID, + sessionId: this.sessionId, + runId, + turnId, + status: 'failed', + ts: terminalAt, + terminalEvent: terminal, + failureClass: 'legacy_terminal', + }); + } + return { turnId, runId, sources: [followup, steering] }; + } finally { + await stores?.sessionStore.close?.(); + await owner.close(); + } + } + + deleteRootSourceProof(messageId: string): void { + const database = new DatabaseSync(join(this.root, OPERATIONAL_STATE_DATABASE_NAME)); + try { + const result = database + .prepare( + 'DELETE FROM core_root_source_message_proofs WHERE session_id = ? AND message_id = ?', + ) + .run(this.sessionId, messageId); + assert.equal(result.changes, 1); + } finally { + database.close(); + } + } + async archiveSession(): Promise { const owner = await tryAcquireInteractiveRootOwner(this.capability); assert.ok(owner); @@ -1000,6 +1124,20 @@ export class ExecutionFixture { } } + async readTurnRuns(turnId: string) { + const reader = await acquireReader(this.capability); + let stores: Awaited> | undefined; + try { + stores = await openInteractiveExecutionStoresForRead(reader.lease); + return (await stores.agentRunStore.listSessionRuns(this.sessionId)).filter( + (candidate) => candidate.turnId === turnId, + ); + } finally { + await stores?.sessionStore.close?.(); + await reader.close(); + } + } + async readSessionUserMessages(): Promise>> { const reader = await acquireReader(this.capability); let stores: Awaited> | undefined; diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts index 6ff54ecd0d..b03a4c579e 100644 --- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts @@ -21,7 +21,9 @@ import assert from 'node:assert/strict'; import { test } from 'node:test'; import { messageContentDigest, type MessageContent } from '@maka/core/events'; import type { RuntimeEvent } from '@maka/core/runtime-event'; +import { RuntimeMessageAuthorityInvariantError } from '@maka/runtime/message-authority'; import type { + MarkMessagesHandedOffInput, MessageAdmissionStore, PendingMessageAdmission, RootTurnSourceMessageReceipt, @@ -1690,10 +1692,91 @@ test('run settlement hands off only steering admissions with immutable proof', a }); assert.equal(fixture.readMessageAdmission('steer-proved'), undefined); + assert.deepEqual(fixture.handoffCalls, [ + { + sessionId: ROOT.sessionId, + messageIds: ['steer-proved'], + turnId: ROOT.turnId, + }, + ]); const batch = fixture.coordinator.beginTerminalTransition(ROOT); fixture.coordinator.completeIdle(batch); }); +test('run materialization preserves exact Root source receipt fallback order', async () => { + const fixture = createFixture(); + fixture.receipts.set('exact-root', matchingSourceReceipt('exact-root', 42)); + fixture.receipts.set('exact-second', matchingSourceReceipt('exact-second', 43)); + + await fixture.coordinator.materializeMessageHandoffsForRun({ + ...ROOT, + messageIds: ['exact-root', 'exact-second', 'exact-root'], + }); + + assert.deepEqual(fixture.handoffCalls, [ + { + sessionId: ROOT.sessionId, + turnId: ROOT.turnId, + messageIds: ['exact-root', 'exact-second'], + provenRootMessages: [ + { + messageId: 'exact-root', + content: { text: 'canonical exact-root' }, + admittedAt: 42, + }, + { + messageId: 'exact-second', + content: { text: 'canonical exact-second' }, + admittedAt: 43, + }, + ], + }, + ]); +}); + +test('run materialization rejects the whole requested Root batch when any receipt mismatches', async () => { + const mismatches: Array<{ + messageId: string; + receipt?: RootTurnSourceMessageReceipt; + }> = [ + { messageId: 'proof-less' }, + { + messageId: 'wrong-session', + receipt: matchingSourceReceipt('wrong-session', 10, { sessionId: 'other' }), + }, + { + messageId: 'wrong-turn', + receipt: matchingSourceReceipt('wrong-turn', 11, { turnId: 'other' }), + }, + { + messageId: 'wrong-run', + receipt: matchingSourceReceipt('wrong-run', 12, { runId: 'other' }), + }, + { + messageId: 'wrong-message', + receipt: matchingSourceReceipt('other-message', 13), + }, + ]; + + for (const mismatch of mismatches) { + const fixture = createFixture(); + fixture.receipts.set('exact-root', matchingSourceReceipt('exact-root', 42)); + if (mismatch.receipt) fixture.receipts.set(mismatch.messageId, mismatch.receipt); + + await assert.rejects( + () => + fixture.coordinator.materializeMessageHandoffsForRun({ + ...ROOT, + messageIds: ['exact-root', mismatch.messageId], + }), + (error: unknown) => + error instanceof RuntimeMessageAuthorityInvariantError && + error.message === `Root admission does not prove Message handoff ${mismatch.messageId}`, + ); + assert.deepEqual(fixture.handoffCalls, []); + } +}); + test('a failed terminal root leaves no handed-off payload for restart recovery', async () => { const fixture = createFixture(); await fixture.admissions.commitMessageAdmission({ @@ -2277,7 +2360,10 @@ function createFixture( state: 'accepted' | 'handed_off' | 'executed' | 'cancelled'; } >(); - const admissions = memoryMessageAdmissionStore(messageAdmissions); + const handoffCalls: MarkMessagesHandedOffInput[] = []; + const admissions = memoryMessageAdmissionStore(messageAdmissions, (input) => { + handoffCalls.push(input); + }); const stopClaimed = deferred(); const terminal = deferred(); let coordinator: HostMessageCoordinator; @@ -2417,6 +2503,7 @@ function createFixture( events, receipts, recoveredBatches, + handoffCalls, readMessageAdmission: (messageId: string) => messageAdmissions.get(messageId)?.admission, stopClaimed, resolveTerminal: terminal.resolve, @@ -2441,6 +2528,7 @@ function memoryMessageAdmissionStore( state: 'accepted' | 'handed_off' | 'executed' | 'cancelled'; } >, + onMessagesHandedOff?: (input: MarkMessagesHandedOffInput) => void, ): MessageAdmissionStore { return { commitMessageAdmission: async (admission) => { @@ -2470,7 +2558,9 @@ function memoryMessageAdmissionStore( } } }, - markMessagesHandedOff: async ({ messageIds }) => { + markMessagesHandedOff: async (input) => { + onMessagesHandedOff?.(input); + const { messageIds } = input; for (const messageId of messageIds) admissions.delete(messageId); }, }; @@ -2541,6 +2631,29 @@ function sourceReceipt( }; } +function matchingSourceReceipt( + messageId: string, + admittedAt: number, + overrides: Partial = {}, +): RootTurnSourceMessageReceipt { + const base = sourceReceipt( + messageId, + { text: `canonical ${messageId}` }, + 'current_turn', + 'steering', + ROOT.turnId, + ); + return { + ...base, + admission: { + ...base.admission, + runId: ROOT.runId, + admittedAt, + ...overrides, + }, + }; +} + function steeringEvent( messageId: string, content: MessageContent | string, diff --git a/packages/runtime-host/src/server/hosted-execution-recovery.ts b/packages/runtime-host/src/server/hosted-execution-recovery.ts index d720c185d5..489ba242f7 100644 --- a/packages/runtime-host/src/server/hosted-execution-recovery.ts +++ b/packages/runtime-host/src/server/hosted-execution-recovery.ts @@ -97,7 +97,7 @@ export async function prepareHostedExecutionRecovery( } if (admission.userMessageId === null) { if (admission.sourceMessages.length > 0) { - verifyQueueSourceMessages(admission, messageIndex); + await verifyQueueSourceMessages(admission, messageIndex, input.stores.agentRunStore); } if (rootUserMessages.length > 0) { throw new Error(`Admitted Turn ${admission.turnId} must not record a UserMessage`); @@ -307,12 +307,35 @@ function verifyOrRecoverUserMessage( indexRecoveryMessage(index, recoveredMessage); } -function verifyQueueSourceMessages( +async function verifyQueueSourceMessages( admission: RootTurnAdmission, index: RecoveryMessageIndex, -): void { + proofReader: Pick< + ExecutionStoresWriter<'interactive'>['agentRunStore'], + 'readRootTurnSourceMessageReceipt' + >, +): Promise { for (const source of admission.sourceMessages) { const owners = index.messagesById.get(source.messageId) ?? []; + if (owners.length === 0) { + const proof = await proofReader.readRootTurnSourceMessageReceipt( + admission.sessionId, + source.messageId, + ); + if ( + !proof || + proof.admission.sessionId !== admission.sessionId || + proof.admission.turnId !== admission.turnId || + proof.admission.runId !== admission.runId || + proof.sourceMessage.messageId !== source.messageId || + !messageContentsEqual(proof.sourceMessage.content, source.content) + ) { + throw new Error( + `Admitted Turn ${admission.turnId} has no durable proof for queue source ${source.messageId}`, + ); + } + continue; + } if ( owners.length !== 1 || owners[0]?.type !== 'user' || diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts index 0ba526392f..cf11cda26a 100644 --- a/packages/runtime-host/src/server/message-coordinator.ts +++ b/packages/runtime-host/src/server/message-coordinator.ts @@ -40,6 +40,7 @@ import { normalizeRootTurnAdmissionPayload, submittedTurnIntentsEqual, type ImmutableSteeringMessageProof, + type MarkMessagesHandedOffInput, type MessageAdmissionStore, type PendingMessageAdmission, type RootTurnSourceMessage, @@ -573,27 +574,18 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { messageIds: readonly string[]; }): Promise { const handoff: string[] = []; + const provenRootMessages: Array< + NonNullable[number] + > = []; for (const messageId of new Set(input.messageIds)) { - const proof = await this.#durableProof.readRootTurnSourceMessageReceipt( - input.sessionId, - messageId, - ); - if ( - !proof || - proof.admission.turnId !== input.turnId || - proof.admission.runId !== input.runId || - proof.sourceMessage.messageId !== messageId - ) { - throw new RuntimeMessageAuthorityInvariantError( - `Root admission does not prove Message handoff ${messageId}`, - ); - } handoff.push(messageId); + provenRootMessages.push(await this.#readProvenRootMessage(input, messageId)); } await this.#admissions.markMessagesHandedOff({ sessionId: input.sessionId, messageIds: handoff, turnId: input.turnId, + ...(provenRootMessages.length > 0 ? { provenRootMessages } : {}), }); } @@ -605,19 +597,13 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { messageIds: readonly string[]; }): Promise { const messageIds = new Set(); + const provenRootMessages: Array< + NonNullable[number] + > = []; const admissions = await this.#admissions.listMessageAdmissions(input.sessionId); for (const messageId of new Set(input.messageIds)) { - const proof = await this.#durableProof.readRootTurnSourceMessageReceipt( - input.sessionId, - messageId, - ); - if ( - proof?.admission.turnId === input.turnId && - proof.admission.runId === input.runId && - proof.sourceMessage.messageId === messageId - ) { - messageIds.add(messageId); - } + messageIds.add(messageId); + provenRootMessages.push(await this.#readProvenRootMessage(input, messageId)); } for (const admission of admissions) { if ( @@ -639,9 +625,36 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { sessionId: input.sessionId, messageIds: [...messageIds], turnId: input.turnId, + ...(provenRootMessages.length > 0 ? { provenRootMessages } : {}), }); } + async #readProvenRootMessage( + input: { readonly sessionId: string; readonly turnId: string; readonly runId: string }, + messageId: string, + ): Promise[number]> { + const proof = await this.#durableProof.readRootTurnSourceMessageReceipt( + input.sessionId, + messageId, + ); + if ( + !proof || + proof.admission.sessionId !== input.sessionId || + proof.admission.turnId !== input.turnId || + proof.admission.runId !== input.runId || + proof.sourceMessage.messageId !== messageId + ) { + throw new RuntimeMessageAuthorityInvariantError( + `Root admission does not prove Message handoff ${messageId}`, + ); + } + return { + messageId, + content: proof.sourceMessage.content, + admittedAt: proof.admission.admittedAt, + }; + } + async cancelMessages(sessionId: string, messageIds: readonly string[]): Promise { await this.#admissions.cancelMessageAdmissions(sessionId, messageIds); } @@ -680,7 +693,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority { sessionId, turnId: admission.turnId, runId: admission.runId, - messageIds: [admission.messageId], + messageIds: [], }); } else { pending.push(admission); diff --git a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts index 4411168568..89d8e712f0 100644 --- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts @@ -25,7 +25,7 @@ import { DatabaseSync } from 'node:sqlite'; import { describe, test } from 'node:test'; import { Worker } from 'node:worker_threads'; import { AgentGraphClientTerminalCursorError } from '@maka/core/agent-graph-client-projection'; -import { messageContentDigest } from '@maka/core/events'; +import { messageContentDigest, type MessageContent } from '@maka/core/events'; import { canReadPath, createReadOnlyPermissionProfile, @@ -46,7 +46,11 @@ import { type SessionConfigurationMetadataUpdate, type SqliteSessionMetadataStoreFailpoint, } from '../sqlite-session-metadata-store.js'; -import type { PendingMessageAdmission } from '../message-admission-store.js'; +import type { + MarkMessagesHandedOffInput, + PendingMessageAdmission, + ProvenRootMessageHandoff, +} from '../message-admission-store.js'; import { createSqliteRuntimeStore, SQLITE_RUNTIME_SCHEMA_VERSION, @@ -380,6 +384,722 @@ describe('SqliteSessionMetadataStore', () => { } }); + test('materializes a proven Root message when its admission is absent', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-legacy-root' })); + + await markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-legacy-root', + messageIds: ['message-legacy-root'], + turnId: 'turn-legacy-root', + provenRootMessages: [ + { + messageId: 'message-legacy-root', + content: { text: 'retained by the legacy Root', displayText: 'legacy display' }, + admittedAt: 17, + }, + ], + }); + + assert.deepEqual(await store.readMessages('session-legacy-root'), [ + { + type: 'user', + id: 'message-legacy-root', + turnId: 'turn-legacy-root', + ts: 17, + text: 'retained by the legacy Root', + displayText: 'legacy display', + steeringEventId: 'message-legacy-root', + }, + ]); + assert.deepEqual(await store.listMessageAdmissions('session-legacy-root'), []); + } finally { + store.close(); + } + }); + + test('inserts proven Root messages before existing output from their Turn', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-legacy-root-order-')); + const path = join(root, 'state.sqlite'); + const store = createSqliteSessionMetadataStore(path); + try { + await store.create(fullHeader({ id: 'session-legacy-order' })); + const legacyOutput = 'existing chunked output '.repeat(4_096); + await store.appendMessages( + 'session-legacy-order', + [ + { + type: 'assistant', + id: 'message-prior-output', + turnId: 'turn-prior', + ts: 10, + text: 'prior output', + modelId: 'fake-model', + }, + { + type: 'assistant', + id: 'message-legacy-output', + turnId: 'turn-legacy-order', + ts: 18, + text: legacyOutput, + modelId: 'fake-model', + }, + { + type: 'user', + id: 'message-newer-user', + turnId: 'turn-newer', + ts: 30, + text: 'newest preview', + }, + ], + { lastMessageAt: 30, lastMessagePreview: 'newest preview' }, + ); + + await markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-legacy-order', + messageIds: ['message-legacy-followup', 'message-legacy-steering'], + turnId: 'turn-legacy-order', + provenRootMessages: [ + { + messageId: 'message-legacy-followup', + content: { text: 'legacy follow-up' }, + admittedAt: 17, + }, + { + messageId: 'message-legacy-steering', + content: { text: 'legacy steering' }, + admittedAt: 17, + }, + ], + }); + + assert.deepEqual( + (await store.readMessages('session-legacy-order')).map((message) => message.id), + [ + 'message-prior-output', + 'message-legacy-followup', + 'message-legacy-steering', + 'message-legacy-output', + 'message-newer-user', + ], + ); + assert.equal((await store.read('session-legacy-order')).header.lastMessageAt, 30); + const shiftedOutput = (await store.readMessages('session-legacy-order')).find( + (message) => message.id === 'message-legacy-output', + ); + assert.equal(shiftedOutput?.type, 'assistant'); + assert.equal( + shiftedOutput?.type === 'assistant' ? shiftedOutput.text : undefined, + legacyOutput, + ); + assert.equal( + (await store.readCatalogRecord('session-legacy-order')).lastMessagePreview, + 'newest preview', + ); + const audit = new DatabaseSync(path, { readOnly: true }); + try { + assert.deepEqual(audit.prepare('PRAGMA foreign_key_check').all(), []); + } finally { + audit.close(); + } + } finally { + store.close(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('places a proven Root message before an equally-timed newer transcript row', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-legacy-time-tie' })); + await store.appendMessages( + 'session-legacy-time-tie', + [ + { + type: 'user', + id: 'message-newer-time-tie', + turnId: 'turn-newer-time-tie', + ts: 17, + text: 'newer same-millisecond preview', + }, + ], + { lastMessageAt: 17, lastMessagePreview: 'newer same-millisecond preview' }, + ); + + await markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-legacy-time-tie', + messageIds: ['message-legacy-time-tie'], + turnId: 'turn-legacy-time-tie', + provenRootMessages: [ + { + messageId: 'message-legacy-time-tie', + content: { text: 'legacy same-millisecond source' }, + admittedAt: 17, + }, + ], + }); + + assert.deepEqual( + (await store.readMessages('session-legacy-time-tie')).map((message) => message.id), + ['message-legacy-time-tie', 'message-newer-time-tie'], + ); + assert.equal( + (await store.readCatalogRecord('session-legacy-time-tie')).lastMessagePreview, + 'newer same-millisecond preview', + ); + } finally { + store.close(); + } + }); + + test('keeps ordinary admission-only handoff append semantics', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-ordinary-handoff-order' })); + await store.appendMessages( + 'session-ordinary-handoff-order', + [ + { + type: 'assistant', + id: 'message-existing-ordinary-output', + turnId: 'turn-ordinary-handoff-order', + ts: 20, + text: 'existing output', + modelId: 'fake-model', + }, + ], + { lastMessageAt: 20, lastMessagePreview: 'existing output' }, + ); + await store.commitMessageAdmission({ + sessionId: 'session-ordinary-handoff-order', + turnId: 'turn-ordinary-handoff-order', + runId: 'run-ordinary-handoff-order', + messageId: 'message-ordinary-admission', + content: { text: 'ordinary admission' }, + submittedContentDigest: messageContentDigest({ text: 'ordinary admission' }), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 10, + }); + + await store.markMessagesHandedOff({ + sessionId: 'session-ordinary-handoff-order', + messageIds: ['message-ordinary-admission'], + turnId: 'turn-ordinary-handoff-order', + }); + + assert.deepEqual( + (await store.readMessages('session-ordinary-handoff-order')).map((message) => message.id), + ['message-existing-ordinary-output', 'message-ordinary-admission'], + ); + } finally { + store.close(); + } + }); + + test('rejects an admission handed off to a different Turn', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-admission-turn-conflict' })); + await store.commitMessageAdmission({ + sessionId: 'session-admission-turn-conflict', + turnId: 'turn-admission-authority', + runId: 'run-admission-turn-conflict', + messageId: 'message-admission-turn-conflict', + content: { text: 'turn-owned admission' }, + submittedContentDigest: messageContentDigest({ text: 'turn-owned admission' }), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 24, + }); + + await assert.rejects( + store.markMessagesHandedOff({ + sessionId: 'session-admission-turn-conflict', + messageIds: ['message-admission-turn-conflict'], + turnId: 'turn-different', + }), + /Turn conflict/, + ); + assert.deepEqual(await store.readMessages('session-admission-turn-conflict'), []); + assert.equal( + (await store.listMessageAdmissions('session-admission-turn-conflict')).length, + 1, + ); + } finally { + store.close(); + } + }); + + test('rejects fully materialized proven Root sources in a conflicting order', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-existing-source-order' })); + await store.appendMessages( + 'session-existing-source-order', + [ + { + type: 'user', + id: 'message-existing-source-b', + turnId: 'turn-existing-source-order', + ts: 25, + text: 'source b', + }, + { + type: 'user', + id: 'message-existing-source-a', + turnId: 'turn-existing-source-order', + ts: 25, + text: 'source a', + }, + ], + { lastMessageAt: 25, lastMessagePreview: 'source a' }, + ); + + await assert.rejects( + markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-existing-source-order', + messageIds: ['message-existing-source-a', 'message-existing-source-b'], + turnId: 'turn-existing-source-order', + provenRootMessages: [ + { + messageId: 'message-existing-source-a', + content: { text: 'source a' }, + admittedAt: 25, + }, + { + messageId: 'message-existing-source-b', + content: { text: 'source b' }, + admittedAt: 25, + }, + ], + }), + /source order conflict/, + ); + } finally { + store.close(); + } + }); + + test('rejects a partial proven Root group that already crosses newer history', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-partial-source-order' })); + await store.appendMessages( + 'session-partial-source-order', + [ + { + type: 'user', + id: 'message-partial-source-a', + turnId: 'turn-partial-source-order', + ts: 15, + text: 'source a', + }, + { + type: 'user', + id: 'message-partial-newer-tail', + turnId: 'turn-partial-newer', + ts: 30, + text: 'newer tail', + }, + { + type: 'user', + id: 'message-partial-source-c', + turnId: 'turn-partial-source-order', + ts: 15, + text: 'source c', + }, + ], + { lastMessageAt: 30, lastMessagePreview: 'newer tail' }, + ); + + await assert.rejects( + markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-partial-source-order', + messageIds: [ + 'message-partial-source-a', + 'message-partial-source-b', + 'message-partial-source-c', + ], + turnId: 'turn-partial-source-order', + provenRootMessages: [ + { + messageId: 'message-partial-source-a', + content: { text: 'source a' }, + admittedAt: 15, + }, + { + messageId: 'message-partial-source-b', + content: { text: 'source b' }, + admittedAt: 15, + }, + { + messageId: 'message-partial-source-c', + content: { text: 'source c' }, + admittedAt: 15, + }, + ], + }), + /source order conflict/, + ); + assert.deepEqual( + (await store.readMessages('session-partial-source-order')).map((message) => message.id), + ['message-partial-source-a', 'message-partial-newer-tail', 'message-partial-source-c'], + ); + } finally { + store.close(); + } + }); + + test('rejects an unsafe proven Root tail insertion range', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-legacy-root-overflow-')); + const path = join(root, 'state.sqlite'); + const store = createSqliteSessionMetadataStore(path); + try { + await store.create(fullHeader({ id: 'session-legacy-overflow' })); + await store.appendMessages( + 'session-legacy-overflow', + [ + { + type: 'assistant', + id: 'message-overflow-anchor', + turnId: 'turn-overflow-anchor', + ts: 1, + text: 'anchor', + modelId: 'fake-model', + }, + ], + { lastMessageAt: 1, lastMessagePreview: 'anchor' }, + ); + const database = new DatabaseSync(path); + try { + database + .prepare('UPDATE session_messages SET sequence = ? WHERE session_id = ? AND sequence = 0') + .run(Number.MAX_SAFE_INTEGER - 1, 'session-legacy-overflow'); + } finally { + database.close(); + } + + await assert.rejects( + markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-legacy-overflow', + messageIds: ['message-overflow-a', 'message-overflow-b'], + turnId: 'turn-legacy-overflow', + provenRootMessages: [ + { messageId: 'message-overflow-a', content: { text: 'a' }, admittedAt: 2 }, + { messageId: 'message-overflow-b', content: { text: 'b' }, admittedAt: 2 }, + ], + }), + /sequence overflow/, + ); + assert.deepEqual( + (await store.readMessages('session-legacy-overflow')).map((message) => message.id), + ['message-overflow-anchor'], + ); + } finally { + store.close(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('repeats a proven Root message handoff without duplicating its transcript', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-legacy-repeat' })); + const input = { + sessionId: 'session-legacy-repeat', + messageIds: ['message-legacy-repeat'], + turnId: 'turn-legacy-repeat', + provenRootMessages: [ + { + messageId: 'message-legacy-repeat', + content: { text: 'a single durable transcript message' }, + admittedAt: 18, + }, + ], + }; + + await markMessagesHandedOffWithProvenRoots(store, input); + await markMessagesHandedOffWithProvenRoots(store, input); + + assert.deepEqual( + (await store.readMessages('session-legacy-repeat')).map((message) => ({ + id: message.id, + turnId: message.turnId, + ts: message.ts, + })), + [{ id: 'message-legacy-repeat', turnId: 'turn-legacy-repeat', ts: 18 }], + ); + } finally { + store.close(); + } + }); + + test('rejects an admission-less handoff without a proven Root message', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-no-legacy-proof' })); + + await assert.rejects( + store.markMessagesHandedOff({ + sessionId: 'session-no-legacy-proof', + messageIds: ['message-no-legacy-proof'], + turnId: 'turn-no-legacy-proof', + }), + /Message admission does not exist/, + ); + } finally { + store.close(); + } + }); + + test('rejects a proven Root handoff for a cancelled admission', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-legacy-cancelled' })); + await store.commitMessageAdmission({ + sessionId: 'session-legacy-cancelled', + turnId: 'turn-legacy-cancelled', + runId: 'run-legacy-cancelled', + messageId: 'message-legacy-cancelled', + content: { text: 'cancelled' }, + submittedContentDigest: messageContentDigest({ text: 'cancelled' }), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 19, + }); + await store.cancelMessageAdmissions('session-legacy-cancelled', ['message-legacy-cancelled']); + + await assert.rejects( + markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-legacy-cancelled', + messageIds: ['message-legacy-cancelled'], + turnId: 'turn-legacy-cancelled', + provenRootMessages: [ + { + messageId: 'message-legacy-cancelled', + content: { text: 'cancelled' }, + admittedAt: 19, + }, + ], + }), + /already cancelled/, + ); + } finally { + store.close(); + } + }); + + test('rejects proven Root repeats with an existing transcript content or Turn conflict', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-legacy-conflict' })); + await markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-legacy-conflict', + messageIds: ['message-legacy-conflict'], + turnId: 'turn-legacy-conflict', + provenRootMessages: [ + { + messageId: 'message-legacy-conflict', + content: { text: 'canonical text' }, + admittedAt: 20, + }, + ], + }); + + await assert.rejects( + markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-legacy-conflict', + messageIds: ['message-legacy-conflict'], + turnId: 'turn-legacy-conflict', + provenRootMessages: [ + { + messageId: 'message-legacy-conflict', + content: { text: 'different text' }, + admittedAt: 20, + }, + ], + }), + /transcript identity conflict/, + ); + await assert.rejects( + markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-legacy-conflict', + messageIds: ['message-legacy-conflict'], + turnId: 'turn-legacy-conflict-different', + provenRootMessages: [ + { + messageId: 'message-legacy-conflict', + content: { text: 'canonical text' }, + admittedAt: 20, + }, + ], + }), + /transcript Turn conflict/, + ); + } finally { + store.close(); + } + }); + + test('keeps an admission as the content and timestamp authority during handoff', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-admission-authority' })); + await store.commitMessageAdmission({ + sessionId: 'session-admission-authority', + turnId: 'turn-admission-authority', + runId: 'run-admission-authority', + messageId: 'message-admission-authority', + content: { text: 'admission authority', displayText: 'submitted display' }, + submittedContentDigest: messageContentDigest({ text: 'admission authority' }), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 21, + }); + + await markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-admission-authority', + messageIds: ['message-admission-authority'], + turnId: 'turn-admission-authority', + provenRootMessages: [ + { + messageId: 'message-admission-authority', + content: { text: 'admission authority', displayText: 'submitted display' }, + admittedAt: 99, + }, + ], + }); + + assert.deepEqual( + (await store.readMessages('session-admission-authority')).map((message) => ({ + text: message.type === 'user' ? message.text : undefined, + ts: message.ts, + })), + [{ text: 'admission authority', ts: 21 }], + ); + assert.deepEqual(await store.listMessageAdmissions('session-admission-authority'), []); + } finally { + store.close(); + } + }); + + test('rejects proven Root fallback content that drifts from an admission', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-admission-drift' })); + await store.commitMessageAdmission({ + sessionId: 'session-admission-drift', + turnId: 'turn-admission-drift', + runId: 'run-admission-drift', + messageId: 'message-admission-drift', + content: { text: 'admitted content' }, + submittedContentDigest: messageContentDigest({ text: 'admitted content' }), + submittedPlacement: 'current_turn', + placement: 'current_turn', + disposition: 'steering', + admittedAt: 22, + }); + + await assert.rejects( + markMessagesHandedOffWithProvenRoots(store, { + sessionId: 'session-admission-drift', + messageIds: ['message-admission-drift'], + turnId: 'turn-admission-drift', + provenRootMessages: [ + { + messageId: 'message-admission-drift', + content: { text: 'drifted content' }, + admittedAt: 22, + }, + ], + }), + /fallback content conflict/, + ); + assert.deepEqual(await store.readMessages('session-admission-drift'), []); + assert.equal((await store.listMessageAdmissions('session-admission-drift')).length, 1); + } finally { + store.close(); + } + }); + + test('validates proven Root fallback identities and timestamps before handoff', async () => { + const store = createSqliteSessionMetadataStore(':memory:'); + try { + await store.create(fullHeader({ id: 'session-legacy-validation' })); + const base = { + sessionId: 'session-legacy-validation', + messageIds: ['message-legacy-validation'], + turnId: 'turn-legacy-validation', + }; + + await assert.rejects( + markMessagesHandedOffWithProvenRoots(store, { + ...base, + provenRootMessages: [ + { + messageId: 'message-legacy-validation', + content: { text: 'first' }, + admittedAt: 23, + }, + { + messageId: 'message-legacy-validation', + content: { text: 'second' }, + admittedAt: 24, + }, + ], + }), + /duplicate identities/, + ); + await assert.rejects( + markMessagesHandedOffWithProvenRoots(store, { + ...base, + provenRootMessages: [ + { + messageId: 'message-not-requested', + content: { text: 'not requested' }, + admittedAt: 23, + }, + ], + }), + /not present in messageIds/, + ); + await assert.rejects( + markMessagesHandedOffWithProvenRoots(store, { + ...base, + provenRootMessages: [ + { + messageId: 'message-legacy-validation', + content: { text: 'bad timestamp' }, + admittedAt: -1, + }, + ], + }), + /timestamp/, + ); + await assert.rejects( + markMessagesHandedOffWithProvenRoots(store, { + ...base, + provenRootMessages: [ + { + messageId: 'message-not-requested', + content: { text: 23 } as unknown as MessageContent, + admittedAt: 23, + }, + ], + }), + /Invalid MessageContent/, + ); + } finally { + store.close(); + } + }); + test('removes the accepted payload after transcript handoff', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-message-handoff-')); const path = join(root, 'state.sqlite'); @@ -3365,6 +4085,17 @@ function fullHeader(overrides: Partial = {}): SessionHeader { }; } +type ProvenRootHandoffInput = MarkMessagesHandedOffInput & { + readonly provenRootMessages: readonly ProvenRootMessageHandoff[]; +}; + +async function markMessagesHandedOffWithProvenRoots( + store: ReturnType, + input: ProvenRootHandoffInput, +): Promise { + return store.markMessagesHandedOff(input); +} + function graphRootHeader(id: string): SessionHeader { return fullHeader({ id, diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index 14462e5363..db18c8938d 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -111,6 +111,7 @@ export type { RuntimeEventScanResult, } from './agent-run-store.js'; export type { + MarkMessagesHandedOffInput, MessageAdmissionStore, PendingMessageAdmission, } from './message-admission-store.js'; diff --git a/packages/storage/src/message-admission-store.ts b/packages/storage/src/message-admission-store.ts index f5512adcc4..8bfacbb2e8 100644 --- a/packages/storage/src/message-admission-store.ts +++ b/packages/storage/src/message-admission-store.ts @@ -18,7 +18,11 @@ */ import { isDeepStrictEqual } from 'node:util'; -import { normalizeMessageContent, type MessageContent } from '@maka/core/events'; +import { + decodeMessageContent, + normalizeMessageContent, + type MessageContent, +} from '@maka/core/events'; import { normalizeSubmittedTurnIntent, submittedTurnIntentsEqual, @@ -49,6 +53,19 @@ export interface PendingMessageAdmission { readonly admittedAt: number; } +export interface ProvenRootMessageHandoff { + readonly messageId: string; + readonly content: MessageContent; + readonly admittedAt: number; +} + +export interface MarkMessagesHandedOffInput { + readonly sessionId: string; + readonly messageIds: readonly string[]; + readonly turnId: string; + readonly provenRootMessages?: readonly ProvenRootMessageHandoff[]; +} + export interface MessageAdmissionStore { commitMessageAdmission(admission: PendingMessageAdmission): Promise; readMessageAdmission( @@ -62,11 +79,7 @@ export interface MessageAdmissionStore { */ hasCancelledMessageAdmission(sessionId: string, messageId: string): Promise; listMessageAdmissions(sessionId: string): Promise; - markMessagesHandedOff(input: { - sessionId: string; - messageIds: readonly string[]; - turnId: string; - }): Promise; + markMessagesHandedOff(input: MarkMessagesHandedOffInput): Promise; updateMessageAdmission(admission: PendingMessageAdmission): Promise; reorderMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise; cancelMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise; @@ -108,6 +121,19 @@ export function normalizePendingMessageAdmission( return normalized; } +export function normalizeProvenRootMessageHandoff( + handoff: ProvenRootMessageHandoff, +): ProvenRootMessageHandoff { + assertSafeId(handoff.messageId, 'Invalid proven Root Message identity'); + if (!Number.isSafeInteger(handoff.admittedAt) || handoff.admittedAt < 0) { + throw new Error('Invalid proven Root Message timestamp'); + } + return Object.freeze({ + ...handoff, + content: decodeMessageContent(handoff.content), + }); +} + export function samePendingMessageAdmission( left: PendingMessageAdmission, right: PendingMessageAdmission, diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index 10d045a4d5..0207af2590 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -83,7 +83,11 @@ import { type TurnStateMessage, type UserMessage, } from '@maka/core/session'; -import type { MessageAdmissionStore, PendingMessageAdmission } from './message-admission-store.js'; +import type { + MarkMessagesHandedOffInput, + MessageAdmissionStore, + PendingMessageAdmission, +} from './message-admission-store.js'; import { isVisibleSessionMessage, lastMessagePreviewForMessages, @@ -898,11 +902,7 @@ class SqliteSessionStore implements SessionAuthorityStore { return this.metadata.listMessageAdmissions(sessionId); } - async markMessagesHandedOff(input: { - sessionId: string; - messageIds: readonly string[]; - turnId: string; - }): Promise { + async markMessagesHandedOff(input: MarkMessagesHandedOffInput): Promise { await this.ensureReady(); await this.metadata.markMessagesHandedOff(input); for (const listener of this.transcriptChangeListeners) listener(input.sessionId); diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index 59b34eae88..18d18a92bb 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -98,8 +98,11 @@ import { import { markPersisted } from '@maka/core/persisted-value'; import { normalizePendingMessageAdmission, + normalizeProvenRootMessageHandoff, samePendingMessageAdmission, + type MarkMessagesHandedOffInput, type PendingMessageAdmission, + type ProvenRootMessageHandoff, } from './message-admission-store.js'; import { normalizeSubmittedTurnIntent } from './submitted-turn-intent.js'; import { messageContentsEqual, normalizeMessageContent } from '@maka/core/events'; @@ -1699,16 +1702,26 @@ export class SqliteSessionMetadataStore { }); } - async markMessagesHandedOff(input: { - sessionId: string; - messageIds: readonly string[]; - turnId: string; - }): Promise { + async markMessagesHandedOff(input: MarkMessagesHandedOffInput): Promise { this.assertOpen(); assertSafeSessionId(input.sessionId); assertSafeSessionId(input.turnId); const unique = [...new Set(input.messageIds)]; for (const messageId of unique) assertSafeSessionId(messageId); + const requestedMessageIds = new Set(unique); + const provenRootMessages = new Map(); + for (const fallback of input.provenRootMessages ?? []) { + const normalized = normalizeProvenRootMessageHandoff(fallback); + if (!requestedMessageIds.has(normalized.messageId)) { + throw new SessionMetadataConflictError( + 'Proven Root Message identity is not present in messageIds', + ); + } + if (provenRootMessages.has(normalized.messageId)) { + throw new SessionMetadataConflictError('Proven Root Messages contain duplicate identities'); + } + provenRootMessages.set(normalized.messageId, normalized); + } this.transaction(() => { const lastSequenceRow = this.db .prepare( @@ -1717,13 +1730,23 @@ export class SqliteSessionMetadataStore { .get(input.sessionId) as { last_sequence?: unknown }; if ( typeof lastSequenceRow.last_sequence !== 'number' || - !Number.isSafeInteger(lastSequenceRow.last_sequence) + !Number.isSafeInteger(lastSequenceRow.last_sequence) || + lastSequenceRow.last_sequence < -1 ) { throw new SessionMetadataConflictError('Invalid Session message sequence'); } - let nextSequence = lastSequenceRow.last_sequence + 1; - const materialized: StoredMessage[] = []; + const lastSequence = lastSequenceRow.last_sequence; + const historicalMissingMessages = new Map< + string, + { readonly message: StoredMessage; readonly json: string } + >(); + const ordinaryMissingMessages = new Map< + string, + { readonly message: StoredMessage; readonly json: string } + >(); + const existingSequences = new Map(); for (const messageId of unique) { + const fallback = provenRootMessages.get(messageId); const admissionRow = this.db .prepare( ` @@ -1738,6 +1761,20 @@ export class SqliteSessionMetadataStore { const admission = admissionRow ? decodeMessageAdmissionRow(input.sessionId, admissionRow) : undefined; + if ( + admission !== undefined && + admission.turnId !== input.turnId && + admission.disposition !== 'followup' + ) { + throw new SessionMetadataConflictError('Message admission Turn conflict'); + } + if ( + admission !== undefined && + fallback !== undefined && + !messageContentsEqual(admission.content, fallback.content) + ) { + throw new SessionMetadataConflictError('Message admission fallback content conflict'); + } if ( !admission && this.db @@ -1770,20 +1807,21 @@ export class SqliteSessionMetadataStore { ); } if (rows.length === 0) { - if (!admission) - throw new SessionMetadataConflictError('Message admission does not exist'); + const source = admission ?? fallback; + if (!source) throw new SessionMetadataConflictError('Message admission does not exist'); const message = decodeCanonicalMessage({ type: 'user', id: messageId, turnId: input.turnId, - ts: admission.admittedAt, - ...admission.content, + ts: source.admittedAt, + ...source.content, steeringEventId: messageId, }); const json = JSON.stringify(message); - this.insertSessionMessagesSync(input.sessionId, nextSequence, [{ message, json }]); - nextSequence += 1; - materialized.push(message); + (fallback ? historicalMissingMessages : ordinaryMissingMessages).set(messageId, { + message, + json, + }); } else { const sequence = rows[0]?.sequence; if (typeof sequence !== 'number' || !Number.isSafeInteger(sequence)) { @@ -1795,8 +1833,11 @@ export class SqliteSessionMetadataStore { if ( message.type !== 'user' || message.id !== messageId || - (admission !== undefined && - !messageContentsEqual(normalizeMessageContent(message), admission.content)) + ((admission !== undefined || fallback !== undefined) && + !messageContentsEqual( + normalizeMessageContent(message), + (admission ?? fallback)!.content, + )) ) { throw new SessionMetadataConflictError( 'Message admission transcript identity conflict', @@ -1805,6 +1846,7 @@ export class SqliteSessionMetadataStore { if (message.turnId !== input.turnId) { throw new SessionMetadataConflictError('Message admission transcript Turn conflict'); } + existingSequences.set(messageId, sequence); } if (admission) { const deleted = this.db @@ -1815,17 +1857,120 @@ export class SqliteSessionMetadataStore { } } } - const latest = materialized.at(-1); - if (latest?.type === 'user') { + let tailLatest: StoredMessage | undefined; + if (provenRootMessages.size > 0) { + const transcript = this.readSessionMessageOrderingSync(input.sessionId); + let previousExistingSequence = -1; + const historicalMessageIds = unique.filter((messageId) => + provenRootMessages.has(messageId), + ); + const historicalMessageIdSet = new Set(historicalMessageIds); + for (const messageId of historicalMessageIds) { + const sequence = existingSequences.get(messageId); + if (sequence === undefined) continue; + const admittedAt = provenRootMessages.get(messageId)!.admittedAt; + const blockingRow = transcript.find( + ({ sequence: candidateSequence, message }) => + candidateSequence > previousExistingSequence && + candidateSequence < sequence && + !historicalMessageIdSet.has(message.id) && + (message.turnId === input.turnId || message.ts >= admittedAt), + ); + if (sequence <= previousExistingSequence || blockingRow !== undefined) { + throw new SessionMetadataConflictError( + 'Message admission transcript source order conflict', + ); + } + previousExistingSequence = sequence; + } + + const insertionGroups: Array<{ + readonly boundary: number; + readonly entries: Array<{ readonly message: StoredMessage; readonly json: string }>; + }> = []; + let pending: Array<{ readonly message: StoredMessage; readonly json: string }> = []; + let previousAnchor = -1; + for (const messageId of historicalMessageIds) { + const missing = historicalMissingMessages.get(messageId); + if (missing) { + pending.push(missing); + continue; + } + const boundary = existingSequences.get(messageId); + if (pending.length > 0 && boundary !== undefined) { + const admittedAt = Math.min(...pending.map(({ message }) => message.ts)); + const earlierBoundary = transcript.find( + ({ sequence, message }) => + sequence > previousAnchor && + sequence < boundary && + !historicalMessageIdSet.has(message.id) && + (message.turnId === input.turnId || message.ts >= admittedAt), + ); + if (earlierBoundary) { + throw new SessionMetadataConflictError( + 'Message admission transcript source order conflict', + ); + } + insertionGroups.push({ boundary, entries: pending }); + pending = []; + } + if (boundary !== undefined) previousAnchor = boundary; + } + if (pending.length > 0) { + const admittedAt = Math.min(...pending.map(({ message }) => message.ts)); + const repairBoundary = transcript.find( + ({ sequence, message }) => + sequence > previousAnchor && + !historicalMessageIdSet.has(message.id) && + (message.turnId === input.turnId || message.ts >= admittedAt), + )?.sequence; + insertionGroups.push({ + boundary: repairBoundary ?? lastSequence + 1, + entries: pending, + }); + } + + for (const group of insertionGroups.reverse()) { + this.shiftSessionMessageSuffixSync(input.sessionId, group.boundary, group.entries.length); + this.insertSessionMessagesSync(input.sessionId, group.boundary, group.entries); + if (group.boundary === lastSequence + 1) { + tailLatest = group.entries.at(-1)?.message; + } + } + } + if (ordinaryMissingMessages.size > 0) { + const ordinaryEntries = unique.flatMap((messageId) => { + const entry = ordinaryMissingMessages.get(messageId); + return entry ? [entry] : []; + }); + const currentLastSequenceRow = this.db + .prepare( + 'SELECT COALESCE(MAX(sequence), -1) AS last_sequence FROM session_messages WHERE session_id = ?', + ) + .get(input.sessionId) as { last_sequence?: unknown }; + const currentLastSequence = currentLastSequenceRow.last_sequence; + if ( + typeof currentLastSequence !== 'number' || + !Number.isSafeInteger(currentLastSequence) || + currentLastSequence < -1 + ) { + throw new SessionMetadataConflictError('Invalid Session message sequence'); + } + this.insertSessionMessagesSync(input.sessionId, currentLastSequence + 1, ordinaryEntries); + tailLatest = ordinaryEntries.at(-1)?.message; + } + if (tailLatest?.type === 'user') { this.updateCatalogProjectionSync( input.sessionId, { - lastMessageAt: latest.ts, - lastMessagePreview: catalogPreviewForUserMessage(latest), + lastMessageAt: tailLatest.ts, + lastMessagePreview: catalogPreviewForUserMessage(tailLatest), }, false, true, ); + } else if (historicalMissingMessages.size > 0 || ordinaryMissingMessages.size > 0) { + this.updateCatalogProjectionSync(input.sessionId, {}, false, true); } }); } @@ -4634,6 +4779,84 @@ export class SqliteSessionMetadataStore { return row ? decodeRecord(row) : undefined; } + private readSessionMessageOrderingSync( + sessionId: string, + ): Array<{ readonly sequence: number; readonly message: StoredMessage }> { + const rows = this.db + .prepare( + ` + SELECT message.sequence, message.record_json, payload.record_bytes, payload.sha256 + FROM session_messages AS message + LEFT JOIN session_message_payloads AS payload + ON payload.session_id = message.session_id AND payload.sequence = message.sequence + WHERE message.session_id = ? + ORDER BY message.sequence + `, + ) + .all(sessionId) as StoredSessionMessagePayloadRow[]; + return rows.map((row) => { + const sequence = requireStoredMessageSequence(row.sequence, sessionId); + const recordJson = readStoredMessageRecordJson(this.db, sessionId, sequence, row); + return { + sequence, + message: decodeStoredMessage(JSON.parse(recordJson) as unknown), + }; + }); + } + + private shiftSessionMessageSuffixSync( + sessionId: string, + firstSequence: number, + amount: number, + ): void { + if (!Number.isSafeInteger(firstSequence) || firstSequence < 0) { + throw new SessionMetadataConflictError('Invalid transcript insertion sequence'); + } + if (!Number.isSafeInteger(amount) || amount < 1) { + throw new SessionMetadataConflictError('Invalid transcript insertion size'); + } + const sequences = ( + this.db + .prepare( + ` + SELECT sequence + FROM session_messages + WHERE session_id = ? AND sequence >= ? + ORDER BY sequence DESC + `, + ) + .all(sessionId, firstSequence) as Array<{ sequence?: unknown }> + ).map((row) => requireStoredMessageSequence(row.sequence, sessionId)); + const highest = sequences[0]; + if (highest !== undefined && highest > Number.MAX_SAFE_INTEGER - amount) { + throw new SessionMetadataConflictError('Session message sequence overflow'); + } + if (sequences.length === 0) return; + + this.db.exec('PRAGMA defer_foreign_keys = ON'); + const moveChunks = this.db.prepare( + 'UPDATE session_message_chunks SET sequence = ? WHERE session_id = ? AND sequence = ?', + ); + const movePayload = this.db.prepare( + 'UPDATE session_message_payloads SET sequence = ? WHERE session_id = ? AND sequence = ?', + ); + const moveMessage = this.db.prepare( + 'UPDATE session_messages SET sequence = ? WHERE session_id = ? AND sequence = ?', + ); + for (const sequence of sequences) { + const shifted = sequence + amount; + moveChunks.run(shifted, sessionId, sequence); + const payload = movePayload.run(shifted, sessionId, sequence); + if (payload.changes !== 0 && payload.changes !== 1) { + throw new SessionMetadataConflictError('Message payload sequence is ambiguous'); + } + const message = moveMessage.run(shifted, sessionId, sequence); + if (message.changes !== 1) { + throw new SessionMetadataConflictError('Message transcript sequence changed during repair'); + } + } + } + private insertSessionMessagesSync( sessionId: string, firstSequence: number, @@ -4642,6 +4865,13 @@ export class SqliteSessionMetadataStore { readonly json: string; }[], ): void { + if ( + !Number.isSafeInteger(firstSequence) || + firstSequence < 0 || + entries.length > Number.MAX_SAFE_INTEGER - firstSequence + 1 + ) { + throw new SessionMetadataConflictError('Session message sequence overflow'); + } const insertMessage = this.db.prepare(` INSERT INTO session_messages( session_id, sequence, message_id, message_type, message_ts, record_json