You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/08/24 17:40:14 UTC
(maka) 01/05: fix(runtime-host): replay admitted roots from durable contracts
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 ee24c6bbe7c5d78f29b636f8bf052bca179c37b1
Author: AstroHan <le...@outlook.com>
AuthorDate: Tue Aug 25 01:26:22 2026 +0800
fix(runtime-host): replay admitted roots from durable contracts
Generated-by: Codex
---
.../src/__tests__/execution-host-queue.test.ts | 23 +++++++++
.../src/__tests__/fixtures/execution-host-suite.ts | 59 +++++++++++++++++++++-
.../src/server/hosted-execution-recovery.ts | 13 ++++-
3 files changed, 93 insertions(+), 2 deletions(-)
diff --git a/packages/runtime-host/src/__tests__/execution-host-queue.test.ts b/packages/runtime-host/src/__tests__/execution-host-queue.test.ts
index fb0cc029b..fe6056285 100644
--- a/packages/runtime-host/src/__tests__/execution-host-queue.test.ts
+++ b/packages/runtime-host/src/__tests__/execution-host-queue.test.ts
@@ -298,6 +298,29 @@ test('a Host crash after queue admission recovers the durable successor once', a
});
});
+test('restart replays an atomically admitted root without duplicating its transcript', async () => {
+ await withExecutionRoot(async (fixture) => {
+ const turnId = randomUUID();
+ const messageId = randomUUID();
+ const content = { text: 'recover the root after admission before Run creation' };
+ await fixture.seedAtomicRootAdmissionWithoutRun({ turnId, messageId, content });
+
+ const host = await fixture.startHost();
+ const client = await connectClient(fixture.root);
+ const terminal = await waitForTerminalTurn(client, fixture.sessionId, turnId);
+ assert.equal(terminal.status, 'completed');
+ await client.close();
+ await fixture.stopHost(host);
+
+ const ledger = await fixture.readTurn(turnId);
+ assert.deepEqual(
+ ledger.userMessages.filter((message) => message.id === messageId).map((message) => message.id),
+ [messageId],
+ );
+ assert.equal(await fixture.readMessageLifecycleState(messageId), 'handed_off');
+ });
+});
+
test('concurrent root admission for one Session has a single winner', async () => {
await withExecutionRoot(async (fixture) => {
const host = await fixture.startHost();
diff --git a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts
index c19755581..d9d68cee9 100644
--- a/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts
+++ b/packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts
@@ -38,7 +38,11 @@ import { test } from 'node:test';
import { TOOL_BOUNDARY_PROTOCOL_V1 } from '@maka/core/runtime-event';
import { canonicalToolArgsHash } from '@maka/core/tool-args-identity';
import type { AgentRunHeader } from '@maka/core/agent-run';
-import { normalizeMessageContent, type MessageContent } from '@maka/core/events';
+import {
+ messageContentDigest,
+ normalizeMessageContent,
+ type MessageContent,
+} from '@maka/core/events';
import type { ConnectionCatalogEntry } from '@maka/core/runtime-policy';
import type { StoredMessage } from '@maka/core/session';
import type { Task } from '@maka/core/task-ledger';
@@ -674,6 +678,59 @@ export class ExecutionFixture {
return this.seedTurnState(turnId, content, false, false);
}
+ async seedAtomicRootAdmissionWithoutRun(input: {
+ turnId: string;
+ messageId: string;
+ content: MessageContent;
+ }): Promise<void> {
+ const owner = await tryAcquireInteractiveRootOwner(this.capability);
+ assert.ok(owner);
+ if (!owner) throw new Error('Unable to acquire execution root for atomic root setup');
+ let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForWrite>> | undefined;
+ try {
+ stores = await openInteractiveExecutionStoresForWrite(owner.lease);
+ const admittedAt = Date.now();
+ const content = normalizeMessageContent(input.content);
+ const contentDigest = messageContentDigest(content);
+ const runId = randomUUID();
+ await stores.sessionStore.commitMessageAdmission({
+ sessionId: this.sessionId,
+ turnId: input.turnId,
+ runId,
+ messageId: input.messageId,
+ content,
+ modelContent: content,
+ submittedPlacement: 'current_turn',
+ placement: 'current_turn',
+ disposition: 'steering',
+ admittedAt,
+ });
+ const result = await stores.agentRunStore.admitRootTurn({
+ sessionId: this.sessionId,
+ turnId: input.turnId,
+ proposedRunId: runId,
+ proposedUserMessageId: input.messageId,
+ execution: { kind: 'external_message', inputDigest: contentDigest },
+ previousRootTurnId: null,
+ normalizedInput: content,
+ sourceMessages: [
+ {
+ messageId: input.messageId,
+ content,
+ submittedContentDigest: contentDigest,
+ placement: 'current_turn',
+ disposition: 'turn_started',
+ },
+ ],
+ admittedAt,
+ });
+ assert.equal(result.kind, 'admitted');
+ } finally {
+ await stores?.sessionStore.close?.();
+ await owner.close();
+ }
+ }
+
async archiveSession(): Promise<void> {
const owner = await tryAcquireInteractiveRootOwner(this.capability);
assert.ok(owner);
diff --git a/packages/runtime-host/src/server/hosted-execution-recovery.ts b/packages/runtime-host/src/server/hosted-execution-recovery.ts
index 674c92e2d..032f745ff 100644
--- a/packages/runtime-host/src/server/hosted-execution-recovery.ts
+++ b/packages/runtime-host/src/server/hosted-execution-recovery.ts
@@ -145,7 +145,16 @@ export async function prepareHostedExecutionRecovery(
continue;
}
if (!run) {
- if (rootUserMessages.length > 0 || messageIdOwner) {
+ if (executionContract.pendingWithoutRun === 'root_replay') {
+ verifyOrRecoverUserMessage(
+ admission,
+ rootUserMessages,
+ messageIdOwner,
+ missingMessages,
+ messageIndex,
+ false,
+ );
+ } else if (rootUserMessages.length > 0 || messageIdOwner) {
throw new Error(`Admitted Turn ${admission.turnId} has a UserMessage but no Run`);
}
replayAdmissions.push(admission);
@@ -263,6 +272,7 @@ function verifyOrRecoverUserMessage(
messageIdOwner: StoredMessage | undefined,
missingMessages: RecoveryUserMessage[],
index: RecoveryMessageIndex,
+ materializeMissing = true,
): void {
if (rootUserMessages.length > 1) {
throw new Error(`Admitted Turn ${admission.turnId} has multiple UserMessages`);
@@ -285,6 +295,7 @@ function verifyOrRecoverUserMessage(
if (messageIdOwner) {
throw new Error(`Admitted Turn ${admission.turnId} reuses another message identity`);
}
+ if (!materializeMissing) return;
const recoveredMessage = recoveryUserMessage(admission);
missingMessages.push(recoveredMessage);
indexRecoveryMessage(index, recoveredMessage);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org