feat: durable message lifecycle from admission to execution

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 &&

packages/runtime-host/src/__tests__/canonical-session-projection.test.ts

@@ -543,8 +543,10 @@ function createMessages(
         stores.agentRunStore.readRootTurnSourceMessageReceipt(requestedSessionId, messageId),
       readImmutableSteeringMessageProof: (requestedSessionId, messageId) =>
         stores.runtimeEventStore.readImmutableSteeringMessageProof(requestedSessionId, messageId),
+      readProviderRequestProof: async () => false,
     },
     receipts: stores.messageReceiptStore,
+    lifecycle: stores.sessionStore,
     sessionAdmission: new SessionAdmissionGate(),
     acquireResidency: () => ({ release: () => undefined }),
     preflightSessionSnapshot: () => true,

packages/runtime-host/src/__tests__/execution-host-message.test.ts

@@ -195,6 +195,7 @@ test('steering becomes durable and ordered followups automatically start the nex
     await first.close();
     await second.close();
     await fixture.stopHost(host);
+    assert.equal(await fixture.readMessageLifecycleState(steeringId), 'handed_off');
 
     const firstLedger = await fixture.readTurn(firstTurnId);
     const steeringEvents = firstLedger.runtimeEvents.filter(
@@ -213,6 +214,7 @@ test('steering becomes durable and ordered followups automatically start the nex
     const chain = await fixture.readAdmissionChain();
     assert.equal(chain.length, 2);
     assert.equal(chain[1]?.previousRootTurnId, firstTurnId);
+    assert.equal(chain[1]?.userMessageId, null);
     assert.deepEqual(
       chain[1]?.sourceMessages.map(({ messageId, content, placement, disposition }) => ({
         messageId,
@@ -236,9 +238,23 @@ test('steering becomes durable and ordered followups automatically start the nex
     assert.ok(followupTurnId);
     const followupLedger = await fixture.readTurn(followupTurnId);
     const expectedQuotes = followupSources.flatMap((source) => source.content.quotes ?? []);
-    assert.equal(followupLedger.userMessages.length, 1);
-    assert.deepEqual(followupLedger.userMessages[0]?.quotes, expectedQuotes);
+    assert.equal(followupLedger.userMessages.length, followupSources.length);
+    assert.deepEqual(
+      followupLedger.userMessages.flatMap((message) => message.quotes ?? []),
+      expectedQuotes,
+    );
     assert.deepEqual(userRuntimeContent(followupLedger.runtimeEvents)?.quotes, expectedQuotes);
+    const sessionUserMessages = await fixture.readSessionUserMessages();
+    for (const source of followupSources) {
+      assert.equal(
+        sessionUserMessages.filter((message) => message.id === source.messageId).length,
+        1,
+      );
+    }
+    assert.equal(
+      sessionUserMessages.filter((message) => message.turnId === followupTurnId).length,
+      followupSources.length,
+    );
   });
 });
 

packages/runtime-host/src/__tests__/execution-host-queue.test.ts

@@ -205,6 +205,7 @@ test('subscribed Clients share one canonical queue and ordered root handoff', as
     await waitForTerminalTurn(tui, fixture.sessionId, successor.snapshot.rootTurn.turnId);
     await tui.close();
     await fixture.stopHost(host);
+    assert.equal(await fixture.readMessageLifecycleState(followupId), 'handed_off');
 
     const chain = await fixture.readAdmissionChain();
     assert.deepEqual(
@@ -215,6 +216,113 @@ test('subscribed Clients share one canonical queue and ordered root handoff', as
   });
 });
 
+test('production UDS admission commits one transcript before the root handoff', async () => {
+  await withExecutionRoot(async (fixture) => {
+    const host = await fixture.startHost();
+    const client = await connectClient(fixture.root);
+    const messageId = randomUUID();
+    const started = await client.request('turn.message.submit', {
+      originHostEpoch: host.hostEpoch,
+      sessionId: fixture.sessionId,
+      messageId,
+      content: { text: FAKE_ASK_USER_QUESTION_PROMPT },
+      placement: 'current_turn',
+    });
+    assert.equal(started.disposition, 'turn_started');
+    if (started.disposition !== 'turn_started') return;
+    const active = await client.queryTurn({ sessionId: fixture.sessionId, turnId: started.turnId });
+    await client.stopTurn({
+      sessionId: fixture.sessionId,
+      turnId: started.turnId,
+      runId: active.runId,
+    });
+    await client.close();
+    await fixture.stopHost(host);
+    const ledger = await fixture.readTurn(started.turnId);
+    assert.deepEqual(
+      ledger.userMessages
+        .filter((message) => message.id === messageId)
+        .map((message) => message.id),
+      [messageId],
+    );
+    assert.equal(await fixture.readMessageLifecycleState(messageId), 'cancelled');
+  });
+});
+
+test('a Host crash after queue admission recovers the durable successor once', async () => {
+  await withExecutionRoot(async (fixture) => {
+    const firstHost = await fixture.startHost();
+    const first = await connectClient(fixture.root);
+    const started = requireStartedTurn(
+      await first.startTurn({
+        sessionId: fixture.sessionId,
+        turnId: randomUUID(),
+        content: { text: FAKE_ASK_USER_QUESTION_PROMPT },
+      }),
+    );
+    const messageId = randomUUID();
+    const queued = await first.request('turn.message.submit', {
+      originHostEpoch: firstHost.hostEpoch,
+      sessionId: fixture.sessionId,
+      messageId,
+      content: { text: 'recover this accepted successor' },
+      placement: 'next_turn',
+    });
+    assert.equal(queued.disposition, 'followup');
+    await fixture.killHost(firstHost);
+    await first.closed;
+
+    const secondHost = await fixture.startHost();
+    const second = await connectClient(fixture.root);
+    const subscription = await second.openSessionSubscription({
+      sessionId: fixture.sessionId,
+      transcript: { kind: 'none' },
+    });
+    const probe = new SubscriptionProbe(subscription);
+    const successor = await probe.waitFor(
+      (frame) =>
+        frame.kind === 'subscription.session_projection' &&
+        frame.snapshot.rootTurn !== null &&
+        frame.snapshot.rootTurn.turnId !== started.turnId,
+      'durable successor was not recovered after the Host crash',
+    );
+    assert.equal(successor.kind, 'subscription.session_projection');
+    if (successor.kind !== 'subscription.session_projection' || !successor.snapshot.rootTurn)
+      return;
+    await waitForTerminalTurn(second, fixture.sessionId, successor.snapshot.rootTurn.turnId);
+    await subscription.close();
+    await probe.done;
+    await second.close();
+    await fixture.stopHost(secondHost);
+    assert.equal(await fixture.readMessageLifecycleState(messageId), 'handed_off');
+  });
+});
+
+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();

packages/runtime-host/src/__tests__/execution-host-recovery.test.ts

@@ -299,6 +299,64 @@ test('same idle Message submit is connection-independent and starts one canonica
   });
 });
 
