fix(runtime): preserve durable RuntimeEvent order

packages/core/src/runtime-event-store.ts

@@ -96,6 +96,10 @@ export interface RuntimeEventStore {
  event: RuntimeEvent,
  ): Promise<void>;
  readRuntimeEvents(sessionId: string, runId: string): Promise<RuntimeEvent[]>;
+ /** Session-wide immutable append order. */
+ readSessionRuntimeEventEntries(
+ sessionId: string,
+ ): Promise<Array<{ readonly ordinal: number; readonly event: RuntimeEvent }>>;
  /** Physical append-log rows only; excludes mutable partial snapshots. */
  readImmutableRuntimeEvents?(sessionId: string, runId: string): Promise<RuntimeEvent[]>;
  /** Versioned physical prefix with event-seq high-water and canonical digest. */

packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts

@@ -30,6 +30,11 @@ import { type AgentGraphOperatorProvisionRequest } from '@maka/core/agent-graph-
 import { type AgentRunHeader } from '@maka/core/agent-run';
 import { type RuntimeEvent } from '@maka/core/runtime-event';
 import { WORKHUB_COORDINATION_SESSION_ID } from '@maka/core/session';
+import {
+ buildHistoryCompactCheckpoint,
+ matchHistoryCompactCheckpointPrefix,
+ validateHistoryCompactCheckpointShape,
+} from '@maka/runtime/history-compact-checkpoint';
 import { agentGraphIdForRootSession } from '@maka/runtime/stream-graph-coordinator';
 import { FAKE_ASK_USER_QUESTION_PROMPT } from '@maka/runtime/test-only/fake-backend';
 import { createAgentGraphControlStore } from '@maka/storage/agent-graph-control-store';
