From 3232efe4cbd4dde08f39d3ab88b376baaf468e0b Mon Sep 17 00:00:00 2001 From: colafornia Date: Wed, 2 Sep 2026 13:35:10 +0800 Subject: [PATCH 1/3] fix(runtime): preserve durable event order Use RuntimeReadModel events for conversation copies so backfilled events are retained without a duplicate session-ledger read. Preserve immutable session order across timestamp conflicts and surface durable-order reader failures. --- packages/core/src/runtime-event-store.ts | 4 + .../session-revision-two-client-uds.test.ts | 82 +++++++++- .../src/__tests__/conversation-copy.test.ts | 61 ++++++++ .../session-manager-terminal-ledger.test.ts | 148 ++++++++++++++++++ packages/runtime/src/runtime-read-model.ts | 96 ++++++++++-- 5 files changed, 378 insertions(+), 13 deletions(-) diff --git a/packages/core/src/runtime-event-store.ts b/packages/core/src/runtime-event-store.ts index 9bb29085ba..2eff1fb16c 100644 --- a/packages/core/src/runtime-event-store.ts +++ b/packages/core/src/runtime-event-store.ts @@ -96,6 +96,10 @@ export interface RuntimeEventStore { event: RuntimeEvent, ): Promise; readRuntimeEvents(sessionId: string, runId: string): Promise; + /** Session-wide immutable append order when the store exposes it. */ + readSessionRuntimeEventEntries?( + sessionId: string, + ): Promise>; /** Physical append-log rows only; excludes mutable partial snapshots. */ readImmutableRuntimeEvents?(sessionId: string, runId: string): Promise; /** Versioned physical prefix with event-seq high-water and canonical digest. */ diff --git a/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts index 432ab86f0e..bf5f2de201 100644 --- a/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts +++ b/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 { + 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'), []); diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts index 5bf195594d..486a0fab47 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -1209,6 +1209,67 @@ 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; + const runtimeEventStore = { + readRuntimeEvents: async () => [], + } as Pick; + 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); diff --git a/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts b/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts index e0303f6b64..c86a328290 100644 --- a/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts +++ b/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts @@ -1913,6 +1913,154 @@ describe('SessionManager terminal ledger invariants', () => { ); }); + test('RuntimeReadModel preserves per-run event order when timestamps disagree', async () => { + const runStore = new TinyAgentRunStore(); + const run = makeRunHeader({ + sessionId: 'session-durable-order', + runId: 'run-durable-order', + turnId: 'turn-durable-order', + status: 'completed', + }); + await runStore.createRun(run); + for (const event of [ + runtimeEvent({ + id: 'rt-user-durable-order', + sessionId: run.sessionId, + runId: run.runId, + turnId: run.turnId, + ts: 2, + role: 'user', + author: 'user', + content: { kind: 'text', text: 'hello' }, + }), + runtimeEvent({ + id: 'rt-assistant-durable-order', + sessionId: run.sessionId, + runId: run.runId, + turnId: run.turnId, + ts: 1, + role: 'model', + author: 'agent', + content: { kind: 'text', text: 'world' }, + }), + runtimeEvent({ + id: 'rt-terminal-durable-order', + sessionId: run.sessionId, + runId: run.runId, + turnId: run.turnId, + ts: 3, + status: 'completed', + actions: { endInvocation: true }, + }), + ]) { + await runStore.appendRuntimeEvent(run.sessionId, run.runId, event); + } + + const view = await new RuntimeReadModel({ + runStore, + runtimeEventStore: runStore, + }).getSessionView(run.sessionId); + + assert.deepStrictEqual( + view.events.map((event) => event.id), + ['rt-user-durable-order', 'rt-assistant-durable-order', 'rt-terminal-durable-order'], + ); + }); + + test('RuntimeReadModel preserves durable order around backfilled events', async () => { + const sessionId = 'session-mixed-durable-order'; + const firstRun = makeRunHeader({ + sessionId, + runId: 'run-first-durable', + turnId: 'turn-first-durable', + status: 'running', + createdAt: 1, + }); + const backfilledRun = makeRunHeader({ + sessionId, + runId: 'run-backfilled', + turnId: 'turn-backfilled', + status: 'completed', + createdAt: 2, + }); + const lastRun = makeRunHeader({ + sessionId, + runId: 'run-last-durable', + turnId: 'turn-last-durable', + status: 'completed', + createdAt: 3, + }); + const runStore = new TinyAgentRunStore(); + for (const run of [firstRun, backfilledRun, lastRun]) await runStore.createRun(run); + + const firstEvent = runtimeEvent({ + id: 'rt-first-durable', + invocationId: 'inv-first-durable', + sessionId, + runId: firstRun.runId, + turnId: firstRun.turnId, + ts: 100, + status: 'completed', + actions: { endInvocation: true }, + }); + const lastEvent = runtimeEvent({ + id: 'rt-last-durable', + invocationId: 'inv-last-durable', + sessionId, + runId: lastRun.runId, + turnId: lastRun.turnId, + ts: 1, + status: 'completed', + actions: { endInvocation: true }, + }); + await runStore.appendRuntimeEvent(sessionId, firstRun.runId, firstEvent); + await runStore.appendRuntimeEvent(sessionId, lastRun.runId, lastEvent); + + const runtimeEventStore = Object.assign(runStore, { + readSessionRuntimeEventEntries: async () => [ + { ordinal: 1, event: firstEvent }, + { ordinal: 2, event: lastEvent }, + ], + }); + const legacyMessages: StoredMessage[] = [ + { + type: 'turn_state', + id: 'legacy-state', + turnId: backfilledRun.turnId, + ts: 50, + status: 'completed', + partialOutputRetained: false, + }, + ]; + + const view = await new RuntimeReadModel({ + runStore, + runtimeEventStore, + projectionCache: { readMessages: async () => legacyMessages }, + }).getSessionView(sessionId); + + assert.deepStrictEqual( + view.events.map((event) => event.runId), + [firstRun.runId, backfilledRun.runId, lastRun.runId], + ); + }); + + test('RuntimeReadModel rejects a failing durable-order reader', async () => { + const runStore = new TinyAgentRunStore(); + const run = makeRunHeader({ status: 'completed' }); + await runStore.createRun(run); + const runtimeEventStore = Object.assign(runStore, { + readSessionRuntimeEventEntries: async () => { + throw new Error('ordinal read rejected'); + }, + }); + + await assert.rejects( + new RuntimeReadModel({ runStore, runtimeEventStore }).getSessionView(run.sessionId), + /RuntimeEvent session order read failed/, + ); + }); + test('RuntimeReadModel rejects terminal headers when the ledger has no valid terminal fact', async () => { const runStore = new TinyAgentRunStore(); const run = makeRunHeader({ diff --git a/packages/runtime/src/runtime-read-model.ts b/packages/runtime/src/runtime-read-model.ts index 4a11aa446b..6a524ed77b 100644 --- a/packages/runtime/src/runtime-read-model.ts +++ b/packages/runtime/src/runtime-read-model.ts @@ -111,7 +111,22 @@ export class RuntimeReadModel { return this.buildView({ runs: inlineRuns, events: [], diagnostics }); } - const ordered: Array<{ event: RuntimeEvent; runIndex: number; eventIndex: number }> = []; + const durableEventOrdinals = await this.readSessionRuntimeEventOrdinals(sessionId); + const durableEventOrdinalById = new Map( + durableEventOrdinals.map(({ event, ordinal }) => [event.id, ordinal]), + ); + const eventsByRun = new Map(); + for (const { event } of durableEventOrdinals) { + const runEvents = eventsByRun.get(event.runId) ?? []; + runEvents.push(event); + eventsByRun.set(event.runId, runEvents); + } + const ordered: Array<{ + event: RuntimeEvent; + runIndex: number; + eventIndex: number; + ordinal?: number; + }> = []; const terminalFacts: RuntimeEventTerminalFact[] = []; for (let runIndex = 0; runIndex < inlineRuns.length; runIndex += 1) { const run = inlineRuns[runIndex]!; @@ -121,9 +136,7 @@ export class RuntimeReadModel { inlineRuns[runIndex] = effectiveRunHeaderFromTerminalFact(run, activeRunContext.fact); terminalFacts.push(activeRunContext.fact); diagnostics.push(...activeRunContext.fact.diagnostics); - for (let eventIndex = 0; eventIndex < activeRunContext.events.length; eventIndex += 1) { - ordered.push({ event: activeRunContext.events[eventIndex]!, runIndex, eventIndex }); - } + appendOrderedEvents(ordered, activeRunContext.events, runIndex, durableEventOrdinalById); continue; } @@ -152,15 +165,15 @@ export class RuntimeReadModel { ]); } const overlayEvents = activeRunContext?.events.flatMap(activeInteractionOverlayEvent) ?? []; - for (let eventIndex = 0; eventIndex < overlayEvents.length; eventIndex += 1) { - ordered.push({ event: overlayEvents[eventIndex]!, runIndex, eventIndex }); - } + appendOrderedEvents(ordered, overlayEvents, runIndex); continue; } let runEvents: RuntimeEvent[]; try { - runEvents = await this.deps.runtimeEventStore.readRuntimeEvents(sessionId, run.runId); + runEvents = + eventsByRun.get(run.runId) ?? + (await this.deps.runtimeEventStore.readRuntimeEvents(sessionId, run.runId)); } catch (error) { throw new RuntimeReadModelError('RuntimeEvent ledger read failed', [ readModelDiagnostic('unsupported_event', 'RuntimeEventStore.readRuntimeEvents failed', { @@ -237,11 +250,16 @@ export class RuntimeReadModel { inlineRuns[runIndex] = effectiveRunHeaderFromTerminalFact(run, terminalFact.fact); terminalFacts.push(terminalFact.fact); - for (let eventIndex = 0; eventIndex < runEvents.length; eventIndex += 1) { - ordered.push({ event: runEvents[eventIndex]!, runIndex, eventIndex }); - } + appendOrderedEvents(ordered, runEvents, runIndex, durableEventOrdinalById); } + const fallbackEventsByRun = new Map(); + for (const item of ordered) { + if (item.ordinal !== undefined) continue; + const runEvents = fallbackEventsByRun.get(item.runIndex) ?? []; + runEvents.push(item); + fallbackEventsByRun.set(item.runIndex, runEvents); + } ordered.sort( (a, b) => a.event.ts - b.event.ts || @@ -250,6 +268,24 @@ export class RuntimeReadModel { a.event.id.localeCompare(b.event.id), ); + const nextFallbackEventByRun = new Map(); + for (let index = 0; index < ordered.length; index += 1) { + const item = ordered[index]!; + if (item.ordinal !== undefined) continue; + const next = nextFallbackEventByRun.get(item.runIndex) ?? 0; + ordered[index] = fallbackEventsByRun.get(item.runIndex)![next]!; + nextFallbackEventByRun.set(item.runIndex, next + 1); + } + + const durableEvents = ordered + .filter((item) => item.ordinal !== undefined) + .sort((a, b) => a.ordinal! - b.ordinal! || a.event.id.localeCompare(b.event.id)); + let nextDurableEvent = 0; + for (let index = 0; index < ordered.length; index += 1) { + if (ordered[index]!.ordinal === undefined) continue; + ordered[index] = durableEvents[nextDurableEvent++]!; + } + return this.buildView({ runs: inlineRuns, events: ordered.map((item) => item.event), @@ -276,6 +312,24 @@ export class RuntimeReadModel { }; } + private async readSessionRuntimeEventOrdinals( + sessionId: string, + ): Promise> { + const reader = this.deps.runtimeEventStore.readSessionRuntimeEventEntries; + if (!reader) return []; + try { + return await reader.call(this.deps.runtimeEventStore, sessionId); + } catch (error) { + throw new RuntimeReadModelError('RuntimeEvent session order read failed', [ + readModelDiagnostic( + 'unsupported_event', + 'RuntimeEventStore.readSessionRuntimeEventEntries failed', + { error: errorMessage(error) }, + ), + ]); + } + } + private async backfillMissingRuntimeEvents( sessionId: string, run: AgentRunHeader, @@ -504,3 +558,23 @@ function isTerminalRunStatus(status: AgentRunHeader['status']): boolean { function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } + +function appendOrderedEvents( + ordered: Array<{ + event: RuntimeEvent; + runIndex: number; + eventIndex: number; + ordinal?: number; + }>, + events: readonly RuntimeEvent[], + runIndex: number, + ordinals?: ReadonlyMap, +): void { + const hasCompleteDurableOrder = + ordinals !== undefined && events.every((event) => ordinals.has(event.id)); + for (let eventIndex = 0; eventIndex < events.length; eventIndex += 1) { + const event = events[eventIndex]!; + const ordinal = hasCompleteDurableOrder ? ordinals.get(event.id) : undefined; + ordered.push({ event, runIndex, eventIndex, ...(ordinal !== undefined ? { ordinal } : {}) }); + } +} From 60c4a3fd53dde93d0f4e99276d2ae35a02a252d1 Mon Sep 17 00:00:00 2001 From: colafornia Date: Wed, 2 Sep 2026 18:07:42 +0800 Subject: [PATCH 2/3] fix(runtime): keep copied event sequence Preserve mutable partial snapshots and copy interleaved RuntimeEvents in session ordinal order. Generated-by: OpenAI Codex --- packages/core/src/runtime-event-store.ts | 4 +- .../src/__tests__/agent-run-inspect.test.ts | 10 ++ .../src/__tests__/conversation-copy.test.ts | 49 ++++++-- .../session-manager-terminal-ledger.test.ts | 73 +++++++++++- .../src/__tests__/session-manager.test.ts | 28 ++++- packages/runtime/src/conversation-copy.ts | 21 +++- packages/runtime/src/runtime-read-model.ts | 109 ++++++++---------- packages/storage/src/sqlite-runtime-store.ts | 21 ++-- 8 files changed, 229 insertions(+), 86 deletions(-) diff --git a/packages/core/src/runtime-event-store.ts b/packages/core/src/runtime-event-store.ts index 2eff1fb16c..1593689f0e 100644 --- a/packages/core/src/runtime-event-store.ts +++ b/packages/core/src/runtime-event-store.ts @@ -96,8 +96,8 @@ export interface RuntimeEventStore { event: RuntimeEvent, ): Promise; readRuntimeEvents(sessionId: string, runId: string): Promise; - /** Session-wide immutable append order when the store exposes it. */ - readSessionRuntimeEventEntries?( + /** Session-wide immutable append order. */ + readSessionRuntimeEventEntries( sessionId: string, ): Promise>; /** Physical append-log rows only; excludes mutable partial snapshots. */ diff --git a/packages/runtime/src/__tests__/agent-run-inspect.test.ts b/packages/runtime/src/__tests__/agent-run-inspect.test.ts index 6b586ffd81..a1eba13ae2 100644 --- a/packages/runtime/src/__tests__/agent-run-inspect.test.ts +++ b/packages/runtime/src/__tests__/agent-run-inspect.test.ts @@ -182,6 +182,7 @@ class MemoryAgentRunStore implements AgentRunStore, RuntimeEventStore { private headers = new Map(); private events = new Map(); private runtimeEvents = new Map(); + 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 { const ordered: Array<{ event: RuntimeEvent; runId: string; eventIndex: number }> = []; for (const [eventKey, events] of this.runtimeEvents.entries()) { diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts index 486a0fab47..e34c6301bf 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -1249,6 +1249,7 @@ test('conversation copy can use RuntimeEvents backfilled by the read model', asy } as Pick; const runtimeEventStore = { readRuntimeEvents: async () => [], + readSessionRuntimeEventEntries: async () => [], } as Pick; const source = await new RuntimeReadModel({ runStore: runStore as AgentRunStore, @@ -2423,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', @@ -2445,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', @@ -2478,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), @@ -2527,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', diff --git a/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts b/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts index c86a328290..156eae2614 100644 --- a/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts +++ b/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts @@ -1967,7 +1967,7 @@ describe('SessionManager terminal ledger invariants', () => { ); }); - test('RuntimeReadModel preserves durable order around backfilled events', async () => { + test('RuntimeReadModel places backfilled events after durable session order', async () => { const sessionId = 'session-mixed-durable-order'; const firstRun = makeRunHeader({ sessionId, @@ -2041,7 +2041,60 @@ describe('SessionManager terminal ledger invariants', () => { assert.deepStrictEqual( view.events.map((event) => event.runId), - [firstRun.runId, backfilledRun.runId, lastRun.runId], + [firstRun.runId, lastRun.runId, backfilledRun.runId], + ); + }); + + test('RuntimeReadModel retains terminal partial snapshots alongside durable events', async () => { + const runStore = new TinyAgentRunStore(); + const run = makeRunHeader({ status: 'cancelled', abortSource: 'user' }); + await runStore.createRun(run); + const opening = runtimeEvent({ + id: 'rt-partial-opening', + sessionId: run.sessionId, + runId: run.runId, + turnId: run.turnId, + ts: 1, + role: 'user', + author: 'user', + content: { kind: 'text', text: 'hello' }, + }); + const partial = runtimeEvent({ + id: 'rt-partial-snapshot', + sessionId: run.sessionId, + runId: run.runId, + turnId: run.turnId, + ts: 2, + partial: true, + role: 'model', + author: 'agent', + content: { kind: 'text', text: 'wor' }, + }); + const terminal = runtimeEvent({ + id: 'rt-partial-terminal', + sessionId: run.sessionId, + runId: run.runId, + turnId: run.turnId, + ts: 3, + status: 'cancelled', + actions: { endInvocation: true }, + }); + const runtimeEventStore = Object.assign(runStore, { + readRuntimeEvents: async () => [opening, partial, terminal], + readSessionRuntimeEventEntries: async () => [ + { ordinal: 1, event: opening }, + { ordinal: 2, event: terminal }, + ], + }); + + const view = await new RuntimeReadModel({ + runStore, + runtimeEventStore, + }).getSessionView(run.sessionId); + + assert.deepStrictEqual( + view.events.map((event) => event.id), + [opening.id, partial.id, terminal.id], ); }); @@ -2571,6 +2624,7 @@ class TinyAgentRunStore implements AgentRunStore, RuntimeEventStore { private headers = new Map(); private events = new Map(); private runtimeEvents = new Map(); + private runtimeEventEntries: RuntimeEvent[] = []; /** One-shot append rejections, for latching the store availability. */ failNextRuntimeEventAppends = 0; /** While true every runtime-event read rejects, a store that is down. */ @@ -2665,6 +2719,9 @@ class TinyAgentRunStore implements AgentRunStore, RuntimeEventStore { if (isTerminalRuntimeEvent(event)) await this.options.beforeTerminalRuntimeEventAppend?.(); const eventKey = key(sessionId, runId); this.runtimeEvents.set(eventKey, [...(this.runtimeEvents.get(eventKey) ?? []), clone(event)]); + if (event.partial !== true && !this.runtimeEventEntries.some(({ id }) => id === event.id)) { + this.runtimeEventEntries.push(clone(event)); + } } async ensureTerminalRuntimeEventDurable( @@ -2693,6 +2750,12 @@ class TinyAgentRunStore implements AgentRunStore, RuntimeEventStore { return clone(this.runtimeEvents.get(key(sessionId, runId)) ?? []); } + async readSessionRuntimeEventEntries(sessionId: string) { + return this.runtimeEventEntries + .filter((event) => event.sessionId === sessionId) + .map((event, index) => ({ ordinal: index + 1, event: clone(event) })); + } + async readSessionRuntimeEvents(sessionId: string): Promise { const ordered: Array<{ event: RuntimeEvent; runId: string; eventIndex: number }> = []; for (const [eventKey, events] of this.runtimeEvents.entries()) { @@ -2749,6 +2812,12 @@ class BatchingRuntimeEventStore implements RuntimeEventStore { return clone(this.events); } + async readSessionRuntimeEventEntries(sessionId: string) { + return this.events + .filter((event) => event.sessionId === sessionId && event.partial !== true) + .map((event, index) => ({ ordinal: index + 1, event: clone(event) })); + } + async readSessionRuntimeEvents(): Promise { return clone(this.events); } diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index 4d77f850d0..8b8c0e0a17 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -4936,6 +4936,8 @@ describe('SessionManager permission mode updates', () => { ensureTerminalRuntimeEventDurable: (sessionId, runId, event) => durableEvents.ensureTerminalRuntimeEventDurable(sessionId, runId, event), readRuntimeEvents: (sessionId, runId) => durableEvents.readRuntimeEvents(sessionId, runId), + readSessionRuntimeEventEntries: (sessionId) => + durableEvents.readSessionRuntimeEventEntries(sessionId), readSessionRuntimeEvents: (sessionId) => durableEvents.readSessionRuntimeEvents(sessionId), }; const backends = new BackendRegistry(); @@ -8876,7 +8878,7 @@ describe('SessionManager permission mode updates', () => { assert.deepStrictEqual(messages, seeded.projectedMessages); }); - test('getMessages orders RuntimeEvent-primary reads by session event chronology across runs', async () => { + test('getMessages orders RuntimeEvent-primary reads by durable session chronology', async () => { const store = new MemorySessionStore(); const runStore = new MemoryAgentRunStore(); const manager = makeManagerForReadCutover(store, runStore); @@ -8986,11 +8988,11 @@ describe('SessionManager permission mode updates', () => { ), [ 'user:slow:101', + 'assistant:slow:106', + 'turn_state:slow:107', 'user:fast:103', 'assistant:fast:104', 'turn_state:fast:105', - 'assistant:slow:106', - 'turn_state:slow:107', ], ); }); @@ -13524,6 +13526,7 @@ class MemoryAgentRunStore private headers = new Map(); private events = new Map(); private runtimeEvents = new Map(); + private runtimeEventEntries: RuntimeEvent[] = []; private continuationClaims = new Map(); private continuationStartKinds = new Map(); private runtimeEventAppendCount = 0; @@ -13662,6 +13665,9 @@ class MemoryAgentRunStore ...(this.runtimeEvents.get(eventKey) ?? []), copyRuntimeEvent(event), ]); + if (event.partial !== true && !this.runtimeEventEntries.some(({ id }) => id === event.id)) { + this.runtimeEventEntries.push(copyRuntimeEvent(event)); + } } async ensureTerminalRuntimeEventDurable( @@ -13695,6 +13701,12 @@ class MemoryAgentRunStore .map(copyRuntimeEvent); } + async readSessionRuntimeEventEntries(sessionId: string) { + return this.runtimeEventEntries + .filter((event) => event.sessionId === sessionId) + .map((event, index) => ({ ordinal: index + 1, event: copyRuntimeEvent(event) })); + } + replaceRuntimeEvent( sessionId: string, runId: string, @@ -14016,6 +14028,7 @@ class MissingCheckpointProjectionAgentRunStore extends MemoryAgentRunStore { class MemoryRuntimeEventStore implements RuntimeEventStore { private runtimeEvents = new Map(); + private runtimeEventEntries: RuntimeEvent[] = []; constructor( private readonly options: { @@ -14031,6 +14044,9 @@ class MemoryRuntimeEventStore implements 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( @@ -14055,6 +14071,12 @@ class MemoryRuntimeEventStore implements 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 { const ordered: Array<{ event: RuntimeEvent; runId: string; eventIndex: number }> = []; for (const [eventKey, events] of this.runtimeEvents.entries()) { diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index 15bc6f8824..b51fa52197 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -600,12 +600,25 @@ export async function cloneConversationRuntimeLedger( } if (input.runtimeEventStore.importConversationCopyRuntimeEvents) { + const importedSourceEventIds = new Set(); + const orderedBatches = input.plan.inlineRuntimeEvents.flatMap((event) => { + const cloned = clonedEventBySourceId.get(event.id); + const runId = runIds.get(event.runId); + if (!cloned || !runId) return []; + importedSourceEventIds.add(event.id); + return [{ runId, events: [cloned] }]; + }); + for (const { plan, runId } of preparedPlans) { + const remaining = plan.events.filter((event) => !importedSourceEventIds.has(event.id)); + if (remaining.length === 0) continue; + orderedBatches.push({ + runId, + events: remaining.map((event) => clonedEventBySourceId.get(event.id)!), + }); + } await input.runtimeEventStore.importConversationCopyRuntimeEvents( input.referenceMap.targetSessionId, - preparedPlans.map(({ runId, clonedRuntimeEvents }) => ({ - runId, - events: clonedRuntimeEvents, - })), + orderedBatches, ); } else { for (const { runId, clonedRuntimeEvents } of preparedPlans) { diff --git a/packages/runtime/src/runtime-read-model.ts b/packages/runtime/src/runtime-read-model.ts index 6a524ed77b..ee6dbbb8b3 100644 --- a/packages/runtime/src/runtime-read-model.ts +++ b/packages/runtime/src/runtime-read-model.ts @@ -115,18 +115,7 @@ export class RuntimeReadModel { const durableEventOrdinalById = new Map( durableEventOrdinals.map(({ event, ordinal }) => [event.id, ordinal]), ); - const eventsByRun = new Map(); - for (const { event } of durableEventOrdinals) { - const runEvents = eventsByRun.get(event.runId) ?? []; - runEvents.push(event); - eventsByRun.set(event.runId, runEvents); - } - const ordered: Array<{ - event: RuntimeEvent; - runIndex: number; - eventIndex: number; - ordinal?: number; - }> = []; + const ordered: OrderedRuntimeEvent[] = []; const terminalFacts: RuntimeEventTerminalFact[] = []; for (let runIndex = 0; runIndex < inlineRuns.length; runIndex += 1) { const run = inlineRuns[runIndex]!; @@ -171,9 +160,7 @@ export class RuntimeReadModel { let runEvents: RuntimeEvent[]; try { - runEvents = - eventsByRun.get(run.runId) ?? - (await this.deps.runtimeEventStore.readRuntimeEvents(sessionId, run.runId)); + runEvents = await this.deps.runtimeEventStore.readRuntimeEvents(sessionId, run.runId); } catch (error) { throw new RuntimeReadModelError('RuntimeEvent ledger read failed', [ readModelDiagnostic('unsupported_event', 'RuntimeEventStore.readRuntimeEvents failed', { @@ -253,38 +240,7 @@ export class RuntimeReadModel { appendOrderedEvents(ordered, runEvents, runIndex, durableEventOrdinalById); } - const fallbackEventsByRun = new Map(); - for (const item of ordered) { - if (item.ordinal !== undefined) continue; - const runEvents = fallbackEventsByRun.get(item.runIndex) ?? []; - runEvents.push(item); - fallbackEventsByRun.set(item.runIndex, runEvents); - } - ordered.sort( - (a, b) => - a.event.ts - b.event.ts || - a.runIndex - b.runIndex || - a.eventIndex - b.eventIndex || - a.event.id.localeCompare(b.event.id), - ); - - const nextFallbackEventByRun = new Map(); - for (let index = 0; index < ordered.length; index += 1) { - const item = ordered[index]!; - if (item.ordinal !== undefined) continue; - const next = nextFallbackEventByRun.get(item.runIndex) ?? 0; - ordered[index] = fallbackEventsByRun.get(item.runIndex)![next]!; - nextFallbackEventByRun.set(item.runIndex, next + 1); - } - - const durableEvents = ordered - .filter((item) => item.ordinal !== undefined) - .sort((a, b) => a.ordinal! - b.ordinal! || a.event.id.localeCompare(b.event.id)); - let nextDurableEvent = 0; - for (let index = 0; index < ordered.length; index += 1) { - if (ordered[index]!.ordinal === undefined) continue; - ordered[index] = durableEvents[nextDurableEvent++]!; - } + ordered.sort(compareOrderedRuntimeEvents); return this.buildView({ runs: inlineRuns, @@ -315,10 +271,8 @@ export class RuntimeReadModel { private async readSessionRuntimeEventOrdinals( sessionId: string, ): Promise> { - const reader = this.deps.runtimeEventStore.readSessionRuntimeEventEntries; - if (!reader) return []; try { - return await reader.call(this.deps.runtimeEventStore, sessionId); + return await this.deps.runtimeEventStore.readSessionRuntimeEventEntries(sessionId); } catch (error) { throw new RuntimeReadModelError('RuntimeEvent session order read failed', [ readModelDiagnostic( @@ -559,22 +513,57 @@ function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } +interface OrderedRuntimeEvent { + event: RuntimeEvent; + runIndex: number; + eventIndex: number; + ordinal?: number; + ordinalPhase?: -1 | 0 | 1; +} + function appendOrderedEvents( - ordered: Array<{ - event: RuntimeEvent; - runIndex: number; - eventIndex: number; - ordinal?: number; - }>, + ordered: OrderedRuntimeEvent[], events: readonly RuntimeEvent[], runIndex: number, ordinals?: ReadonlyMap, ): void { - const hasCompleteDurableOrder = - ordinals !== undefined && events.every((event) => ordinals.has(event.id)); + const nextOrdinals: Array = new Array(events.length); + let nextOrdinal: number | undefined; + for (let eventIndex = events.length - 1; eventIndex >= 0; eventIndex -= 1) { + nextOrdinal = ordinals?.get(events[eventIndex]!.id) ?? nextOrdinal; + nextOrdinals[eventIndex] = nextOrdinal; + } + let previousOrdinal: number | undefined; for (let eventIndex = 0; eventIndex < events.length; eventIndex += 1) { const event = events[eventIndex]!; - const ordinal = hasCompleteDurableOrder ? ordinals.get(event.id) : undefined; - ordered.push({ event, runIndex, eventIndex, ...(ordinal !== undefined ? { ordinal } : {}) }); + const durableOrdinal = ordinals?.get(event.id); + if (durableOrdinal !== undefined) previousOrdinal = durableOrdinal; + const ordinal = durableOrdinal ?? previousOrdinal ?? nextOrdinals[eventIndex]; + const ordinalPhase = durableOrdinal !== undefined ? 0 : previousOrdinal !== undefined ? 1 : -1; + ordered.push({ + event, + runIndex, + eventIndex, + ...(ordinal !== undefined ? { ordinal, ordinalPhase } : {}), + }); + } +} + +function compareOrderedRuntimeEvents(a: OrderedRuntimeEvent, b: OrderedRuntimeEvent): number { + if (a.ordinal !== undefined || b.ordinal !== undefined) { + if (a.ordinal === undefined) return 1; + if (b.ordinal === undefined) return -1; + return ( + a.ordinal - b.ordinal || + a.ordinalPhase! - b.ordinalPhase! || + a.eventIndex - b.eventIndex || + a.event.id.localeCompare(b.event.id) + ); } + return ( + a.event.ts - b.event.ts || + a.runIndex - b.runIndex || + a.eventIndex - b.eventIndex || + a.event.id.localeCompare(b.event.id) + ); } diff --git a/packages/storage/src/sqlite-runtime-store.ts b/packages/storage/src/sqlite-runtime-store.ts index c2c459e46c..a25a0b7cb3 100644 --- a/packages/storage/src/sqlite-runtime-store.ts +++ b/packages/storage/src/sqlite-runtime-store.ts @@ -462,19 +462,17 @@ export class SqliteRuntimeStore batches: readonly ConversationCopyRuntimeEventBatch[], ): Promise { assertRuntimeStorageSafeId(sessionId, 'Invalid session id'); - const runIds = new Set(); const canonicalBatches = batches.map(({ runId, events }) => { assertRuntimeStorageSafeId(runId, 'Invalid run id'); - if (runIds.has(runId)) { - throw new Error(`Conversation copy contains duplicate run ${runId}`); - } - runIds.add(runId); return { runId, events: events.map(canonicalizeRuntimeEventForStorage), }; }); const canonicalEvents = canonicalBatches.flatMap(({ events }) => events); + if (new Set(canonicalEvents.map(({ id }) => id)).size !== canonicalEvents.length) { + throw new Error('Conversation copy contains duplicate RuntimeEvents'); + } for (const { runId, events } of canonicalBatches) { for (const event of events) { assertNoReservedWorkspaceAuthorityAppend(event); @@ -493,7 +491,12 @@ export class SqliteRuntimeStore ); } this.transaction(() => { + const eventsByRun = new Map(); for (const { runId, events } of canonicalBatches) { + eventsByRun.set(runId, [...(eventsByRun.get(runId) ?? []), ...events]); + } + const newRunIds = new Set(); + for (const [runId, events] of eventsByRun) { const existing = ( this.db .prepare(` @@ -507,9 +510,11 @@ export class SqliteRuntimeStore if (existing.length > 0 && !isDeepStrictEqual(existing, events)) { throw new Error(`Conversation copy RuntimeEvent identity conflict for run ${runId}`); } - if (existing.length === 0) { - for (const event of events) this.insertRuntimeEvent(event, event.ts, true); - } + if (existing.length === 0) newRunIds.add(runId); + } + for (const { runId, events } of canonicalBatches) { + if (!newRunIds.has(runId)) continue; + for (const event of events) this.insertRuntimeEvent(event, event.ts, true); } if (canonicalEvents.some(isToolLedgerBearingEvent)) { this.rebuildToolProjectionsFromRuntimeEventsSync(sessionId); From c60400d8020f41fce38d3bea289ceaedbd44acce Mon Sep 17 00:00:00 2001 From: colafornia Date: Thu, 3 Sep 2026 11:28:54 +0800 Subject: [PATCH 3/3] refactor(runtime): remove ordering fallbacks Require the copy importer that preserves session order and remove the redundant ordinal phase. Generated-by: OpenAI Codex --- packages/runtime/src/conversation-copy.ts | 51 ++++++++-------------- packages/runtime/src/runtime-read-model.ts | 9 +--- 2 files changed, 21 insertions(+), 39 deletions(-) diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index b51fa52197..d35910590f 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -138,7 +138,7 @@ export interface CloneConversationRuntimeLedgerInput { readonly referenceMap: ConversationCopyArtifactReferenceMap; readonly runStore: AgentRunStore; readonly runtimeEventStore: RuntimeEventStore & { - importConversationCopyRuntimeEvents?( + importConversationCopyRuntimeEvents( sessionId: string, batches: readonly { readonly runId: string; @@ -586,7 +586,6 @@ export async function cloneConversationRuntimeLedger( invocationId, references, ), - clonedRuntimeEvents: plan.events.map((event) => clonedEventBySourceId.get(event.id)!), clonedOperationalEvents, terminalEvent, }; @@ -599,38 +598,26 @@ export async function cloneConversationRuntimeLedger( await input.runStore.createRun(clonedRun); } - if (input.runtimeEventStore.importConversationCopyRuntimeEvents) { - const importedSourceEventIds = new Set(); - const orderedBatches = input.plan.inlineRuntimeEvents.flatMap((event) => { - const cloned = clonedEventBySourceId.get(event.id); - const runId = runIds.get(event.runId); - if (!cloned || !runId) return []; - importedSourceEventIds.add(event.id); - return [{ runId, events: [cloned] }]; + const importedSourceEventIds = new Set(); + const orderedBatches = input.plan.inlineRuntimeEvents.flatMap((event) => { + const cloned = clonedEventBySourceId.get(event.id); + const runId = runIds.get(event.runId); + if (!cloned || !runId) return []; + importedSourceEventIds.add(event.id); + return [{ runId, events: [cloned] }]; + }); + for (const { plan, runId } of preparedPlans) { + const remaining = plan.events.filter((event) => !importedSourceEventIds.has(event.id)); + if (remaining.length === 0) continue; + orderedBatches.push({ + runId, + events: remaining.map((event) => clonedEventBySourceId.get(event.id)!), }); - for (const { plan, runId } of preparedPlans) { - const remaining = plan.events.filter((event) => !importedSourceEventIds.has(event.id)); - if (remaining.length === 0) continue; - orderedBatches.push({ - runId, - events: remaining.map((event) => clonedEventBySourceId.get(event.id)!), - }); - } - await input.runtimeEventStore.importConversationCopyRuntimeEvents( - input.referenceMap.targetSessionId, - orderedBatches, - ); - } else { - for (const { runId, clonedRuntimeEvents } of preparedPlans) { - for (const clonedEvent of clonedRuntimeEvents) { - await input.runtimeEventStore.appendRuntimeEvent( - input.referenceMap.targetSessionId, - runId, - clonedEvent, - ); - } - } } + await input.runtimeEventStore.importConversationCopyRuntimeEvents( + input.referenceMap.targetSessionId, + orderedBatches, + ); for (const { plan, runId, clonedOperationalEvents, terminalEvent } of preparedPlans) { for (const clonedEvent of clonedOperationalEvents) { diff --git a/packages/runtime/src/runtime-read-model.ts b/packages/runtime/src/runtime-read-model.ts index ee6dbbb8b3..eca372de48 100644 --- a/packages/runtime/src/runtime-read-model.ts +++ b/packages/runtime/src/runtime-read-model.ts @@ -518,7 +518,6 @@ interface OrderedRuntimeEvent { runIndex: number; eventIndex: number; ordinal?: number; - ordinalPhase?: -1 | 0 | 1; } function appendOrderedEvents( @@ -539,12 +538,11 @@ function appendOrderedEvents( const durableOrdinal = ordinals?.get(event.id); if (durableOrdinal !== undefined) previousOrdinal = durableOrdinal; const ordinal = durableOrdinal ?? previousOrdinal ?? nextOrdinals[eventIndex]; - const ordinalPhase = durableOrdinal !== undefined ? 0 : previousOrdinal !== undefined ? 1 : -1; ordered.push({ event, runIndex, eventIndex, - ...(ordinal !== undefined ? { ordinal, ordinalPhase } : {}), + ...(ordinal !== undefined ? { ordinal } : {}), }); } } @@ -554,10 +552,7 @@ function compareOrderedRuntimeEvents(a: OrderedRuntimeEvent, b: OrderedRuntimeEv if (a.ordinal === undefined) return 1; if (b.ordinal === undefined) return -1; return ( - a.ordinal - b.ordinal || - a.ordinalPhase! - b.ordinalPhase! || - a.eventIndex - b.eventIndex || - a.event.id.localeCompare(b.event.id) + a.ordinal - b.ordinal || a.eventIndex - b.eventIndex || a.event.id.localeCompare(b.event.id) ); } return (