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/25 03:40:20 UTC
(maka) 01/01: refactor(runtime-host): materialize messages at handoff
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 48b162326f39ac1c83d7ebd8277a33ffd56ddba3
Author: AstroHan <le...@outlook.com>
AuthorDate: Tue Aug 25 11:40:05 2026 +0800
refactor(runtime-host): materialize messages at handoff
Generated-by: Codex
---
.../src/__tests__/execution-host-message.test.ts | 50 ++++-
.../src/__tests__/execution-host-queue.test.ts | 6 +
.../src/__tests__/message-coordinator.test.ts | 3 +-
.../src/__tests__/root-turn-coordinator.test.ts | 15 +-
.../runtime-host/src/server/message-coordinator.ts | 12 +-
.../sqlite-session-metadata-store.test.ts | 43 ++--
packages/storage/src/execution-stores.ts | 5 +-
packages/storage/src/message-receipt-store.ts | 4 +-
packages/storage/src/session-message-projection.ts | 79 ++++++++
packages/storage/src/session-store.ts | 76 +------
.../storage/src/sqlite-session-metadata-store.ts | 221 ++++++++-------------
11 files changed, 258 insertions(+), 256 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 15bbebde4..e197d940f 100644
--- a/packages/runtime-host/src/__tests__/execution-host-message.test.ts
+++ b/packages/runtime-host/src/__tests__/execution-host-message.test.ts
@@ -168,6 +168,23 @@ test('steering becomes durable and ordered followups automatically start the nex
'followup',
);
}
+ const queueSubscription = await second.openSessionSubscription({
+ sessionId: fixture.sessionId,
+ transcript: { kind: 'none' },
+ });
+ const queuedFollowups = queueSubscription.snapshot.queue.followup;
+ assert.deepEqual(
+ queuedFollowups.map((entry) => entry.messageId),
+ followupSources.map((source) => source.messageId),
+ );
+ await second.request('queue.entries.reorder', {
+ originHostEpoch: host.hostEpoch,
+ sessionId: fixture.sessionId,
+ reorderId: randomUUID(),
+ entryIds: queuedFollowups.map((entry) => entry.entryId).reverse(),
+ });
+ await queueSubscription.close();
+ const orderedFollowupSources = [...followupSources].reverse();
assert.equal(
(
await second.request('turn.message.submit', {
@@ -222,22 +239,22 @@ test('steering becomes durable and ordered followups automatically start the nex
placement,
disposition,
})),
- followupSources.map((source) => ({
+ orderedFollowupSources.map((source) => ({
...source,
placement: 'next_turn',
disposition: 'followup',
})),
);
assert.deepEqual(chain[1]?.normalizedInput, {
- text: `${followupSources[0].content.text}\n\n${followupSources[1].content.text}`,
- displayText: `${followupSources[0].content.displayText}\n\n${followupSources[1].content.text}`,
- attachments: followupSources[0].content.attachments,
- quotes: followupSources.flatMap((source) => source.content.quotes ?? []),
+ text: `${orderedFollowupSources[0].content.text}\n\n${orderedFollowupSources[1].content.text}`,
+ displayText: `${orderedFollowupSources[0].content.text}\n\n${orderedFollowupSources[1].content.displayText}`,
+ attachments: orderedFollowupSources[1].content.attachments,
+ quotes: orderedFollowupSources.flatMap((source) => source.content.quotes ?? []),
});
const followupTurnId = chain[1]?.turnId;
assert.ok(followupTurnId);
const followupLedger = await fixture.readTurn(followupTurnId);
- const expectedQuotes = followupSources.flatMap((source) => source.content.quotes ?? []);
+ const expectedQuotes = orderedFollowupSources.flatMap((source) => source.content.quotes ?? []);
assert.equal(followupLedger.userMessages.length, followupSources.length);
assert.deepEqual(
followupLedger.userMessages.flatMap((message) => message.quotes ?? []),
@@ -245,7 +262,7 @@ test('steering becomes durable and ordered followups automatically start the nex
);
assert.deepEqual(userRuntimeContent(followupLedger.runtimeEvents)?.quotes, expectedQuotes);
const sessionUserMessages = await fixture.readSessionUserMessages();
- for (const source of followupSources) {
+ for (const source of orderedFollowupSources) {
assert.equal(
sessionUserMessages.filter((message) => message.id === source.messageId).length,
1,
@@ -253,7 +270,13 @@ test('steering becomes durable and ordered followups automatically start the nex
}
assert.equal(
sessionUserMessages.filter((message) => message.turnId === followupTurnId).length,
- followupSources.length,
+ orderedFollowupSources.length,
+ );
+ assert.deepEqual(
+ sessionUserMessages
+ .filter((message) => message.turnId === followupTurnId)
+ .map((message) => message.id),
+ orderedFollowupSources.map((source) => source.messageId),
);
});
});
@@ -306,6 +329,11 @@ test('explicit retract is durable across connections and prevents successor admi
await retrying.close();
await fixture.stopHost(host);
+ assert.equal(
+ (await fixture.readSessionUserMessages()).some((message) => message.id === messageId),
+ false,
+ 'a retracted draft must not remain in the durable transcript',
+ );
const chain = await fixture.readAdmissionChain();
assert.deepEqual(
chain.map((admission) => admission.turnId),
@@ -375,6 +403,12 @@ test('interrupt atomically retracts queued followup, stops the exact run, and is
await second.close();
await fixture.stopHost(host);
+ assert.equal(
+ (await fixture.readSessionUserMessages()).some((message) => message.id === followupId),
+ false,
+ 'an interrupted draft must not remain in the durable transcript',
+ );
+
const chain = await fixture.readAdmissionChain();
assert.equal(chain.length, 1);
assert.equal(chain[0]?.turnId, turnId);
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 1f911bce9..0c8000ea3 100644
--- a/packages/runtime-host/src/__tests__/execution-host-queue.test.ts
+++ b/packages/runtime-host/src/__tests__/execution-host-queue.test.ts
@@ -295,6 +295,12 @@ test('a Host crash after queue admission recovers the durable successor once', a
await second.close();
await fixture.stopHost(secondHost);
assert.equal(await fixture.readMessageLifecycleState(messageId), 'handed_off');
+ assert.deepEqual(
+ (await fixture.readSessionUserMessages())
+ .filter((message) => message.id === messageId)
+ .map((message) => message.id),
+ [messageId],
+ );
});
});
diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts
index a5ddf5805..79fed8583 100644
--- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts
@@ -2475,7 +2475,6 @@ function memoryMessageLifecycleStore(
admission.sessionId === sessionId && (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}`);
@@ -2490,7 +2489,7 @@ function memoryMessageLifecycleStore(
}
}
},
- markMessagesHandedOff: async (_sessionId, messageIds) => {
+ markMessagesHandedOff: async ({ messageIds }) => {
for (const messageId of messageIds) {
const existing = admissions.get(messageId);
if (existing?.state === 'accepted') existing.state = 'handed_off';
diff --git a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts
index 0fc991a4b..5e10392c3 100644
--- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts
@@ -812,7 +812,7 @@ test('idle turn.message.submit applies hosted Skill preparation before durable a
}
});
-test('idle Skill admission persists only canonical content before root handoff', async () => {
+test('idle Skill admission persists a canonical draft without history before root handoff', async () => {
const canonicalText = '<invoked-skill>Write clearly.</invoked-skill>\n\nDraft this.';
const fixture = await createFailureFixture({
registerBackend: (backends) =>
@@ -860,12 +860,13 @@ test('idle Skill admission persists only canonical content before root handoff',
displayText: '/skill:writer Draft this.',
inlineReferences: [],
});
- assert.deepEqual(
- (await fixture.stores.sessionStore.readMessages(fixture.sessionId)).map((message) => ({
- text: message.type === 'user' ? message.text : undefined,
- displayText: message.type === 'user' ? message.displayText : undefined,
- })),
- [{ text: canonicalText, displayText: '/skill:writer Draft this.' }],
+ assert.deepEqual(await fixture.stores.sessionStore.readMessages(fixture.sessionId), []);
+ assert.equal(
+ await fixture.stores.sessionStore.readMessageLifecycleState(
+ fixture.sessionId,
+ 'idle-skill-before-handoff',
+ ),
+ 'accepted',
);
} finally {
await fixture.dispose();
diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts
index c72595bf9..5b98f9219 100644
--- a/packages/runtime-host/src/server/message-coordinator.ts
+++ b/packages/runtime-host/src/server/message-coordinator.ts
@@ -569,13 +569,11 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
);
}
}
- await this.#lifecycle.rebindMessageAdmissionTranscript({
+ await this.#lifecycle.markMessagesHandedOff({
sessionId: input.sessionId,
- messageIds: [...new Set(input.messageIds)],
+ messageIds: handoff,
turnId: input.turnId,
- previousRootTurnId: input.previousRootTurnId,
});
- await this.#lifecycle.markMessagesHandedOff(input.sessionId, handoff);
}
/**
@@ -632,7 +630,11 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
handedOff.push(messageId);
}
}
- await this.#lifecycle.markMessagesHandedOff(input.sessionId, handedOff);
+ await this.#lifecycle.markMessagesHandedOff({
+ sessionId: input.sessionId,
+ messageIds: handedOff,
+ turnId: input.turnId,
+ });
const proved: string[] = [];
for (const messageId of messageIds) {
if (
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 3e3d0be12..6edbb861c 100644
--- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts
+++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts
@@ -239,10 +239,10 @@ describe('SqliteSessionMetadataStore', () => {
}
});
- test('atomically accepts a steering message and its canonical transcript', async () => {
+ test('materializes an accepted steering draft when it is handed off', async () => {
const store = createSqliteSessionMetadataStore(':memory:');
try {
- await store.create(fullHeader({ id: 'session-1' }));
+ await store.create(fullHeader({ id: 'session-1', connectionLocked: false }));
const admission: PendingMessageAdmission = {
sessionId: 'session-1',
turnId: 'turn-1',
@@ -265,6 +265,20 @@ describe('SqliteSessionMetadataStore', () => {
await store.readMessageAdmission('session-1', 'message-1'),
normalizedAdmission,
);
+ assert.deepEqual(await store.readMessages('session-1'), []);
+ assert.equal(await store.readMessageLifecycleState('session-1', 'message-1'), 'accepted');
+ assert.equal((await store.read('session-1')).header.lastMessageAt, 3);
+ assert.equal((await store.readCatalogRecord('session-1')).lastMessagePreview, undefined);
+ assert.equal((await store.read('session-1')).header.connectionLocked, false);
+ assert.deepEqual(
+ (await store.listMessageAdmissions('session-1')).map((entry) => entry.messageId),
+ ['message-1'],
+ );
+ await store.markMessagesHandedOff({
+ sessionId: 'session-1',
+ messageIds: ['message-1'],
+ turnId: 'turn-1',
+ });
assert.deepEqual(
(await store.readMessages('session-1')).map((message) => ({
id: message.id,
@@ -283,13 +297,10 @@ describe('SqliteSessionMetadataStore', () => {
},
],
);
- assert.equal(await store.readMessageLifecycleState('session-1', 'message-1'), 'accepted');
- assert.deepEqual(
- (await store.listMessageAdmissions('session-1')).map((entry) => entry.messageId),
- ['message-1'],
- );
- await store.markMessagesHandedOff('session-1', ['message-1']);
assert.equal(await store.readMessageLifecycleState('session-1', 'message-1'), 'handed_off');
+ assert.equal((await store.read('session-1')).header.lastMessageAt, 10);
+ assert.equal((await store.readCatalogRecord('session-1')).lastMessagePreview, 'submitted');
+ assert.equal((await store.read('session-1')).header.connectionLocked, true);
assert.deepEqual(await store.listMessageAdmissions('session-1'), []);
assert.deepEqual(
(await store.listUnsettledMessageAdmissions('session-1')).map((entry) => entry.messageId),
@@ -303,7 +314,7 @@ describe('SqliteSessionMetadataStore', () => {
}
});
- test('atomically accepts a follow-up message and its canonical transcript', async () => {
+ test('materializes an accepted follow-up under its successor root', async () => {
const store = createSqliteSessionMetadataStore(':memory:');
try {
await store.create(fullHeader({ id: 'session-followup-admission' }));
@@ -322,24 +333,16 @@ describe('SqliteSessionMetadataStore', () => {
admittedAt: 11,
});
assert.equal(admission.disposition, 'followup');
- assert.deepEqual(
- (await store.readMessages('session-followup-admission')).map((message) => ({
- id: message.id,
- turnId: message.turnId,
- })),
- [{ id: 'message-followup', turnId: 'turn-current' }],
- );
- await store.rebindMessageAdmissionTranscript({
+ assert.deepEqual(await store.readMessages('session-followup-admission'), []);
+ await store.markMessagesHandedOff({
sessionId: 'session-followup-admission',
messageIds: ['message-followup'],
turnId: 'turn-successor',
- previousRootTurnId: 'turn-current',
});
- await store.rebindMessageAdmissionTranscript({
+ await store.markMessagesHandedOff({
sessionId: 'session-followup-admission',
messageIds: ['message-followup'],
turnId: 'turn-successor',
- previousRootTurnId: 'turn-current',
});
assert.deepEqual(
(await store.readMessages('session-followup-admission')).map((message) => ({
diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts
index fcbae1a96..54688e694 100644
--- a/packages/storage/src/execution-stores.ts
+++ b/packages/storage/src/execution-stores.ts
@@ -438,16 +438,13 @@ async function createExecutionStoresForWrite<K extends StorageRootKind, E extend
run(() => sessionStore.listMessageAdmissions(sessionId)),
listUnsettledMessageAdmissions: (sessionId) =>
run(() => sessionStore.listUnsettledMessageAdmissions(sessionId)),
- rebindMessageAdmissionTranscript: (input) =>
- run(() => sessionStore.rebindMessageAdmissionTranscript(input)),
+ markMessagesHandedOff: (input) => run(() => sessionStore.markMessagesHandedOff(input)),
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),
diff --git a/packages/storage/src/message-receipt-store.ts b/packages/storage/src/message-receipt-store.ts
index 8fafab521..47c3558b0 100644
--- a/packages/storage/src/message-receipt-store.ts
+++ b/packages/storage/src/message-receipt-store.ts
@@ -57,16 +57,14 @@ export interface MessageLifecycleStore {
): Promise<MessageLifecycleState | undefined>;
listMessageAdmissions(sessionId: string): Promise<readonly PendingMessageAdmission[]>;
listUnsettledMessageAdmissions(sessionId: string): Promise<readonly PendingMessageAdmission[]>;
- rebindMessageAdmissionTranscript(input: {
+ markMessagesHandedOff(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>;
- markMessagesHandedOff(sessionId: string, messageIds: readonly string[]): Promise<void>;
markMessagesExecuted(sessionId: string, messageIds: readonly string[]): Promise<void>;
}
diff --git a/packages/storage/src/session-message-projection.ts b/packages/storage/src/session-message-projection.ts
new file mode 100644
index 000000000..11ba43fde
--- /dev/null
+++ b/packages/storage/src/session-message-projection.ts
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import type { StoredMessage, UserMessage } from '@maka/core/session';
+
+export function projectSessionCatalogMessages(messages: readonly StoredMessage[]): {
+ readonly lastMessageAt?: number;
+ readonly lastMessagePreview?: string;
+} {
+ const lastMessageAt = latestVisibleMessageAt(messages);
+ const lastMessagePreview = lastMessagePreviewForMessages(messages);
+ return {
+ ...(lastMessageAt === undefined ? {} : { lastMessageAt }),
+ ...(lastMessagePreview === undefined ? {} : { lastMessagePreview }),
+ };
+}
+
+export function catalogPreviewForUserMessage(message: UserMessage): string | undefined {
+ const text = normalizePreviewText(message.displayText ?? message.text);
+ if (text) return truncatePreview(text);
+ return message.attachments && message.attachments.length > 0 ? '附件' : undefined;
+}
+
+export function latestVisibleMessageAt(messages: readonly StoredMessage[]): number | undefined {
+ for (let index = messages.length - 1; index >= 0; index -= 1) {
+ const message = messages[index]!;
+ if (message.type === 'user' || message.type === 'assistant') return message.ts;
+ }
+ return undefined;
+}
+
+export function isVisibleSessionMessage(
+ message: StoredMessage,
+): message is Extract<StoredMessage, { type: 'user' | 'assistant' }> {
+ return message.type === 'user' || message.type === 'assistant';
+}
+
+export function lastMessagePreviewForMessages(
+ messages: readonly StoredMessage[],
+): string | undefined {
+ for (let index = messages.length - 1; index >= 0; index -= 1) {
+ const message = messages[index]!;
+ if (message.type === 'user') {
+ const preview = catalogPreviewForUserMessage(message);
+ if (preview !== undefined) return preview;
+ }
+ if (message.type === 'assistant') {
+ const text = normalizePreviewText(message.text);
+ if (text) return truncatePreview(text);
+ }
+ }
+ return undefined;
+}
+
+function normalizePreviewText(text: string): string {
+ return text.replace(/\s+/g, ' ').trim();
+}
+
+function truncatePreview(text: string, maxLength = 96): string {
+ const chars = Array.from(text);
+ if (chars.length <= maxLength) return text;
+ return `${chars.slice(0, maxLength - 1).join('')}…`;
+}
diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts
index c6bc1a59f..6325b8fc3 100644
--- a/packages/storage/src/session-store.ts
+++ b/packages/storage/src/session-store.ts
@@ -81,6 +81,13 @@ import {
type UserMessage,
} from '@maka/core/session';
import type { MessageLifecycleStore, PendingMessageAdmission } from './message-receipt-store.js';
+import {
+ isVisibleSessionMessage,
+ lastMessagePreviewForMessages,
+ latestVisibleMessageAt,
+ projectSessionCatalogMessages,
+} from './session-message-projection.js';
+export { projectSessionCatalogMessages };
const SESSION_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/;
@@ -862,9 +869,7 @@ class SqliteSessionStore implements SessionAuthorityStore {
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;
+ return this.metadata.commitMessageAdmission(admission);
}
async readMessageAdmission(
@@ -887,14 +892,13 @@ class SqliteSessionStore implements SessionAuthorityStore {
return this.metadata.listUnsettledMessageAdmissions(sessionId);
}
- async rebindMessageAdmissionTranscript(input: {
+ async markMessagesHandedOff(input: {
sessionId: string;
messageIds: readonly string[];
turnId: string;
- previousRootTurnId: string | null;
}): Promise<void> {
await this.ensureReady();
- await this.metadata.rebindMessageAdmissionTranscript(input);
+ await this.metadata.markMessagesHandedOff(input);
for (const listener of this.transcriptChangeListeners) listener(input.sessionId);
}
@@ -909,7 +913,6 @@ class SqliteSessionStore implements SessionAuthorityStore {
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> {
@@ -922,11 +925,6 @@ class SqliteSessionStore implements SessionAuthorityStore {
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);
@@ -1435,32 +1433,6 @@ function toCatalogSummary(
};
}
-export function projectSessionCatalogMessages(messages: readonly StoredMessage[]): {
- readonly lastMessageAt?: number;
- readonly lastMessagePreview?: string;
-} {
- const lastMessageAt = latestVisibleMessageAt(messages);
- const lastMessagePreview = lastMessagePreviewForMessages(messages);
- return {
- ...(lastMessageAt === undefined ? {} : { lastMessageAt }),
- ...(lastMessagePreview === undefined ? {} : { lastMessagePreview }),
- };
-}
-
-function latestVisibleMessageAt(messages: readonly StoredMessage[]): number | undefined {
- for (let index = messages.length - 1; index >= 0; index -= 1) {
- const message = messages[index]!;
- if (isVisibleSessionMessage(message)) return message.ts;
- }
- return undefined;
-}
-
-function isVisibleSessionMessage(
- message: StoredMessage,
-): message is Extract<StoredMessage, { type: 'user' | 'assistant' }> {
- return message.type === 'user' || message.type === 'assistant';
-}
-
function maxTimestamp(left: number | undefined, right: number | undefined): number | undefined {
if (left === undefined) return right;
if (right === undefined) return left;
@@ -1471,34 +1443,6 @@ function normalizeSessionName(name: string): string {
return name === 'New Session' ? DEFAULT_SESSION_NAME : name;
}
-function lastMessagePreviewForMessages(messages: readonly StoredMessage[]): string | undefined {
- for (let index = messages.length - 1; index >= 0; index -= 1) {
- const message = messages[index]!;
- if (message.type === 'user') {
- // Prefer the human-facing view when the stored model text is a composed
- // envelope (e.g. explicit skill invocation).
- const text = normalizePreviewText(message.displayText ?? message.text);
- if (text) return truncatePreview(text);
- if (message.attachments && message.attachments.length > 0) return '附件';
- }
- if (message.type === 'assistant') {
- const text = normalizePreviewText(message.text);
- if (text) return truncatePreview(text);
- }
- }
- return undefined;
-}
-
-function normalizePreviewText(text: string): string {
- return text.replace(/\s+/g, ' ').trim();
-}
-
-function truncatePreview(text: string, maxLength = 96): string {
- const chars = Array.from(text);
- if (chars.length <= maxLength) return text;
- return `${chars.slice(0, maxLength - 1).join('')}…`;
-}
-
export function createUserMessage(input: {
turnId: string;
text: string;
diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts
index 6954c2b59..6066af774 100644
--- a/packages/storage/src/sqlite-session-metadata-store.ts
+++ b/packages/storage/src/sqlite-session-metadata-store.ts
@@ -134,6 +134,7 @@ import {
isDiscardableConversationCopy,
isValidConversationCopyTransition,
} from './session-conversation-copy.js';
+import { catalogPreviewForUserMessage } from './session-message-projection.js';
import {
configureSqliteSessionMetadataDatabase,
migrateSqliteSessionMetadataDatabase,
@@ -1545,8 +1546,7 @@ export class SqliteSessionMetadataStore {
this.assertOpen();
const stored = normalizePendingMessageAdmission(admission);
return this.transaction(() => {
- const record = this.readRecordSync(stored.sessionId);
- if (!record) throw new SessionNotFoundError(stored.sessionId);
+ if (!this.readRecordSync(stored.sessionId)) throw new SessionNotFoundError(stored.sessionId);
const existingRow = this.db
.prepare(
`
@@ -1602,52 +1602,6 @@ export class SqliteSessionMetadataStore {
stored.admittedAt,
);
- if (stored.disposition === 'steering' || stored.disposition === 'followup') {
- const message = decodeCanonicalMessage({
- type: 'user',
- id: stored.messageId,
- turnId: stored.turnId,
- ts: stored.admittedAt,
- ...stored.content,
- steeringEventId: stored.messageId,
- });
- const existingMessages = this.readMessagesWith(
- stored.sessionId,
- decodeStoredMessage,
- ).filter((candidate) => candidate.id === stored.messageId);
- if (existingMessages.length > 1) {
- throw new SessionMetadataConflictError(
- 'Message admission transcript identity is ambiguous',
- );
- }
- const existingMessage = existingMessages[0];
- if (existingMessage && !isDeepStrictEqual(existingMessage, message)) {
- throw new SessionMetadataConflictError('Message admission transcript identity conflict');
- }
- if (!existingMessage) {
- 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');
- }
- const json = JSON.stringify(message);
- this.insertSessionMessagesSync(stored.sessionId, row.last_sequence + 1, [
- { message, json },
- ]);
- this.updateCatalogProjectionSync(
- stored.sessionId,
- {
- lastMessageAt: stored.admittedAt,
- lastMessagePreview: message.type === 'user' ? message.displayText : undefined,
- },
- false,
- !record.header.connectionLocked,
- );
- }
- }
return stored;
});
}
@@ -1745,21 +1699,30 @@ export class SqliteSessionMetadataStore {
});
}
- async rebindMessageAdmissionTranscript(input: {
+ async markMessagesHandedOff(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(() => {
+ const lastSequenceRow = this.db
+ .prepare(
+ 'SELECT COALESCE(MAX(sequence), -1) AS last_sequence FROM session_messages WHERE session_id = ?',
+ )
+ .get(input.sessionId) as { last_sequence?: unknown };
+ if (
+ typeof lastSequenceRow.last_sequence !== 'number' ||
+ !Number.isSafeInteger(lastSequenceRow.last_sequence)
+ ) {
+ throw new SessionMetadataConflictError('Invalid Session message sequence');
+ }
+ let nextSequence = lastSequenceRow.last_sequence + 1;
+ const materialized: StoredMessage[] = [];
for (const messageId of unique) {
const admissionRow = this.db
.prepare(
@@ -1775,11 +1738,7 @@ export class SqliteSessionMetadataStore {
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'
- ) {
+ if (admission.lifecycleState === 'cancelled') {
throw new SessionMetadataConflictError('Message admission is already cancelled');
}
const rows = this.db
@@ -1798,36 +1757,74 @@ export class SqliteSessionMetadataStore {
record_bytes?: unknown;
sha256?: unknown;
}>;
- if (rows.length !== 1) {
+ if (rows.length > 1) {
throw new SessionMetadataConflictError(
- rows.length === 0
- ? 'Message admission transcript is missing'
- : 'Message admission transcript identity is ambiguous',
+ '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 (rows.length === 0) {
+ if (admission.lifecycleState !== 'accepted') {
+ throw new SessionMetadataConflictError('Handed-off Message transcript is missing');
+ }
+ const message = decodeCanonicalMessage({
+ type: 'user',
+ id: messageId,
+ turnId: input.turnId,
+ ts: admission.admission.admittedAt,
+ ...admission.admission.content,
+ steeringEventId: messageId,
+ });
+ const json = JSON.stringify(message);
+ this.insertSessionMessagesSync(input.sessionId, nextSequence, [{ message, json }]);
+ nextSequence += 1;
+ materialized.push(message);
+ } else {
+ 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) {
+ throw new SessionMetadataConflictError('Message admission transcript Turn conflict');
+ }
}
- if (message.turnId === input.turnId) continue;
- if (
- message.turnId !== admission.admission.turnId &&
- (input.previousRootTurnId === null || message.turnId !== input.previousRootTurnId)
- ) {
- throw new SessionMetadataConflictError('Message admission transcript Turn conflict');
+ if (admission.lifecycleState === 'accepted') {
+ const transitioned = this.db
+ .prepare(
+ `
+ UPDATE message_admissions
+ SET lifecycle_state = 'handed_off'
+ WHERE session_id = ? AND message_id = ? AND lifecycle_state = 'accepted'
+ `,
+ )
+ .run(input.sessionId, messageId);
+ if (transitioned.changes !== 1) {
+ throw new SessionMetadataConflictError('Message admission lifecycle identity conflict');
+ }
}
- const rebound = decodeCanonicalMessage({ ...message, turnId: input.turnId });
- this.replaceSessionMessageSync(input.sessionId, sequence, rebound);
+ }
+ const latest = materialized.at(-1);
+ if (latest?.type === 'user') {
+ this.updateCatalogProjectionSync(
+ input.sessionId,
+ {
+ lastMessageAt: latest.ts,
+ lastMessagePreview: catalogPreviewForUserMessage(latest),
+ },
+ false,
+ true,
+ );
}
});
}
@@ -1875,60 +1872,6 @@ export class SqliteSessionMetadataStore {
stored.sessionId,
stored.messageId,
);
- if (stored.disposition !== 'steering' && stored.disposition !== 'followup') 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.replaceSessionMessageSync(stored.sessionId, sequence, message, json);
- }
- this.updateCatalogProjectionSync(
- stored.sessionId,
- {
- lastMessageAt: stored.admittedAt,
- lastMessagePreview: message.type === 'user' ? message.displayText : undefined,
- },
- true,
- );
});
}
@@ -2003,10 +1946,6 @@ export class SqliteSessionMetadataStore {
});
}
- 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');
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org