@@ -104,6 +109,7 @@ test('two Clients share exact retryable Session branch and revision authority',
  await stopHost(host);
  host = undefined;
 
+ await seedDurableOrderCheckpoint(capability, sourceSessionId);
  host = await startHost(root, capability.rootId);
  await verifyRestartRecoveryAndAdmission(root, sourceSessionId);
  await stopHost(host);
@@ -1011,7 +1017,7 @@ async function seedSource(
  const sourceRuntimeEvents = [
  runtimeEvent(source.id, 'run-turn-1', 'invocation-turn-1', 'turn-1', {
  id: 'user-1',
- ts: 1,
+ ts: 2,
  role: 'user',
  author: 'user',
  content: {
@@ -1034,7 +1040,7 @@ async function seedSource(
  }),
  runtimeEvent(source.id, 'run-turn-1', 'invocation-turn-1', 'turn-1', {
  id: 'assistant-1',
- ts: 2,
+ ts: 1,
  role: 'model',
  author: 'agent',
  content: { kind: 'text', text: 'first response' },
@@ -1604,6 +1610,51 @@ async function seedSource(
  }
 }
 
+async function seedDurableOrderCheckpoint(
+ capability: StorageRootCapability<'interactive'>,
+ sourceSessionId: string,
+): Promise<void> {
+ const owner = await tryAcquireInteractiveRootOwner(capability);
+ assert.ok(owner);
+ if (!owner) throw new Error('Unable to acquire execution root for checkpoint setup');
+ try {
+ const execution = await openInteractiveExecutionStoresForWrite(owner.lease);
+ const coveredRuntimeEvents = (
+ await execution.runtimeEventStore.readSessionRuntimeEventEntries(sourceSessionId)
+ )
+ .map(({ event }) => event)
+ .filter((event) => event.id === 'user-1' || event.id === 'assistant-1');
+ assert.deepEqual(
+ coveredRuntimeEvents.map((event) => event.id),
+ ['user-1', 'assistant-1'],
+ );
+ const checkpoint = buildHistoryCompactCheckpoint({
+ sessionId: sourceSessionId,
+ coveredRuntimeEvents,
+ summary: 'The first turn completed.',
+ summaryFormat: 'legacy_freeform',
+ highWaterSeq: 2,
+ });
+ await execution.agentRunStore.appendEvent(sourceSessionId, 'run-turn-1', {
+ type: 'history_compact_checkpoint_recorded',
+ id: 'checkpoint-turn-1',
+ runId: 'run-turn-1',
+ sessionId: sourceSessionId,
+ turnId: 'turn-1',
+ ts: 2,
+ data: {
+ checkpointId: checkpoint.checkpointId,
+ highWaterName: checkpoint.highWaterName,
+ highWaterSeq: checkpoint.highWaterSeq,
+ boundaryKind: 'historyCompact',
+ checkpoint,
+ },
+ });
+ } finally {
+ await owner.close();
+ }
+}
+
 async function verifyDurableBranch(
  capability: StorageRootCapability<'interactive'>,
  sourceSessionId: string,
@@ -1702,6 +1753,33 @@ async function verifyDurableBranch(
  );
  assert.ok(copiedProjectionArtifact);
  assert.equal(copiedProjectionPart.ref.relativePath, copiedProjectionArtifact.id);
+ const durableCopiedRuns =
+ await execution.agentRunStore.listSessionRuns(admittedRevisionTargetId);
+ const durableCopiedParent = durableCopiedRuns.find((run) => run.turnId === 'turn-1');
+ assert.ok(durableCopiedParent);
+ const copiedParentEvents = (
+ await execution.runtimeEventStore.readSessionRuntimeEventEntries(admittedRevisionTargetId)
+ )
+ .map(({ event }) => event)
+ .filter(
+ (event) => event.runId === durableCopiedParent.runId && event.content?.kind === 'text',
+ );
+ assert.deepEqual(
+ copiedParentEvents.map((event) => event.ts),
+ [2, 1],
+ );
+ const copiedCheckpoint = await execution.agentRunStore.readEventProjection?.(
+ admittedRevisionTargetId,
+ 'history_compact_checkpoint_recorded',
+ );
+ const copiedCheckpointData = copiedCheckpoint?.data?.checkpoint;
+ assert.ok(
+ validateHistoryCompactCheckpointShape(copiedCheckpointData, admittedRevisionTargetId),
+ );
+ assert.equal(
+ matchHistoryCompactCheckpointPrefix(copiedCheckpointData, copiedParentEvents).reason,
+ undefined,
+ );
  assert.equal((await artifacts.listPage('revision-target', { offset: 0, limit: 10 })).total, 0);
  assert.deepEqual(await todos.readOrBootstrap('revision-target'), { items: [] });
  assert.deepEqual(await execution.agentRunStore.listSessionRuns('revision-target'), []);

packages/runtime/src/__tests__/agent-run-inspect.test.ts

@@ -182,6 +182,7 @@ class MemoryAgentRunStore implements AgentRunStore, RuntimeEventStore {
  private headers = new Map<string, AgentRunHeader>();
  private events = new Map<string, AgentRunEvent[]>();
  private runtimeEvents = new Map<string, RuntimeEvent[]>();
+ private runtimeEventEntries: RuntimeEvent[] = [];
 
  constructor(private readonly options: { failRuntimeEventReads?: boolean } = {}) {}
 
@@ -229,6 +230,9 @@ class MemoryAgentRunStore implements AgentRunStore, RuntimeEventStore {
  ...(this.runtimeEvents.get(eventKey) ?? []),
  copyRuntimeEvent(event),
  ]);
+ if (event.partial !== true && !this.runtimeEventEntries.some(({ id }) => id === event.id)) {
+ this.runtimeEventEntries.push(copyRuntimeEvent(event));
+ }
  }
 
  async ensureTerminalRuntimeEventDurable(
@@ -253,6 +257,12 @@ class MemoryAgentRunStore implements AgentRunStore, RuntimeEventStore {
  return (this.runtimeEvents.get(key(sessionId, runId)) ?? []).map(copyRuntimeEvent);
  }
 
+ async readSessionRuntimeEventEntries(sessionId: string) {
+ return this.runtimeEventEntries
+ .filter((event) => event.sessionId === sessionId)
+ .map((event, index) => ({ ordinal: index + 1, event: copyRuntimeEvent(event) }));
+ }
+
  async readSessionRuntimeEvents(sessionId: string): Promise<RuntimeEvent[]> {
  const ordered: Array<{ event: RuntimeEvent; runId: string; eventIndex: number }> = [];
  for (const [eventKey, events] of this.runtimeEvents.entries()) {

packages/runtime/src/__tests__/conversation-copy.test.ts

@@ -1209,6 +1209,68 @@ test('conversation copy rejects a retained AgentRun without RuntimeEvent facts',
  }
 });
 
+test('conversation copy can use RuntimeEvents backfilled by the read model', async () => {
+ const run = agentRunHeader({
+ runId: 'run-backfilled',
+ invocationId: 'invocation-backfilled',
+ turnId: 'turn-backfilled',
+ status: 'completed',
+ updatedAt: 3,
+ completedAt: 3,
+ });
+ const legacyMessages: StoredMessage[] = [
+ {
+ type: 'user',
+ id: 'legacy-user',
+ turnId: run.turnId,
+ ts: 1,
+ text: 'hello',
+ },
+ {
+ type: 'assistant',
+ id: 'legacy-assistant',
+ turnId: run.turnId,
+ ts: 2,
+ text: 'world',
+ modelId: 'fake-model',
+ },
+ {
+ type: 'turn_state',
+ id: 'legacy-state',
+ turnId: run.turnId,
+ ts: 3,
+ status: 'completed',
+ partialOutputRetained: false,
+ },
+ ];
+ const runStore = {
+ listSessionRuns: async () => [run],
+ readEvents: async () => [],
+ } as Pick<AgentRunStore, 'listSessionRuns' | 'readEvents'>;
+ const runtimeEventStore = {
+ readRuntimeEvents: async () => [],
+ readSessionRuntimeEventEntries: async () => [],
+ } as Pick<RuntimeEventStore, 'readRuntimeEvents'>;
+ const source = await new RuntimeReadModel({
+ runStore: runStore as AgentRunStore,
+ runtimeEventStore: runtimeEventStore as RuntimeEventStore,
+ projectionCache: { readMessages: async () => legacyMessages },
+ }).getSessionView(run.sessionId);
+
+ const plan = await prepareConversationRuntimeLedgerCopy({
+ sourceSessionId: run.sessionId,
+ sourceEvents: source.events,
+ copiedMessages: source.messages,
+ runStore,
+ runtimeEventStore,
+ });
+
+ assert.deepEqual(
+ plan.runs[0]?.runtimeEvents.map((event) => event.content?.kind ?? event.status),
+ ['text', 'text', 'completed'],
+ );
+});
+
 test('conversation copy rewrites a complete tool recovery bundle atomically', async () => {
  const root = await mkdtemp(join(tmpdir(), 'maka-conversation-recovery-copy-'));
  const runStore = createSqliteAgentRunStore(root);
@@ -2362,6 +2424,16 @@ test('conversation copy rebuilds an inline checkpoint without legacy child event
  author: 'user',
  content: { kind: 'text', text: 'first' },
  }),
+ runtimeEvent({
+ id: 'event-1-assistant',
+ invocationId: 'invocation-1',
+ runId: 'run-1',
+ turnId: 'turn-1',
+ ts: 4,
+ role: 'model',
+ author: 'agent',
+ content: { kind: 'text', text: 'first response' },
+ }),
  runtimeEvent({
  id: 'event-1-terminal',
  invocationId: 'invocation-1',
@@ -2384,6 +2456,16 @@ test('conversation copy rebuilds an inline checkpoint without legacy child event
  author: 'user',
  content: { kind: 'text', text: 'second' },
  }),
+ runtimeEvent({
+ id: 'event-2-assistant',
+ invocationId: 'invocation-2',
+ runId: 'run-2',
+ turnId: 'turn-2',
+ ts: 4.5,
+ role: 'model',
+ author: 'agent',
+ content: { kind: 'text', text: 'second response' },
+ }),
  runtimeEvent({
  id: 'event-2-terminal',
  invocationId: 'invocation-2',
@@ -2417,10 +2499,26 @@ test('conversation copy rebuilds an inline checkpoint without legacy child event
  status: 'completed',
  }),
  ];
- for (const event of [...firstEvents, ...childEvents, ...secondEvents]) {
+ for (const event of [
+ firstEvents[0]!,
+ childEvents[0]!,
+ secondEvents[0]!,
+ firstEvents[1]!,
+ secondEvents[1]!,
+ firstEvents[2]!,
+ childEvents[1]!,
+ secondEvents[2]!,
+ ]) {
  await runtimeEventStore.appendRuntimeEvent(event.sessionId, event.runId, event);
  }
- const sourceEvents = [...firstEvents, ...secondEvents];
+ const sourceEvents = [
+ firstEvents[0]!,
+ secondEvents[0]!,
+ firstEvents[1]!,
+ secondEvents[1]!,
+ firstEvents[2]!,
+ secondEvents[2]!,
+ ];
  const checkpoint = buildHistoryCompactCheckpoint({
  sessionId: 'session-source',
  coveredRuntimeEvents: sourceEvents.filter(isHistoryCompactContentEvent),
@@ -2466,14 +2564,12 @@ test('conversation copy rebuilds an inline checkpoint without legacy child event
  });
 
  const targetRuns = await runStore.listSessionRuns('session-target');
- const targetEvents = (
- await Promise.all(
- targetRuns.map((run) => runtimeEventStore.readRuntimeEvents('session-target', run.runId)),
- )
- ).flat();
  const targetInlineRunIds = new Set(
  targetRuns.filter(isSessionInlineRun).map((run) => run.runId),
  );
+ const targetEvents = (await runtimeEventStore.readSessionRuntimeEventEntries('session-target'))
+ .map(({ event }) => event)
+ .filter((event) => targetInlineRunIds.has(event.runId));
  assert.ok(targetRuns.some((run) => !isSessionInlineRun(run)));
  const projectedCheckpoint = await runStore.readEventProjection?.(
  'session-target',