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 17:40:16 UTC
(maka) 03/05: fix(runtime-host): keep one canonical follow-up transcript
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 f59edcc9c53e3a9d33cf637285e62ed8105a8d37
Author: AstroHan <le...@outlook.com>
AuthorDate: Tue Aug 25 01:36:27 2026 +0800
fix(runtime-host): keep one canonical follow-up transcript
Generated-by: Codex
---
.../src/__tests__/execution-host-message.test.ts | 18 ++++-
.../src/__tests__/fixtures/execution-host-suite.ts | 14 ++++
.../src/__tests__/message-coordinator.test.ts | 1 +
.../runtime-host/src/server/message-coordinator.ts | 7 ++
.../src/server/root-turn-coordinator.ts | 8 +-
.../sqlite-session-metadata-store.test.ts | 19 +++++
packages/storage/src/execution-stores.ts | 2 +
packages/storage/src/message-receipt-store.ts | 6 ++
packages/storage/src/session-store.ts | 11 +++
.../storage/src/sqlite-session-metadata-store.ts | 92 ++++++++++++++++++++++
10 files changed, 173 insertions(+), 5 deletions(-)
diff --git a/packages/runtime-host/src/__tests__/execution-host-message.test.ts b/packages/runtime-host/src/__tests__/execution-host-message.test.ts
index 01ac3e4af..61c1ee357 100644
--- a/packages/runtime-host/src/__tests__/execution-host-message.test.ts
+++ b/packages/runtime-host/src/__tests__/execution-host-message.test.ts
@@ -237,9 +237,23 @@ test('steering becomes durable and ordered followups automatically start the nex
assert.ok(followupTurnId);
const followupLedger = await fixture.readTurn(followupTurnId);
const expectedQuotes = followupSources.flatMap((source) => source.content.quotes ?? []);
- assert.equal(followupLedger.userMessages.length, 1);
- assert.deepEqual(followupLedger.userMessages[0]?.quotes, expectedQuotes);
+ assert.equal(followupLedger.userMessages.length, followupSources.length);
+ assert.deepEqual(
+ followupLedger.userMessages.flatMap((message) => message.quotes ?? []),
+ expectedQuotes,
+ );
assert.deepEqual(userRuntimeContent(followupLedger.runtimeEvents)?.quotes, expectedQuotes);
+ const sessionUserMessages = await fixture.readSessionUserMessages();
+ for (const source of followupSources) {
+ assert.equal(
+ sessionUserMessages.filter((message) => message.id === source.messageId).length,
+ 1,
+ );
+ }
+ assert.equal(
+ sessionUserMessages.filter((message) => message.turnId === followupTurnId).length,
+ followupSources.length,
+ );
});
});
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 d9d68cee9..3175bd3a2 100644
--- a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts
+++ b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts
@@ -1002,6 +1002,20 @@ export class ExecutionFixture {
}
}
+ async readSessionUserMessages(): Promise<Array<Extract<StoredMessage, { type: 'user' }>>> {
+ const reader = await acquireReader(this.capability);
+ let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForRead>> | undefined;
+ try {
+ stores = await openInteractiveExecutionStoresForRead(reader.lease);
+ return (await stores.sessionStore.readMessages(this.sessionId)).filter(
+ (message): message is Extract<StoredMessage, { type: 'user' }> => message.type === 'user',
+ );
+ } finally {
+ await stores?.sessionStore.close?.();
+ await reader.close();
+ }
+ }
+
async readAdmissionChain() {
const owner = await tryAcquireInteractiveRootOwner(this.capability);
assert.ok(owner);
diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts
index 6982e9e06..b3a39b368 100644
--- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts
@@ -2474,6 +2474,7 @@ function memoryMessageLifecycleStore(
(state === 'accepted' || state === 'handed_off'),
)
.map(({ admission }) => admission),
+ rebindMessageAdmissionTranscript: async () => undefined,
updateMessageAdmission: async (admission) => {
const existing = admissions.get(admission.messageId);
if (!existing) throw new Error(`Missing admission ${admission.messageId}`);
diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts
index 31299af8f..f9db15b10 100644
--- a/packages/runtime-host/src/server/message-coordinator.ts
+++ b/packages/runtime-host/src/server/message-coordinator.ts
@@ -539,6 +539,7 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
sessionId: string;
turnId: string;
runId: string;
+ previousRootTurnId: string | null;
messageIds: readonly string[];
}): Promise<void> {
const handoff: string[] = [];
@@ -566,6 +567,12 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
);
}
}
+ await this.#lifecycle.rebindMessageAdmissionTranscript({
+ sessionId: input.sessionId,
+ messageIds: [...new Set(input.messageIds)],
+ turnId: input.turnId,
+ previousRootTurnId: input.previousRootTurnId,
+ });
await this.#lifecycle.markMessagesHandedOff(input.sessionId, handoff);
}
diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts
index 6bb9241ee..c2614ad3c 100644
--- a/packages/runtime-host/src/server/root-turn-coordinator.ts
+++ b/packages/runtime-host/src/server/root-turn-coordinator.ts
@@ -413,6 +413,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
sessionId,
turnId: admission.turnId,
runId: admission.runId,
+ previousRootTurnId: admission.previousRootTurnId,
messageIds: admission.sourceMessages.map((source) => source.messageId),
});
return this.prepareAdmittedTurn(
@@ -1084,6 +1085,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
sessionId: input.sessionId,
turnId,
runId,
+ previousRootTurnId: admitted.admission.previousRootTurnId,
messageIds: [input.sourceMessage.messageId],
});
const disposition = await this.prepareAdmittedTurn(
@@ -1146,6 +1148,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
sessionId: input.sessionId,
turnId,
runId: admitted.admission.runId,
+ previousRootTurnId: admitted.admission.previousRootTurnId,
messageIds: input.sources.map((source) => source.messageId),
});
const disposition = await this.prepareAdmittedTurn(
@@ -2009,9 +2012,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
userMessageId: admission.userMessageId,
execution: admission.execution,
});
- const initialUserMessagesMaterialized =
- admission.sourceMessages.length > 0 &&
- admission.sourceMessages.every((source) => source.disposition === 'turn_started');
+ const initialUserMessagesMaterialized = admission.sourceMessages.length > 0;
if (initialUserMessagesMaterialized) {
await this.manager.materializeRootSourceMessages({
sessionId: input.sessionId,
@@ -2449,6 +2450,7 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
sessionId: batch.sessionId,
turnId,
runId: admitted.admission.runId,
+ previousRootTurnId: admitted.admission.previousRootTurnId,
messageIds: batch.sources.map((source) => source.messageId),
});
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 6b61abc96..e15b352eb 100644
--- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts
+++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts
@@ -327,6 +327,25 @@ describe('SqliteSessionMetadataStore', () => {
})),
[{ id: 'message-followup', turnId: 'turn-current' }],
);
+ await store.rebindMessageAdmissionTranscript({
+ sessionId: 'session-followup-admission',
+ messageIds: ['message-followup'],
+ turnId: 'turn-successor',
+ previousRootTurnId: 'turn-current',
+ });
+ await store.rebindMessageAdmissionTranscript({
+ sessionId: 'session-followup-admission',
+ messageIds: ['message-followup'],
+ turnId: 'turn-successor',
+ previousRootTurnId: 'turn-current',
+ });
+ assert.deepEqual(
+ (await store.readMessages('session-followup-admission')).map((message) => ({
+ id: message.id,
+ turnId: message.turnId,
+ })),
+ [{ id: 'message-followup', turnId: 'turn-successor' }],
+ );
} finally {
store.close();
}
diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts
index e01a5ec5d..fcbae1a96 100644
--- a/packages/storage/src/execution-stores.ts
+++ b/packages/storage/src/execution-stores.ts
@@ -438,6 +438,8 @@ async function createExecutionStoresForWrite<K extends StorageRootKind, E extend
run(() => sessionStore.listMessageAdmissions(sessionId)),
listUnsettledMessageAdmissions: (sessionId) =>
run(() => sessionStore.listUnsettledMessageAdmissions(sessionId)),
+ rebindMessageAdmissionTranscript: (input) =>
+ run(() => sessionStore.rebindMessageAdmissionTranscript(input)),
updateMessageAdmission: (admission) =>
run(() => sessionStore.updateMessageAdmission(admission)),
reorderMessageAdmissions: (sessionId, messageIds) =>
diff --git a/packages/storage/src/message-receipt-store.ts b/packages/storage/src/message-receipt-store.ts
index cb8de5232..42cd1375f 100644
--- a/packages/storage/src/message-receipt-store.ts
+++ b/packages/storage/src/message-receipt-store.ts
@@ -57,6 +57,12 @@ export interface MessageLifecycleStore {
): Promise<MessageLifecycleState | undefined>;
listMessageAdmissions(sessionId: string): Promise<readonly PendingMessageAdmission[]>;
listUnsettledMessageAdmissions(sessionId: string): Promise<readonly PendingMessageAdmission[]>;
+ rebindMessageAdmissionTranscript(input: {
+ sessionId: string;
+ messageIds: readonly string[];
+ turnId: string;
+ previousRootTurnId: string | null;
+ }): Promise<void>;
updateMessageAdmission(admission: PendingMessageAdmission): Promise<void>;
reorderMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise<void>;
cancelMessageAdmissions(sessionId: string, messageIds: readonly string[]): Promise<void>;
diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts
index c968312b7..c6bc1a59f 100644
--- a/packages/storage/src/session-store.ts
+++ b/packages/storage/src/session-store.ts
@@ -887,6 +887,17 @@ class SqliteSessionStore implements SessionAuthorityStore {
return this.metadata.listUnsettledMessageAdmissions(sessionId);
}
+ async rebindMessageAdmissionTranscript(input: {
+ sessionId: string;
+ messageIds: readonly string[];
+ turnId: string;
+ previousRootTurnId: string | null;
+ }): Promise<void> {
+ await this.ensureReady();
+ await this.metadata.rebindMessageAdmissionTranscript(input);
+ for (const listener of this.transcriptChangeListeners) listener(input.sessionId);
+ }
+
async readMessageLifecycleState(
sessionId: string,
messageId: string,
diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts
index 1ed55331d..d2370a01d 100644
--- a/packages/storage/src/sqlite-session-metadata-store.ts
+++ b/packages/storage/src/sqlite-session-metadata-store.ts
@@ -102,6 +102,7 @@ import {
type MessageLifecycleState,
type PendingMessageAdmission,
} from './message-receipt-store.js';
+import { messageContentsEqual, normalizeMessageContent } from '@maka/core/events';
import {
type AgentGraphIntentAdmissionSnapshot,
type AgentGraphTimelineMetadataSnapshot,
@@ -1743,6 +1744,97 @@ export class SqliteSessionMetadataStore {
});
}
+ async rebindMessageAdmissionTranscript(input: {
+ sessionId: string;
+ messageIds: readonly string[];
+ turnId: string;
+ previousRootTurnId: string | null;
+ }): Promise<void> {
+ this.assertOpen();
+ assertSafeSessionId(input.sessionId);
+ assertSafeSessionId(input.turnId);
+ if (input.previousRootTurnId !== null) {
+ assertSafeSessionId(input.previousRootTurnId);
+ }
+ const unique = [...new Set(input.messageIds)];
+ for (const messageId of unique) assertSafeSessionId(messageId);
+ this.transaction(() => {
+ for (const messageId of unique) {
+ const admissionRow = 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(input.sessionId, messageId) as MessageAdmissionRow | undefined;
+ if (!admissionRow) {
+ throw new SessionMetadataConflictError('Message admission does not exist');
+ }
+ const admission = decodeMessageAdmissionRow(input.sessionId, admissionRow);
+ if (
+ admission.lifecycleState !== 'accepted' &&
+ admission.lifecycleState !== 'handed_off' &&
+ admission.lifecycleState !== 'executed'
+ ) {
+ throw new SessionMetadataConflictError('Message admission is already cancelled');
+ }
+ 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 = ? AND message.message_id = ?
+ `,
+ )
+ .all(input.sessionId, messageId) as Array<{
+ sequence?: unknown;
+ record_json?: unknown;
+ record_bytes?: unknown;
+ sha256?: unknown;
+ }>;
+ if (rows.length !== 1) {
+ throw new SessionMetadataConflictError(
+ rows.length === 0
+ ? 'Message admission transcript is missing'
+ : 'Message admission transcript identity is ambiguous',
+ );
+ }
+ const sequence = rows[0]?.sequence;
+ if (typeof sequence !== 'number' || !Number.isSafeInteger(sequence)) {
+ throw new SessionMetadataConflictError('Invalid Message transcript sequence');
+ }
+ const row = rows[0]!;
+ const recordJson = readStoredMessageRecordJson(this.db, input.sessionId, sequence, row);
+ const message = decodeStoredMessage(JSON.parse(recordJson) as unknown);
+ if (
+ message.type !== 'user' ||
+ message.id !== messageId ||
+ !messageContentsEqual(normalizeMessageContent(message), admission.admission.content)
+ ) {
+ throw new SessionMetadataConflictError('Message admission transcript identity conflict');
+ }
+ if (message.turnId === input.turnId) continue;
+ if (
+ input.previousRootTurnId === null ||
+ message.turnId !== input.previousRootTurnId
+ ) {
+ throw new SessionMetadataConflictError('Message admission transcript Turn conflict');
+ }
+ const rebound = decodeCanonicalMessage({ ...message, turnId: input.turnId });
+ this.db
+ .prepare(
+ 'UPDATE session_messages SET record_json = ? WHERE session_id = ? AND sequence = ?',
+ )
+ .run(JSON.stringify(rebound), input.sessionId, sequence);
+ }
+ });
+ }
+
async updateMessageAdmission(admission: PendingMessageAdmission): Promise<void> {
this.assertOpen();
const stored = normalizePendingMessageAdmission(admission);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org