Persist RuntimeEvents per AgentRun

packages/core/src/agent-run.ts

@@ -1,4 +1,5 @@
 import type { PermissionMode } from './permission.js';
+import type { RuntimeEvent } from './runtime-event.js';
 import type { BackendKind } from './session.js';
 
 export const AGENT_RUN_STATUSES = [
@@ -82,4 +83,6 @@ export interface AgentRunStore {
  listSessionRuns(sessionId: string): Promise<AgentRunHeader[]>;
  appendEvent(sessionId: string, runId: string, event: AgentRunEvent): Promise<void>;
  readEvents(sessionId: string, runId: string): Promise<AgentRunEvent[]>;
+ appendRuntimeEvent(sessionId: string, runId: string, event: RuntimeEvent): Promise<void>;
+ readRuntimeEvents(sessionId: string, runId: string): Promise<RuntimeEvent[]>;
 }

packages/runtime/src/__tests__/session-manager.test.ts

@@ -7,6 +7,7 @@ import type {
  AgentRunEvent,
  AgentRunHeader,
  AgentRunStore,
+ RuntimeEvent,
  SessionEvent,
  SessionHeader,
  SessionListFilter,
@@ -251,6 +252,42 @@ describe('SessionManager permission mode updates', () => {
  expect(result.events[0]?.content).toEqual({ kind: 'text', text: 'hello' });
  expect(result.events[1]?.content).toEqual({ kind: 'text', text: 'ok' });
  expect(result.events[2]?.status).toBe('completed');
+
+ const runtimeEvents = await runStore.readRuntimeEvents(session.id, run.runId);
+ expect(runtimeEvents.map((event) => event.id)).toEqual(['id-7', 'turn-1-delta', 'turn-1-complete']);
+ expect(runtimeEvents.map((event) => event.runId)).toEqual([run.runId, run.runId, run.runId]);
+ expect(runtimeEvents.map((event) => event.sessionId)).toEqual([session.id, session.id, session.id]);
+ expect(runtimeEvents.map((event) => event.turnId)).toEqual(['turn-1', 'turn-1', 'turn-1']);
+ expect(runtimeEvents.map((event) => event.role)).toEqual(['user', 'model', 'system']);
+ expect(runtimeEvents[0]?.content).toEqual({ kind: 'text', text: 'hello' });
+ expect(runtimeEvents[1]?.content).toEqual({ kind: 'text', text: 'ok' });
+ expect(runtimeEvents[2]?.status).toBe('completed');
+ });
+
+ test('runtime event ledger write failure does not fail sendMessage', async () => {
+ const store = new MemorySessionStore();
+ const runStore = new MemoryAgentRunStore({ failRuntimeEventAppends: true });
+ const backends = new BackendRegistry();
+ backends.register('fake', (ctx) => new TestBackend(ctx));
+ const manager = new SessionManager({
+ store,
+ runStore,
+ backends,
+ newId: nextId(),
+ now: nextNow(6_750),
+ runtimeSource: 'test',
+ });
+ const session = await manager.createSession(makeInput());
+
+ const sessionEvents = await collectSessionEvents(
+ manager.sendMessage(session.id, { turnId: 'turn-1', text: 'hello' }),
+ );
+
+ expect(sessionEvents.map((event) => event.type)).toEqual(['text_delta', 'complete']);
+ expect(sessionEvents.map((event) => event.id)).toEqual(['turn-1-delta', 'turn-1-complete']);
+ const [run] = await runStore.listSessionRuns(session.id);
+ if (!run) throw new Error('AgentRunStore run was not created');
+ expect(await runStore.readRuntimeEvents(session.id, run.runId)).toEqual([]);
  });
 
  test('sendMessage production source uses AiSdkFlow instead of an inline mapper flow', async () => {
@@ -1101,6 +1138,9 @@ class MemorySessionStore implements SessionStore {
 class MemoryAgentRunStore implements AgentRunStore {
  private headers = new Map<string, AgentRunHeader>();
  private events = new Map<string, AgentRunEvent[]>();
+ private runtimeEvents = new Map<string, RuntimeEvent[]>();
+
+ constructor(private readonly options: { failRuntimeEventAppends?: boolean } = {}) {}
 
  async createRun(header: AgentRunHeader): Promise<AgentRunHeader> {
  this.headers.set(key(header.sessionId, header.runId), { ...header });
@@ -1135,6 +1175,16 @@ class MemoryAgentRunStore implements AgentRunStore {
  async readEvents(sessionId: string, runId: string): Promise<AgentRunEvent[]> {
  return (this.events.get(key(sessionId, runId)) ?? []).map(copyEvent);
  }
+
+ async appendRuntimeEvent(sessionId: string, runId: string, event: RuntimeEvent): Promise<void> {
+ if (this.options.failRuntimeEventAppends) throw new Error('runtime event append failed');
+ const eventKey = key(sessionId, runId);
+ this.runtimeEvents.set(eventKey, [...(this.runtimeEvents.get(eventKey) ?? []), copyRuntimeEvent(event)]);
+ }
+
+ async readRuntimeEvents(sessionId: string, runId: string): Promise<RuntimeEvent[]> {
+ return (this.runtimeEvents.get(key(sessionId, runId)) ?? []).map(copyRuntimeEvent);
+ }
 }
 
 interface Gate {
@@ -1254,3 +1304,7 @@ function copyEvent(event: AgentRunEvent): AgentRunEvent {
  ...(event.data ? { data: { ...event.data } } : {}),
  };
 }
+
+function copyRuntimeEvent(event: RuntimeEvent): RuntimeEvent {
+ return JSON.parse(JSON.stringify(event)) as RuntimeEvent;
+}

packages/runtime/src/agent-run.ts

@@ -1,4 +1,4 @@
-import type { AgentRunEvent, AgentRunHeader, AgentRunStore } from '@maka/core';
+import type { AgentRunEvent, AgentRunHeader, AgentRunStore, RuntimeEvent } from '@maka/core';
 import { redactSecrets } from '@maka/core/redaction';
 import type {
  SessionBlockedReason,
@@ -189,6 +189,15 @@ export class AgentRun {
  }
  }
 
+ async recordRuntimeEvents(events: readonly RuntimeEvent[]): Promise<void> {
+ if (!this.input.runStore || !this.runStoreAvailable || events.length === 0) return;
+ for (const event of events) {
+ await this.enqueueRunStore('append runtime event', async () => {
+ await this.input.runStore?.appendRuntimeEvent(this.sessionId, this.runId, event);
+ });
+ }
+ }
+
  async recordFailure(error: unknown): Promise<void> {
  this.finalStatus = { status: 'blocked', blockedReason: 'unknown' };
  await this.input.hooks.appendTurnState(this.sessionId, this.turnId, 'failed', this.lineage, {

packages/runtime/src/session-manager.ts

@@ -399,6 +399,7 @@ export class SessionManager {
  lineage: run.lineage,
  abortSignal: abortController.signal,
  }).then(async (result) => {
+ await run.recordRuntimeEvents(result.events);
  await this.deps.runtimeInvocationObserver?.(result);
  return result;
  }, (error) => {

packages/storage/src/__tests__/agent-run-store.test.ts

@@ -4,7 +4,7 @@ import { tmpdir } from 'node:os';
 import { describe, it } from 'node:test';
 import assert from 'node:assert/strict';
 import { createAgentRunStore } from '../agent-run-store.js';
-import type { AgentRunEvent, AgentRunHeader } from '@maka/core';
+import type { AgentRunEvent, AgentRunHeader, RuntimeEvent } from '@maka/core';
 
 describe('AgentRunStore', () => {
  it('creates, reads, updates, and lists runs under a session', async () => {
@@ -85,6 +85,50 @@ describe('AgentRunStore', () => {
  assert.equal(events[1]?.data?.lineNumber, 2);
  });
  });
+
+ it('appends and reads runtime events from a separate per-run ledger', async () => {
+ await withStore(async (store, root) => {
+ await store.createRun(makeHeader());
+ await store.appendEvent('session-1', 'run-1', makeEvent({ id: 'operational-event' }));
+ await store.appendRuntimeEvent('session-1', 'run-1', makeRuntimeEvent({ id: 'runtime-1', role: 'user' }));
+ await store.appendRuntimeEvent('session-1', 'run-1', makeRuntimeEvent({ id: 'runtime-2', role: 'model' }));
+
+ const runtimeEvents = await store.readRuntimeEvents('session-1', 'run-1');
+ assert.deepEqual(runtimeEvents.map((event) => event.id), ['runtime-1', 'runtime-2']);
+ assert.deepEqual(runtimeEvents.map((event) => event.role), ['user', 'model']);
+ assert.deepEqual((await store.readEvents('session-1', 'run-1')).map((event) => event.id), ['operational-event']);
+
+ const runtimeEventsPath = join(root, 'sessions', 'session-1', 'runs', 'run-1', 'runtime-events.jsonl');
+ const operationalEventsPath = join(root, 'sessions', 'session-1', 'runs', 'run-1', 'events.jsonl');
+ assert.match(await readFile(runtimeEventsPath, 'utf8'), /"id":"runtime-1"/);
+ assert.match(await readFile(operationalEventsPath, 'utf8'), /"id":"operational-event"/);
+ });
+ });
+
+ it('returns an empty runtime event list when the runtime ledger is missing', async () => {
+ await withStore(async (store) => {
+ await store.createRun(makeHeader());
+
+ assert.deepEqual(await store.readRuntimeEvents('session-1', 'run-1'), []);
+ });
+ });
+
+ it('skips corrupt runtime event lines and ignores a partial corrupt tail', async () => {
+ await withStore(async (store, root) => {
+ await store.createRun(makeHeader());
+ const runtimeEventsPath = join(root, 'sessions', 'session-1', 'runs', 'run-1', 'runtime-events.jsonl');
+ await writeFile(
+ runtimeEventsPath,
+ JSON.stringify(makeRuntimeEvent({ id: 'runtime-1' })) +
+ '\n{"id":"corrupt"\n' +
+ JSON.stringify(makeRuntimeEvent({ id: 'runtime-2' })) +
+ '\n{"id":"partial"',
+ );
+
+ const events = await store.readRuntimeEvents('session-1', 'run-1');
+ assert.deepEqual(events.map((event) => event.id), ['runtime-1', 'runtime-2']);
+ });
+ });
 });
 
 async function withStore(fn: (store: ReturnType<typeof createAgentRunStore>, root: string) => Promise<void>): Promise<void> {
@@ -124,3 +168,19 @@ function makeEvent(overrides: Partial<AgentRunEvent> = {}): AgentRunEvent {
  ...overrides,
  };
 }
+
+function makeRuntimeEvent(overrides: Partial<RuntimeEvent> = {}): RuntimeEvent {
+ return {
+ id: 'runtime-1',
+ invocationId: 'turn-1',
+ runId: 'run-1',
+ sessionId: 'session-1',
+ turnId: 'turn-1',
+ ts: 1,
+ partial: false,
+ role: 'user',
+ author: 'user',
+ content: { kind: 'text', text: 'hello' },
+ ...overrides,
+ };
+}

packages/storage/src/agent-run-store.ts

@@ -1,7 +1,7 @@
 import { randomUUID } from 'node:crypto';
 import { appendFile, mkdir, readFile, readdir, rename, writeFile } from 'node:fs/promises';
 import { dirname, join } from 'node:path';
-import type { AgentRunEvent, AgentRunHeader, AgentRunStore } from '@maka/core';
+import type { AgentRunEvent, AgentRunHeader, AgentRunStore, RuntimeEvent } from '@maka/core';
 
 const SAFE_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/;
 
@@ -111,6 +111,42 @@ class FileAgentRunStore implements AgentRunStore {
  return events;
  }
 
+ async appendRuntimeEvent(sessionId: string, runId: string, event: RuntimeEvent): Promise<void> {
+ assertSafeId(sessionId, 'Invalid session id');
+ assertSafeId(runId, 'Invalid run id');
+ await this.withQueue(sessionId, runId, async () => {
+ await mkdir(this.runDir(sessionId, runId), { recursive: true });
+ await appendFile(this.runtimeEventsPath(sessionId, runId), JSON.stringify(event, sanitizeJson) + '\n', 'utf8');
+ });
+ }
+
+ async readRuntimeEvents(sessionId: string, runId: string): Promise<RuntimeEvent[]> {
+ assertSafeId(sessionId, 'Invalid session id');
+ assertSafeId(runId, 'Invalid run id');
+ let text: string;
+ try {
+ text = await readFile(this.runtimeEventsPath(sessionId, runId), 'utf8');
+ } catch (error) {
+ if ((error as NodeJS.ErrnoException).code === 'ENOENT') return [];
+ throw error;
+ }
+ const rawLines = text.split('\n');
+ const endsWithNewline = text.endsWith('\n');
+ const lines = rawLines
+ .map((line, index) => ({ line, lineNumber: index + 1 }))
+ .filter((entry) => entry.line.trim().length > 0);
+ const lastLineNumber = lines.at(-1)?.lineNumber;
+ const events: RuntimeEvent[] = [];
+ for (const entry of lines) {
+ try {
+ events.push(JSON.parse(entry.line) as RuntimeEvent);
+ } catch {
+ if (!endsWithNewline && entry.lineNumber === lastLineNumber) continue;
+ }
+ }
+ return events;
+ }
+
  private async readRunUnlocked(sessionId: string, runId: string): Promise<AgentRunHeader> {
  assertSafeId(sessionId, 'Invalid session id');
  assertSafeId(runId, 'Invalid run id');
@@ -135,6 +171,10 @@ class FileAgentRunStore implements AgentRunStore {
  return join(this.runDir(sessionId, runId), 'events.jsonl');
  }
 
+ private runtimeEventsPath(sessionId: string, runId: string): string {
+ return join(this.runDir(sessionId, runId), 'runtime-events.jsonl');
+ }
+
  private withQueue(sessionId: string, runId: string, operation: () => Promise<void>): Promise<void> {
  assertSafeId(sessionId, 'Invalid session id');
  assertSafeId(runId, 'Invalid run id');