fix(runtime-host): recover legacy Root source messages

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

@@ -217,6 +217,151 @@ test('startup recovery replays an admitted regenerate with its source lineage',
   });
 });
 
+test('startup recovery materializes legacy terminal Root sources exactly once', async () => {
+  await withExecutionRoot(async (fixture) => {
+    const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts();
+    assert.deepEqual(
+      (await fixture.readSessionUserMessages()).filter((message) =>
+        legacy.sources.some((source) => source.messageId === message.id),
+      ),
+      [],
+    );
+
+    const firstHost = await fixture.startHost();
+    await fixture.stopHost(firstHost);
+    assert.deepEqual(
+      (await fixture.readSessionUserMessages())
+        .filter((message) => legacy.sources.some((source) => source.messageId === message.id))
+        .map(({ id, turnId, ts, text }) => ({ id, turnId, ts, text })),
+      legacy.sources.map((source) => ({
+        id: source.messageId,
+        turnId: legacy.turnId,
+        ts: source.admittedAt,
+        text: source.content.text,
+      })),
+    );
+
+    const secondHost = await fixture.startHost();
+    await fixture.stopHost(secondHost);
+    assert.deepEqual(
+      (await fixture.readSessionUserMessages())
+        .filter((message) => legacy.sources.some((source) => source.messageId === message.id))
+        .map(({ id, turnId, ts, text }) => ({ id, turnId, ts, text })),
+      legacy.sources.map((source) => ({
+        id: source.messageId,
+        turnId: legacy.turnId,
+        ts: source.admittedAt,
+        text: source.content.text,
+      })),
+    );
+  });
+});
+
+test('startup recovery replays a legacy Root without a Run before materializing its sources', async () => {
+  await withExecutionRoot(async (fixture) => {
+    const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts('missing');
+
+    const firstHost = await fixture.startHost();
+    await fixture.stopHost(firstHost);
+    const secondHost = await fixture.startHost();
+    await fixture.stopHost(secondHost);
+
+    assert.deepEqual(
+      (await fixture.readSessionUserMessages())
+        .filter((message) => legacy.sources.some((source) => source.messageId === message.id))
+        .map(({ id, turnId, ts, text }) => ({ id, turnId, ts, text })),
+      legacy.sources.map((source) => ({
+        id: source.messageId,
+        turnId: legacy.turnId,
+        ts: source.admittedAt,
+        text: source.content.text,
+      })),
+    );
+    const ledger = await fixture.readTurn(legacy.turnId);
+    assert.equal(ledger.runs.length, 1);
+    assert.equal(ledger.terminalEvents.length, 1);
+  });
+});
+
+test('startup recovery closes a legacy non-terminal Run before materializing its sources', async () => {
+  await withExecutionRoot(async (fixture) => {
+    const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts('created');
+
+    const firstHost = await fixture.startHost();
+    await fixture.stopHost(firstHost);
+    const secondHost = await fixture.startHost();
+    await fixture.stopHost(secondHost);
+
+    assert.deepEqual(
+      (await fixture.readSessionUserMessages())
+        .filter((message) => legacy.sources.some((source) => source.messageId === message.id))
+        .map(({ id, turnId, ts, text }) => ({ id, turnId, ts, text })),
+      legacy.sources.map((source) => ({
+        id: source.messageId,
+        turnId: legacy.turnId,
+        ts: source.admittedAt,
+        text: source.content.text,
+      })),
+    );
+    const ledger = await fixture.readTurn(legacy.turnId);
+    assert.equal(ledger.runs.length, 1);
+    assert.equal(ledger.terminalEvents.length, 1);
+  });
+});
+
+test('startup recovery rejects an unproven legacy Root without creating its missing Run', async () => {
+  await withExecutionRoot(async (fixture) => {
+    const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts('missing');
+    fixture.deleteRootSourceProof(legacy.sources[1].messageId);
+
+    await fixture.expectHostStartupFailure();
+    await fixture.assertOwnerAvailable();
+    assert.deepEqual(await fixture.readTurnRuns(legacy.turnId), []);
+    assert.deepEqual(
+      (await fixture.readSessionUserMessages()).filter((message) =>
+        legacy.sources.some((source) => source.messageId === message.id),
+      ),
+      [],
+    );
+  });
+});
+
+test('startup recovery rejects an unproven legacy non-terminal Run before closing it', async () => {
+  await withExecutionRoot(async (fixture) => {
+    const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts('created');
+    fixture.deleteRootSourceProof(legacy.sources[1].messageId);
+
+    await fixture.expectHostStartupFailure();
+    await fixture.assertOwnerAvailable();
+    const ledger = await fixture.readTurn(legacy.turnId);
+    assert.equal(ledger.runs.length, 1);
+    assert.equal(ledger.runs[0]?.status, 'created');
+    assert.equal(ledger.terminalEvents.length, 0);
+    assert.deepEqual(
+      (await fixture.readSessionUserMessages()).filter((message) =>
+        legacy.sources.some((source) => source.messageId === message.id),
+      ),
+      [],
+    );
+  });
+});
+
+test('startup recovery rejects a legacy terminal Root source without its durable receipt', async () => {
+  await withExecutionRoot(async (fixture) => {
+    const legacy = await fixture.seedLegacyRootWithoutSourceTranscripts();
+    fixture.deleteRootSourceProof(legacy.sources[1].messageId);
+
+    await fixture.expectHostStartupFailure();
+    await fixture.assertOwnerAvailable();
+    assert.deepEqual(
+      (await fixture.readSessionUserMessages()).filter((message) =>
+        legacy.sources.some((source) => source.messageId === message.id),
+      ),
+      [],
+    );
+  });
+});
+
 test('a fresh quoted Turn preserves durable and Runtime handoff content', async () => {
   await withExecutionRoot(async (fixture) => {
     const host = await fixture.startHost();

packages/runtime-host/src/__tests__/fixtures/execution-host-suite.ts

@@ -34,11 +34,13 @@ import { createServer, type Server } from 'node:http';
 import { connect, type Socket } from 'node:net';
 import { tmpdir } from 'node:os';
 import { dirname, join } from 'node:path';
+import { DatabaseSync } from 'node:sqlite';
 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 {
+  aggregateMessageContents,
   messageContentDigest,
   normalizeMessageContent,
   type MessageContent,
@@ -65,6 +67,7 @@ import {
   openInteractiveExecutionStoresForRead,
   openInteractiveExecutionStoresForWrite,
 } from '@maka/storage/execution-stores';
+import { OPERATIONAL_STATE_DATABASE_NAME } from '@maka/storage/operational-state-store';
 import { openInteractiveRuntimePolicyStoresForWrite } from '@maka/storage/runtime-policy-stores';
 import {
   resolveRootControlNamespace,
@@ -729,6 +732,127 @@ export class ExecutionFixture {
     }
   }
 
+  async seedLegacyRootWithoutSourceTranscripts(
+    runState: 'missing' | 'created' | 'terminal' = 'terminal',
+  ): Promise<{
+    turnId: string;
+    runId: string;
+    sources: readonly [
+      { messageId: string; content: MessageContent; admittedAt: number },
+      { messageId: string; content: MessageContent; admittedAt: number },
+    ];
+  }> {
+    const owner = await tryAcquireInteractiveRootOwner(this.capability);
+    assert.ok(owner);
+    if (!owner) throw new Error('Unable to acquire execution root for legacy Root setup');
+    let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForWrite>> | undefined;
+    try {
+      stores = await openInteractiveExecutionStoresForWrite(owner.lease);
+      const turnId = randomUUID();
+      const runId = randomUUID();
+      const admittedAt = Date.now();
+      const followup = {
+        messageId: randomUUID(),
+        content: { text: 'legacy follow-up source' },
+        admittedAt,
+      };
+      const steering = {
+        messageId: randomUUID(),
+        content: { text: 'legacy steering source' },
+        admittedAt,
+      };
+      const normalizedInput = aggregateMessageContents([followup.content, steering.content]);
+      const admission = await stores.agentRunStore.admitRootTurn({
+        sessionId: this.sessionId,
+        turnId,
+        proposedRunId: runId,
+        proposedUserMessageId: null,
+        execution: {
+          kind: 'external_message',
+          inputDigest: messageContentDigest(normalizedInput),
+        },
+        previousRootTurnId: null,
+        normalizedInput,
+        sourceMessages: [
+          {
+            messageId: followup.messageId,
+            content: followup.content,
+            submittedContentDigest: messageContentDigest(followup.content),
+            placement: 'next_turn',
+            disposition: 'followup',
+          },
+          {
+            messageId: steering.messageId,
+            content: steering.content,
+            submittedContentDigest: messageContentDigest(steering.content),
+            placement: 'current_turn',
+            disposition: 'steering',
+          },
+        ],
+        admittedAt,
+      });
+      assert.equal(admission.kind, 'admitted');
+      const run: AgentRunHeader = {
+        runId,
+        invocationId: runId,
+        sessionId: this.sessionId,
+        turnId,
+        status: 'created',
+        backendKind: 'fake',
+        llmConnectionSlug: 'fake',
+        modelId: 'fake-model',
+        cwd: this.root,
+        permissionMode: 'ask',
+        createdAt: admittedAt,
+        updatedAt: admittedAt,
+      };
+      if (runState !== 'missing') {
+        await stores.agentRunStore.createRun(run, { durable: true });
+      }
+      if (runState === 'terminal') {
+        const terminalAt = admittedAt + 1;
+        const terminal = buildRecoveredTerminalRuntimeEvent({
+          id: randomUUID(),
+          run,
+          status: 'failed',
+          ts: terminalAt,
+          failureClass: 'legacy_terminal',
+          recoveryReason: 'test_legacy_terminal_root',
+        });
+        await commitTerminalRunWithRuntimeFact({
+          runStore: stores.agentRunStore,
+          runtimeEventStore: stores.runtimeEventStore,
+          newId: randomUUID,
+          sessionId: this.sessionId,
+          runId,
+          turnId,
+          status: 'failed',
+          ts: terminalAt,
+          terminalEvent: terminal,
+          failureClass: 'legacy_terminal',
+        });
+      }
+      return { turnId, runId, sources: [followup, steering] };
+    } finally {
+      await stores?.sessionStore.close?.();
+      await owner.close();
+    }
+  }
+
+  deleteRootSourceProof(messageId: string): void {
+    const database = new DatabaseSync(join(this.root, OPERATIONAL_STATE_DATABASE_NAME));
+    try {
+      const result = database
+        .prepare(
+          'DELETE FROM core_root_source_message_proofs WHERE session_id = ? AND message_id = ?',
+        )
+        .run(this.sessionId, messageId);
+      assert.equal(result.changes, 1);
+    } finally {
+      database.close();
+    }
+  }
+
   async archiveSession(): Promise<void> {
     const owner = await tryAcquireInteractiveRootOwner(this.capability);
     assert.ok(owner);
@@ -1000,6 +1124,20 @@ export class ExecutionFixture {
     }
   }
 
+  async readTurnRuns(turnId: string) {
+    const reader = await acquireReader(this.capability);
+    let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForRead>> | undefined;
+    try {
+      stores = await openInteractiveExecutionStoresForRead(reader.lease);
+      return (await stores.agentRunStore.listSessionRuns(this.sessionId)).filter(
+        (candidate) => candidate.turnId === turnId,
+      );
+    } finally {
+      await stores?.sessionStore.close?.();
+      await reader.close();
+    }
+  }
+
   async readSessionUserMessages(): Promise<Array<Extract<StoredMessage, { type: 'user' }>>> {
     const reader = await acquireReader(this.capability);
     let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForRead>> | undefined;