You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/08/29 14:55:42 UTC
(maka) branch main updated: fix(runtime-host): recover legacy Root source messages (#3923)
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/main by this push:
new 6b75a725c3 fix(runtime-host): recover legacy Root source messages (#3923)
6b75a725c3 is described below
commit 6b75a725c32da19e4300e1fdd18c7145d3acef8b
Author: ooopSok <81...@users.noreply.github.com>
AuthorDate: Sat Aug 29 22:55:38 2026 +0800
fix(runtime-host): recover legacy Root source messages (#3923)
Recover trusted legacy Root source messages that predate message-admission persistence. Validate exact receipt ownership, materialize missing transcript entries atomically in source order, and limit transcript scans to genuine legacy recovery.
Closes #3902
Generated-by: OpenAI Codex
---
.../src/__tests__/execution-host-recovery.test.ts | 145 ++++
.../src/__tests__/fixtures/execution-host-suite.ts | 138 ++++
.../src/__tests__/message-coordinator.test.ts | 119 +++-
.../src/server/hosted-execution-recovery.ts | 29 +-
.../runtime-host/src/server/message-coordinator.ts | 65 +-
.../sqlite-session-metadata-store.test.ts | 742 ++++++++++++++++++++-
packages/storage/src/execution-stores.ts | 1 +
packages/storage/src/message-admission-store.ts | 38 +-
packages/storage/src/session-store.ts | 12 +-
.../storage/src/sqlite-session-metadata-store.ts | 275 +++++++-
10 files changed, 1498 insertions(+), 66 deletions(-)
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 a60d5adf2a..32f8ea3ffb 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,
@@ -736,6 +739,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<ReturnType<typeof openInteractiveExecutionStoresForWrite>> | 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<void> {
const owner = await tryAcquireInteractiveRootOwner(this.capability);
assert.ok(owner);
@@ -1021,6 +1145,20 @@ export class ExecutionFixture {
}
}
+ async readTurnRuns(turnId: string) {
+ const reader = await acquireReader(this.capability);
+ let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForRead>> | 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<Array<Extract<StoredMessage, { type: 'user' }>>> {
const reader = await acquireReader(this.capability);
let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForRead>> | undefined;
diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts
index 9b30146285..576ee0d51f 100644
--- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts
@@ -29,7 +29,9 @@ import {
WORKHUB_COORDINATION_SESSION_ID,
WORKHUB_COORDINATION_SESSION_ROLE,
} from '@maka/core/session';
+import { RuntimeMessageAuthorityInvariantError } from '@maka/runtime/message-authority';
import type {
+ MarkMessagesHandedOffInput,
MessageAdmissionStore,
PendingMessageAdmission,
RootTurnSourceMessageReceipt,
@@ -1888,10 +1890,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({
@@ -2476,7 +2559,12 @@ function createFixture(
state: 'accepted' | 'handed_off' | 'executed' | 'cancelled';
}
>();
- const admissions = admissionsOverride ?? memoryMessageAdmissionStore(messageAdmissions);
+ const handoffCalls: MarkMessagesHandedOffInput[] = [];
+ const admissions =
+ admissionsOverride ??
+ memoryMessageAdmissionStore(messageAdmissions, (input) => {
+ handoffCalls.push(input);
+ });
const sessionAdmission = new SessionAdmissionGate();
const stopClaimed = deferred<void>();
const terminal = deferred<TurnSnapshot>();
@@ -2618,6 +2706,7 @@ function createFixture(
events,
receipts,
recoveredBatches,
+ handoffCalls,
readMessageAdmission: (messageId: string) => messageAdmissions.get(messageId)?.admission,
stopClaimed,
resolveTerminal: terminal.resolve,
@@ -2642,6 +2731,7 @@ function memoryMessageAdmissionStore(
state: 'accepted' | 'handed_off' | 'executed' | 'cancelled';
}
>,
+ onMessagesHandedOff?: (input: MarkMessagesHandedOffInput) => void,
): MessageAdmissionStore {
return {
commitMessageAdmission: async (admission) => {
@@ -2671,7 +2761,9 @@ function memoryMessageAdmissionStore(
}
}
},
- markMessagesHandedOff: async ({ messageIds }) => {
+ markMessagesHandedOff: async (input) => {
+ onMessagesHandedOff?.(input);
+ const { messageIds } = input;
for (const messageId of messageIds) admissions.delete(messageId);
},
};
@@ -2742,6 +2834,29 @@ function sourceReceipt(
};
}
+function matchingSourceReceipt(
+ messageId: string,
+ admittedAt: number,
+ overrides: Partial<RootTurnSourceMessageReceipt['admission']> = {},
+): 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 4efa59bdc9..1fac7a4c72 100644
--- a/packages/runtime-host/src/server/hosted-execution-recovery.ts
+++ b/packages/runtime-host/src/server/hosted-execution-recovery.ts
@@ -106,7 +106,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`);
@@ -316,12 +316,35 @@ function verifyOrRecoverUserMessage(
indexRecoveryMessage(index, recoveredMessage);
}
-function verifyQueueSourceMessages(
+async function verifyQueueSourceMessages(
admission: RootTurnAdmission,
index: RecoveryMessageIndex,
-): void {
+ proofReader: Pick<
+ ExecutionStoresWriter<'interactive'>['agentRunStore'],
+ 'readRootTurnSourceMessageReceipt'
+ >,
+): Promise<void> {
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 0f3ec3f083..8df4f86848 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,
@@ -639,27 +640,18 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
messageIds: readonly string[];
}): Promise<void> {
const handoff: string[] = [];
+ const provenRootMessages: Array<
+ NonNullable<MarkMessagesHandedOffInput['provenRootMessages']>[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 } : {}),
});
}
@@ -671,19 +663,13 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
messageIds: readonly string[];
}): Promise<void> {
const messageIds = new Set<string>();
+ const provenRootMessages: Array<
+ NonNullable<MarkMessagesHandedOffInput['provenRootMessages']>[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 (
@@ -705,9 +691,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<NonNullable<MarkMessagesHandedOffInput['provenRootMessages']>[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<void> {
await this.#admissions.cancelMessageAdmissions(sessionId, messageIds);
}
@@ -761,7 +774,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 363688af7c..bd8a6aac81 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,
@@ -434,6 +438,729 @@ 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 handoff append semantics when Root proof is also supplied', 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 markMessagesHandedOffWithProvenRoots(store, {
+ sessionId: 'session-ordinary-handoff-order',
+ messageIds: ['message-ordinary-admission'],
+ turnId: 'turn-ordinary-handoff-order',
+ provenRootMessages: [
+ {
+ messageId: 'message-ordinary-admission',
+ content: { text: 'ordinary admission' },
+ admittedAt: 10,
+ },
+ ],
+ });
+
+ 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');
@@ -3422,6 +4149,17 @@ function fullHeader(overrides: Partial<SessionHeader> = {}): SessionHeader {
};
}
+type ProvenRootHandoffInput = MarkMessagesHandedOffInput & {
+ readonly provenRootMessages: readonly ProvenRootMessageHandoff[];
+};
+
+async function markMessagesHandedOffWithProvenRoots(
+ store: ReturnType<typeof createSqliteSessionMetadataStore>,
+ input: ProvenRootHandoffInput,
+): Promise<void> {
+ 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 56cd4ef14c..047e5ca4eb 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<PendingMessageAdmission>;
readMessageAdmission(
@@ -62,11 +79,7 @@ export interface MessageAdmissionStore {
*/
hasCancelledMessageAdmission(sessionId: string, messageId: string): Promise<boolean>;
listMessageAdmissions(sessionId: string): Promise<readonly PendingMessageAdmission[]>;
- markMessagesHandedOff(input: {
- sessionId: string;
- messageIds: readonly string[];
- turnId: string;
- }): Promise<void>;
+ markMessagesHandedOff(input: MarkMessagesHandedOffInput): Promise<void>;
updateMessageAdmission(admission: PendingMessageAdmission): Promise<void>;
reorderMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise<void>;
cancelMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise<void>;
@@ -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 5f1e4819e7..5605ed6ccc 100644
--- a/packages/storage/src/session-store.ts
+++ b/packages/storage/src/session-store.ts
@@ -85,7 +85,11 @@ import {
type UserMessage,
type WorkHubDelegationAssignedMessage,
} 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,
@@ -976,11 +980,7 @@ class SqliteSessionStore implements SessionAuthorityStore {
return this.metadata.listMessageAdmissions(sessionId);
}
- async markMessagesHandedOff(input: {
- sessionId: string;
- messageIds: readonly string[];
- turnId: string;
- }): Promise<void> {
+ async markMessagesHandedOff(input: MarkMessagesHandedOffInput): Promise<void> {
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 f2f92162d0..07d238eb16 100644
--- a/packages/storage/src/sqlite-session-metadata-store.ts
+++ b/packages/storage/src/sqlite-session-metadata-store.ts
@@ -101,8 +101,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 {
@@ -1878,16 +1881,26 @@ export class SqliteSessionMetadataStore {
});
}
- async markMessagesHandedOff(input: {
- sessionId: string;
- messageIds: readonly string[];
- turnId: string;
- }): Promise<void> {
+ async markMessagesHandedOff(input: MarkMessagesHandedOffInput): Promise<void> {
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<string, ProvenRootMessageHandoff>();
+ 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(
@@ -1896,13 +1909,24 @@ 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 historicalMessageIdSet = new Set<string>();
+ const existingSequences = new Map<string, number>();
for (const messageId of unique) {
+ const fallback = provenRootMessages.get(messageId);
const admissionRow = this.db
.prepare(
`
@@ -1917,6 +1941,23 @@ 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 === undefined && fallback !== undefined) {
+ historicalMessageIdSet.add(messageId);
+ }
if (
!admission &&
this.db
@@ -1949,20 +1990,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);
+ (historicalMessageIdSet.has(messageId)
+ ? historicalMissingMessages
+ : ordinaryMissingMessages
+ ).set(messageId, { message, json });
} else {
const sequence = rows[0]?.sequence;
if (typeof sequence !== 'number' || !Number.isSafeInteger(sequence)) {
@@ -1974,8 +2016,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',
@@ -1984,6 +2029,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
@@ -1994,17 +2040,119 @@ export class SqliteSessionMetadataStore {
}
}
}
- const latest = materialized.at(-1);
- if (latest?.type === 'user') {
+ let tailLatest: StoredMessage | undefined;
+ if (historicalMessageIdSet.size > 0) {
+ const transcript = this.readSessionMessageOrderingSync(input.sessionId);
+ let previousExistingSequence = -1;
+ const historicalMessageIds = unique.filter((messageId) =>
+ historicalMessageIdSet.has(messageId),
+ );
+ 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);
}
});
}
@@ -4828,6 +4976,84 @@ export class SqliteSessionMetadataStore {
return row ? decodeStoredMessageRecordRow(this.db, sessionId, 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,
@@ -4836,6 +5062,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
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org