You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/09/03 05:19:22 UTC
(maka) branch main updated: fix(runtime): preserve durable RuntimeEvent order (#4445)
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/main by this push:
new 01b4f8c9d5 fix(runtime): preserve durable RuntimeEvent order (#4445)
01b4f8c9d5 is described below
commit 01b4f8c9d5b220bddade70216d0efc1bef9d8576
Author: Colafornia <me...@colafornia.me>
AuthorDate: Thu Sep 3 13:19:16 2026 +0800
fix(runtime): preserve durable RuntimeEvent order (#4445)
`RuntimeReadModel` presented a session by RuntimeEvent timestamp, while the durable ledger appends by ordinal. When the two disagreed, a branch or revision copy could silently drop a valid history-compaction checkpoint: since #4348 the copy completes rather than reporting `persistence_failed`, so the loss was invisible.
Immutable session ordinals become the required ordering authority. The production view reads append order, mutable partial snapshots keep their per-run position next to their neighbours, and legacy events reconstructed only in memory with no ordinal follow the durable sequence in deterministic timestamp order. Conversation copies now require the ordered bulk importer and write inline events in source order, which is what stops the target SQLite ledger from regrouping by run and invali [...]
With one authority in place the fallbacks that existed to reconcile the two are gone: `ordinalPhase`, the per-event append fallback in the copy path, and the `clonedRuntimeEvents` field that existed only to feed it. `importConversationCopyRuntimeEvents` is required on the store input, so a store that cannot preserve order no longer typechecks.
Behavior note: repair writes take new tail ordinals, so a run repaired after a later run has already appended has its repaired tail sit at its repair-time position rather than back among its siblings, and that position copies into branch and revision targets. This is deliberate. Timestamps give no trustworthy cross-run insertion point, and grouping a run by its first ordinal would destroy interleaving that is genuinely part of the append history.
`readSessionRuntimeEvents()` stays timestamp-based. It carries mutable presentation snapshots and serves repair and trace projection, so moving it is a separate change.
Fixes #4103
Generated-by: OpenAI Codex
---
packages/core/src/runtime-event-store.ts | 4 +
.../session-revision-two-client-uds.test.ts | 82 +++++++-
.../src/__tests__/agent-run-inspect.test.ts | 10 +
.../src/__tests__/conversation-copy.test.ts | 110 ++++++++++-
.../session-manager-terminal-ledger.test.ts | 217 +++++++++++++++++++++
.../runtime/src/__tests__/session-manager.test.ts | 28 ++-
packages/runtime/src/conversation-copy.ts | 40 ++--
packages/runtime/src/runtime-read-model.ts | 92 +++++++--
packages/storage/src/sqlite-runtime-store.ts | 21 +-
9 files changed, 547 insertions(+), 57 deletions(-)
diff --git a/packages/core/src/runtime-event-store.ts b/packages/core/src/runtime-event-store.ts
index 9bb29085ba..1593689f0e 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<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. */
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<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'), []);
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<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()) {
diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts
index 5bf195594d..e34c6301bf 100644
--- a/packages/runtime/src/__tests__/conversation-copy.test.ts
+++ b/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',
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..156eae2614 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,207 @@ 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 places backfilled events after durable session order', 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, 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],
+ );
+ });
+
+ 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({
@@ -2423,6 +2624,7 @@ class TinyAgentRunStore implements AgentRunStore, RuntimeEventStore {
private headers = new Map<string, AgentRunHeader>();
private events = new Map<string, AgentRunEvent[]>();
private runtimeEvents = new Map<string, RuntimeEvent[]>();
+ 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. */
@@ -2517,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(
@@ -2545,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<RuntimeEvent[]> {
const ordered: Array<{ event: RuntimeEvent; runId: string; eventIndex: number }> = [];
for (const [eventKey, events] of this.runtimeEvents.entries()) {
@@ -2601,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<RuntimeEvent[]> {
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<string, AgentRunHeader>();
private events = new Map<string, AgentRunEvent[]>();
private runtimeEvents = new Map<string, RuntimeEvent[]>();
+ private runtimeEventEntries: RuntimeEvent[] = [];
private continuationClaims = new Map<string, ContinuationClaimV1>();
private continuationStartKinds = new Map<string, 'runtime_admission' | 'claim_repair'>();
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<string, RuntimeEvent[]>();
+ 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<RuntimeEvent[]> {
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..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,25 +598,26 @@ export async function cloneConversationRuntimeLedger(
await input.runStore.createRun(clonedRun);
}
- if (input.runtimeEventStore.importConversationCopyRuntimeEvents) {
- await input.runtimeEventStore.importConversationCopyRuntimeEvents(
- input.referenceMap.targetSessionId,
- preparedPlans.map(({ runId, clonedRuntimeEvents }) => ({
- runId,
- events: clonedRuntimeEvents,
- })),
- );
- } else {
- for (const { runId, clonedRuntimeEvents } of preparedPlans) {
- for (const clonedEvent of clonedRuntimeEvents) {
- await input.runtimeEventStore.appendRuntimeEvent(
- input.referenceMap.targetSessionId,
- runId,
- clonedEvent,
- );
- }
- }
+ const importedSourceEventIds = new Set<string>();
+ 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,
+ 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 4a11aa446b..eca372de48 100644
--- a/packages/runtime/src/runtime-read-model.ts
+++ b/packages/runtime/src/runtime-read-model.ts
@@ -111,7 +111,11 @@ 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 ordered: OrderedRuntimeEvent[] = [];
const terminalFacts: RuntimeEventTerminalFact[] = [];
for (let runIndex = 0; runIndex < inlineRuns.length; runIndex += 1) {
const run = inlineRuns[runIndex]!;
@@ -121,9 +125,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,9 +154,7 @@ 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;
}
@@ -237,18 +237,10 @@ 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);
}
- 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),
- );
+ ordered.sort(compareOrderedRuntimeEvents);
return this.buildView({
runs: inlineRuns,
@@ -276,6 +268,22 @@ export class RuntimeReadModel {
};
}
+ private async readSessionRuntimeEventOrdinals(
+ sessionId: string,
+ ): Promise<ReadonlyArray<{ ordinal: number; event: RuntimeEvent }>> {
+ try {
+ return await this.deps.runtimeEventStore.readSessionRuntimeEventEntries(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 +512,53 @@ function isTerminalRunStatus(status: AgentRunHeader['status']): boolean {
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}
+
+interface OrderedRuntimeEvent {
+ event: RuntimeEvent;
+ runIndex: number;
+ eventIndex: number;
+ ordinal?: number;
+}
+
+function appendOrderedEvents(
+ ordered: OrderedRuntimeEvent[],
+ events: readonly RuntimeEvent[],
+ runIndex: number,
+ ordinals?: ReadonlyMap<string, number>,
+): void {
+ const nextOrdinals: Array<number | undefined> = 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 durableOrdinal = ordinals?.get(event.id);
+ if (durableOrdinal !== undefined) previousOrdinal = durableOrdinal;
+ const ordinal = durableOrdinal ?? previousOrdinal ?? nextOrdinals[eventIndex];
+ ordered.push({
+ event,
+ runIndex,
+ eventIndex,
+ ...(ordinal !== undefined ? { ordinal } : {}),
+ });
+ }
+}
+
+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.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<void> {
assertRuntimeStorageSafeId(sessionId, 'Invalid session id');
- const runIds = new Set<string>();
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<string, RuntimeEvent[]>();
for (const { runId, events } of canonicalBatches) {
+ eventsByRun.set(runId, [...(eventsByRun.get(runId) ?? []), ...events]);
+ }
+ const newRunIds = new Set<string>();
+ 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);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org