+test('a rejected idle Message submit leaves no durable transcript entry', async () => {
+  await withExecutionRoot(async (fixture) => {
+    const host = await fixture.startHost();
+    const client = await connectClient(fixture.root);
+    const messageId = randomUUID();
+    try {
+      await assert.rejects(
+        () =>
+          client.request('turn.message.submit', {
+            originHostEpoch: host.hostEpoch,
+            sessionId: fixture.sessionId,
+            messageId,
+            content: { text: '/skill:missing reject this submit' },
+            placement: 'current_turn',
+          }),
+        operationError('operation_conflict'),
+      );
+    } finally {
+      await client.close();
+      await fixture.stopHost(host);
+    }
+
+    assert.deepEqual(
+      (await fixture.readSessionUserMessages())
+        .filter((message) => message.id === messageId)
+        .map((message) => message.id),
+      [],
+    );
+  });
+});
+
+test('an allowed 32 KiB idle Message crosses the durable admission boundary', async () => {
+  await withExecutionRoot(async (fixture) => {
+    const host = await fixture.startHost();
+    const client = await connectClient(fixture.root);
+    const messageId = randomUUID();
+    try {
+      const started = await client.request('turn.message.submit', {
+        originHostEpoch: host.hostEpoch,
+        sessionId: fixture.sessionId,
+        messageId,
+        content: { text: 'x'.repeat(32 * 1024) },
+        placement: 'current_turn',
+      });
+      assert.equal(started.disposition, 'turn_started');
+    } finally {
+      await client.close();
+      await fixture.stopHost(host);
+    }
+    assert.deepEqual(
+      (await fixture.readSessionUserMessages())
+        .filter((message) => message.id === messageId)
+        .map((message) => message.text),
+      ['x'.repeat(32 * 1024)],
+    );
+  });
+});
+
 test('stale Session operations return not_found across the SQLite-backed UDS Host boundary', async () => {
   await withExecutionRoot(async (fixture) => {
     const host = await fixture.startHost();

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';
@@ -672,6 +676,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,
+        submittedContentDigest: contentDigest,
+        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);
@@ -943,6 +1000,20 @@ export class ExecutionFixture {
     }
   }
 
+  async readSessionUserMessages(): Promise<Array<Extract<StoredMessage, { type: 'user' }>>> {
+    const reader = await acquireReader(this.capability);
+    let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForRead>> | undefined;
+    try {
+      stores = await openInteractiveExecutionStoresForRead(reader.lease);
+      return (await stores.sessionStore.readMessages(this.sessionId)).filter(
+        (message): message is Extract<StoredMessage, { type: 'user' }> => message.type === 'user',
+      );
+    } finally {
+      await stores?.sessionStore.close?.();
+      await reader.close();
+    }
+  }
+
   async readAdmissionChain() {
     const owner = await tryAcquireInteractiveRootOwner(this.capability);
     assert.ok(owner);
@@ -957,6 +1028,18 @@ export class ExecutionFixture {
     }
   }
 
+  async readMessageLifecycleState(messageId: string) {
+    const reader = await acquireReader(this.capability);
+    let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForRead>> | undefined;
+    try {
+      stores = await openInteractiveExecutionStoresForRead(reader.lease);
+      return await stores.sessionStore.readMessageLifecycleState(this.sessionId, messageId);
+    } finally {
+      await stores?.sessionStore.close?.();
+      await reader.close();
+    }
+  }
+
   async readTurnFootprint(turnId: string): Promise<{
     admitted: boolean;
     runCount: number;