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/24 19:27:37 UTC
(maka) 02/21: feat(runtime-host): wire durable message lifecycle
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch fix/durable-message-lifecycle
in repository https://gitbox.apache.org/repos/asf/maka.git
commit a1011c7317198cd8e6b583312cf304ca3aec838d
Author: AstroHan <le...@outlook.com>
AuthorDate: Mon Aug 24 21:41:14 2026 +0800
feat(runtime-host): wire durable message lifecycle
Generated-by: Codex
---
.../src/__tests__/execution-host-queue.test.ts | 80 +++++++
.../src/__tests__/fixtures/execution-host-suite.ts | 12 +
.../src/server/execution-composition.ts | 4 +
.../runtime-host/src/server/message-coordinator.ts | 194 ++++++++++++++++
.../src/server/root-turn-coordinator.ts | 128 ++++++++++-
packages/runtime/src/agent-run.ts | 26 ++-
.../sqlite-session-metadata-store.test.ts | 14 ++
packages/storage/src/execution-stores.ts | 24 ++
packages/storage/src/message-receipt-store.ts | 18 ++
packages/storage/src/session-store.ts | 60 ++++-
.../storage/src/sqlite-session-metadata-store.ts | 253 +++++++++++++++++++++
11 files changed, 799 insertions(+), 14 deletions(-)
diff --git a/packages/runtime-host/src/__tests__/execution-host-queue.test.ts b/packages/runtime-host/src/__tests__/execution-host-queue.test.ts
index 6cc97ee96..77703f95e 100644
--- a/packages/runtime-host/src/__tests__/execution-host-queue.test.ts
+++ b/packages/runtime-host/src/__tests__/execution-host-queue.test.ts
@@ -205,6 +205,7 @@ test('subscribed Clients share one canonical queue and ordered root handoff', as
await waitForTerminalTurn(tui, fixture.sessionId, successor.snapshot.rootTurn.turnId);
await tui.close();
await fixture.stopHost(host);
+ assert.equal(await fixture.readMessageLifecycleState(followupId), 'handed_off');
const chain = await fixture.readAdmissionChain();
assert.deepEqual(
@@ -215,6 +216,85 @@ test('subscribed Clients share one canonical queue and ordered root handoff', as
});
});
+test('production UDS admission commits one transcript before the root handoff', async () => {
+ await withExecutionRoot(async (fixture) => {
+ const host = await fixture.startHost();
+ const client = await connectClient(fixture.root);
+ const messageId = randomUUID();
+ const started = await client.request('turn.message.submit', {
+ originHostEpoch: host.hostEpoch,
+ sessionId: fixture.sessionId,
+ messageId,
+ content: { text: FAKE_ASK_USER_QUESTION_PROMPT },
+ placement: 'current_turn',
+ });
+ assert.equal(started.disposition, 'turn_started');
+ if (started.disposition !== 'turn_started') return;
+ const active = await client.queryTurn({ sessionId: fixture.sessionId, turnId: started.turnId });
+ await client.stopTurn({
+ sessionId: fixture.sessionId,
+ turnId: started.turnId,
+ runId: active.runId,
+ });
+ await client.close();
+ await fixture.stopHost(host);
+ const ledger = await fixture.readTurn(started.turnId);
+ assert.deepEqual(
+ ledger.userMessages.filter((message) => message.id === messageId).map((message) => message.id),
+ [messageId],
+ );
+ assert.equal(await fixture.readMessageLifecycleState(messageId), 'handed_off');
+ });
+});
+
+test('a Host crash after queue admission recovers the durable successor once', async () => {
+ await withExecutionRoot(async (fixture) => {
+ const firstHost = await fixture.startHost();
+ const first = await connectClient(fixture.root);
+ const started = requireStartedTurn(
+ await first.startTurn({
+ sessionId: fixture.sessionId,
+ turnId: randomUUID(),
+ content: { text: FAKE_ASK_USER_QUESTION_PROMPT },
+ }),
+ );
+ const messageId = randomUUID();
+ const queued = await first.request('turn.message.submit', {
+ originHostEpoch: firstHost.hostEpoch,
+ sessionId: fixture.sessionId,
+ messageId,
+ content: { text: 'recover this accepted successor' },
+ placement: 'next_turn',
+ });
+ assert.equal(queued.disposition, 'followup');
+ await fixture.killHost(firstHost);
+ await first.closed;
+
+ const secondHost = await fixture.startHost();
+ const second = await connectClient(fixture.root);
+ const subscription = await second.openSessionSubscription({
+ sessionId: fixture.sessionId,
+ transcript: { kind: 'none' },
+ });
+ const probe = new SubscriptionProbe(subscription);
+ const successor = await probe.waitFor(
+ (frame) =>
+ frame.kind === 'subscription.session_projection' &&
+ frame.snapshot.rootTurn !== null &&
+ frame.snapshot.rootTurn.turnId !== started.turnId,
+ 'durable successor was not recovered after the Host crash',
+ );
+ assert.equal(successor.kind, 'subscription.session_projection');
+ if (successor.kind !== 'subscription.session_projection' || !successor.snapshot.rootTurn) return;
+ await waitForTerminalTurn(second, fixture.sessionId, successor.snapshot.rootTurn.turnId);
+ await subscription.close();
+ await probe.done;
+ await second.close();
+ await fixture.stopHost(secondHost);
+ assert.equal(await fixture.readMessageLifecycleState(messageId), 'handed_off');
+ });
+});
+
test('concurrent root admission for one Session has a single winner', 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 f44e88b8f..23d8b942a 100644
--- a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts
+++ b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts
@@ -957,6 +957,18 @@ export class ExecutionFixture {
}
}
+ async readMessageLifecycleState(messageId: string) {
+ const reader = await acquireReader(this.capability);
+ let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForRead>> | undefined;
+ try {
+ stores = await openInteractiveExecutionStoresForRead(reader.lease);
+ return await stores.sessionStore.readMessageLifecycleState(this.sessionId, messageId);
+ } finally {
+ await stores?.sessionStore.close?.();
+ await reader.close();
+ }
+ }
+
async readTurnFootprint(turnId: string): Promise<{
admitted: boolean;
runCount: number;
diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts
index 639a8beb7..c19fc0e4d 100644
--- a/packages/runtime-host/src/server/execution-composition.ts
+++ b/packages/runtime-host/src/server/execution-composition.ts
@@ -468,6 +468,8 @@ export async function createExecutionRuntimeHostComposition(
requireRootCoordinator(rootCoordinator).claimStopFence(input, commitQueueFence, admission),
startFromMessage: (input, admission) =>
requireRootCoordinator(rootCoordinator).startFromMessage(input, admission),
+ startRecoveredMessages: (input, admission) =>
+ requireRootCoordinator(rootCoordinator).startRecoveredMessages(input, admission),
prepareMessage: (input) => requireRootCoordinator(rootCoordinator).prepareMessage(input),
claimStop: (input, commitQueueFence, admission) =>
requireRootCoordinator(rootCoordinator).claimStop(input, commitQueueFence, admission),
@@ -482,6 +484,7 @@ export async function createExecutionRuntimeHostComposition(
stores.runtimeEventStore.readImmutableSteeringMessageProof(sessionId, messageId),
},
receipts: stores.messageReceiptStore,
+ lifecycle: stores.sessionStore,
sessionAdmission,
acquireResidency: () => context.acquireResidency('message-queue'),
requestDrain: context.requestDrain,
@@ -1485,6 +1488,7 @@ export async function createExecutionRuntimeHostComposition(
),
);
await coordinator.recover();
+ await messages.recoverPendingAfterHostRestart(recoverySessions.map((session) => session.id));
rootRecoveryCompleted = true;
},
},
diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts
index 8d8ac47e9..dab6ddc67 100644
--- a/packages/runtime-host/src/server/message-coordinator.ts
+++ b/packages/runtime-host/src/server/message-coordinator.ts
@@ -36,8 +36,10 @@ import {
import {
normalizeRootTurnAdmissionPayload,
type ImmutableSteeringMessageProof,
+ type MessageLifecycleStore,
type MessageReceiptOperation,
type MessageReceiptStore,
+ type PendingMessageAdmission,
type RootTurnSourceMessage,
type RootTurnSourceMessageReceipt,
} from '@maka/storage/execution-stores';
@@ -104,6 +106,15 @@ export interface HostMessageStartInput {
readonly content: MessageContent;
readonly sourceMessage: RootTurnSourceMessage;
readonly initiatingConnectionId: string;
+ readonly turnId?: string;
+ readonly runId?: string;
+}
+
+export interface HostMessageRecoveryBatch {
+ readonly sessionId: string;
+ readonly content: MessageContent;
+ readonly submittedContent: MessageContent;
+ readonly sources: readonly RootTurnSourceMessage[];
}
export interface HostMessagePreparationInput {
@@ -137,6 +148,10 @@ export interface HostMessageRootPort {
input: HostMessageStartInput,
admission: SessionAdmissionLease,
): Promise<{ readonly turnId: string } | { readonly error: string }>;
+ startRecoveredMessages?(
+ input: HostMessageRecoveryBatch,
+ admission: SessionAdmissionLease,
+ ): Promise<{ readonly turnId: string } | { readonly error: string }>;
prepareMessage(
input: HostMessagePreparationInput,
): Promise<
@@ -167,6 +182,7 @@ export interface HostMessageCoordinatorOptions {
readonly root: HostMessageRootPort;
readonly durableProof: HostMessageDurableProofReader;
readonly receipts: MessageReceiptStore;
+ readonly lifecycle?: MessageLifecycleStore;
readonly sessionAdmission: SessionAdmissionGate;
readonly acquireResidency: () => RuntimeHostResidency;
readonly requestDrain?: () => void;
@@ -186,6 +202,9 @@ export type CandidateSnapshotPreflight = (
interface LiveEntry {
readonly entryId: string;
readonly messageId: string;
+ readonly turnId: string;
+ readonly runId: string;
+ readonly admittedAt: number;
content: MessageContent;
modelContent: MessageContent;
readonly initiatingConnectionId: string;
@@ -310,6 +329,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
readonly #root: HostMessageRootPort;
readonly #durableProof: HostMessageDurableProofReader;
readonly #receipts: MessageReceiptStore;
+ readonly #lifecycle?: MessageLifecycleStore;
readonly #sessionAdmission: SessionAdmissionGate;
readonly #acquireResidency: () => RuntimeHostResidency;
readonly #requestDrain: () => void;
@@ -330,6 +350,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
this.#root = options.root;
this.#durableProof = options.durableProof;
this.#receipts = options.receipts;
+ this.#lifecycle = options.lifecycle;
this.#sessionAdmission = options.sessionAdmission;
this.#acquireResidency = options.acquireResidency;
this.#requestDrain = options.requestDrain ?? (() => undefined);
@@ -503,6 +524,90 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
this.#draining = true;
}
+ async markMessagesHandedOff(sessionId: string, messageIds: readonly string[]): Promise<void> {
+ await this.#lifecycle?.markMessagesHandedOff(sessionId, messageIds);
+ }
+
+ async markMessagesExecuted(sessionId: string, messageIds: readonly string[]): Promise<void> {
+ await this.#lifecycle?.markMessagesExecuted(sessionId, messageIds);
+ }
+
+ async cancelMessages(sessionId: string, messageIds: readonly string[]): Promise<void> {
+ await this.#lifecycle?.cancelMessageAdmissions(sessionId, messageIds);
+ }
+
+ async recoverPendingAfterHostRestart(sessionIds: readonly string[]): Promise<void> {
+ if (!this.#lifecycle) return;
+ for (const sessionId of sessionIds) {
+ const admissions = await this.#lifecycle.listMessageAdmissions(sessionId);
+ if (admissions.length === 0) continue;
+ const rootState = await this.#root.readRootState(sessionId);
+ const pending = [] as PendingMessageAdmission[];
+ for (const admission of admissions) {
+ const source = await this.#durableProof.readRootTurnSourceMessageReceipt(
+ sessionId,
+ admission.messageId,
+ );
+ if (source) {
+ await this.#lifecycle.markMessagesHandedOff(sessionId, [admission.messageId]);
+ } else {
+ pending.push(admission);
+ }
+ }
+ if (pending.length === 0) continue;
+ if (rootState.kind !== 'active') {
+ if (rootState.kind !== 'idle') continue;
+ if (!this.#root.startRecoveredMessages) {
+ throw new RuntimeMessageAuthorityInvariantError(
+ 'Message recovery authority is unavailable',
+ );
+ }
+ await this.#sessionAdmission.run(sessionId, (admission) =>
+ this.#root.startRecoveredMessages!(
+ {
+ sessionId,
+ content: aggregateMessageContents(pending.map((entry) => entry.modelContent)),
+ submittedContent: aggregateMessageContents(pending.map((entry) => entry.content)),
+ sources: pending.map(pendingMessageSource),
+ },
+ admission,
+ ),
+ );
+ continue;
+ }
+ if (!this.#sessions.has(sessionId)) this.#state(sessionId);
+ const state = this.#requireState(sessionId);
+ if (!state.reservedRoot) this.reserveRootTurn(rootState);
+ if (!sameRun(state.reservedRoot!, rootState)) continue;
+ for (const admission of admissions) {
+ if (admission.turnId !== rootState.turnId || admission.runId !== rootState.runId) continue;
+ const existing = allLiveEntries(state).find(
+ (entry) => entry.messageId === admission.messageId,
+ );
+ if (existing) continue;
+ const residency = this.#acquireResidency();
+ const entry: LiveEntry = {
+ entryId: this.#createId(),
+ messageId: admission.messageId,
+ turnId: admission.turnId,
+ runId: admission.runId,
+ admittedAt: admission.admittedAt,
+ content: admission.content,
+ modelContent: admission.modelContent,
+ initiatingConnectionId: '',
+ placement: admission.placement,
+ disposition: admission.disposition,
+ generation: state.generation,
+ residency,
+ state: 'queued',
+ };
+ if (entry.disposition === 'steering') state.steering.push(entry);
+ else state.followup.push(entry);
+ this.#mutated(state);
+ }
+ }
+ }
+
commitStopFence(identity: RuntimeMessageRunIdentity): QueueFenceResult {
return this.#commitQueueFence(identity);
}
@@ -621,16 +726,45 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
placement: input.placement,
disposition: 'turn_started',
};
+ const pendingAdmission = await this.#lifecycle?.readMessageAdmission(
+ input.sessionId,
+ input.messageId,
+ );
+ if (
+ pendingAdmission &&
+ (!messageContentsEqual(pendingAdmission.content, payload.content) ||
+ pendingAdmission.submittedPlacement !== input.placement)
+ ) {
+ return failure('operation_conflict', 'Message admission has a different payload');
+ }
+ const turnId = pendingAdmission?.turnId ?? this.#createId();
+ const runId = pendingAdmission?.runId ?? this.#createId();
+ const messageAdmission: PendingMessageAdmission = {
+ sessionId: input.sessionId,
+ turnId,
+ runId,
+ messageId: input.messageId,
+ content: payload.content,
+ modelContent: payload.content,
+ submittedPlacement: input.placement,
+ placement: 'current_turn',
+ disposition: 'steering',
+ admittedAt: pendingAdmission?.admittedAt ?? Date.now(),
+ };
+ await this.#lifecycle?.commitMessageAdmission(messageAdmission);
const started = await this.#root.startFromMessage(
{
sessionId: input.sessionId,
content: payload.content,
sourceMessage,
initiatingConnectionId,
+ turnId,
+ runId,
},
admission,
);
if ('error' in started) {
+ await this.#lifecycle?.cancelMessageAdmissions(input.sessionId, [input.messageId]);
return failure('operation_conflict', started.error);
}
if (!isEntityId(started.turnId)) {
@@ -638,6 +772,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
'Started Turn identity is not encodable',
);
}
+ await this.#lifecycle?.markMessagesHandedOff(input.sessionId, [input.messageId]);
const result = { disposition: 'turn_started', turnId: started.turnId } as const;
return success(result);
}
@@ -734,10 +869,26 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
continue;
}
const result = { disposition, queueRevision: candidateRevision + 1 } as const;
+ const messageAdmission: PendingMessageAdmission = {
+ sessionId: input.sessionId,
+ turnId: rootState.turnId,
+ runId: rootState.runId,
+ messageId: input.messageId,
+ content: payload.content,
+ modelContent: prepared.content,
+ submittedPlacement: input.placement,
+ placement: input.placement,
+ disposition,
+ admittedAt: Date.now(),
+ };
+ await this.#lifecycle?.commitMessageAdmission(messageAdmission);
const residency = this.#acquireResidency();
const entry: LiveEntry = {
entryId,
messageId: input.messageId,
+ turnId: rootState.turnId,
+ runId: rootState.runId,
+ admittedAt: messageAdmission.admittedAt,
content: payload.content,
modelContent: prepared.content,
initiatingConnectionId,
@@ -794,6 +945,10 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
queueRevision: state.revision + (queued.length > 0 ? 1 : 0),
retracted: queued.map(retractedSnapshot),
};
+ await this.#lifecycle?.cancelMessageAdmissions(
+ input.sessionId,
+ queued.map((entry) => entry.messageId),
+ );
const retracted = this.#retractQueued(state);
if (retracted.length > 0) this.#mutated(state);
if (!isDeepStrictEqual(result, { queueRevision: state.revision, retracted })) {
@@ -970,6 +1125,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
}
return failure('not_found', 'Message queue entry does not exist');
}
+ await this.#lifecycle?.cancelMessageAdmissions(input.sessionId, [queued.entry.messageId]);
queued.remove();
this.#releaseEntry(queued.entry);
this.#mutated(state);
@@ -1023,6 +1179,18 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
}
return failure('not_found', 'Message queue entry does not exist');
}
+ await this.#lifecycle?.updateMessageAdmission({
+ sessionId: input.sessionId,
+ turnId: entry.turnId,
+ runId: entry.runId,
+ messageId: entry.messageId,
+ content: entry.content,
+ modelContent: entry.modelContent,
+ submittedPlacement: 'next_turn',
+ placement: 'current_turn',
+ disposition: 'steering',
+ admittedAt: entry.admittedAt,
+ });
state.followup.splice(index, 1);
state.steering.push({ ...entry, placement: 'current_turn', disposition: 'steering' });
this.#mutated(state);
@@ -1113,6 +1281,18 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
) {
return failure('session_busy', 'Message queue changed during update');
}
+ await this.#lifecycle?.updateMessageAdmission({
+ sessionId: input.sessionId,
+ turnId: queued.entry.turnId,
+ runId: queued.entry.runId,
+ messageId: queued.entry.messageId,
+ content,
+ modelContent,
+ submittedPlacement: queued.entry.placement,
+ placement: queued.entry.placement,
+ disposition: queued.entry.disposition,
+ admittedAt: queued.entry.admittedAt,
+ });
queued.entry.content = content;
queued.entry.modelContent = modelContent;
this.#mutated(state);
@@ -1153,6 +1333,10 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
reordered.push(entry);
}
if (reordered.some((entry, index) => current[index] !== entry)) {
+ await this.#lifecycle?.reorderMessageAdmissions(
+ input.sessionId,
+ reordered.map((entry) => entry.messageId),
+ );
state.followup = reordered;
this.#mutated(state);
}
@@ -1839,6 +2023,16 @@ function sourceFromEntry(entry: LiveEntry): RootFollowupSource {
};
}
+function pendingMessageSource(admission: PendingMessageAdmission): RootTurnSourceMessage {
+ return {
+ messageId: admission.messageId,
+ content: normalizeMessageContent(admission.modelContent),
+ submittedContentDigest: messageContentDigest(admission.content),
+ placement: admission.placement,
+ disposition: admission.disposition,
+ };
+}
+
function queuedSnapshot(entry: LiveEntry): QueuedMessageSnapshot {
return {
entryId: entry.entryId,
diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts
index c05a8a019..ffd3ed7f2 100644
--- a/packages/runtime-host/src/server/root-turn-coordinator.ts
+++ b/packages/runtime-host/src/server/root-turn-coordinator.ts
@@ -79,6 +79,7 @@ import type { HostInteractionCoordinator } from './interaction-coordinator.js';
import {
type HostMessageRootState,
type HostMessagePreparationInput,
+ type HostMessageRecoveryBatch,
type HostMessageSessionHeader,
type HostMessageStartInput,
type HostMessageStopClaim,
@@ -1003,7 +1004,8 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
const header = await this.stores.sessionStore.readHeaderSnapshot(input.sessionId);
const unavailableReason = runtimeHostExternalTurnUnavailableReason(header);
if (unavailableReason) return { error: unavailableReason };
- const turnId = randomUUID();
+ const turnId = input.turnId ?? randomUUID();
+ const runId = input.runId ?? randomUUID();
const hasSkillInvocation = parseSkillInvocationTokens(content.text).length > 0;
const prepared = hasSkillInvocation
? await this.prepareHostedSkillInvocationContent(
@@ -1040,7 +1042,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
const admitted = await this.rootAdmissionOwner.admitRootTurn({
sessionId: input.sessionId,
turnId,
- proposedRunId: randomUUID(),
+ proposedRunId: runId,
proposedUserMessageId: input.sourceMessage.messageId,
execution: {
kind: 'external_message',
@@ -1085,6 +1087,61 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
});
}
+ startRecoveredMessages(
+ input: HostMessageRecoveryBatch,
+ admissionLease: SessionAdmissionLease,
+ ): Promise<{ readonly turnId: string } | { readonly error: string }> {
+ return this.runCommand(async () => {
+ if (this.#executions.has(input.sessionId)) {
+ return { error: 'A root Turn is still active' };
+ }
+ const header = await this.stores.sessionStore.readHeaderSnapshot(input.sessionId);
+ const unavailableReason = runtimeHostExternalTurnUnavailableReason(header);
+ if (unavailableReason) return { error: unavailableReason };
+ const reservation = this.reserveRootTurn(input.sessionId);
+ if (!reservation) return { error: 'Another root Turn is being admitted' };
+ try {
+ const turnId = randomUUID();
+ const admitted = await this.rootAdmissionOwner.admitRootTurn({
+ sessionId: input.sessionId,
+ turnId,
+ proposedRunId: randomUUID(),
+ proposedUserMessageId: randomUUID(),
+ execution: {
+ kind: 'external_message',
+ inputDigest: messageContentDigest(input.submittedContent),
+ },
+ normalizedInput: input.content,
+ sourceMessages: input.sources,
+ admittedAt: Date.now(),
+ });
+ if (admitted.kind !== 'admitted') {
+ return { error: 'Recovered Message root identity already existed' };
+ }
+ const disposition = await this.prepareAdmittedTurn(
+ { sessionId: input.sessionId, turnId, content: input.content },
+ admitted.admission,
+ this.acquireRecoveryResidency,
+ admissionLease,
+ undefined,
+ undefined,
+ reservation,
+ );
+ if (disposition.kind !== 'await_start') {
+ return { error: 'Recovered Message root did not reserve execution' };
+ }
+ await this.messages.markMessagesHandedOff(
+ input.sessionId,
+ input.sources.map((source) => source.messageId),
+ );
+ return { turnId };
+ } catch (error) {
+ this.#admissions.release(reservation);
+ throw error;
+ }
+ });
+ }
+
prepareMessage(
input: HostMessagePreparationInput,
): Promise<
@@ -1810,7 +1867,11 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
if (active.startSettled.phase === 'rejected') {
return { active, deliverStop: () => Promise.resolve() };
}
- commitQueueFence();
+ const fence = commitQueueFence();
+ await this.messages.cancelMessages(
+ input.sessionId,
+ fence.retracted.map((message) => message.messageId),
+ );
await this.interactions.claimRunClosure(input, 'turn_stopped', admission);
const shouldDeliverStop = !active.stopRequested;
active.stopRequested = true;
@@ -1845,7 +1906,11 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
const active = this.#executions.get(input.sessionId);
if (isTerminalSnapshot(snapshot)) {
if (active?.turnId === input.turnId && active.runId === input.runId) {
- commitQueueFence();
+ const fence = commitQueueFence();
+ await this.messages.cancelMessages(
+ input.sessionId,
+ fence.retracted.map((message) => message.messageId),
+ );
active.stopRequested = true;
return { kind: 'await_terminal', active };
}
@@ -1870,7 +1935,11 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
};
}
- commitQueueFence();
+ const fence = commitQueueFence();
+ await this.messages.cancelMessages(
+ input.sessionId,
+ fence.retracted.map((message) => message.messageId),
+ );
await this.interactions.claimRunClosure(input, 'turn_stopped', admissionLease);
const shouldRequestStop = !active.stopRequested;
active.stopRequested = true;
@@ -2137,6 +2206,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
}
this.observeExecutionCompletion(active, { kind: 'terminal', snapshot });
await this.interruptPlanAfterUnsuccessfulTurn(input.sessionId, active, snapshot.status);
+ await this.settleExecutedMessageSources(active);
terminalTransitionStarted = true;
await this.completeTerminalTransition(input.sessionId, active);
} catch (error) {
@@ -2160,6 +2230,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
snapshot,
});
await this.interruptPlanAfterUnsuccessfulTurn(input.sessionId, active, snapshot.status);
+ await this.settleExecutedMessageSources(active);
terminalTransitionStarted = true;
await this.completeTerminalTransition(input.sessionId, active);
containedRunFailure =
@@ -2215,6 +2286,37 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
}
}
+ private async settleExecutedMessageSources(active: ActiveRootTurn): Promise<void> {
+ const admission = await this.stores.agentRunStore.readRootTurnAdmission(
+ active.sessionId,
+ active.turnId,
+ );
+ if (!admission || admission.sourceMessages.length === 0) return;
+ const events = await this.stores.agentRunStore.readEvents(active.sessionId, active.runId);
+ if (
+ !events.some(
+ (event) =>
+ event.type === 'provider_request_captured' ||
+ event.type === 'provider_request_attempt_recorded' ||
+ event.type === 'model_call_attempt_recorded',
+ )
+ ) {
+ return;
+ }
+ const executed = [] as string[];
+ for (const source of admission.sourceMessages) {
+ if (
+ (await this.stores.sessionStore.readMessageLifecycleState(
+ active.sessionId,
+ source.messageId,
+ )) === 'handed_off'
+ ) {
+ executed.push(source.messageId);
+ }
+ }
+ await this.messages.markMessagesExecuted(active.sessionId, executed);
+ }
+
private observeExecutionCompletion(
active: ActiveRootTurn,
completion: HostedExecutionCompletion,
@@ -2276,15 +2378,15 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
admissionLease: SessionAdmissionLease,
): Promise<void> {
const initiatingConnectionId = batch.initiatingConnectionId;
- if (!initiatingConnectionId) {
- throw new RuntimeMessageAuthorityInvariantError(
- 'Follow-up batch lost its initiating Client identity',
- );
- }
// A confirmed follow-up must become a durable root even when a Session
// provider is unavailable. Lost tools are omitted while ephemeral
// capabilities bind to the Client that submitted this follow-up.
- await this.clientCapabilities?.bindConfirmedFollowup(batch.sessionId, initiatingConnectionId);
+ if (initiatingConnectionId) {
+ await this.clientCapabilities?.bindConfirmedFollowup(
+ batch.sessionId,
+ initiatingConnectionId,
+ );
+ }
const turnId = randomUUID();
const header = await this.stores.sessionStore.readHeaderSnapshot(batch.sessionId);
@@ -2307,6 +2409,10 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
'Fresh follow-up root Turn identity already existed',
);
}
+ await this.messages.markMessagesHandedOff(
+ batch.sessionId,
+ batch.sources.map((source) => source.messageId),
+ );
const nextIdentity = {
sessionId: batch.sessionId,
diff --git a/packages/runtime/src/agent-run.ts b/packages/runtime/src/agent-run.ts
index fcbf0102a..a5a2ef633 100644
--- a/packages/runtime/src/agent-run.ts
+++ b/packages/runtime/src/agent-run.ts
@@ -58,7 +58,7 @@ import {
resolveEffectiveOrchestration,
type EffectiveOrchestration,
} from '@maka/core/orchestration';
-import type { SessionEvent } from '@maka/core/events';
+import { messageContentsEqual, type SessionEvent } from '@maka/core/events';
import type { AgentBackend, BackendSendInput } from '@maka/core/backend-types';
import type { RunTraceEvent } from './run-trace.js';
import type { StopSessionInput } from './session-manager.js';
@@ -667,7 +667,7 @@ export class AgentRun {
: {}),
...(this.input.userInput.origin ? { origin: this.input.userInput.origin } : {}),
});
- await this.input.store.appendMessage(this.sessionId, userMsg);
+ await appendUserMessageOnce(this.input.store, this.sessionId, userMsg);
await this.input.hooks.appendTurnState(this.sessionId, this.turnId, 'running', this.lineage);
this.lastTs = userMessageTs;
} else {
@@ -1896,6 +1896,28 @@ function redactTraceString(value: string): string {
function errorMessage(error: unknown): string {
return redactTraceString(error instanceof Error ? error.message : String(error));
}
+
+async function appendUserMessageOnce(
+ store: AgentRunSessionStore,
+ sessionId: string,
+ message: UserMessage,
+): Promise<void> {
+ const existing = (await store.readMessages(sessionId)).find(
+ (candidate) => candidate.id === message.id,
+ );
+ if (!existing) {
+ await store.appendMessage(sessionId, message);
+ return;
+ }
+ if (
+ existing.type !== 'user' ||
+ existing.turnId !== message.turnId ||
+ !messageContentsEqual(existing, message)
+ ) {
+ throw new Error(`Durable UserMessage identity ${message.id} has conflicting content`);
+ }
+}
+
function isInteractionResumeAck(event: SessionEvent): boolean {
return (
event.type === 'sandbox_boundary_decision_ack' || event.type === 'user_question_answer_ack'
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 6f796e9a5..5f52a4ad4 100644
--- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts
+++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts
@@ -283,6 +283,20 @@ describe('SqliteSessionMetadataStore', () => {
},
],
);
+ assert.equal(
+ await store.readMessageLifecycleState('session-1', 'message-1'),
+ 'accepted',
+ );
+ await store.markMessagesHandedOff('session-1', ['message-1']);
+ assert.equal(
+ await store.readMessageLifecycleState('session-1', 'message-1'),
+ 'handed_off',
+ );
+ await store.markMessagesExecuted('session-1', ['message-1']);
+ assert.equal(
+ await store.readMessageLifecycleState('session-1', 'message-1'),
+ 'executed',
+ );
} finally {
store.close();
}
diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts
index ea8e49bcd..e0b134120 100644
--- a/packages/storage/src/execution-stores.ts
+++ b/packages/storage/src/execution-stores.ts
@@ -115,9 +115,12 @@ export type {
RuntimeEventScanResult,
} from './agent-run-store.js';
export type {
+ MessageLifecycleState,
+ MessageLifecycleStore,
MessageOperationReceipt,
MessageReceiptOperation,
MessageReceiptStore,
+ PendingMessageAdmission,
} from './message-receipt-store.js';
export type {
ProbeSessionRemovalResult,
@@ -173,6 +176,10 @@ export interface ExecutionSessionReader {
list(filter?: SessionListFilter): Promise<SessionSummary[]>;
readHeader(sessionId: string): Promise<SessionHeader>;
readMessages(sessionId: string): Promise<StoredMessage[]>;
+ readMessageLifecycleState(
+ sessionId: string,
+ messageId: string,
+ ): Promise<import('./message-receipt-store.js').MessageLifecycleState | undefined>;
listTurns(sessionId: string): Promise<TurnRecord[]>;
close?(): Promise<void>;
}
@@ -421,6 +428,21 @@ async function createExecutionStoresForWrite<K extends StorageRootKind, E extend
run(() => sessionStore.appendMessage(sessionId, message)),
appendMessages: (sessionId, messages) =>
run(() => sessionStore.appendMessages(sessionId, messages)),
+ commitMessageAdmission: (admission) => run(() => sessionStore.commitMessageAdmission(admission)),
+ readMessageAdmission: (sessionId, messageId) =>
+ run(() => sessionStore.readMessageAdmission(sessionId, messageId)),
+ readMessageLifecycleState: (sessionId, messageId) =>
+ run(() => sessionStore.readMessageLifecycleState(sessionId, messageId)),
+ listMessageAdmissions: (sessionId) => run(() => sessionStore.listMessageAdmissions(sessionId)),
+ updateMessageAdmission: (admission) => run(() => sessionStore.updateMessageAdmission(admission)),
+ reorderMessageAdmissions: (sessionId, messageIds) =>
+ run(() => sessionStore.reorderMessageAdmissions(sessionId, messageIds)),
+ cancelMessageAdmissions: (sessionId, messageIds) =>
+ run(() => sessionStore.cancelMessageAdmissions(sessionId, messageIds)),
+ markMessagesHandedOff: (sessionId, messageIds) =>
+ run(() => sessionStore.markMessagesHandedOff(sessionId, messageIds)),
+ markMessagesExecuted: (sessionId, messageIds) =>
+ run(() => sessionStore.markMessagesExecuted(sessionId, messageIds)),
subscribeTranscriptChanges: (listener) => sessionStore.subscribeTranscriptChanges(listener),
updateHeader: (sessionId, patch) => run(() => sessionStore.updateHeader(sessionId, patch)),
updateHeaderVersioned: (sessionId, patch, expectedRevision) =>
@@ -608,6 +630,8 @@ async function openExecutionStoresForRead<K extends StorageRootKind, E extends o
list: (filter) => run(() => sessionStore.list(filter)),
readHeader: (sessionId) => run(() => sessionStore.readHeaderSnapshot(sessionId)),
readMessages: (sessionId) => run(() => sessionStore.readMessagesSnapshot(sessionId)),
+ readMessageLifecycleState: (sessionId, messageId) =>
+ run(() => sessionStore.readMessageLifecycleState(sessionId, messageId)),
listTurns: (sessionId) => run(() => sessionStore.listTurnsSnapshot(sessionId)),
close: () =>
closeExecutionStorePersistence(sessionStore, runtimePersistence, {
diff --git a/packages/storage/src/message-receipt-store.ts b/packages/storage/src/message-receipt-store.ts
index b9dfeb320..d0b5c40a1 100644
--- a/packages/storage/src/message-receipt-store.ts
+++ b/packages/storage/src/message-receipt-store.ts
@@ -45,6 +45,24 @@ export interface PendingMessageAdmission {
readonly admittedAt: number;
}
+export interface MessageLifecycleStore {
+ commitMessageAdmission(admission: PendingMessageAdmission): Promise<PendingMessageAdmission>;
+ readMessageAdmission(
+ sessionId: string,
+ messageId: string,
+ ): Promise<PendingMessageAdmission | undefined>;
+ readMessageLifecycleState(
+ sessionId: string,
+ messageId: string,
+ ): Promise<MessageLifecycleState | undefined>;
+ listMessageAdmissions(sessionId: string): Promise<readonly PendingMessageAdmission[]>;
+ updateMessageAdmission(admission: PendingMessageAdmission): Promise<void>;
+ reorderMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise<void>;
+ cancelMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise<void>;
+ markMessagesHandedOff(sessionId: string, messageIds: readonly string[]): Promise<void>;
+ markMessagesExecuted(sessionId: string, messageIds: readonly string[]): Promise<void>;
+}
+
export function normalizePendingMessageAdmission(
admission: PendingMessageAdmission,
): PendingMessageAdmission {
diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts
index a731808e9..5b310c076 100644
--- a/packages/storage/src/session-store.ts
+++ b/packages/storage/src/session-store.ts
@@ -80,6 +80,10 @@ import {
type TurnStateMessage,
type UserMessage,
} from '@maka/core/session';
+import type {
+ MessageLifecycleStore,
+ PendingMessageAdmission,
+} from './message-receipt-store.js';
const SESSION_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/;
@@ -299,7 +303,7 @@ export interface SessionStore {
close?(): Promise<void>;
}
-export interface SessionAuthorityStore extends SessionStore {
+export interface SessionAuthorityStore extends SessionStore, MessageLifecycleStore {
/** Read a bounded set of durable messages at an inclusive transcript watermark. */
readTranscriptMessagesSnapshot(
sessionId: string,
@@ -857,6 +861,60 @@ class SqliteSessionStore implements SessionAuthorityStore {
for (const listener of this.transcriptChangeListeners) listener(sessionId);
}
+ async commitMessageAdmission(admission: PendingMessageAdmission): Promise<PendingMessageAdmission> {
+ await this.ensureReady();
+ const committed = await this.metadata.commitMessageAdmission(admission);
+ for (const listener of this.transcriptChangeListeners) listener(admission.sessionId);
+ return committed;
+ }
+
+ async readMessageAdmission(
+ sessionId: string,
+ messageId: string,
+ ): Promise<PendingMessageAdmission | undefined> {
+ await this.ensureReady();
+ return this.metadata.readMessageAdmission(sessionId, messageId);
+ }
+
+ async listMessageAdmissions(sessionId: string): Promise<readonly PendingMessageAdmission[]> {
+ await this.ensureReady();
+ return this.metadata.listMessageAdmissions(sessionId);
+ }
+
+ async readMessageLifecycleState(
+ sessionId: string,
+ messageId: string,
+ ): Promise<import('./message-receipt-store.js').MessageLifecycleState | undefined> {
+ await this.ensureReady();
+ return this.metadata.readMessageLifecycleState(sessionId, messageId);
+ }
+
+ async updateMessageAdmission(admission: PendingMessageAdmission): Promise<void> {
+ await this.ensureReady();
+ await this.metadata.updateMessageAdmission(admission);
+ for (const listener of this.transcriptChangeListeners) listener(admission.sessionId);
+ }
+
+ async reorderMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise<void> {
+ await this.ensureReady();
+ await this.metadata.reorderMessageAdmissions(sessionId, messageIds);
+ }
+
+ async cancelMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise<void> {
+ await this.ensureReady();
+ await this.metadata.cancelMessageAdmissions(sessionId, messageIds);
+ }
+
+ async markMessagesHandedOff(sessionId: string, messageIds: readonly string[]): Promise<void> {
+ await this.ensureReady();
+ await this.metadata.markMessagesHandedOff(sessionId, messageIds);
+ }
+
+ async markMessagesExecuted(sessionId: string, messageIds: readonly string[]): Promise<void> {
+ await this.ensureReady();
+ await this.metadata.markMessagesExecuted(sessionId, messageIds);
+ }
+
subscribeTranscriptChanges(listener: (sessionId: string) => void): () => void {
this.transcriptChangeListeners.add(listener);
return () => this.transcriptChangeListeners.delete(listener);
diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts
index 03395bb0a..99437bdf2 100644
--- a/packages/storage/src/sqlite-session-metadata-store.ts
+++ b/packages/storage/src/sqlite-session-metadata-store.ts
@@ -1670,6 +1670,259 @@ export class SqliteSessionMetadataStore {
});
}
+ async listMessageAdmissions(sessionId: string): Promise<readonly PendingMessageAdmission[]> {
+ this.assertOpen();
+ assertSafeSessionId(sessionId);
+ return this.readTransaction(() => {
+ const rows = this.db
+ .prepare(
+ `
+ SELECT turn_id, run_id, message_id, content_json, model_content_json,
+ submitted_placement, placement, disposition, lifecycle_state, queue_order, admitted_at
+ FROM message_admissions
+ WHERE session_id = ? AND lifecycle_state = 'accepted'
+ ORDER BY queue_order, sequence
+ `,
+ )
+ .all(sessionId) as MessageAdmissionRow[];
+ return rows.map((row) => decodeMessageAdmissionRow(sessionId, row).admission);
+ });
+ }
+
+ async readMessageLifecycleState(
+ sessionId: string,
+ messageId: string,
+ ): Promise<MessageLifecycleState | undefined> {
+ this.assertOpen();
+ assertSafeSessionId(sessionId);
+ assertSafeSessionId(messageId);
+ return this.readTransaction(() => {
+ const row = this.db
+ .prepare(
+ `
+ SELECT lifecycle_state
+ FROM message_admissions
+ WHERE session_id = ? AND message_id = ?
+ `,
+ )
+ .get(sessionId, messageId) as { lifecycle_state?: unknown } | undefined;
+ if (!row) return undefined;
+ if (
+ row.lifecycle_state !== 'accepted' &&
+ row.lifecycle_state !== 'handed_off' &&
+ row.lifecycle_state !== 'executed' &&
+ row.lifecycle_state !== 'cancelled'
+ ) {
+ throw new SessionMetadataConflictError('Invalid Message admission lifecycle state');
+ }
+ return row.lifecycle_state;
+ });
+ }
+
+ async updateMessageAdmission(admission: PendingMessageAdmission): Promise<void> {
+ this.assertOpen();
+ const stored = normalizePendingMessageAdmission(admission);
+ this.transaction(() => {
+ const currentRow = this.db
+ .prepare(
+ `
+ SELECT turn_id, run_id, message_id, content_json, model_content_json,
+ submitted_placement, placement, disposition, lifecycle_state, queue_order, admitted_at
+ FROM message_admissions
+ WHERE session_id = ? AND message_id = ?
+ `,
+ )
+ .get(stored.sessionId, stored.messageId) as MessageAdmissionRow | undefined;
+ if (!currentRow) throw new SessionMetadataConflictError('Message admission does not exist');
+ const current = decodeMessageAdmissionRow(stored.sessionId, currentRow);
+ if (current.lifecycleState !== 'accepted') {
+ throw new SessionMetadataConflictError('Message admission is already settled');
+ }
+ if (
+ current.admission.turnId !== stored.turnId ||
+ current.admission.runId !== stored.runId ||
+ current.admission.submittedPlacement !== stored.submittedPlacement ||
+ current.admission.admittedAt !== stored.admittedAt
+ ) {
+ throw new SessionMetadataConflictError('Message admission update identity conflict');
+ }
+ this.db
+ .prepare(
+ `
+ UPDATE message_admissions
+ SET content_json = ?, model_content_json = ?, placement = ?, disposition = ?
+ WHERE session_id = ? AND message_id = ? AND lifecycle_state = 'accepted'
+ `,
+ )
+ .run(
+ JSON.stringify(stored.content),
+ JSON.stringify(stored.modelContent),
+ stored.placement,
+ stored.disposition,
+ stored.sessionId,
+ stored.messageId,
+ );
+ if (stored.disposition !== 'steering') return;
+ const message = decodeCanonicalMessage({
+ type: 'user',
+ id: stored.messageId,
+ turnId: stored.turnId,
+ ts: stored.admittedAt,
+ ...stored.content,
+ steeringEventId: stored.messageId,
+ });
+ const rows = this.db
+ .prepare(
+ `
+ SELECT sequence, record_json
+ FROM session_messages
+ WHERE session_id = ? AND message_id = ?
+ `,
+ )
+ .all(stored.sessionId, stored.messageId) as Array<{
+ sequence?: unknown;
+ record_json?: unknown;
+ }>;
+ if (rows.length > 1) {
+ throw new SessionMetadataConflictError('Message admission transcript identity is ambiguous');
+ }
+ const json = JSON.stringify(message);
+ if (rows.length === 0) {
+ const row = this.db
+ .prepare(
+ 'SELECT COALESCE(MAX(sequence), -1) AS last_sequence FROM session_messages WHERE session_id = ?',
+ )
+ .get(stored.sessionId) as { last_sequence?: unknown };
+ if (typeof row.last_sequence !== 'number' || !Number.isSafeInteger(row.last_sequence)) {
+ throw new SessionMetadataConflictError('Invalid Session message sequence');
+ }
+ this.insertSessionMessagesSync(stored.sessionId, row.last_sequence + 1, [
+ { message, json },
+ ]);
+ } else {
+ const sequence = rows[0]?.sequence;
+ if (typeof sequence !== 'number' || !Number.isSafeInteger(sequence)) {
+ throw new SessionMetadataConflictError('Invalid Message transcript sequence');
+ }
+ this.db
+ .prepare(
+ 'UPDATE session_messages SET record_json = ? WHERE session_id = ? AND sequence = ?',
+ )
+ .run(json, stored.sessionId, sequence);
+ }
+ this.updateCatalogProjectionSync(
+ stored.sessionId,
+ {
+ lastMessageAt: stored.admittedAt,
+ lastMessagePreview: message.type === 'user' ? message.displayText : undefined,
+ },
+ true,
+ );
+ });
+ }
+
+ async cancelMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise<void> {
+ this.assertOpen();
+ assertSafeSessionId(sessionId);
+ const unique = [...new Set(messageIds)];
+ for (const messageId of unique) assertSafeSessionId(messageId);
+ this.transaction(() => {
+ const statement = this.db.prepare(
+ `
+ UPDATE message_admissions
+ SET lifecycle_state = 'cancelled'
+ WHERE session_id = ? AND message_id = ? AND lifecycle_state = 'accepted'
+ `,
+ );
+ for (const messageId of unique) {
+ const result = statement.run(sessionId, messageId);
+ if (result.changes !== 1) {
+ throw new SessionMetadataConflictError('Message admission cancellation identity conflict');
+ }
+ }
+ });
+ }
+
+ async reorderMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise<void> {
+ this.assertOpen();
+ assertSafeSessionId(sessionId);
+ const unique = [...new Set(messageIds)];
+ if (unique.length !== messageIds.length) {
+ throw new SessionMetadataConflictError('Message admission reorder contains duplicate identities');
+ }
+ for (const messageId of unique) assertSafeSessionId(messageId);
+ this.transaction(() => {
+ const rows = this.db
+ .prepare(
+ `
+ SELECT message_id
+ FROM message_admissions
+ WHERE session_id = ? AND lifecycle_state = 'accepted' AND disposition = 'followup'
+ ORDER BY queue_order, sequence
+ `,
+ )
+ .all(sessionId) as Array<{ message_id?: unknown }>;
+ const current = rows.map((row) => row.message_id);
+ if (
+ current.length !== unique.length ||
+ current.some((messageId, index) => messageId !== unique[index])
+ ) {
+ throw new SessionMetadataConflictError('Message admission reorder identity conflict');
+ }
+ const update = this.db.prepare(
+ `
+ UPDATE message_admissions
+ SET queue_order = ?
+ WHERE session_id = ? AND message_id = ? AND lifecycle_state = 'accepted'
+ `,
+ );
+ unique.forEach((messageId, index) => update.run(index, sessionId, messageId));
+ });
+ }
+
+ async markMessagesHandedOff(sessionId: string, messageIds: readonly string[]): Promise<void> {
+ return this.markMessageLifecycle(sessionId, messageIds, 'handed_off');
+ }
+
+ async markMessagesExecuted(sessionId: string, messageIds: readonly string[]): Promise<void> {
+ return this.markMessageLifecycle(sessionId, messageIds, 'executed');
+ }
+
+ private markMessageLifecycle(
+ sessionId: string,
+ messageIds: readonly string[],
+ state: 'handed_off' | 'executed',
+ ): Promise<void> {
+ this.assertOpen();
+ assertSafeSessionId(sessionId);
+ const unique = [...new Set(messageIds)];
+ for (const messageId of unique) assertSafeSessionId(messageId);
+ this.transaction(() => {
+ const statement = this.db.prepare(
+ `
+ UPDATE message_admissions
+ SET lifecycle_state = ?
+ WHERE session_id = ? AND message_id = ?
+ AND lifecycle_state IN ('accepted', 'handed_off')
+ `,
+ );
+ for (const messageId of unique) {
+ const result = statement.run(state, sessionId, messageId);
+ if (result.changes !== 1) {
+ const existing = this.db
+ .prepare(
+ 'SELECT lifecycle_state FROM message_admissions WHERE session_id = ? AND message_id = ?',
+ )
+ .get(sessionId, messageId) as { lifecycle_state?: unknown } | undefined;
+ if (existing?.lifecycle_state !== state) {
+ throw new SessionMetadataConflictError('Message admission lifecycle identity conflict');
+ }
+ }
+ }
+ });
+ return Promise.resolve();
+ }
+
async readMessages(sessionId: string): Promise<StoredMessage[]> {
return this.readMessagesWith(sessionId, decodeStoredMessage);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org