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

@@ -544,7 +544,7 @@ function createMessages(
       readImmutableSteeringMessageProof: (requestedSessionId, messageId) =>
         stores.runtimeEventStore.readImmutableSteeringMessageProof(requestedSessionId, messageId),
     },
-    receipts: stores.messageReceiptStore,
+    admissions: stores.sessionStore,
     sessionAdmission: new SessionAdmissionGate(),
     acquireResidency: () => ({ release: () => undefined }),
     preflightSessionSnapshot: () => true,
@@ -650,7 +650,6 @@ async function withStores(
   if (!owner) throw new Error('Unable to acquire test root');
   try {
     const stores = await openInteractiveExecutionStoresForWrite(owner.lease);
-    await stores.messageReceiptStore.beginHostEpoch('epoch-1');
     await run(capability.canonicalPath, stores);
   } finally {
     await owner.close();

packages/runtime-host/src/__tests__/client-capability-coordinator.test.ts

@@ -676,6 +676,26 @@ describe('Host Client Capability coordinator', () => {
     await coordinator.close();
   });
 
+  test('rebuilds a multi-source external root binding from its durable execution contract', async () => {
+    const coordinator = createCoordinator();
+    const connection = coordinator.attachConnection(
+      clientCapabilityConnectionIdentity('connection-a'),
+      { send: async () => {} },
+    );
+    await replace(coordinator, 'connection-a', 'registration-a', 'inspect');
+
+    await coordinator.bindDurableRoot({
+      sessionId: 'session-a',
+      execution: { kind: 'external_message' },
+    });
+
+    const snapshot = coordinator.snapshotForSession('session-a');
+    assert.deepEqual(snapshot?.registrationIds, ['registration-a']);
+    snapshot?.release();
+    await connection.close();
+    await coordinator.close();
+  });
+
   test('retires Session bindings after explicit replacement and unregister', async () => {
     const coordinator = createCoordinator();
     const connection = coordinator.attachConnection(

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', {
@@ -195,7 +212,6 @@ test('steering becomes durable and ordered followups automatically start the nex
     await first.close();
     await second.close();
     await fixture.stopHost(host);
-
     const firstLedger = await fixture.readTurn(firstTurnId);
     const steeringEvents = firstLedger.runtimeEvents.filter(
       (event) =>
@@ -213,32 +229,53 @@ 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,
         content,
         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 ?? []);
-    assert.equal(followupLedger.userMessages.length, 1);
-    assert.deepEqual(followupLedger.userMessages[0]?.quotes, expectedQuotes);
+    const expectedQuotes = orderedFollowupSources.flatMap((source) => source.content.quotes ?? []);
+    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 orderedFollowupSources) {
+      assert.equal(
+        sessionUserMessages.filter((message) => message.id === source.messageId).length,
+        1,
+      );
+    }
+    assert.equal(
+      sessionUserMessages.filter((message) => message.turnId === followupTurnId).length,
+      orderedFollowupSources.length,
+    );
+    assert.deepEqual(
+      sessionUserMessages
+        .filter((message) => message.turnId === followupTurnId)
+        .map((message) => message.id),
+      orderedFollowupSources.map((source) => source.messageId),
+    );
   });
 });
 
@@ -290,6 +327,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),
@@ -359,6 +401,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);

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

@@ -156,22 +156,33 @@ test('subscribed Clients share one canonical queue and ordered root handoff', as
       }
     }
 
-    const followupId = randomUUID();
-    const followupContent = { text: 'continue after the first root completes' };
-    const queued = await tui.request('turn.message.submit', {
+    const desktopFollowupId = randomUUID();
+    const desktopFollowupContent = { text: 'continue from the desktop' };
+    const desktopQueued = await desktop.request('turn.message.submit', {
       originHostEpoch: host.hostEpoch,
       sessionId: fixture.sessionId,
-      messageId: followupId,
-      content: followupContent,
+      messageId: desktopFollowupId,
+      content: desktopFollowupContent,
       placement: 'next_turn',
     });
-    assert.equal(queued.disposition, 'followup');
+    assert.equal(desktopQueued.disposition, 'followup');
+    const tuiFollowupId = randomUUID();
+    const tuiFollowupContent = { text: 'continue from the terminal' };
+    const tuiQueued = await tui.request('turn.message.submit', {
+      originHostEpoch: host.hostEpoch,
+      sessionId: fixture.sessionId,
+      messageId: tuiFollowupId,
+      content: tuiFollowupContent,
+      placement: 'next_turn',
+    });
+    assert.equal(tuiQueued.disposition, 'followup');
     for (const probe of [desktopProbe, tuiProbe]) {
       const queueProjection = await probe.waitFor(
         (frame) =>
           frame.kind === 'subscription.session_projection' &&
-          frame.snapshot.queue.followup.some((entry) => entry.messageId === followupId),
-        'continuity did not publish the accepted follow-up',
+          frame.snapshot.queue.followup.some((entry) => entry.messageId === desktopFollowupId) &&
+          frame.snapshot.queue.followup.some((entry) => entry.messageId === tuiFollowupId),
+        'continuity did not publish both accepted follow-ups',
       );
       assert.equal(queueProjection.kind, 'subscription.session_projection');
     }
@@ -205,13 +216,128 @@ 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);
-
     const chain = await fixture.readAdmissionChain();
     assert.deepEqual(
       chain.map((admission) => admission.turnId),
       [firstTurnId, successor.snapshot.rootTurn.turnId],
     );
-    assert.deepEqual(chain[1]?.normalizedInput, followupContent);
+    assert.deepEqual(
+      chain[1]?.sourceMessages.map((source) => source.messageId),
+      [desktopFollowupId, tuiFollowupId],
+    );
+    assert.deepEqual(chain[1]?.normalizedInput, {
+      text: `${desktopFollowupContent.text}\n\n${tuiFollowupContent.text}`,
+    });
+  });
+});
+
+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],
+    );
+  });
+});
+
+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.deepEqual(
+      (await fixture.readSessionUserMessages())
+        .filter((message) => message.id === messageId)
+        .map((message) => message.id),
+      [messageId],
+    );
+  });
+});
+
+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],
+    );
   });
 });