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 16:06:38 UTC
(maka) 01/03: fix(runtime): prove prepared root sources by submitted digest
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 9d10944844145d46654bf2691d2ffa798bd138a1
Author: AstroHan <le...@outlook.com>
AuthorDate: Mon Aug 24 23:47:12 2026 +0800
fix(runtime): prove prepared root sources by submitted digest
Generated-by: Codex
---
packages/core/src/events.ts | 19 ++++++++++
.../src/__tests__/message-coordinator.test.ts | 3 +-
.../src/server/message-content-digest.ts | 38 -------------------
.../runtime-host/src/server/message-coordinator.ts | 2 +-
.../src/server/root-turn-coordinator.ts | 5 ++-
.../__tests__/runtime-kernel-interaction.test.ts | 43 +++++++++++++++++++++-
packages/runtime/src/runtime-kernel.ts | 8 +++-
packages/runtime/src/session-manager.ts | 1 +
8 files changed, 75 insertions(+), 44 deletions(-)
diff --git a/packages/core/src/events.ts b/packages/core/src/events.ts
index 4a58287e4..15c6170fb 100644
--- a/packages/core/src/events.ts
+++ b/packages/core/src/events.ts
@@ -26,6 +26,7 @@
* Connection-setup events live in ./connections.ts (separate channel).
*/
+import * as nodeCrypto from 'node:crypto';
import type {
AdditionalPermissionRequest,
PermissionMode,
@@ -398,6 +399,24 @@ export function messageContentsEqual(left: MessageContent, right: MessageContent
);
}
+export function messageContentDigest(content: MessageContent): `sha256:${string}` {
+ return `sha256:${nodeCrypto
+ .createHash('sha256')
+ .update(JSON.stringify(canonicalizeMessageContent(normalizeMessageContent(content))))
+ .digest('hex')}`;
+}
+
+function canonicalizeMessageContent(value: unknown): unknown {
+ if (Array.isArray(value)) return value.map(canonicalizeMessageContent);
+ if (value === null || typeof value !== 'object') return value;
+ return Object.fromEntries(
+ Object.entries(value)
+ .filter(([, entry]) => entry !== undefined)
+ .sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0))
+ .map(([key, entry]) => [key, canonicalizeMessageContent(entry)]),
+ );
+}
+
function inlineReferencesEqual(left: InlineReference, right: InlineReference): boolean {
return (
left.kind === right.kind &&
diff --git a/packages/runtime-host/src/__tests__/message-coordinator.test.ts b/packages/runtime-host/src/__tests__/message-coordinator.test.ts
index 934227bd5..62024d88d 100644
--- a/packages/runtime-host/src/__tests__/message-coordinator.test.ts
+++ b/packages/runtime-host/src/__tests__/message-coordinator.test.ts
@@ -19,7 +19,7 @@
import assert from 'node:assert/strict';
import { test } from 'node:test';
-import type { MessageContent } from '@maka/core/events';
+import { messageContentDigest, type MessageContent } from '@maka/core/events';
import type { RuntimeEvent } from '@maka/core/runtime-event';
import type {
MessageOperationReceipt,
@@ -39,7 +39,6 @@ import {
type HostMessageRootPort,
type HostMessageRootState,
} from '../server/message-coordinator.js';
-import { messageContentDigest } from '../server/message-content-digest.js';
import { SessionAdmissionGate } from '../server/session-admission-gate.js';
const ROOT = { sessionId: 'session-1', turnId: 'turn-1', runId: 'run-1' } as const;
diff --git a/packages/runtime-host/src/server/message-content-digest.ts b/packages/runtime-host/src/server/message-content-digest.ts
deleted file mode 100644
index 03f88d598..000000000
--- a/packages/runtime-host/src/server/message-content-digest.ts
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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 { createHash } from 'node:crypto';
-import { normalizeMessageContent, type MessageContent } from '@maka/core/events';
-
-export function messageContentDigest(content: MessageContent): `sha256:${string}` {
- return `sha256:${createHash('sha256')
- .update(JSON.stringify(canonicalize(normalizeMessageContent(content))))
- .digest('hex')}`;
-}
-
-function canonicalize(value: unknown): unknown {
- if (Array.isArray(value)) return value.map(canonicalize);
- if (value === null || typeof value !== 'object') return value;
- return Object.fromEntries(
- Object.entries(value)
- .filter(([, entry]) => entry !== undefined)
- .sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0))
- .map(([key, entry]) => [key, canonicalize(entry)]),
- );
-}
diff --git a/packages/runtime-host/src/server/message-coordinator.ts b/packages/runtime-host/src/server/message-coordinator.ts
index f0b2916d9..32669e814 100644
--- a/packages/runtime-host/src/server/message-coordinator.ts
+++ b/packages/runtime-host/src/server/message-coordinator.ts
@@ -22,6 +22,7 @@ import { isDeepStrictEqual } from 'node:util';
import type { SteeringLease } from '@maka/core/backend-types';
import {
aggregateMessageContents,
+ messageContentDigest,
messageContentsEqual,
normalizeMessageContent,
type MessageContent,
@@ -72,7 +73,6 @@ import type { RuntimeHostResidency } from './host-kernel.js';
import { worstCaseFailedTurnSnapshot } from './canonical-turn-snapshot.js';
import { worstCaseMessageQueueProjection } from './message-queue-capacity.js';
import type { ConnectionContext, MessageOperationHandlerMap } from './operation-dispatcher.js';
-import { messageContentDigest } from './message-content-digest.js';
import { type SessionAdmissionLease, SessionAdmissionGate } from './session-admission-gate.js';
type MessageOperationErrorCode =
diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts
index 9dd00b692..12474ec8e 100644
--- a/packages/runtime-host/src/server/root-turn-coordinator.ts
+++ b/packages/runtime-host/src/server/root-turn-coordinator.ts
@@ -23,6 +23,7 @@ import type { BackendStopMode } from '@maka/core/backend-types';
import type { AgentRunHeader, RootExecutionDescriptor } from '@maka/core/agent-run';
import {
INLINE_REFERENCE_MAX_COUNT,
+ messageContentDigest,
messageContentsEqual,
normalizeMessageContent,
type AttachmentRef,
@@ -88,7 +89,6 @@ import {
type QueueFenceResult,
type RootFollowupBatch,
} from './message-coordinator.js';
-import { messageContentDigest } from './message-content-digest.js';
import type { ConnectionContext, TurnOperationHandlerMap } from './operation-dispatcher.js';
import { RootAdmissionOwner } from './root-admission-owner.js';
import { type SessionAdmissionLease, SessionAdmissionGate } from './session-admission-gate.js';
@@ -1995,6 +1995,9 @@ export class RootTurnCoordinator implements HostedExecutionAuthority {
messages: admission.sourceMessages.map((source) => ({
messageId: source.messageId,
content: source.content,
+ ...(source.submittedContentDigest
+ ? { submittedContentDigest: source.submittedContentDigest }
+ : {}),
disposition: source.disposition,
})),
});
diff --git a/packages/runtime/src/__tests__/runtime-kernel-interaction.test.ts b/packages/runtime/src/__tests__/runtime-kernel-interaction.test.ts
index 4eab4a25e..df2d5ed93 100644
--- a/packages/runtime/src/__tests__/runtime-kernel-interaction.test.ts
+++ b/packages/runtime/src/__tests__/runtime-kernel-interaction.test.ts
@@ -20,7 +20,7 @@
import assert from 'node:assert/strict';
import { describe, test } from 'node:test';
-import type { SessionEvent } from '@maka/core/events';
+import { messageContentDigest, type SessionEvent } from '@maka/core/events';
import type { SessionHeader, StoredMessage } from '@maka/core/session';
import type { SandboxBoundaryResponse } from '@maka/core/sandbox-boundary';
@@ -39,6 +39,47 @@ import {
import { BackendRegistry, type SessionStore } from '../session-manager.js';
describe('RuntimeKernel Interaction close cleanup', () => {
+ test('accepts submitted transcript content when the root source is model-prepared', async () => {
+ const store = memoryStore();
+ const submitted = { text: '/skill:writer inspect' };
+ await store.appendMessage(SESSION_ID, {
+ type: 'user',
+ id: 'submitted-message',
+ turnId: 'prepared-turn',
+ ts: 1,
+ ...submitted,
+ });
+ const kernel = new RuntimeKernel({
+ store,
+ backends: new BackendRegistry(),
+ newId: () => 'materialize-id',
+ now: () => 1,
+ });
+
+ await kernel.materializeRootSourceMessages({
+ sessionId: SESSION_ID,
+ turnId: 'prepared-turn',
+ previousRootTurnId: null,
+ messages: [
+ {
+ messageId: 'submitted-message',
+ content: { text: '<prepared-skill>inspect</prepared-skill>' },
+ submittedContentDigest: messageContentDigest(submitted),
+ disposition: 'turn_started',
+ },
+ ],
+ });
+ assert.deepEqual(await store.readMessages(SESSION_ID), [
+ {
+ type: 'user',
+ id: 'submitted-message',
+ turnId: 'prepared-turn',
+ ts: 1,
+ ...submitted,
+ },
+ ]);
+ });
+
test('reserve followed by begin failure settles a concurrent stop claim', async () => {
const store = memoryStore();
const updateHeader = store.updateHeader;
diff --git a/packages/runtime/src/runtime-kernel.ts b/packages/runtime/src/runtime-kernel.ts
index 591fb82e9..58da67793 100644
--- a/packages/runtime/src/runtime-kernel.ts
+++ b/packages/runtime/src/runtime-kernel.ts
@@ -26,6 +26,7 @@ import type {
} from '@maka/core/runtime-event-store';
import { isSessionInlineRun } from '@maka/core/agent-run';
import {
+ messageContentDigest,
messageContentsEqual,
normalizeMessageContent,
type ActiveInteractionRequestEvent,
@@ -177,6 +178,7 @@ export interface RuntimeKernelLike {
messages: readonly {
messageId: string;
content: MessageContent;
+ submittedContentDigest?: `sha256:${string}`;
disposition: 'steering' | 'followup' | 'turn_started';
}[];
}): Promise<void>;
@@ -2322,6 +2324,7 @@ export class RuntimeKernel implements RuntimeKernelLike {
messages: readonly {
messageId: string;
content: MessageContent;
+ submittedContentDigest?: `sha256:${string}`;
disposition: 'steering' | 'followup' | 'turn_started';
}[];
}): Promise<void> {
@@ -2333,7 +2336,10 @@ export class RuntimeKernel implements RuntimeKernelLike {
if (existing) {
if (
existing.type !== 'user' ||
- !messageContentsEqual(normalizeMessageContent(existing), message.content) ||
+ (!messageContentsEqual(normalizeMessageContent(existing), message.content) &&
+ (message.submittedContentDigest === undefined ||
+ messageContentDigest(normalizeMessageContent(existing)) !==
+ message.submittedContentDigest)) ||
(existing.turnId !== input.turnId &&
((message.disposition !== 'steering' && message.disposition !== 'followup') ||
existing.turnId !== input.previousRootTurnId))
diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts
index b556f87db..2065665f7 100644
--- a/packages/runtime/src/session-manager.ts
+++ b/packages/runtime/src/session-manager.ts
@@ -4830,6 +4830,7 @@ export class SessionManager {
messages: readonly {
messageId: string;
content: import('@maka/core/events').MessageContent;
+ submittedContentDigest?: `sha256:${string}`;
disposition: 'steering' | 'followup' | 'turn_started';
}[];
}): Promise<void> {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org