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/26 14:58:05 UTC
(maka) 14/26: fix: reconcile transient messages from durable proof
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch fix/canonical-message-projection
in repository https://gitbox.apache.org/repos/asf/maka.git
commit a1324abaa1aebad3dc68afcb583d4d4b9b112370
Author: AstroHan <le...@outlook.com>
AuthorDate: Wed Aug 26 10:10:05 2026 +0800
fix: reconcile transient messages from durable proof
Generated-by: Codex
---
...runtime-host-session-execution-ipc-main.test.ts | 36 ++++++++++--
.../__tests__/transient-message-projection.test.ts | 20 +++++++
apps/desktop/src/main/runtime-host-client.ts | 6 ++
.../runtime-host-session-execution-ipc-main.ts | 16 ++++--
apps/desktop/src/preload/bridge-contract.d.ts | 9 +--
apps/desktop/src/preload/preload.ts | 16 ++++--
apps/desktop/src/renderer/app-shell.tsx | 2 +
.../src/renderer/transient-message-projection.ts | 10 ++++
.../renderer/use-app-shell-session-workspace.ts | 23 ++++++++
packages/cli/src/__tests__/pi-transcript.test.ts | 19 +++++++
packages/cli/src/__tests__/pi-tui-runner.test.ts | 7 +--
packages/cli/src/__tests__/pi-tui-turn.test.ts | 7 +--
.../__tests__/runtime-host-session-driver.test.ts | 28 +++------
packages/cli/src/pi-transcript.ts | 15 ++++-
packages/cli/src/pi-tui-runner.ts | 63 ++++++++++-----------
packages/cli/src/pi-tui-turn.ts | 7 +--
packages/cli/src/runtime-host-session-driver.ts | 19 ++++---
packages/cli/src/session-driver.ts | 14 ++---
.../src/__tests__/message-coordinator.test.ts | 63 +++++++++++++++++++++
.../runtime-host/src/__tests__/protocol.test.ts | 13 +++++
packages/runtime-host/src/protocol/index.ts | 5 +-
packages/runtime-host/src/protocol/message.ts | 66 ++++++++++++++++++++++
packages/runtime-host/src/protocol/operations.ts | 1 +
.../runtime-host/src/server/message-coordinator.ts | 44 +++++++++++++++
.../src/server/operation-dispatcher.ts | 1 +
.../sqlite-session-metadata-store.test.ts | 5 ++
packages/storage/src/execution-stores.ts | 2 +
packages/storage/src/message-admission-store.ts | 10 ++++
packages/storage/src/session-store.ts | 5 ++
.../storage/src/sqlite-session-metadata-store.ts | 42 ++++++++++++++
30 files changed, 469 insertions(+), 105 deletions(-)
diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts
index 4ae67b122..165367d3e 100644
--- a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts
+++ b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts
@@ -624,7 +624,6 @@ test("submits an ordinary composer message once under its stable message identit
assert.deepEqual(result, {
ok: true,
disposition: "turn_started",
- messageId: "message-1",
turnId: "host-turn",
attachments: [],
inlineReferences: [],
@@ -632,6 +631,36 @@ test("submits an ordinary composer message once under its stable message identit
});
});
+test('returns Host-owned Message lifecycle proof to the renderer', async () => {
+ const ipc = ipcHarness();
+ registerExecutionIpc(
+ {
+ client: executionClient({
+ queryMessages: async (input) => ({
+ messages: input.messageIds.map((messageId) => ({
+ messageId,
+ status: messageId === 'message-cancelled' ? 'cancelled' : 'accepted',
+ })),
+ }),
+ }),
+ },
+ ipc,
+ );
+
+ assert.deepEqual(
+ await ipc.invoke('sessions:queryMessageStatuses', 'session-1', [
+ 'message-accepted',
+ 'message-cancelled',
+ ]),
+ {
+ messages: [
+ { messageId: 'message-accepted', status: 'accepted' },
+ { messageId: 'message-cancelled', status: 'cancelled' },
+ ],
+ },
+ );
+});
+
test('keeps slash Skill sends on the exact-Turn path with their stable message identity', async () => {
const starts: unknown[] = [];
const ipc = ipcHarness();
@@ -938,7 +967,6 @@ test("retries a dispatched busy fallback with its original message identity", as
{
ok: false,
reason: "outcome_unknown",
- messageId: "turn-unknown",
skillInvocation: { loaded: [], failed: [], receipts: [] },
},
);
@@ -1138,7 +1166,6 @@ test("queues explicit Desktop follow-ups", async () => {
}),
{
kind: "queued",
- messageId: "followup-message",
attachments: [
{
kind: "other",
@@ -1211,7 +1238,7 @@ test('keeps an unknown Desktop follow-up admission available for reconciliation'
messageId: 'followup-unknown',
text: 'keep this visible',
}),
- { kind: 'outcome_unknown', messageId: 'followup-unknown' },
+ { kind: 'outcome_unknown' },
);
});
@@ -1536,6 +1563,7 @@ function executionClient(overrides: Partial<ExecutionClient>): ExecutionClient {
interruptTurn: unavailable,
listSessionTurnLandmarks: unavailable,
listSessionTurns: unavailable,
+ queryMessages: unavailable,
queryTurnResume: unavailable,
readExecutionBoundary: unavailable,
regenerateTurn: unavailable,
diff --git a/apps/desktop/src/main/__tests__/transient-message-projection.test.ts b/apps/desktop/src/main/__tests__/transient-message-projection.test.ts
index ee8447b22..120bbd6a2 100644
--- a/apps/desktop/src/main/__tests__/transient-message-projection.test.ts
+++ b/apps/desktop/src/main/__tests__/transient-message-projection.test.ts
@@ -23,6 +23,7 @@ import type { StoredMessage } from '@maka/core/session';
import {
mergeTransientMessageProjection,
projectQueuedTransientMessages,
+ reconcileTransientMessageLifecycle,
reconcileTransientMessages,
} from '../../renderer/transient-message-projection.js';
@@ -67,6 +68,25 @@ test('replaces a transient message by canonical message id exactly once', () =>
assert.equal(pending.size, 0);
});
+test('removes only messages with durable cancellation proof after reconnect', () => {
+ const accepted = { ...transient, id: 'message-accepted' };
+ const handedOff = { ...transient, id: 'message-handed-off' };
+ const cancelled = { ...transient, id: 'message-cancelled' };
+ const unknown = { ...transient, id: 'message-unknown' };
+ const pending = new Map(
+ [accepted, handedOff, cancelled, unknown].map((message) => [message.id, message]),
+ );
+
+ reconcileTransientMessageLifecycle(pending, [
+ { messageId: accepted.id, status: 'accepted' },
+ { messageId: handedOff.id, status: 'handed_off' },
+ { messageId: cancelled.id, status: 'cancelled' },
+ { messageId: unknown.id, status: 'unknown' },
+ ]);
+
+ assert.deepEqual([...pending.keys()], [accepted.id, handedOff.id, unknown.id]);
+});
+
test('canonicalizing one send does not hide a later transient send', () => {
const second = {
...transient,
diff --git a/apps/desktop/src/main/runtime-host-client.ts b/apps/desktop/src/main/runtime-host-client.ts
index c1b539c1a..e07d62512 100644
--- a/apps/desktop/src/main/runtime-host-client.ts
+++ b/apps/desktop/src/main/runtime-host-client.ts
@@ -1094,6 +1094,12 @@ export class DesktopRuntimeHostClient {
});
}
+ queryMessages(
+ input: OperationInput<'turn.message.query'>,
+ ): Promise<OperationOutput<'turn.message.query'>> {
+ return this.request('turn.message.query', input);
+ }
+
retractQueueEntry(
input: Omit<QueueEntryRetractInput, "originHostEpoch">,
): Promise<QueueMutationResult> {
diff --git a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts
index dd68f8420..6ffd2e96f 100644
--- a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts
+++ b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts
@@ -97,6 +97,7 @@ type RuntimeHostSessionExecutionClient = Pick<
| "interruptTurn"
| 'listSessionTurns'
| 'listSessionTurnLandmarks'
+ | 'queryMessages'
| "queryTurnResume"
| "readExecutionBoundary"
| "regenerateTurn"
@@ -191,6 +192,14 @@ export function registerRuntimeHostSessionExecutionIpc(
const newId = deps.newId ?? randomUUID;
const stopSession = createRuntimeHostSessionStop(deps, newId);
+ ipcMain.handle(
+ 'sessions:queryMessageStatuses',
+ async (_event, sessionId: string, messageIds: unknown) => {
+ if (!Array.isArray(messageIds)) throw new Error('Invalid Message identities');
+ return deps.client.queryMessages({ sessionId, messageIds });
+ },
+ );
+
handleReconnectableRead(
ipcMain,
"sessions:observe",
@@ -340,7 +349,6 @@ export function registerRuntimeHostSessionExecutionIpc(
return {
ok: false as const,
reason: 'outcome_unknown' as const,
- messageId: command.messageId,
skillInvocation,
};
}
@@ -351,7 +359,6 @@ export function registerRuntimeHostSessionExecutionIpc(
return {
ok: true as const,
disposition: submitted.disposition,
- messageId: command.messageId,
turnId: submitted.turnId,
attachments,
inlineReferences,
@@ -362,7 +369,6 @@ export function registerRuntimeHostSessionExecutionIpc(
return {
ok: true as const,
disposition: submitted.disposition,
- messageId: command.messageId,
attachments,
inlineReferences,
skillInvocation,
@@ -548,7 +554,7 @@ export function registerRuntimeHostSessionExecutionIpc(
inlineReferences,
},
});
- if (!result) return { kind: 'outcome_unknown' as const, messageId };
+ if (!result) return { kind: 'outcome_unknown' as const };
if (result.disposition === "turn_started") {
deps.emitSessionsChanged("status-change", sessionId, {
turnId: result.turnId,
@@ -556,14 +562,12 @@ export function registerRuntimeHostSessionExecutionIpc(
return {
kind: "started" as const,
turnId: result.turnId,
- messageId,
attachments,
inlineReferences,
};
}
return {
kind: "queued" as const,
- messageId,
attachments,
inlineReferences,
};
diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts
index c804ee2d2..f5d768b79 100644
--- a/apps/desktop/src/preload/bridge-contract.d.ts
+++ b/apps/desktop/src/preload/bridge-contract.d.ts
@@ -822,7 +822,6 @@ export interface MakaBridge {
| {
ok: true;
disposition: 'turn_started' | 'steering' | 'followup';
- messageId: string;
turnId?: string;
attachments: import('@maka/core/events').AttachmentRef[];
inlineReferences: import('@maka/core/events').InlineReference[];
@@ -836,7 +835,6 @@ export interface MakaBridge {
| {
ok: false;
reason: 'outcome_unknown';
- messageId: string;
skillInvocation: import('@maka/runtime/skill-invocation').SkillInvocationResult;
}
>;
@@ -927,12 +925,15 @@ export interface MakaBridge {
| {
kind: 'queued' | 'started';
turnId?: string;
- messageId: string;
attachments: import('@maka/core/events').AttachmentRef[];
inlineReferences: import('@maka/core/events').InlineReference[];
}
- | { kind: 'outcome_unknown'; messageId: string }
+ | { kind: 'outcome_unknown' }
>;
+ queryMessageStatuses(
+ sessionId: string,
+ messageIds: readonly string[],
+ ): Promise<import('@maka/runtime-host/protocol').TurnMessageQueryResult>;
retractQueueEntry(sessionId: string, entryId: string): Promise<void>;
promoteQueueEntry(sessionId: string, entryId: string): Promise<void>;
updateQueueEntry(
diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts
index 9578adf94..f8cb0ac41 100644
--- a/apps/desktop/src/preload/preload.ts
+++ b/apps/desktop/src/preload/preload.ts
@@ -1737,11 +1737,10 @@ const makaBridge = {
| {
kind: 'queued' | 'started';
turnId?: string;
- messageId: string;
attachments: AttachmentRef[];
inlineReferences: InlineReference[];
}
- | { kind: 'outcome_unknown'; messageId: string }
+ | { kind: 'outcome_unknown' }
> {
const session = await runtimeHostSessionRef(sessionId);
const attachmentItems = command.attachmentItems
@@ -1760,17 +1759,26 @@ const makaBridge = {
| {
kind: 'queued' | 'started';
turnId?: string;
- messageId: string;
attachments: AttachmentRef[];
inlineReferences: InlineReference[];
}
- | { kind: 'outcome_unknown'; messageId: string };
+ | { kind: 'outcome_unknown' };
if (result.kind === 'outcome_unknown') return result;
return {
...result,
attachments: projectDesktopAttachmentRefs(session.scope, result.attachments),
};
},
+ queryMessageStatuses(
+ sessionId: string,
+ messageIds: readonly string[],
+ ): Promise<import('@maka/runtime-host/protocol').TurnMessageQueryResult> {
+ return invokeSessionRuntimeHost(
+ 'sessions:queryMessageStatuses',
+ sessionId,
+ messageIds,
+ );
+ },
retractQueueEntry(sessionId: string, entryId: string): Promise<void> {
return invokeSessionRuntimeHost('sessions:retractQueueEntry', sessionId, entryId);
},
diff --git a/apps/desktop/src/renderer/app-shell.tsx b/apps/desktop/src/renderer/app-shell.tsx
index 86966881b..3df0adf91 100644
--- a/apps/desktop/src/renderer/app-shell.tsx
+++ b/apps/desktop/src/renderer/app-shell.tsx
@@ -352,6 +352,7 @@ function AppShellContent({
addTransientMessage,
updateTransientMessage,
projectQueuedTransientMessages,
+ reconcileTransientMessageStatuses,
removeTransientMessage,
transcriptRangeRef,
messageLoadPending,
@@ -2308,6 +2309,7 @@ function AppShellContent({
const next = completeLiveContentSeed(current, sessionId, expected);
activeEventSeedRef.current = next;
setActiveEventSeed(next);
+ void reconcileTransientMessageStatuses(sessionId);
};
useActiveSessionEvents({
uiLocale,
diff --git a/apps/desktop/src/renderer/transient-message-projection.ts b/apps/desktop/src/renderer/transient-message-projection.ts
index cc4d93594..7c89f0136 100644
--- a/apps/desktop/src/renderer/transient-message-projection.ts
+++ b/apps/desktop/src/renderer/transient-message-projection.ts
@@ -18,6 +18,7 @@
*/
import type { StoredMessage } from '@maka/core/session';
+import type { MessageLifecycleStatus } from '@maka/runtime-host/protocol';
import type { TransientUserMessageProjection } from '@maka/ui';
type TransientUserMessage = TransientUserMessageProjection;
@@ -51,6 +52,15 @@ export function mergeTransientMessageProjection(
return hostBoundCurrentTurn ? { ...update, turnId: current.turnId } : update;
}
+export function reconcileTransientMessageLifecycle(
+ transient: Map<string, TransientUserMessage>,
+ messages: readonly { messageId: string; status: MessageLifecycleStatus }[],
+): void {
+ for (const message of messages) {
+ if (message.status === 'cancelled') transient.delete(message.messageId);
+ }
+}
+
/**
* Project renderer-only messages beside the canonical transcript until the
* canonical transcript carries the same message id. Keeping the two arrays
diff --git a/apps/desktop/src/renderer/use-app-shell-session-workspace.ts b/apps/desktop/src/renderer/use-app-shell-session-workspace.ts
index 0c10cb1f1..06223d38f 100644
--- a/apps/desktop/src/renderer/use-app-shell-session-workspace.ts
+++ b/apps/desktop/src/renderer/use-app-shell-session-workspace.ts
@@ -32,6 +32,7 @@ import type { DesktopTranscriptRangeController } from './desktop-transcript-rang
import {
mergeTransientMessageProjection,
projectQueuedTransientMessages as applyQueuedTransientProjection,
+ reconcileTransientMessageLifecycle,
reconcileTransientMessages,
} from './transient-message-projection.js';
@@ -122,6 +123,7 @@ export function useAppShellSessionWorkspace(toastApi: ToastApi) {
messages: readonly TransientUserMessage[],
): void {
let pending = transientMessagesBySessionRef.current.get(sessionId);
+ if (!pending && messages.length === 0) return;
if (!pending) {
pending = new Map();
transientMessagesBySessionRef.current.set(sessionId, pending);
@@ -132,6 +134,26 @@ export function useAppShellSessionWorkspace(toastApi: ToastApi) {
}
}
+ async function reconcileTransientMessageStatuses(sessionId: string): Promise<void> {
+ const pending = transientMessagesBySessionRef.current.get(sessionId);
+ if (!pending || pending.size === 0) return;
+ try {
+ const result = await window.maka.sessions.queryMessageStatuses(
+ sessionId,
+ [...pending.keys()],
+ );
+ const current = transientMessagesBySessionRef.current.get(sessionId);
+ if (!current) return;
+ reconcileTransientMessageLifecycle(current, result.messages);
+ if (current.size === 0) transientMessagesBySessionRef.current.delete(sessionId);
+ if (activeIdRef.current === sessionId) {
+ setTransientMessages(projectTransientMessages(sessionId, messagesRef.current));
+ }
+ } catch {
+ // A failed proof query leaves presentation intact until canonical proof arrives.
+ }
+ }
+
function removeTransientMessage(sessionId: string, messageId: string): void {
const pending = transientMessagesBySessionRef.current.get(sessionId);
if (!pending?.delete(messageId)) return;
@@ -197,6 +219,7 @@ export function useAppShellSessionWorkspace(toastApi: ToastApi) {
addTransientMessage,
updateTransientMessage,
projectQueuedTransientMessages,
+ reconcileTransientMessageStatuses,
removeTransientMessage,
transcriptRangeRef,
messageLoadPending,
diff --git a/packages/cli/src/__tests__/pi-transcript.test.ts b/packages/cli/src/__tests__/pi-transcript.test.ts
index e06e35c19..84fda1ae2 100644
--- a/packages/cli/src/__tests__/pi-transcript.test.ts
+++ b/packages/cli/src/__tests__/pi-transcript.test.ts
@@ -37,6 +37,7 @@ import {
refreshRunningShellRunElapsed,
hydrateToolsWithStoredMessages,
makaPiToolPresentationStatus,
+ reconcileTransientMessageLifecycle,
replaceTranscriptWithStoredMessages,
submitCompactToTranscript,
toggleAllThinkingExpansion,
@@ -358,6 +359,24 @@ describe('Maka Pi TUI transcript', () => {
]);
});
+ test('removes only transient rows with durable cancellation proof', () => {
+ const state = createMakaPiTranscriptState();
+ appendUserPrompt(state, 'accepted', 'message-accepted', true);
+ appendUserPrompt(state, 'handed off', 'message-handed-off', true);
+ appendUserPrompt(state, 'cancelled', 'message-cancelled', true);
+
+ reconcileTransientMessageLifecycle(state, [
+ { messageId: 'message-accepted', status: 'accepted' },
+ { messageId: 'message-handed-off', status: 'handed_off' },
+ { messageId: 'message-cancelled', status: 'cancelled' },
+ ]);
+
+ assert.deepEqual(
+ state.entries.map((entry) => ('messageId' in entry ? entry.messageId : undefined)),
+ ['message-accepted', 'message-handed-off'],
+ );
+ });
+
test('keeps a transient user row before later durable output in a sparse replacement', () => {
const state = createMakaPiTranscriptState();
replaceTranscriptWithStoredMessages(state, [
diff --git a/packages/cli/src/__tests__/pi-tui-runner.test.ts b/packages/cli/src/__tests__/pi-tui-runner.test.ts
index afae12a8d..dfb74021a 100644
--- a/packages/cli/src/__tests__/pi-tui-runner.test.ts
+++ b/packages/cli/src/__tests__/pi-tui-runner.test.ts
@@ -7008,7 +7008,7 @@ class SteeringTurnDriver implements MakaSessionDriver {
summary: fakeSessionSummary(turn.sessionId),
}),
);
- return { messageId: options.messageId, disposition: 'turn_started' as const };
+ return;
}
if (options.placement === 'current_turn') {
this.steered.push(text);
@@ -7018,11 +7018,6 @@ class SteeringTurnDriver implements MakaSessionDriver {
this.followup.push({ messageId: options.messageId, text });
}
this.emitQueueUpdate();
- return {
- messageId: options.messageId,
- disposition:
- options.placement === 'current_turn' ? ('steering' as const) : ('followup' as const),
- };
}
subscribeStartedTurns(listener: (turn: MakaAttachedSessionTurn) => void): () => void {
diff --git a/packages/cli/src/__tests__/pi-tui-turn.test.ts b/packages/cli/src/__tests__/pi-tui-turn.test.ts
index 8606611c1..f3a61af77 100644
--- a/packages/cli/src/__tests__/pi-tui-turn.test.ts
+++ b/packages/cli/src/__tests__/pi-tui-turn.test.ts
@@ -39,7 +39,6 @@ describe('Maka Pi TUI turn', () => {
placement: 'current_turn',
modelText: 'expanded prompt',
});
- return { messageId: options.messageId, disposition: 'turn_started' };
},
},
turnActivity: { activities: new SessionActivityRegistry() },
@@ -57,11 +56,7 @@ describe('Maka Pi TUI turn', () => {
},
});
- assert.deepEqual(outcome, {
- kind: 'admitted',
- messageId: 'message-1',
- disposition: 'turn_started',
- });
+ assert.deepEqual(outcome, { kind: 'admitted' });
assert.deepEqual(sequence, ['start', 'submit']);
});
diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts
index 14b37b200..70687991a 100644
--- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts
+++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts
@@ -1108,13 +1108,10 @@ describe('Runtime Host Maka Session driver', () => {
});
await driver.switchSession('session-1');
- assert.deepEqual(
- await driver.submitMessage!('Later', {
- messageId: 'message-1',
- placement: 'next_turn',
- }),
- { messageId: 'message-1', disposition: 'followup' },
- );
+ await driver.submitMessage!('Later', {
+ messageId: 'message-1',
+ placement: 'next_turn',
+ });
assert.deepEqual(await driver.retractQueued!(), {
text: 'Later',
messageIds: ['message-1'],
@@ -1159,16 +1156,11 @@ describe('Runtime Host Maka Session driver', () => {
});
await driver.switchSession('session-1');
- const admission = await driver.submitMessage!('Visible prompt', {
+ await driver.submitMessage!('Visible prompt', {
messageId: 'message-1',
placement: 'current_turn',
modelText: 'Expanded prompt',
});
-
- assert.deepEqual(admission, {
- messageId: 'message-1',
- disposition: 'steering',
- });
assert.deepEqual(connection.requests.at(-1), {
operation: 'turn.message.submit',
input: {
@@ -1199,12 +1191,11 @@ describe('Runtime Host Maka Session driver', () => {
});
await driver.switchSession('session-1');
- assert.deepEqual(
- await driver.submitMessage!('Keep this visible', {
+ await assert.doesNotReject(() =>
+ driver.submitMessage!('Keep this visible', {
messageId: 'message-unknown',
placement: 'current_turn',
}),
- { messageId: 'message-unknown', disposition: 'outcome_unknown' },
);
});
@@ -1227,12 +1218,11 @@ describe('Runtime Host Maka Session driver', () => {
});
await driver.switchSession('session-1');
- assert.deepEqual(
- await driver.submitMessage!('Keep this visible', {
+ await assert.doesNotReject(() =>
+ driver.submitMessage!('Keep this visible', {
messageId: 'message-interrupted',
placement: 'current_turn',
}),
- { messageId: 'message-interrupted', disposition: 'outcome_unknown' },
);
});
diff --git a/packages/cli/src/pi-transcript.ts b/packages/cli/src/pi-transcript.ts
index 7c731355f..6a6564a6f 100644
--- a/packages/cli/src/pi-transcript.ts
+++ b/packages/cli/src/pi-transcript.ts
@@ -60,7 +60,7 @@ import { goalStatusLineText, isLiveGoalStatus } from './pi-goal.js';
import { renderToolBlock } from './pi-transcript-tools.js';
import { getTuiPrimaryGuidance } from './tui-primary-guidance.js';
import { renderTuiShortcutCopy } from './tui-shortcut-copy.js';
-import type { GoalProjection } from '@maka/runtime-host/protocol';
+import type { GoalProjection, MessageLifecycleStatus } from '@maka/runtime-host/protocol';
export interface MakaPiUsageSummary {
/** Cumulative cost in USD across the session. */
@@ -259,6 +259,19 @@ export function appendUserPrompt(
state.entries.push(entry);
}
+export function reconcileTransientMessageLifecycle(
+ state: MakaPiTranscriptState,
+ messages: readonly { messageId: string; status: MessageLifecycleStatus }[],
+): void {
+ const cancelled = new Set(
+ messages.filter(({ status }) => status === 'cancelled').map(({ messageId }) => messageId),
+ );
+ if (cancelled.size === 0) return;
+ state.entries = state.entries.filter(
+ (entry) => entry.kind !== 'user' || entry.transient !== true || !cancelled.has(entry.messageId),
+ );
+}
+
export function appendTurnFailureToTranscript(state: MakaPiTranscriptState, error: unknown): void {
clearPendingInteractions(state);
state.entries.push({
diff --git a/packages/cli/src/pi-tui-runner.ts b/packages/cli/src/pi-tui-runner.ts
index 4e6f0a897..1ee582295 100644
--- a/packages/cli/src/pi-tui-runner.ts
+++ b/packages/cli/src/pi-tui-runner.ts
@@ -95,6 +95,7 @@ import {
completePendingInteraction,
applyShellRunViewUpdateToTranscript,
permissionModeLabel,
+ reconcileTransientMessageLifecycle,
replaceTranscriptWithStoredMessages,
hydrateToolsWithStoredMessages,
submitCompactToTranscript,
@@ -570,6 +571,19 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise<void> {
replaceTranscript(messages, { preserveTransientMessages: true });
shellRunElapsedTicker.sync();
requestRender();
+ const messageIds = state.entries.flatMap((entry) =>
+ entry.kind === 'user' && entry.transient === true ? [entry.messageId] : [],
+ );
+ if (messageIds.length > 0) {
+ void input.driver
+ .queryMessageStatuses?.(messageIds)
+ .then((result) => {
+ if (closed || input.driver.getSessionId() !== sessionId) return;
+ reconcileTransientMessageLifecycle(state, result.messages);
+ requestRender();
+ })
+ .catch(() => undefined);
+ }
return;
}
rememberTranscriptModel(messages);
@@ -934,14 +948,7 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise<void> {
refillEditorFromQueues(retracted.text);
};
- // Enter during a turn asks the Host to place the message at the current
- // step boundary. The Host alone decides whether it steers or starts a
- // successor Turn if the previous Turn settled during admission.
- const steerRunningTurn = (text: string) => {
- if (!text.trim()) {
- requestRender();
- return;
- }
+ const submitRunningMessage = (text: string, placement: 'current_turn' | 'next_turn') => {
editor.addToHistory(text);
const submitMessage = input.driver.submitMessage;
if (!submitMessage) {
@@ -952,12 +959,8 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise<void> {
appendUserPrompt(state, text, messageId, true);
requestRender();
const task = submitMessage
- .call(input.driver, text, { messageId, placement: 'current_turn' })
- .then(() => {
- // The subscription projects queue and Turn state; admission only
- // confirms that this stable message identity belongs to the Host.
- requestRender();
- })
+ .call(input.driver, text, { messageId, placement })
+ .then(requestRender)
.catch((error) => {
removeTransientUserMessage(messageId);
refillEditorFromQueues(text);
@@ -966,6 +969,17 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise<void> {
trackEnqueue(task);
};
+ // Enter during a turn asks the Host to place the message at the current
+ // step boundary. The Host alone decides whether it steers or starts a
+ // successor Turn if the previous Turn settled during admission.
+ const steerRunningTurn = (text: string) => {
+ if (!text.trim()) {
+ requestRender();
+ return;
+ }
+ submitRunningMessage(text, 'current_turn');
+ };
+
// Alt+Enter: during a turn, queue the text to open the next turn; when idle,
// it submits like Enter.
const handleAltEnter = () => {
@@ -983,26 +997,7 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise<void> {
submitPrompt(text);
return;
}
- editor.addToHistory(text);
- const submitMessage = input.driver.submitMessage;
- if (!submitMessage) {
- refillEditorFromQueues(text);
- return;
- }
- const messageId = randomUUID();
- appendUserPrompt(state, text, messageId, true);
- requestRender();
- const task = submitMessage
- .call(input.driver, text, { messageId, placement: 'next_turn' })
- .then(() => {
- requestRender();
- })
- .catch((error) => {
- removeTransientUserMessage(messageId);
- refillEditorFromQueues(text);
- reportError(error);
- });
- trackEnqueue(task);
+ submitRunningMessage(text, 'next_turn');
};
// Alt+↑: take back every queued message from the Runtime Host, joined and
diff --git a/packages/cli/src/pi-tui-turn.ts b/packages/cli/src/pi-tui-turn.ts
index 9b1553bc6..957dc8cc6 100644
--- a/packages/cli/src/pi-tui-turn.ts
+++ b/packages/cli/src/pi-tui-turn.ts
@@ -30,7 +30,6 @@ import { type GoalTurnOutcome } from '@maka/runtime/goal-continuation';
import {
SkillInvocationBlockedError,
type MakaPreparedSessionTurn,
- type MakaMessageAdmission,
type MakaSessionDriver,
} from './session-driver.js';
@@ -69,7 +68,7 @@ export interface RunMakaPiTuiTurnInput {
onFailure?: (error: unknown) => void | Promise<void>;
}
-export type MakaPiTuiTurnOutcome = GoalTurnOutcome | ({ kind: 'admitted' } & MakaMessageAdmission);
+export type MakaPiTuiTurnOutcome = GoalTurnOutcome | { kind: 'admitted' };
/**
* Owns one visible TUI turn from activity reservation through full stream drain.
@@ -109,12 +108,12 @@ export async function runMakaPiTuiTurn(
request.turnOrchestration === undefined &&
input.driver.submitMessage
) {
- const admission = await input.driver.submitMessage(request.prompt, {
+ await input.driver.submitMessage(request.prompt, {
messageId: externalTurnId!,
placement: 'current_turn',
...(request.sendText !== undefined ? { modelText: request.sendText } : {}),
});
- return finishBeforeDrain({ kind: 'admitted', ...admission });
+ return finishBeforeDrain({ kind: 'admitted' });
}
const turn =
diff --git a/packages/cli/src/runtime-host-session-driver.ts b/packages/cli/src/runtime-host-session-driver.ts
index 07f1e773b..9f3fe4f8f 100644
--- a/packages/cli/src/runtime-host-session-driver.ts
+++ b/packages/cli/src/runtime-host-session-driver.ts
@@ -82,7 +82,6 @@ import type {
MakaSideConversationCloseResult,
MakaSideConversationOpenResult,
MakaSideConversationParentStatus,
- MakaMessageAdmission,
MakaRetractedMessages,
MakaPreparePromptOptions,
MakaPreparedSessionTurn,
@@ -391,10 +390,7 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver {
yield* events;
}
- async submitMessage(
- text: string,
- options: MakaSubmitMessageOptions,
- ): Promise<MakaMessageAdmission> {
+ async submitMessage(text: string, options: MakaSubmitMessageOptions): Promise<void> {
const sessionId = await this.#ensureSession();
const sessionGeneration = this.#sessionGeneration;
const configuration = await this.#loadConfiguration(sessionId);
@@ -403,9 +399,8 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver {
this.#assertCurrentSession(sessionId, sessionGeneration);
this.#adoptLoadedConfiguration(configuration);
const modelText = options.modelText ?? text;
- let result;
try {
- result = await this.#request('turn.message.submit', {
+ await this.#request('turn.message.submit', {
originHostEpoch: this.#connection.hostEpoch,
sessionId,
messageId: options.messageId,
@@ -420,11 +415,17 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver {
(error instanceof RuntimeHostOperationError && error.code === 'outcome_unknown') ||
(error instanceof RuntimeHostRequestInterruptedError && error.dispatch === 'dispatched')
) {
- return { messageId: options.messageId, disposition: 'outcome_unknown' };
+ return;
}
throw error;
}
- return { messageId: options.messageId, disposition: result.disposition };
+ }
+
+ async queryMessageStatuses(
+ messageIds: readonly string[],
+ ): Promise<OperationOutput<'turn.message.query'>> {
+ const sessionId = await this.#ensureSession();
+ return this.#request('turn.message.query', { sessionId, messageIds });
}
async retractQueued(): Promise<MakaRetractedMessages> {
diff --git a/packages/cli/src/session-driver.ts b/packages/cli/src/session-driver.ts
index d2065adc3..3ec422861 100644
--- a/packages/cli/src/session-driver.ts
+++ b/packages/cli/src/session-driver.ts
@@ -28,7 +28,11 @@ import type { CreateSessionInput, TurnOrchestration } from '@maka/core/runtime-i
import type { UserQuestionResponse } from '@maka/core/user-question';
import type { ContextDiagnostics } from '@maka/runtime/context-diagnostics';
import type { SkillInvocationResult } from '@maka/core/skill-invocation';
-import type { GoalControlAction, GoalProjection } from '@maka/runtime-host/protocol';
+import type {
+ GoalControlAction,
+ GoalProjection,
+ TurnMessageQueryResult,
+} from '@maka/runtime-host/protocol';
export interface MakaSessionMoveResult {
previousCwd: string;
@@ -104,11 +108,6 @@ export interface MakaSubmitMessageOptions {
modelText?: string;
}
-export interface MakaMessageAdmission {
- messageId: string;
- disposition: 'steering' | 'followup' | 'turn_started' | 'outcome_unknown';
-}
-
export interface MakaRetractedMessages {
text: string;
messageIds: readonly string[];
@@ -128,7 +127,8 @@ export interface MakaSessionDriver {
prompt: string,
options?: MakaPreparePromptOptions,
): Promise<MakaPreparedSessionTurn>;
- submitMessage?(text: string, options: MakaSubmitMessageOptions): Promise<MakaMessageAdmission>;
+ submitMessage?(text: string, options: MakaSubmitMessageOptions): Promise<void>;
+ queryMessageStatuses?(messageIds: readonly string[]): Promise<TurnMessageQueryResult>;
compactSession(): AsyncIterable<SessionEvent>;
resumeLatest?(): AsyncIterable<SessionEvent>;
retractQueued?(): Promise<MakaRetractedMessages>;
diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts
index 4763c1267..df3c9ed76 100644
--- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts
@@ -72,6 +72,59 @@ test('idle submit starts exactly one root Turn and retry identity is connection-
assert.equal(fixture.liveResidencies(), 0);
});
+test('message query returns durable cancellation proof after the live queue disappears', async () => {
+ const fixture = createFixture();
+ fixture.coordinator.reserveRootTurn(ROOT);
+ await submit(fixture, 'cancelled-message', 'discard me', 'next_turn');
+ await fixture.coordinator.cancelMessages(ROOT.sessionId, ['cancelled-message']);
+
+ const result = await fixture.coordinator.handlers['turn.message.query'](
+ {
+ sessionId: ROOT.sessionId,
+ messageIds: ['cancelled-message', 'unknown-message'],
+ },
+ operationContext(),
+ );
+
+ assert.deepEqual(result, {
+ ok: true,
+ result: {
+ messages: [
+ { messageId: 'cancelled-message', status: 'cancelled' },
+ { messageId: 'unknown-message', status: 'unknown' },
+ ],
+ },
+ });
+});
+
+test('message query distinguishes a live admission from durable handoff proof', async () => {
+ const fixture = createFixture();
+ fixture.coordinator.reserveRootTurn(ROOT);
+ await submit(fixture, 'accepted-message', 'waiting', 'next_turn');
+ fixture.receipts.set(
+ 'handed-off-message',
+ sourceReceipt('handed-off-message', 'delivered', 'current_turn', 'steering'),
+ );
+
+ const result = await fixture.coordinator.handlers['turn.message.query'](
+ {
+ sessionId: ROOT.sessionId,
+ messageIds: ['accepted-message', 'handed-off-message'],
+ },
+ operationContext(),
+ );
+
+ assert.deepEqual(result, {
+ ok: true,
+ result: {
+ messages: [
+ { messageId: 'accepted-message', status: 'accepted' },
+ { messageId: 'handed-off-message', status: 'handed_off' },
+ ],
+ },
+ });
+});
+
test('submit re-runs admission when the queue revision moves during preflight', async () => {
let preflightCalls = 0;
const fixture = createFixture(undefined, async () => {
@@ -2267,6 +2320,16 @@ function memoryMessageAdmissionStore(
return admission;
},
readMessageAdmission: async (_sessionId, messageId) => admissions.get(messageId)?.admission,
+ readCancelledMessageAdmission: async (_sessionId, messageId) => {
+ const entry = admissions.get(messageId);
+ return entry?.state === 'cancelled'
+ ? {
+ messageId,
+ submittedContentDigest: entry.admission.submittedContentDigest,
+ submittedPlacement: entry.admission.submittedPlacement,
+ }
+ : undefined;
+ },
listMessageAdmissions: async (sessionId) =>
[...admissions.values()]
.filter(({ admission, state }) => admission.sessionId === sessionId && state === 'accepted')
diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts
index c4f2fad83..ad488a531 100644
--- a/packages/runtime-host/src/__tests__/protocol.test.ts
+++ b/packages/runtime-host/src/__tests__/protocol.test.ts
@@ -242,6 +242,10 @@ describe('Runtime Host bootstrap protocol', () => {
assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 43);
});
+ test('publishes a new compatibility epoch for durable Message lifecycle queries', () => {
+ assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 50);
+ });
+
test('selects the highest mutually supported protocol and rejects a gap', () => {
assert.equal(negotiateProtocol({ min: 0, max: 0 }, { min: 0, max: 0 }), 0);
assert.equal(negotiateProtocol({ min: 1, max: 3 }, { min: 2, max: 4 }), 3);
@@ -969,6 +973,14 @@ describe('Runtime Host bootstrap protocol', () => {
});
test('requires stable Message command identities, origin Host Epoch, and exact inputs', () => {
+ const query = {
+ requestId: 'query-request-1',
+ operation: 'turn.message.query' as const,
+ input: {
+ sessionId: 'session-1',
+ messageIds: ['message-1', 'message-2'],
+ },
+ };
const submit = {
requestId: 'submit-request-1',
operation: 'turn.message.submit' as const,
@@ -996,6 +1008,7 @@ describe('Runtime Host bootstrap protocol', () => {
runId: 'run-1',
},
};
+ assert.deepEqual(decodeClientFrame(query), query);
assert.deepEqual(decodeClientFrame(submit), submit);
assert.deepEqual(decodeClientFrame(retract), retract);
assert.deepEqual(decodeClientFrame(interrupt), interrupt);
diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts
index 414fec8c3..a8dc23998 100644
--- a/packages/runtime-host/src/protocol/index.ts
+++ b/packages/runtime-host/src/protocol/index.ts
@@ -92,7 +92,10 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const;
export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const;
// Increment when the same protocol version no longer guarantees safe Client-Host
// interoperability. Mismatches are rejected before domain commands are admitted.
-export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 51 as const;
+export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 52 as const;
+// 52: Message admission answers `turn.message.submit` with an explicit
+// disposition, and queued Messages can be proven cancelled. Older peers read the
+// answer as a bare acknowledgement and cannot reconcile their own projection.
// 51: WorkHub exposes bounded coordination candidates and admits only typed
// actions through the deterministic Runtime Host Action Gate.
// 50: WorkHub can append durable coordination summaries and admit tool-free
diff --git a/packages/runtime-host/src/protocol/message.ts b/packages/runtime-host/src/protocol/message.ts
index e72c171b0..b8c48908d 100644
--- a/packages/runtime-host/src/protocol/message.ts
+++ b/packages/runtime-host/src/protocol/message.ts
@@ -91,6 +91,20 @@ export type TurnMessageSubmitResult =
| { readonly disposition: 'followup'; readonly queueRevision: number }
| { readonly disposition: 'turn_started'; readonly turnId: string };
+export type MessageLifecycleStatus = 'accepted' | 'handed_off' | 'cancelled' | 'unknown';
+
+export interface TurnMessageQueryInput {
+ readonly sessionId: string;
+ readonly messageIds: readonly string[];
+}
+
+export interface TurnMessageQueryResult {
+ readonly messages: readonly {
+ readonly messageId: string;
+ readonly status: MessageLifecycleStatus;
+ }[];
+}
+
export interface QueueRetractInput {
readonly originHostEpoch: string;
readonly sessionId: string;
@@ -163,6 +177,13 @@ const MESSAGE_OPERATION_ERRORS = [
] as const;
export const MESSAGE_OPERATION_SPECS = {
+ 'turn.message.query': defineOperation({
+ mode: 'query',
+ availability: 'ready',
+ errors: MESSAGE_OPERATION_ERRORS,
+ decodeInput: decodeTurnMessageQueryInput,
+ decodeOutput: decodeTurnMessageQueryResult,
+ }),
'turn.message.submit': defineOperation({
mode: 'command',
availability: 'ready',
@@ -258,6 +279,51 @@ function decodeTurnMessageSubmitInput(value: unknown): TurnMessageSubmitInput {
};
}
+function decodeTurnMessageQueryInput(value: unknown): TurnMessageQueryInput {
+ const record = requireExactRecord(value, 'turn.message.query input', ['sessionId', 'messageIds']);
+ if (!Array.isArray(record.messageIds) || record.messageIds.length > MESSAGE_QUEUE_MAX_ENTRIES) {
+ throw invalidProtocolFrame('Invalid turn.message.query messageIds');
+ }
+ const messageIds = record.messageIds.map((messageId) => requireEntityId(messageId, 'messageId'));
+ if (new Set(messageIds).size !== messageIds.length) {
+ throw invalidProtocolFrame('Duplicate turn.message.query messageId');
+ }
+ return {
+ sessionId: requireEntityId(record.sessionId, 'sessionId'),
+ messageIds,
+ };
+}
+
+function decodeTurnMessageQueryResult(value: unknown): TurnMessageQueryResult {
+ const record = requireExactRecord(value, 'turn.message.query result', ['messages']);
+ if (!Array.isArray(record.messages) || record.messages.length > MESSAGE_QUEUE_MAX_ENTRIES) {
+ throw invalidProtocolFrame('Invalid turn.message.query messages');
+ }
+ const messages = record.messages.map((candidate) => {
+ const message = requireExactRecord(candidate, 'turn.message.query message', [
+ 'messageId',
+ 'status',
+ ]);
+ if (
+ message.status !== 'accepted' &&
+ message.status !== 'handed_off' &&
+ message.status !== 'cancelled' &&
+ message.status !== 'unknown'
+ ) {
+ throw invalidProtocolFrame('Invalid turn.message.query status');
+ }
+ const status = message.status as MessageLifecycleStatus;
+ return {
+ messageId: requireEntityId(message.messageId, 'messageId'),
+ status,
+ };
+ });
+ if (new Set(messages.map(({ messageId }) => messageId)).size !== messages.length) {
+ throw invalidProtocolFrame('Duplicate turn.message.query result messageId');
+ }
+ return { messages };
+}
+
function decodeTurnMessageSubmitResult(value: unknown): TurnMessageSubmitResult {
const record = requireRecord(value, 'turn.message.submit result');
if (record.disposition === 'turn_started') {
diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts
index 745d53e0e..84595571b 100644
--- a/packages/runtime-host/src/protocol/operations.ts
+++ b/packages/runtime-host/src/protocol/operations.ts
@@ -316,6 +316,7 @@ export const REMOTE_OWNER_OPERATION_GRANTS = Object.freeze([
'subscription.open',
'task.ledger.query',
'turn.interrupt',
+ 'turn.message.query',
'turn.message.submit',
'turn.query',
'turn.regenerate',
diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts
index 066d3d5e0..acd667d75 100644
--- a/packages/runtime-host/src/server/message-coordinator.ts
+++ b/packages/runtime-host/src/server/message-coordinator.ts
@@ -315,6 +315,7 @@ const HOST_EPOCH_PATTERN = /^[A-Za-z0-9_-]{1,128}$/u;
/** The sole in-memory message authority for one Runtime Host Epoch. */
export class HostMessageCoordinator implements RuntimeMessageAuthority {
readonly handlers: MessageOperationHandlerMap = {
+ 'turn.message.query': (input) => this.queryMessages(input),
'turn.message.submit': (input, context) => this.submit(input, context),
'queue.retract': (input) => this.retract(input),
'queue.entry.retract': (input) => this.retractQueuedEntry(input),
@@ -370,6 +371,49 @@ export class HostMessageCoordinator implements RuntimeMessageAuthority {
return state ? hasLiveMessageState(state) : false;
}
+ async queryMessages(input: { sessionId: string; messageIds: readonly string[] }): Promise<
+ MessageOutcome<{
+ messages: Array<{
+ messageId: string;
+ status: 'accepted' | 'handed_off' | 'cancelled' | 'unknown';
+ }>;
+ }>
+ > {
+ const messages = [] as Array<{
+ messageId: string;
+ status: 'accepted' | 'handed_off' | 'cancelled' | 'unknown';
+ }>;
+ for (const messageId of input.messageIds) {
+ const cancelled = await this.#admissions.readCancelledMessageAdmission(
+ input.sessionId,
+ messageId,
+ );
+ if (cancelled) {
+ messages.push({ messageId, status: 'cancelled' });
+ continue;
+ }
+ const accepted = await this.#admissions.readMessageAdmission(input.sessionId, messageId);
+ if (accepted) {
+ messages.push({ messageId, status: 'accepted' });
+ continue;
+ }
+ const root = await this.#durableProof.readRootTurnSourceMessageReceipt(
+ input.sessionId,
+ messageId,
+ );
+ if (root) {
+ messages.push({ messageId, status: 'handed_off' });
+ continue;
+ }
+ const steering = await this.#durableProof.readImmutableSteeringMessageProof(
+ input.sessionId,
+ messageId,
+ );
+ messages.push({ messageId, status: steering ? 'handed_off' : 'unknown' });
+ }
+ return success({ messages });
+ }
+
retireSessions(sessionIds: readonly string[]): void {
for (const sessionId of new Set(sessionIds)) {
const state = this.#sessions.get(sessionId);
diff --git a/packages/runtime-host/src/server/operation-dispatcher.ts b/packages/runtime-host/src/server/operation-dispatcher.ts
index 208befbb2..49876f799 100644
--- a/packages/runtime-host/src/server/operation-dispatcher.ts
+++ b/packages/runtime-host/src/server/operation-dispatcher.ts
@@ -85,6 +85,7 @@ export type ConnectionEffectOperationKey = Extract<
>;
export type MessageOperationKey = Extract<
OperationKey,
+ | 'turn.message.query'
| 'turn.message.submit'
| 'queue.retract'
| 'queue.entry.retract'
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 f939ab609..93027587f 100644
--- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts
+++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts
@@ -445,6 +445,11 @@ describe('SqliteSessionMetadataStore', () => {
await store.commitMessageAdmission(admission);
await store.cancelMessageAdmissions('session-1', ['message-1']);
assert.deepEqual(await store.listMessageAdmissions('session-1'), []);
+ assert.deepEqual(await store.readCancelledMessageAdmission('session-1', 'message-1'), {
+ messageId: 'message-1',
+ submittedContentDigest: messageContentDigest({ text: 'discard this draft' }),
+ submittedPlacement: 'next_turn',
+ });
await assert.rejects(
store.commitMessageAdmission(admission),
/identity is already cancelled/,
diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts
index 32996da17..88201cf48 100644
--- a/packages/storage/src/execution-stores.ts
+++ b/packages/storage/src/execution-stores.ts
@@ -415,6 +415,8 @@ async function createExecutionStoresForWrite<K extends StorageRootKind, E extend
run(() => sessionStore.commitMessageAdmission(admission)),
readMessageAdmission: (sessionId, messageId) =>
run(() => sessionStore.readMessageAdmission(sessionId, messageId)),
+ readCancelledMessageAdmission: (sessionId, messageId) =>
+ run(() => sessionStore.readCancelledMessageAdmission(sessionId, messageId)),
listMessageAdmissions: (sessionId) =>
run(() => sessionStore.listMessageAdmissions(sessionId)),
markMessagesHandedOff: (input) => run(() => sessionStore.markMessagesHandedOff(input)),
diff --git a/packages/storage/src/message-admission-store.ts b/packages/storage/src/message-admission-store.ts
index a73a3f43e..e0b728e10 100644
--- a/packages/storage/src/message-admission-store.ts
+++ b/packages/storage/src/message-admission-store.ts
@@ -35,12 +35,22 @@ export interface PendingMessageAdmission {
readonly admittedAt: number;
}
+export interface CancelledMessageAdmission {
+ readonly messageId: string;
+ readonly submittedContentDigest: `sha256:${string}`;
+ readonly submittedPlacement: 'current_turn' | 'next_turn';
+}
+
export interface MessageAdmissionStore {
commitMessageAdmission(admission: PendingMessageAdmission): Promise<PendingMessageAdmission>;
readMessageAdmission(
sessionId: string,
messageId: string,
): Promise<PendingMessageAdmission | undefined>;
+ readCancelledMessageAdmission(
+ sessionId: string,
+ messageId: string,
+ ): Promise<CancelledMessageAdmission | undefined>;
listMessageAdmissions(sessionId: string): Promise<readonly PendingMessageAdmission[]>;
markMessagesHandedOff(input: {
sessionId: string;
diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts
index 536131893..3d69a5267 100644
--- a/packages/storage/src/session-store.ts
+++ b/packages/storage/src/session-store.ts
@@ -888,6 +888,11 @@ class SqliteSessionStore implements SessionAuthorityStore {
return this.metadata.readMessageAdmission(sessionId, messageId);
}
+ async readCancelledMessageAdmission(sessionId: string, messageId: string) {
+ await this.ensureReady();
+ return this.metadata.readCancelledMessageAdmission(sessionId, messageId);
+ }
+
async listMessageAdmissions(sessionId: string): Promise<readonly PendingMessageAdmission[]> {
await this.ensureReady();
return this.metadata.listMessageAdmissions(sessionId);
diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts
index 8943dea71..a2ed11dfc 100644
--- a/packages/storage/src/sqlite-session-metadata-store.ts
+++ b/packages/storage/src/sqlite-session-metadata-store.ts
@@ -1656,6 +1656,48 @@ export class SqliteSessionMetadataStore {
});
}
+ async readCancelledMessageAdmission(
+ sessionId: string,
+ messageId: string,
+ ): Promise<
+ | {
+ messageId: string;
+ submittedContentDigest: `sha256:${string}`;
+ submittedPlacement: 'current_turn' | 'next_turn';
+ }
+ | undefined
+ > {
+ this.assertOpen();
+ assertSafeSessionId(sessionId);
+ assertSafeSessionId(messageId);
+ return this.readTransaction(() => {
+ const row = this.db
+ .prepare(
+ `
+ SELECT submitted_content_digest, submitted_placement
+ FROM cancelled_message_admissions
+ WHERE session_id = ? AND message_id = ?
+ `,
+ )
+ .get(sessionId, messageId) as
+ | { submitted_content_digest?: unknown; submitted_placement?: unknown }
+ | undefined;
+ if (!row) return undefined;
+ if (
+ typeof row.submitted_content_digest !== 'string' ||
+ !/^sha256:[a-f0-9]{64}$/u.test(row.submitted_content_digest) ||
+ (row.submitted_placement !== 'current_turn' && row.submitted_placement !== 'next_turn')
+ ) {
+ throw new SessionMetadataConflictError('Invalid cancelled Message admission identity');
+ }
+ return {
+ messageId,
+ submittedContentDigest: row.submitted_content_digest as `sha256:${string}`,
+ submittedPlacement: row.submitted_placement,
+ };
+ });
+ }
+
async listMessageAdmissions(sessionId: string): Promise<readonly PendingMessageAdmission[]> {
this.assertOpen();
assertSafeSessionId(sessionId);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org