diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml
index b19278dea..30ee5aaef 100644
--- a/.github/workflows/run-tests.yaml
+++ b/.github/workflows/run-tests.yaml
@@ -115,7 +115,7 @@ jobs:
- batch: graphile-unit
packages: 'graphile/graphile-plugin-utils graphile/graphile-realtime-subscriptions graphile/graphile-sql-expression-validator graphile/graphile-upload-plugin'
- batch: agentic
- packages: 'agentic/protocol agentic/agentic-kit agentic/agent agentic/harness agentic/chat agentic/cli agentic/pi agentic/react agentic/agentic-server agentic/anthropic agentic/openai agentic/ollama'
+ packages: 'agentic/protocol agentic/agentic-kit agentic/agent agentic/harness agentic/chat agentic/cli agentic/pi agentic/react agentic/agentic-server agentic/anthropic agentic/openai agentic/ollama agentic/run-log agentic/pi-ext-run-log'
- batch: pgpm-unit
packages: 'pgpm/types pgpm/naming-spec pgpm/diff pgpm/import pgpm/slice pgpm/transform'
- batch: pglite
diff --git a/agentic/pi-ext-run-log/README.md b/agentic/pi-ext-run-log/README.md
new file mode 100644
index 000000000..295ed4549
--- /dev/null
+++ b/agentic/pi-ext-run-log/README.md
@@ -0,0 +1,48 @@
+
+ 
+
+
+# @agentic-kit/pi-ext-run-log
+
+The write side of the [`@agentic-kit/run-log`](../run-log) for a pi coding-agent session: a pi extension that mirrors every session entry into a run-log store, verbatim and in order.
+
+The same extension runs in both placements — a local session in Constructive Desktop and a cloud session in a long-running job. Only `runId` and the store differ, which is what makes a run observable from anywhere without a second transcript format.
+
+## Usage
+
+```ts
+import { createRunLogExtension } from '@agentic-kit/pi-ext-run-log';
+
+const { extension, flush } = createRunLogExtension({
+ runId,
+ store // any RunLogAppendStore: memory, JSONL, or the API/Postgres-backed one
+});
+
+const session = await AgentSession.create({ extensions: [extension] });
+// …
+await flush(); // before the host exits
+```
+
+## How it mirrors
+
+pi owns the session — an append-only tree of entries, persisted as JSONL — and exposes no "entry appended" event. So the extension *drains*: after each event that could have appended, it takes the entries it has not seen yet and appends them.
+
+- Index-based draining is sound because the session is append-only: entries are never rewritten or removed, only branched from.
+- The session header is mirrored once, as the first record of that session.
+- Drains are serialized, so concurrent events cannot interleave batches and break run-log ordering.
+- A drain advances its read position only after a successful append, so a failed append is retried whole rather than leaving a hole.
+- Read position is keyed to the session header id: a switch/fork/new-session re-mirrors from the start, and entries carried over are absorbed by the store's idempotency (pi entry ids). The same property makes resume free — a restarted host re-appends its history and writes nothing new.
+
+`MIRROR_EVENTS` is the default event list; narrow it with `events` if a host wants fewer drains.
+
+## Failure behavior
+
+A store failure is **rethrown into pi's event dispatch** by default. A run log that silently stops recording is worse than a loud one, so losing entries is never the default. Pass `onError` if the host would rather log and continue.
+
+## Testing
+
+`SessionMirror` is the whole mechanism and knows nothing about pi's extension API, so mirroring is tested against a fake session; the extension test drives the registered handlers with a fake `ExtensionAPI`.
+
+```sh
+pnpm test
+```
diff --git a/agentic/pi-ext-run-log/__tests__/extension.test.ts b/agentic/pi-ext-run-log/__tests__/extension.test.ts
new file mode 100644
index 000000000..9c7797b18
--- /dev/null
+++ b/agentic/pi-ext-run-log/__tests__/extension.test.ts
@@ -0,0 +1,114 @@
+import { MemoryRunLogStore, type RunLogAppendStore, START } from '@agentic-kit/run-log';
+import type { ExtensionAPI } from '@earendil-works/pi-coding-agent';
+
+import { createRunLogExtension, MIRROR_EVENTS } from '../src/extension';
+
+const header = { type: 'session', version: 3, id: 'sess-1', timestamp: '2026-01-01T00:00:00.000Z', cwd: '/w' };
+
+const entry = (id: string, parentId: string | null) => ({
+ type: 'message',
+ id,
+ parentId,
+ timestamp: '2026-01-01T00:00:01.000Z',
+ message: { role: 'assistant', content: [{ type: 'text', text: id }] }
+});
+
+interface FakePi {
+ api: ExtensionAPI;
+ emit(event: string): Promise;
+ registered: string[];
+ entries: unknown[];
+}
+
+const fakePi = (): FakePi => {
+ const handlers = new Map Promise | void>();
+ const entries: unknown[] = [];
+ const sessionManager = { getHeader: () => header, getEntries: () => entries };
+ const api = {
+ on: (event: string, handler: (event: unknown, ctx: unknown) => Promise | void) => {
+ handlers.set(event, handler);
+ }
+ } as unknown as ExtensionAPI;
+
+ return {
+ api,
+ registered: [],
+ entries,
+ emit: async (event) => {
+ const handler = handlers.get(event);
+ if (!handler) throw new Error(`no handler registered for ${event}`);
+ await handler({ type: event }, { sessionManager });
+ }
+ };
+};
+
+describe('createRunLogExtension', () => {
+ it('drains the session on each mirrored event', async () => {
+ const store = new MemoryRunLogStore();
+ const pi = fakePi();
+ const { extension } = createRunLogExtension({ runId: 'run-1', store });
+ extension(pi.api);
+
+ pi.entries.push(entry('e1', null));
+ await pi.emit('session_start');
+ pi.entries.push(entry('e2', 'e1'));
+ await pi.emit('message_end');
+
+ const page = await store.read('run-1', START);
+ expect(page.records.map((r) => r.seq)).toEqual([1, 2, 3]);
+ });
+
+ it('registers every mirrored event, and only those', async () => {
+ const seen: string[] = [];
+ const api = { on: (event: string) => seen.push(event) } as unknown as ExtensionAPI;
+ createRunLogExtension({ runId: 'run-1', store: new MemoryRunLogStore() }).extension(api);
+ expect(seen).toEqual([...MIRROR_EVENTS]);
+ });
+
+ it('honours a narrowed event list', () => {
+ const seen: string[] = [];
+ const api = { on: (event: string) => seen.push(event) } as unknown as ExtensionAPI;
+ createRunLogExtension({ runId: 'run-1', store: new MemoryRunLogStore(), events: ['turn_end'] }).extension(api);
+ expect(seen).toEqual(['turn_end']);
+ });
+
+ it('throws into pi by default when the store fails', async () => {
+ const store: RunLogAppendStore = {
+ append: async () => {
+ throw new Error('store offline');
+ }
+ };
+ const pi = fakePi();
+ createRunLogExtension({ runId: 'run-1', store }).extension(pi.api);
+ pi.entries.push(entry('e1', null));
+
+ await expect(pi.emit('message_end')).rejects.toThrow('store offline');
+ });
+
+ it('routes failures to onError when the host wants to survive them', async () => {
+ const errors: unknown[] = [];
+ const store: RunLogAppendStore = {
+ append: async () => {
+ throw new Error('store offline');
+ }
+ };
+ const pi = fakePi();
+ createRunLogExtension({ runId: 'run-1', store, onError: (error) => errors.push(error) }).extension(pi.api);
+ pi.entries.push(entry('e1', null));
+
+ await pi.emit('message_end');
+ expect(errors).toHaveLength(1);
+ });
+
+ it('flushes on demand for a host that is shutting down', async () => {
+ const store = new MemoryRunLogStore();
+ const pi = fakePi();
+ const { extension, flush, mirror } = createRunLogExtension({ runId: 'run-1', store });
+ extension(pi.api);
+ await pi.emit('session_start');
+
+ pi.entries.push(entry('e1', null));
+ expect(await flush()).toHaveLength(1);
+ expect(await mirror.drain()).toEqual([]);
+ });
+});
diff --git a/agentic/pi-ext-run-log/__tests__/mirror.test.ts b/agentic/pi-ext-run-log/__tests__/mirror.test.ts
new file mode 100644
index 000000000..9b3437eb5
--- /dev/null
+++ b/agentic/pi-ext-run-log/__tests__/mirror.test.ts
@@ -0,0 +1,186 @@
+import { MemoryRunLogStore, projectParts, type RunLogAppendStore, START } from '@agentic-kit/run-log';
+
+import { SessionMirror } from '../src/mirror';
+
+interface FakeSession {
+ getHeader(): unknown;
+ getEntries(): readonly unknown[];
+ push(entry: unknown): void;
+ reset(sessionId: string): void;
+}
+
+const fakeSession = (sessionId = 'sess-1'): FakeSession => {
+ let header: Record = { type: 'session', version: 3, id: sessionId, timestamp: '2026-01-01T00:00:00.000Z', cwd: '/w' };
+ let entries: unknown[] = [];
+ return {
+ getHeader: () => header,
+ getEntries: () => entries,
+ push: (entry) => {
+ entries.push(entry);
+ },
+ reset: (id) => {
+ header = { ...header, id };
+ entries = [];
+ }
+ };
+};
+
+let parent: string | null = null;
+let n = 0;
+const message = (role: 'user' | 'assistant', text: string) => {
+ n += 1;
+ const id = `e${n}`;
+ const entry = {
+ type: 'message',
+ id,
+ parentId: parent,
+ timestamp: `2026-01-01T00:00:0${n}.000Z`,
+ message: { role, content: [{ type: 'text', text }] }
+ };
+ parent = id;
+ return entry;
+};
+
+beforeEach(() => {
+ parent = null;
+ n = 0;
+});
+
+describe('SessionMirror', () => {
+ it('mirrors the header once, then every new entry, in order', async () => {
+ const store = new MemoryRunLogStore();
+ const session = fakeSession();
+ const mirror = new SessionMirror({ runId: 'run-1', store });
+ mirror.bind(session);
+
+ session.push(message('user', 'hi'));
+ expect(await mirror.drain()).toHaveLength(2); // header + entry
+ session.push(message('assistant', 'hello'));
+ expect(await mirror.drain()).toHaveLength(1);
+ expect(await mirror.drain()).toHaveLength(0);
+
+ const page = await store.read('run-1', START);
+ expect(page.records.map((r) => r.seq)).toEqual([1, 2, 3]);
+ expect(page.records[0].entry.type).toBe('session');
+ expect(projectParts(page.records).parts.map((p) => p.kind)).toEqual(['text', 'text']);
+ });
+
+ it('stores entries verbatim, including types it does not understand', async () => {
+ const store = new MemoryRunLogStore();
+ const session = fakeSession();
+ const mirror = new SessionMirror({ runId: 'run-1', store });
+ mirror.bind(session);
+
+ const exotic = { type: 'future_thing', id: 'x1', parentId: null as string | null, timestamp: '2026-01-01T00:00:01.000Z', payload: { deep: [1, 2] } };
+ session.push(exotic);
+ await mirror.drain();
+
+ const page = await store.read('run-1', START);
+ expect(page.records[1].entry).toEqual(exotic);
+ });
+
+ it('does nothing until bound', async () => {
+ const store = new MemoryRunLogStore();
+ const mirror = new SessionMirror({ runId: 'run-1', store });
+ expect(await mirror.drain()).toEqual([]);
+ });
+
+ it('serializes concurrent drains rather than interleaving batches', async () => {
+ const store = new MemoryRunLogStore();
+ const session = fakeSession();
+ const mirror = new SessionMirror({ runId: 'run-1', store });
+ mirror.bind(session);
+
+ session.push(message('user', 'a'));
+ const first = mirror.drain();
+ session.push(message('assistant', 'b'));
+ const second = mirror.drain();
+ const [a, b] = await Promise.all([first, second]);
+
+ const seqs = [...a, ...b].map((r) => r.seq);
+ expect(seqs).toEqual([...seqs].sort((x, y) => x - y));
+ const page = await store.read('run-1', START);
+ expect(page.records).toHaveLength(3);
+ });
+
+ it('retries the whole batch when an append fails, losing nothing', async () => {
+ const inner = new MemoryRunLogStore();
+ let fail = true;
+ const store: RunLogAppendStore = {
+ append: async (runId, entries, options) => {
+ if (fail) {
+ fail = false;
+ throw new Error('transport down');
+ }
+ return inner.append(runId, entries, options);
+ }
+ };
+ const session = fakeSession();
+ const mirror = new SessionMirror({ runId: 'run-1', store });
+ mirror.bind(session);
+
+ session.push(message('user', 'a'));
+ await expect(mirror.drain()).rejects.toThrow('transport down');
+ session.push(message('assistant', 'b'));
+ expect(await mirror.drain()).toHaveLength(3);
+ expect((await inner.read('run-1', START)).records).toHaveLength(3);
+ });
+
+ it('re-mirrors from the start when pi switches to another session', async () => {
+ const store = new MemoryRunLogStore();
+ const session = fakeSession('sess-1');
+ const mirror = new SessionMirror({ runId: 'run-1', store });
+ mirror.bind(session);
+
+ session.push(message('user', 'a'));
+ await mirror.drain();
+
+ session.reset('sess-2');
+ session.push(message('user', 'a-forked'));
+ await mirror.drain();
+
+ const page = await store.read('run-1', START);
+ const headers = page.records.filter((r) => r.entry.type === 'session');
+ expect(headers).toHaveLength(2);
+ expect(page.records.map((r) => r.seq)).toEqual([1, 2, 3, 4]);
+ });
+
+ it('replays a resumed session idempotently', async () => {
+ const store = new MemoryRunLogStore();
+ const session = fakeSession();
+ session.push(message('user', 'a'));
+ session.push(message('assistant', 'b'));
+
+ const first = new SessionMirror({ runId: 'run-1', store });
+ first.bind(session);
+ await first.drain();
+
+ // Fresh process, same session file, same run: nothing is written twice.
+ const resumed = new SessionMirror({ runId: 'run-1', store });
+ resumed.bind(session);
+ expect(await resumed.drain()).toEqual([]);
+ expect((await store.read('run-1', START)).records).toHaveLength(3);
+ });
+
+ it('rejects a malformed entry loudly', async () => {
+ const store = new MemoryRunLogStore();
+ const session = fakeSession();
+ const mirror = new SessionMirror({ runId: 'run-1', store });
+ mirror.bind(session);
+ session.push({ type: 'message', message: {} });
+
+ await expect(mirror.drain()).rejects.toThrow(/id/);
+ });
+
+ it('passes the pi session version through to the records', async () => {
+ const store = new MemoryRunLogStore();
+ const session = fakeSession();
+ const mirror = new SessionMirror({ runId: 'run-1', store, piSessionVersion: 3 });
+ mirror.bind(session);
+ session.push(message('user', 'a'));
+ await mirror.drain();
+
+ const page = await store.read('run-1', START);
+ expect(page.records.every((r) => r.piSessionVersion === 3)).toBe(true);
+ });
+});
diff --git a/agentic/pi-ext-run-log/jest.config.js b/agentic/pi-ext-run-log/jest.config.js
new file mode 100644
index 000000000..8a26efd6d
--- /dev/null
+++ b/agentic/pi-ext-run-log/jest.config.js
@@ -0,0 +1,21 @@
+/** @type {import('ts-jest').JestConfigWithTsJest} */
+module.exports = {
+ preset: 'ts-jest',
+ testEnvironment: 'node',
+ transform: {
+ '^.+\\.tsx?$': [
+ 'ts-jest',
+ {
+ babelConfig: false,
+ tsconfig: 'tsconfig.json',
+ },
+ ],
+ },
+ transformIgnorePatterns: [`/node_modules/*`],
+ testRegex: '(/__tests__/.*\\.(test|spec))\\.(jsx?|tsx?)$',
+ moduleFileExtensions: ['ts', 'tsx', 'js', 'jsx', 'json', 'node'],
+ modulePathIgnorePatterns: ['dist/*'],
+ moduleNameMapper: {
+ '^(\\.{1,2}/.*)\\.js$': '$1',
+ },
+};
diff --git a/agentic/pi-ext-run-log/package.json b/agentic/pi-ext-run-log/package.json
new file mode 100644
index 000000000..532ac6d80
--- /dev/null
+++ b/agentic/pi-ext-run-log/package.json
@@ -0,0 +1,47 @@
+{
+ "name": "@agentic-kit/pi-ext-run-log",
+ "version": "0.1.0",
+ "author": "Constructive ",
+ "description": "pi extension that mirrors every session entry into an @agentic-kit/run-log store, verbatim and in order",
+ "main": "index.js",
+ "module": "esm/index.js",
+ "types": "index.d.ts",
+ "homepage": "https://github.com/constructive-io/constructive",
+ "license": "SEE LICENSE IN LICENSE",
+ "publishConfig": {
+ "access": "public",
+ "directory": "dist"
+ },
+ "repository": {
+ "type": "git",
+ "url": "https://github.com/constructive-io/constructive"
+ },
+ "bugs": {
+ "url": "https://github.com/constructive-io/constructive/issues"
+ },
+ "scripts": {
+ "clean": "makage clean",
+ "prepack": "npm run build",
+ "build": "makage build",
+ "build:dev": "makage build --dev",
+ "lint": "eslint . --fix",
+ "test": "jest",
+ "test:watch": "jest --watch"
+ },
+ "dependencies": {
+ "@agentic-kit/run-log": "workspace:^"
+ },
+ "peerDependencies": {
+ "@earendil-works/pi-coding-agent": ">=0.79.0"
+ },
+ "devDependencies": {
+ "@earendil-works/pi-coding-agent": "0.79.6"
+ },
+ "keywords": [
+ "agentic-kit",
+ "pi",
+ "coding-agent",
+ "run-log",
+ "constructive"
+ ]
+}
diff --git a/agentic/pi-ext-run-log/src/extension.ts b/agentic/pi-ext-run-log/src/extension.ts
new file mode 100644
index 000000000..2f287ab28
--- /dev/null
+++ b/agentic/pi-ext-run-log/src/extension.ts
@@ -0,0 +1,89 @@
+/**
+ * The pi extension: drain the session mirror after anything that can append an
+ * entry. Every listed event is a point where pi has just written to the session,
+ * so the run log trails the session by at most one event.
+ */
+
+import type { RunEventRecord, RunLogAppendStore } from '@agentic-kit/run-log';
+import type { ExtensionAPI, ExtensionFactory } from '@earendil-works/pi-coding-agent';
+
+import { type SessionEntrySource, SessionMirror } from './mirror';
+
+/**
+ * Events after which the session may have grown. `session_start` also covers
+ * resume (pi replays the file before the first event), so a resumed run
+ * re-appends its history and the store's idempotency discards the duplicates.
+ */
+export const MIRROR_EVENTS = [
+ 'session_start',
+ 'session_compact',
+ 'session_tree',
+ 'session_shutdown',
+ 'input',
+ 'before_agent_start',
+ 'message_end',
+ 'tool_execution_end',
+ 'turn_end',
+ 'agent_end',
+ 'model_select',
+ 'thinking_level_select'
+] as const;
+
+export type MirrorEvent = (typeof MIRROR_EVENTS)[number];
+
+export interface RunLogExtensionOptions {
+ /** The run this session belongs to. Local and cloud runs differ only here. */
+ runId: string;
+ store: RunLogAppendStore;
+ piSessionVersion?: number;
+ events?: readonly MirrorEvent[];
+ /**
+ * Called when a drain fails. Without it the failure is rethrown into pi's
+ * event dispatch: a run log that silently stops recording is worse than a
+ * loud one, so losing entries is never the default.
+ */
+ onError?: (error: unknown) => void;
+}
+
+export interface RunLogExtension {
+ extension: ExtensionFactory;
+ /** Drain now — for a host that wants the log flushed before it exits. */
+ flush(): Promise;
+ mirror: SessionMirror;
+}
+
+export function createRunLogExtension(options: RunLogExtensionOptions): RunLogExtension {
+ const mirror = new SessionMirror({
+ runId: options.runId,
+ store: options.store,
+ ...(options.piSessionVersion === undefined ? {} : { piSessionVersion: options.piSessionVersion })
+ });
+ const events = options.events ?? MIRROR_EVENTS;
+
+ const drain = async (): Promise => {
+ try {
+ return await mirror.drain();
+ } catch (error) {
+ if (!options.onError) throw error;
+ options.onError(error);
+ return [];
+ }
+ };
+
+ const extension: ExtensionFactory = (pi: ExtensionAPI) => {
+ for (const event of events) {
+ // Each overload of `on` is typed for its own handler; the handler here
+ // ignores the event and only reads the context, so one cast at the
+ // registration boundary keeps the loop.
+ (pi.on as (name: MirrorEvent, handler: (event: unknown, ctx: { sessionManager: unknown }) => Promise) => void)(
+ event,
+ async (_event, ctx) => {
+ mirror.bind(ctx.sessionManager as SessionEntrySource);
+ await drain();
+ }
+ );
+ }
+ };
+
+ return { extension, flush: drain, mirror };
+}
diff --git a/agentic/pi-ext-run-log/src/index.ts b/agentic/pi-ext-run-log/src/index.ts
new file mode 100644
index 000000000..aefc3fbc3
--- /dev/null
+++ b/agentic/pi-ext-run-log/src/index.ts
@@ -0,0 +1,14 @@
+/**
+ * `@agentic-kit/pi-ext-run-log` — the write side of the run log for a pi
+ * session. Same extension locally and in the cloud; only `runId` and the store
+ * differ.
+ */
+
+export {
+ createRunLogExtension,
+ MIRROR_EVENTS,
+ type MirrorEvent,
+ type RunLogExtension,
+ type RunLogExtensionOptions
+} from './extension';
+export { type SessionEntrySource, SessionMirror, type SessionMirrorOptions } from './mirror';
diff --git a/agentic/pi-ext-run-log/src/mirror.ts b/agentic/pi-ext-run-log/src/mirror.ts
new file mode 100644
index 000000000..d1819da4f
--- /dev/null
+++ b/agentic/pi-ext-run-log/src/mirror.ts
@@ -0,0 +1,107 @@
+/**
+ * Mirroring pi's session into the run log.
+ *
+ * pi owns the session: it appends entries to an in-memory, append-only tree and
+ * (when persisted) a JSONL file. There is no "entry appended" event to subscribe
+ * to, so the mirror drains instead — after anything that could have appended, it
+ * takes the entries it has not seen yet and appends them to the run log
+ * verbatim. Index-based draining is sound precisely because the session is
+ * append-only: entries are never rewritten or removed, only branched from.
+ *
+ * A switch/fork/new-session replaces the entry list under the same manager, so
+ * the read position is keyed to the session header's id and resets when that id
+ * changes; entries carried into the new session are absorbed by the store's
+ * idempotency (pi entry ids).
+ *
+ * This file knows nothing about pi's extension API so it can be tested without a
+ * running agent; `./extension.ts` wires it to the events.
+ */
+
+import { assertPiSessionEntry, type PiSessionEntry, type RunEventRecord, type RunLogAppendStore } from '@agentic-kit/run-log';
+
+/** The slice of pi's `ReadonlySessionManager` the mirror needs. */
+export interface SessionEntrySource {
+ getHeader(): unknown;
+ getEntries(): readonly unknown[];
+}
+
+export interface SessionMirrorOptions {
+ runId: string;
+ store: RunLogAppendStore;
+ /** pi session format version the entries are produced under. */
+ piSessionVersion?: number;
+}
+
+export class SessionMirror {
+ private readonly runId: string;
+ private readonly store: RunLogAppendStore;
+ private readonly piSessionVersion: number | undefined;
+
+ private source: SessionEntrySource | null = null;
+ private sessionId: string | null = null;
+ private consumed = 0;
+ private headerMirrored = false;
+ private tail: Promise = Promise.resolve();
+
+ constructor(options: SessionMirrorOptions) {
+ this.runId = options.runId;
+ this.store = options.store;
+ this.piSessionVersion = options.piSessionVersion;
+ }
+
+ /** Point the mirror at a session source. Safe to call on every event. */
+ bind(source: SessionEntrySource): void {
+ this.source = source;
+ }
+
+ /**
+ * Append everything the mirror has not seen yet. Drains are serialized, so
+ * concurrent callers cannot interleave batches and break run-log ordering.
+ */
+ drain(): Promise {
+ const run = this.tail.then(() => this.flushOnce());
+ this.tail = run.then(
+ (): void => undefined,
+ (): void => undefined
+ );
+ return run;
+ }
+
+ private async flushOnce(): Promise {
+ const source = this.source;
+ if (!source) return [];
+
+ const header = source.getHeader();
+ const sessionId = headerSessionId(header);
+ if (sessionId !== null && sessionId !== this.sessionId) {
+ this.sessionId = sessionId;
+ this.consumed = 0;
+ this.headerMirrored = false;
+ }
+
+ const batch: PiSessionEntry[] = [];
+ if (!this.headerMirrored && header !== null && header !== undefined) batch.push(assertPiSessionEntry(header));
+
+ const entries = source.getEntries();
+ const upto = entries.length;
+ for (let i = this.consumed; i < upto; i += 1) batch.push(assertPiSessionEntry(entries[i]));
+ if (batch.length === 0) return [];
+
+ const written = await this.store.append(
+ this.runId,
+ batch,
+ this.piSessionVersion === undefined ? undefined : { piSessionVersion: this.piSessionVersion }
+ );
+
+ // Advance only after a successful append: a failed drain is retried whole.
+ this.headerMirrored = true;
+ this.consumed = upto;
+ return written;
+ }
+}
+
+function headerSessionId(header: unknown): string | null {
+ if (typeof header !== 'object' || header === null) return null;
+ const id = (header as { id?: unknown }).id;
+ return typeof id === 'string' && id.length > 0 ? id : null;
+}
diff --git a/agentic/pi-ext-run-log/tsconfig.esm.json b/agentic/pi-ext-run-log/tsconfig.esm.json
new file mode 100644
index 000000000..624ab17cf
--- /dev/null
+++ b/agentic/pi-ext-run-log/tsconfig.esm.json
@@ -0,0 +1,7 @@
+{
+ "extends": "./tsconfig.json",
+ "compilerOptions": {
+ "module": "es2022",
+ "outDir": "dist/esm"
+ }
+}
diff --git a/agentic/pi-ext-run-log/tsconfig.json b/agentic/pi-ext-run-log/tsconfig.json
new file mode 100644
index 000000000..df063b5ee
--- /dev/null
+++ b/agentic/pi-ext-run-log/tsconfig.json
@@ -0,0 +1,8 @@
+{
+ "extends": "../../tsconfig.json",
+ "compilerOptions": {
+ "outDir": "dist",
+ "rootDir": "src"
+ },
+ "include": ["src/**/*.ts"]
+}
diff --git a/agentic/run-log/README.md b/agentic/run-log/README.md
new file mode 100644
index 000000000..34fd6cebd
--- /dev/null
+++ b/agentic/run-log/README.md
@@ -0,0 +1,95 @@
+
+
+
+
+# @agentic-kit/run-log
+
+The append-only **run log**: one ordered record of what an agent run did, wherever it ran.
+
+A coding agent run has to be observable from a desktop app, a web UI and a CLI; it has to be resumable after a crash, a restart, or a move from the cloud to a laptop; and it has to be meterable. Those are usually four subsystems. Here they are four *projections* of one log:
+
+```
+pi session entries ──► run log (append-only) ──┬──► transcript parts (what a UI draws)
+ ├──► usage totals (what a run cost)
+ ├──► tool + approval (what is blocked)
+ └──► pi session file (how it resumes)
+```
+
+The log stores **pi entries verbatim** under four platform-owned fields. pi already versions and migrates its own session format, so re-encoding it into a second semantic event model would mean maintaining a translation layer that silently loses whatever pi adds next.
+
+```ts
+{
+ runId, // which run
+ seq, // 1-based, gapless, strictly increasing
+ recordedAt, // when the platform durably recorded it
+ piSessionVersion, // pi's session format version at write time
+ entry // the pi entry, byte-for-byte
+}
+```
+
+## Install
+
+```sh
+npm install @agentic-kit/run-log
+```
+
+## Writing
+
+Writing is an `append` against a store. Appends are **idempotent by pi entry id**, so a writer that restarts and replays its tail does not duplicate history.
+
+```ts
+import { MemoryRunLogStore } from '@agentic-kit/run-log';
+
+const store = new MemoryRunLogStore();
+await store.append('run-1', [entry]); // → the records actually written
+await store.append('run-1', [entry]); // → [] — already present
+```
+
+Two stores ship here: `MemoryRunLogStore` (the reference implementation, and the test double) and `FileRunLogStore` from the node-only entry point:
+
+```ts
+import { FileRunLogStore, writeSessionFile } from '@agentic-kit/run-log/file-store';
+```
+
+The main entry is **browser-safe** — renderers import the projectors, so nothing in it may reach for a node builtin. Anything touching the filesystem lives behind `/file-store` (the same split, for the same reason, as `12factor-env/dotenv`).
+
+The durable store is Postgres, in `constructive-db`: `append` becomes an insert, `read` a keyset scan. Neither is in this package, because the interface is the contract.
+
+## Reading
+
+Every surface is the same reader: hold a cursor, ask for what came after it. That is what makes a local run and a cloud run indistinguishable to a UI.
+
+```ts
+import { follow, projectParts, readAll } from '@agentic-kit/run-log';
+
+const records = await readAll(store, 'run-1');
+
+for await (const batch of follow(store, 'run-1', { waitForChange })) {
+ render(projectParts(batch).parts);
+}
+```
+
+`waitForChange` (Postgres `LISTEN/NOTIFY`, IPC, a websocket) is an *optimisation*: it races the poll delay, and without one `follow` degrades to polling. No new transport is required for a run to stream.
+
+## Projections
+
+| Projection | Answers |
+| --- | --- |
+| `projectParts` | the renderable transcript — a tool call and its later result collapse into one part, so no renderer correlates messages itself |
+| `projectUsage` | tokens and cost, per `provider/model` and per run, including nested tool usage and compaction calls |
+| `projectToolState` | what is running, and what is waiting on a human (approvals ride in the log as pi `custom` messages, so a surface that reconnects hours later still sees a pending request) |
+| `projectSession` | a pi session file — the resume path, cloud → local included |
+
+All four are pure and total: the same records always produce the same output, whichever host wrote them, which is the property the tests assert as *placement invariance*.
+
+Unreadable input fails loudly. A corrupt record, a mixed-version log, or a session whose header is not first throws rather than degrading into an empty transcript — a run that silently renders as blank is worse than one that reports it cannot be read. An *unknown* entry type is different: it is carried through as an `unknown` part, so a log written by a newer pi still renders.
+
+## Related
+
+- [`@agentic-kit/pi`](../pi) — Constructive's typed db tools as a pi extension
+- [`@agentic-kit/harness`](../harness) — host-neutral gates and policy
+- [`agentic-server`](../agentic-server) — the metered inference gateway
+
+## License
+
+SEE LICENSE IN LICENSE
diff --git a/agentic/run-log/__tests__/entry-points.test.ts b/agentic/run-log/__tests__/entry-points.test.ts
new file mode 100644
index 000000000..ca66972f9
--- /dev/null
+++ b/agentic/run-log/__tests__/entry-points.test.ts
@@ -0,0 +1,40 @@
+import { readdirSync, readFileSync, statSync } from 'node:fs';
+import { join } from 'node:path';
+
+/**
+ * The main entry is imported by browsers, Electron renderers and Next client
+ * components. A `node:` import anywhere in its module graph breaks those
+ * bundles, and it breaks them at build time in someone else's repo — so the
+ * boundary is asserted here rather than discovered there.
+ */
+const srcDir = join(__dirname, '..', 'src');
+
+const sources = (dir: string): string[] =>
+ readdirSync(dir).flatMap((name) => {
+ const path = join(dir, name);
+ if (statSync(path).isDirectory()) return sources(path);
+ return name.endsWith('.ts') ? [path] : [];
+ });
+
+/** Only this file may touch the filesystem. */
+const NODE_ONLY = ['file-store.ts'];
+
+describe('entry points', () => {
+ const browserSafe = sources(srcDir).filter((path) => !NODE_ONLY.some((name) => path.endsWith(name)));
+
+ it.each(browserSafe.map((path) => [path.slice(srcDir.length + 1), path]))(
+ 'src/%s imports no node builtin',
+ (_name, path) => {
+ const contents = readFileSync(path, 'utf8');
+ expect(contents).not.toMatch(/from ['"]node:/);
+ expect(contents).not.toMatch(/require\(['"]node:/);
+ }
+ );
+
+ it('keeps the filesystem store out of the main entry', () => {
+ const index = readFileSync(join(srcDir, 'index.ts'), 'utf8');
+ expect(index).not.toMatch(/from ['"]\.\/file-store['"]/);
+ // eslint-disable-next-line @typescript-eslint/no-var-requires
+ expect(Object.keys(require('../src'))).not.toContain('FileRunLogStore');
+ });
+});
diff --git a/agentic/run-log/__tests__/file-store.test.ts b/agentic/run-log/__tests__/file-store.test.ts
new file mode 100644
index 000000000..34e8b593d
--- /dev/null
+++ b/agentic/run-log/__tests__/file-store.test.ts
@@ -0,0 +1,86 @@
+import { appendFileSync, mkdtempSync, readFileSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+
+import { parseSessionJsonl, projectParts, readAll } from '../src';
+import { FileRunLogStore, writeSessionFile } from '../src/file-store';
+import { assistantText, assistantToolCall, header, resetIds, toolResult, userMessage } from './fixtures';
+
+let dir: string;
+let path: string;
+
+beforeEach(() => {
+ resetIds();
+ dir = mkdtempSync(join(tmpdir(), 'run-log-'));
+ path = join(dir, 'nested', 'run.jsonl');
+});
+
+describe('FileRunLogStore', () => {
+ it('creates the log on first append and survives a new store instance', async () => {
+ const writer = new FileRunLogStore({ path });
+ await writer.append('run-1', [header(), userMessage('hi')]);
+ await writer.append('run-1', [assistantText('hello')]);
+
+ const reader = new FileRunLogStore({ path });
+ const records = await readAll(reader, 'run-1');
+ expect(records.map((r) => r.seq)).toEqual([1, 2, 3]);
+ expect(readFileSync(path, 'utf8').trimEnd().split('\n')).toHaveLength(3);
+ });
+
+ it('reads nothing for a log that does not exist yet', async () => {
+ expect((await new FileRunLogStore({ path }).read('run-1')).records).toEqual([]);
+ });
+
+ it('resumes sequence numbering after a process restart', async () => {
+ await new FileRunLogStore({ path }).append('run-1', [userMessage('a')]);
+ const written = await new FileRunLogStore({ path }).append('run-1', [userMessage('b', 2)]);
+ expect(written[0].seq).toBe(2);
+ });
+
+ it('skips entries already on disk, so a restarted writer can replay its tail', async () => {
+ const entries = [header(), userMessage('a'), assistantText('b')];
+ await new FileRunLogStore({ path }).append('run-1', entries);
+ const retry = await new FileRunLogStore({ path }).append('run-1', entries);
+ expect(retry).toEqual([]);
+ expect(readFileSync(path, 'utf8').trimEnd().split('\n')).toHaveLength(3);
+ });
+
+ it('keeps runs separate within one file', async () => {
+ const store = new FileRunLogStore({ path });
+ await store.append('run-1', [userMessage('a')]);
+ await store.append('run-2', [userMessage('b')]);
+ expect((await store.read('run-2')).records[0].seq).toBe(1);
+ expect((await store.read('run-1')).records).toHaveLength(1);
+ });
+
+ it('throws on a truncated or corrupt line instead of returning a partial log', async () => {
+ const store = new FileRunLogStore({ path });
+ await store.append('run-1', [userMessage('a')]);
+ appendFileSync(path, '{"runId":"run-1","seq":2,');
+ await expect(store.read('run-1')).rejects.toThrow(/line 2 is not valid JSON/);
+
+ writeFileSync(path, '{"runId":"run-1","seq":1,"recordedAt":"x","piSessionVersion":3}\n');
+ await expect(store.read('run-1')).rejects.toThrow(/must be an object/);
+ });
+});
+
+describe('writeSessionFile', () => {
+ it('writes a pi session file the log can be resumed from', async () => {
+ const store = new FileRunLogStore({ path });
+ await store.append('run-1', [
+ header(),
+ userMessage('add a test'),
+ assistantToolCall({ id: 'call-1', name: 'write_file' }),
+ toolResult({ toolCallId: 'call-1', toolName: 'write_file', text: 'ok' })
+ ]);
+ const records = await readAll(store, 'run-1');
+
+ const sessionPath = writeSessionFile(join(dir, 'sessions', 'session-1.jsonl'), records);
+ const entries = parseSessionJsonl(readFileSync(sessionPath, 'utf8'));
+
+ expect(entries[0]).toMatchObject({ type: 'session', id: 'session-1' });
+ expect(entries).toHaveLength(4);
+ // The transcript a UI would draw is unchanged by the round trip.
+ expect(projectParts(records).parts).toHaveLength(2);
+ });
+});
diff --git a/agentic/run-log/__tests__/fixtures.ts b/agentic/run-log/__tests__/fixtures.ts
new file mode 100644
index 000000000..b8cf8d322
--- /dev/null
+++ b/agentic/run-log/__tests__/fixtures.ts
@@ -0,0 +1,112 @@
+import type { PiSessionEntry, PiSessionHeader, PiUsage } from '../src/pi-entry';
+
+let counter = 0;
+const nextId = (): string => {
+ counter += 1;
+ return counter.toString(16).padStart(8, '0');
+};
+
+export const resetIds = (): void => {
+ counter = 0;
+};
+
+const at = (n: number): string => new Date(Date.UTC(2026, 0, 1, 0, 0, n)).toISOString();
+
+export const usage = (over: Partial = {}): PiUsage => ({
+ input: 100,
+ output: 20,
+ cacheRead: 0,
+ cacheWrite: 0,
+ totalTokens: 120,
+ cost: { input: 0.001, output: 0.002, cacheRead: 0, cacheWrite: 0, total: 0.003 },
+ ...over
+});
+
+export const header = (over: Partial = {}): PiSessionEntry => ({
+ type: 'session',
+ version: 3,
+ id: 'session-1',
+ timestamp: at(0),
+ cwd: '/repo',
+ ...over
+});
+
+const entry = (type: string, rest: Record, seq: number): PiSessionEntry =>
+ ({ type, id: nextId(), parentId: null, timestamp: at(seq), ...rest }) as PiSessionEntry;
+
+export const userMessage = (text: string, seq = 1): PiSessionEntry =>
+ entry('message', { message: { role: 'user', content: text, timestamp: seq } }, seq);
+
+export const assistantText = (text: string, seq = 2, over: Record = {}): PiSessionEntry =>
+ entry(
+ 'message',
+ {
+ message: {
+ role: 'assistant',
+ content: [{ type: 'text', text }],
+ provider: 'anthropic',
+ model: 'claude-sonnet-4-5',
+ usage: usage(),
+ stopReason: 'stop',
+ ...over
+ }
+ },
+ seq
+ );
+
+export const assistantToolCall = (
+ call: { id: string; name: string; arguments?: Record },
+ seq = 3,
+ over: Record = {}
+): PiSessionEntry =>
+ entry(
+ 'message',
+ {
+ message: {
+ role: 'assistant',
+ content: [{ type: 'toolCall', ...call }],
+ provider: 'anthropic',
+ model: 'claude-sonnet-4-5',
+ usage: usage({ output: 30, totalTokens: 130 }),
+ stopReason: 'toolUse',
+ ...over
+ }
+ },
+ seq
+ );
+
+export const toolResult = (
+ result: { toolCallId: string; toolName: string; text: string; isError?: boolean; usage?: PiUsage },
+ seq = 4
+): PiSessionEntry =>
+ entry(
+ 'message',
+ {
+ message: {
+ role: 'toolResult',
+ toolCallId: result.toolCallId,
+ toolName: result.toolName,
+ content: [{ type: 'text', text: result.text }],
+ isError: result.isError ?? false,
+ ...(result.usage ? { usage: result.usage } : {})
+ }
+ },
+ seq
+ );
+
+export const custom = (
+ message: { customType: string; content: string; details?: unknown; display?: boolean },
+ seq = 5
+): PiSessionEntry => entry('message', { message: { role: 'custom', display: true, ...message } }, seq);
+
+export const bash = (command: string, output: string, exitCode = 0, seq = 6): PiSessionEntry =>
+ entry('message', { message: { role: 'bashExecution', command, output, exitCode } }, seq);
+
+export const compaction = (summary: string, seq = 7, over: Record = {}): PiSessionEntry =>
+ entry('compaction', { summary, tokensBefore: 50_000, ...over }, seq);
+
+export const branchSummary = (summary: string, seq = 8, over: Record = {}): PiSessionEntry =>
+ entry('branch_summary', { summary, fromId: 'aaaaaaaa', ...over }, seq);
+
+/** An entry type this version of the package has never seen. */
+export const futureEntry = (seq = 9): PiSessionEntry => entry('quantum_thought', { intensity: 11 }, seq);
diff --git a/agentic/run-log/__tests__/projectors.test.ts b/agentic/run-log/__tests__/projectors.test.ts
new file mode 100644
index 000000000..e435d1890
--- /dev/null
+++ b/agentic/run-log/__tests__/projectors.test.ts
@@ -0,0 +1,287 @@
+import {
+ APPROVAL_REQUEST_TYPE,
+ APPROVAL_RESOLUTION_TYPE,
+ approvalRequestMessage,
+ approvalResolutionMessage,
+ MemoryRunLogStore,
+ modelKey,
+ parseSessionJsonl,
+ projectParts,
+ projectSession,
+ projectToolState,
+ projectUsage,
+ readAll,
+ type RunEventRecord,
+ wrapEntry
+} from '../src';
+import type { PiSessionEntry } from '../src/pi-entry';
+import {
+ assistantText,
+ assistantToolCall,
+ bash,
+ branchSummary,
+ compaction,
+ custom,
+ futureEntry,
+ header,
+ resetIds,
+ toolResult,
+ usage,
+ userMessage
+} from './fixtures';
+
+beforeEach(resetIds);
+
+const recordsOf = (...entries: PiSessionEntry[]): RunEventRecord[] =>
+ entries.map((entry, i) => wrapEntry({ runId: 'run-1', seq: i + 1, entry, recordedAt: `2026-01-01T00:00:0${String(i)}.000Z` }));
+
+describe('projectParts', () => {
+ it('renders a full turn: user text, assistant text, tool call collapsed with its result', () => {
+ const { parts, sessionId, cwd } = projectParts(
+ recordsOf(
+ header(),
+ userMessage('add a test'),
+ assistantText('on it'),
+ assistantToolCall({ id: 'call-1', name: 'write_file', arguments: { path: 'a.ts' } }),
+ toolResult({ toolCallId: 'call-1', toolName: 'write_file', text: 'wrote a.ts' })
+ )
+ );
+
+ expect(sessionId).toBe('session-1');
+ expect(cwd).toBe('/repo');
+ expect(parts).toEqual([
+ { kind: 'text', role: 'user', text: 'add a test', seq: 2, entryId: expect.any(String) },
+ {
+ kind: 'text',
+ role: 'assistant',
+ text: 'on it',
+ model: 'claude-sonnet-4-5',
+ provider: 'anthropic',
+ seq: 3,
+ entryId: expect.any(String)
+ },
+ {
+ kind: 'tool',
+ toolCallId: 'call-1',
+ name: 'write_file',
+ arguments: { path: 'a.ts' },
+ status: 'completed',
+ output: 'wrote a.ts',
+ settledSeq: 5,
+ seq: 4,
+ entryId: expect.any(String)
+ }
+ ]);
+ });
+
+ it('leaves an unsettled tool call in the requested state', () => {
+ const [, tool] = projectParts(
+ recordsOf(userMessage('go'), assistantToolCall({ id: 'call-1', name: 'bash' }))
+ ).parts;
+ expect(tool).toMatchObject({ kind: 'tool', status: 'requested' });
+ expect((tool as { output?: string }).output).toBeUndefined();
+ });
+
+ it('marks an errored tool result as failed', () => {
+ const { parts } = projectParts(
+ recordsOf(
+ assistantToolCall({ id: 'call-1', name: 'bash' }),
+ toolResult({ toolCallId: 'call-1', toolName: 'bash', text: 'boom', isError: true })
+ )
+ );
+ expect(parts[0]).toMatchObject({ status: 'failed', output: 'boom' });
+ });
+
+ it('keeps a tool result whose call is outside the read window', () => {
+ const { parts } = projectParts(recordsOf(toolResult({ toolCallId: 'call-9', toolName: 'bash', text: 'ok' })));
+ expect(parts).toEqual([
+ expect.objectContaining({ kind: 'tool', toolCallId: 'call-9', status: 'completed', arguments: {} })
+ ]);
+ });
+
+ it('projects thinking, bash, custom and summary entries', () => {
+ const { parts } = projectParts(
+ recordsOf(
+ assistantText('answer', 2, { content: [{ type: 'thinking', thinking: 'hmm' }, { type: 'text', text: 'answer' }] }),
+ bash('ls', 'a.ts\n'),
+ custom({ customType: 'constructive.note', content: 'heads up' }),
+ compaction('summary so far'),
+ branchSummary('branched')
+ )
+ );
+
+ expect(parts.map((p) => p.kind)).toEqual(['thinking', 'text', 'bash', 'custom', 'summary', 'summary']);
+ expect(parts[2]).toMatchObject({ kind: 'bash', command: 'ls', output: 'a.ts\n', exitCode: 0 });
+ expect(parts[3]).toMatchObject({ customType: 'constructive.note', text: 'heads up', display: true });
+ expect(parts[4]).toMatchObject({ reason: 'compaction', summary: 'summary so far' });
+ expect(parts[5]).toMatchObject({ reason: 'branch', summary: 'branched' });
+ });
+
+ it('surfaces an entry type it does not understand instead of dropping it', () => {
+ const { parts } = projectParts(recordsOf(futureEntry()));
+ expect(parts).toEqual([
+ expect.objectContaining({ kind: 'unknown', entryType: 'quantum_thought' })
+ ]);
+ });
+});
+
+describe('projectUsage', () => {
+ it('totals tokens and cost per model and for the run', () => {
+ const totals = projectUsage(
+ recordsOf(
+ userMessage('hi'),
+ assistantText('a'),
+ assistantText('b', 3, { model: 'claude-haiku-4-5', usage: usage({ input: 10, output: 5, totalTokens: 15 }) })
+ )
+ );
+
+ expect(totals).toMatchObject({ input: 110, output: 25, totalTokens: 135, calls: 2 });
+ expect(totals.cost).toBeCloseTo(0.006, 6);
+ expect(Object.keys(totals.byModel)).toEqual([
+ modelKey('anthropic', 'claude-sonnet-4-5'),
+ modelKey('anthropic', 'claude-haiku-4-5')
+ ]);
+ expect(totals.byModel['anthropic/claude-haiku-4-5']).toMatchObject({ input: 10, calls: 1 });
+ });
+
+ it('counts nested tool usage and compaction, attributed to the requesting model', () => {
+ const totals = projectUsage(
+ recordsOf(
+ assistantToolCall({ id: 'call-1', name: 'subagent' }),
+ toolResult({ toolCallId: 'call-1', toolName: 'subagent', text: 'done', usage: usage({ input: 7, output: 3, totalTokens: 10 }) }),
+ compaction('summary', 7, { usage: usage({ input: 1, output: 1, totalTokens: 2 }) })
+ )
+ );
+
+ expect(totals.calls).toBe(3);
+ expect(totals.input).toBe(108);
+ expect(totals.byModel['anthropic/claude-sonnet-4-5'].calls).toBe(3);
+ });
+
+ it('derives a missing total from the parts the provider did report', () => {
+ const totals = projectUsage(
+ recordsOf(assistantText('a', 2, { usage: { input: 4, output: 6, cacheRead: 2, cacheWrite: 1 } }))
+ );
+ expect(totals.totalTokens).toBe(13);
+ expect(totals.cost).toBe(0);
+ });
+
+ it('is zero for a run with no model calls', () => {
+ expect(projectUsage(recordsOf(header(), userMessage('hi')))).toMatchObject({ totalTokens: 0, calls: 0 });
+ });
+});
+
+describe('projectToolState', () => {
+ it('tracks a tool through approval to completion', () => {
+ const records = recordsOf(
+ assistantToolCall({ id: 'call-1', name: 'deploy' }),
+ custom(approvalRequestMessage({ toolCallId: 'call-1', prompt: 'deploy to prod?' })),
+ custom({ ...approvalResolutionMessage({ toolCallId: 'call-1', decision: 'approved', actorId: 'user-1' }) }),
+ toolResult({ toolCallId: 'call-1', toolName: 'deploy', text: 'deployed' })
+ );
+
+ const midway = projectToolState(records.slice(0, 2));
+ expect(midway.tools['call-1'].status).toBe('awaiting-approval');
+ expect(midway.pendingApprovals).toEqual([
+ expect.objectContaining({ toolCallId: 'call-1', prompt: 'deploy to prod?' })
+ ]);
+
+ const approved = projectToolState(records.slice(0, 3));
+ expect(approved.tools['call-1'].status).toBe('running');
+ expect(approved.pendingApprovals).toEqual([]);
+ expect(approved.tools['call-1'].approval).toMatchObject({ decision: 'approved', actorId: 'user-1' });
+
+ const done = projectToolState(records);
+ expect(done.tools['call-1']).toMatchObject({ status: 'completed', output: 'deployed' });
+ });
+
+ it('marks a rejected tool as rejected', () => {
+ const state = projectToolState(
+ recordsOf(
+ assistantToolCall({ id: 'call-1', name: 'deploy' }),
+ custom(approvalRequestMessage({ toolCallId: 'call-1', prompt: 'ok?' })),
+ custom(approvalResolutionMessage({ toolCallId: 'call-1', decision: 'rejected', reason: 'not now' }))
+ )
+ );
+ expect(state.tools['call-1']).toMatchObject({ status: 'rejected' });
+ expect(state.tools['call-1'].approval).toMatchObject({ decision: 'rejected', reason: 'not now' });
+ });
+
+ it('throws on an approval message that cannot be attributed', () => {
+ expect(() =>
+ projectToolState(recordsOf(custom({ customType: APPROVAL_REQUEST_TYPE, content: 'ok?' })))
+ ).toThrow(/no toolCallId/);
+ expect(() =>
+ projectToolState(recordsOf(custom({ customType: APPROVAL_RESOLUTION_TYPE, content: 'yes', details: {} })))
+ ).toThrow(/no toolCallId/);
+ });
+
+ it('orders pending approvals oldest first', () => {
+ const state = projectToolState(
+ recordsOf(
+ assistantToolCall({ id: 'call-1', name: 'a' }),
+ assistantToolCall({ id: 'call-2', name: 'b' }),
+ custom(approvalRequestMessage({ toolCallId: 'call-2', prompt: 'b?' })),
+ custom(approvalRequestMessage({ toolCallId: 'call-1', prompt: 'a?' }))
+ )
+ );
+ expect(state.pendingApprovals.map((a) => a.toolCallId)).toEqual(['call-2', 'call-1']);
+ });
+});
+
+describe('projectSession', () => {
+ it('projects a resumable session file with the header first', () => {
+ const records = recordsOf(header(), userMessage('hi'), assistantText('hello'));
+ const { jsonl, entries, piSessionVersion } = projectSession(records);
+
+ expect(piSessionVersion).toBe(3);
+ expect(entries[0]).toMatchObject({ type: 'session', id: 'session-1' });
+ expect(jsonl.endsWith('\n')).toBe(true);
+ expect(parseSessionJsonl(jsonl)).toEqual(entries);
+ });
+
+ it('synthesises a header when the log has none', () => {
+ const { entries } = projectSession(recordsOf(userMessage('hi')), { sessionId: 's-9', cwd: '/w' });
+ expect(entries[0]).toMatchObject({ type: 'session', id: 's-9', cwd: '/w', version: 3 });
+ expect(entries).toHaveLength(2);
+ });
+
+ it('refuses to project an unloadable session', () => {
+ expect(() => projectSession(recordsOf(userMessage('hi'), header()))).toThrow(/requires it first/);
+
+ const mixed = recordsOf(userMessage('hi'), userMessage('there', 2));
+ mixed[1] = { ...mixed[1], piSessionVersion: 2 };
+ expect(() => projectSession(mixed)).toThrow(/mixes pi session versions/);
+ });
+
+ it('rejects a malformed session file rather than returning a partial one', () => {
+ expect(() => parseSessionJsonl('{"type":"session"}\nnot json\n')).toThrow(/line 2 is not valid JSON/);
+ });
+});
+
+describe('placement invariance', () => {
+ it('projects identically whether entries were appended in one batch or streamed', async () => {
+ const entries = [
+ header(),
+ userMessage('add a test'),
+ assistantToolCall({ id: 'call-1', name: 'write_file', arguments: { path: 'a.ts' } }),
+ toolResult({ toolCallId: 'call-1', toolName: 'write_file', text: 'wrote a.ts' }),
+ assistantText('done', 5)
+ ];
+
+ const cloud = new MemoryRunLogStore();
+ await cloud.append('run-1', entries);
+
+ const local = new MemoryRunLogStore();
+ for (const entry of entries) await local.append('run-1', [entry]);
+
+ const cloudRecords = await readAll(cloud, 'run-1');
+ const localRecords = await readAll(local, 'run-1');
+
+ expect(localRecords.map((r) => r.seq)).toEqual(cloudRecords.map((r) => r.seq));
+ expect(projectParts(localRecords)).toEqual(projectParts(cloudRecords));
+ expect(projectUsage(localRecords)).toEqual(projectUsage(cloudRecords));
+ expect(projectToolState(localRecords)).toEqual(projectToolState(cloudRecords));
+ expect(projectSession(localRecords).jsonl).toEqual(projectSession(cloudRecords).jsonl);
+ });
+});
diff --git a/agentic/run-log/__tests__/record.test.ts b/agentic/run-log/__tests__/record.test.ts
new file mode 100644
index 000000000..3085e5577
--- /dev/null
+++ b/agentic/run-log/__tests__/record.test.ts
@@ -0,0 +1,102 @@
+import {
+ assertOrdered,
+ assertPiSessionEntry,
+ assertRunEventRecord,
+ idempotencyKey,
+ RUN_LOG_WRAPPER_VERSION,
+ SUPPORTED_PI_SESSION_VERSION,
+ wrapEntry
+} from '../src';
+import { assistantText, header, resetIds, userMessage } from './fixtures';
+
+beforeEach(resetIds);
+
+describe('wrapEntry', () => {
+ it('wraps a pi entry in the four platform fields, leaving the entry untouched', () => {
+ const entry = userMessage('hello');
+ const record = wrapEntry({ runId: 'run-1', seq: 1, entry, recordedAt: '2026-01-01T00:00:00.000Z' });
+
+ expect(record).toEqual({
+ runId: 'run-1',
+ seq: 1,
+ recordedAt: '2026-01-01T00:00:00.000Z',
+ piSessionVersion: SUPPORTED_PI_SESSION_VERSION,
+ entry
+ });
+ // Same object, not a copy: the entry is stored verbatim.
+ expect(record.entry).toBe(entry);
+ });
+
+ it('defaults recordedAt and records the pi session version', () => {
+ const record = wrapEntry({ runId: 'run-1', seq: 1, entry: userMessage('hi') });
+ expect(Date.parse(record.recordedAt)).not.toBeNaN();
+ expect(record.piSessionVersion).toBe(3);
+ expect(RUN_LOG_WRAPPER_VERSION).toBe(1);
+ });
+
+ it('rejects a seq that would break ordering', () => {
+ expect(() => wrapEntry({ runId: 'run-1', seq: 0, entry: userMessage('hi') })).toThrow(/positive integer/);
+ expect(() => wrapEntry({ runId: '', seq: 1, entry: userMessage('hi') })).toThrow(/runId/);
+ });
+});
+
+describe('assertPiSessionEntry', () => {
+ it('accepts a session header without tree fields', () => {
+ expect(assertPiSessionEntry(header()).type).toBe('session');
+ });
+
+ it('accepts an entry type it has never seen', () => {
+ const entry = {
+ type: 'quantum_thought',
+ id: 'abc',
+ parentId: null as string | null,
+ timestamp: '2026-01-01T00:00:00.000Z'
+ };
+ expect(assertPiSessionEntry(entry)).toBe(entry);
+ });
+
+ it('throws rather than yielding an unreadable entry', () => {
+ expect(() => assertPiSessionEntry(null)).toThrow(/must be an object/);
+ expect(() => assertPiSessionEntry({})).toThrow(/`type`/);
+ expect(() => assertPiSessionEntry({ type: 'message' })).toThrow(/`id`/);
+ expect(() => assertPiSessionEntry({ type: 'message', id: 'a' })).toThrow(/timestamp/);
+ expect(() => assertPiSessionEntry({ type: 'message', id: 'a', timestamp: 'x' })).toThrow(/parentId/);
+ });
+});
+
+describe('assertRunEventRecord', () => {
+ const valid = wrapEntry({ runId: 'run-1', seq: 1, entry: assistantText('hi') });
+
+ it('round-trips through JSON', () => {
+ expect(assertRunEventRecord(JSON.parse(JSON.stringify(valid)))).toEqual(valid);
+ });
+
+ it.each([
+ [{ ...valid, runId: '' }, /non-empty runId/],
+ [{ ...valid, seq: 0 }, /invalid seq/],
+ [{ ...valid, recordedAt: 5 }, /recordedAt/],
+ [{ ...valid, piSessionVersion: '3' }, /piSessionVersion/],
+ [{ ...valid, entry: 'nope' }, /must be an object/]
+ ])('throws on a corrupt record (%#)', (record, message) => {
+ expect(() => assertRunEventRecord(record)).toThrow(message);
+ });
+});
+
+describe('idempotencyKey', () => {
+ it('is stable per entry so a retried append is recognised', () => {
+ const entry = userMessage('hello');
+ expect(idempotencyKey('run-1', entry)).toBe(idempotencyKey('run-1', { ...entry }));
+ expect(idempotencyKey('run-1', entry)).not.toBe(idempotencyKey('run-2', entry));
+ });
+});
+
+describe('assertOrdered', () => {
+ it('rejects mixed runs and out-of-order sequences', () => {
+ const a = wrapEntry({ runId: 'run-1', seq: 2, entry: userMessage('a') });
+ const b = wrapEntry({ runId: 'run-1', seq: 1, entry: userMessage('b') });
+ const c = wrapEntry({ runId: 'run-2', seq: 3, entry: userMessage('c') });
+ expect(() => assertOrdered([a, b])).toThrow(/out of order/);
+ expect(() => assertOrdered([a, c])).toThrow(/mix runs/);
+ expect(() => assertOrdered([b, a])).not.toThrow();
+ });
+});
diff --git a/agentic/run-log/__tests__/store.test.ts b/agentic/run-log/__tests__/store.test.ts
new file mode 100644
index 000000000..e288ebe37
--- /dev/null
+++ b/agentic/run-log/__tests__/store.test.ts
@@ -0,0 +1,135 @@
+import { cursorAfter, follow, MemoryRunLogStore, readAll, START } from '../src';
+import { assistantText, header, resetIds, userMessage } from './fixtures';
+
+let store: MemoryRunLogStore;
+
+beforeEach(() => {
+ resetIds();
+ store = new MemoryRunLogStore();
+});
+
+describe('MemoryRunLogStore', () => {
+ it('assigns gapless sequences in append order', async () => {
+ await store.append('run-1', [header(), userMessage('hi')]);
+ await store.append('run-1', [assistantText('hello')]);
+ expect(store.snapshot('run-1').map((r) => r.seq)).toEqual([1, 2, 3]);
+ });
+
+ it('keeps runs independent', async () => {
+ await store.append('run-1', [userMessage('a')]);
+ await store.append('run-2', [userMessage('b')]);
+ expect(store.snapshot('run-1')).toHaveLength(1);
+ expect(store.snapshot('run-2')[0].seq).toBe(1);
+ expect(store.runIds()).toEqual(['run-1', 'run-2']);
+ });
+
+ it('skips entries it already holds, so an append can be retried', async () => {
+ const entries = [header(), userMessage('hi'), assistantText('hello')];
+ const first = await store.append('run-1', entries);
+ const retry = await store.append('run-1', entries);
+ const extended = await store.append('run-1', [...entries, userMessage('again', 5)]);
+
+ expect(first).toHaveLength(3);
+ expect(retry).toHaveLength(0);
+ expect(extended).toHaveLength(1);
+ expect(extended[0].seq).toBe(4);
+ });
+
+ it('reads after a cursor', async () => {
+ await store.append('run-1', [userMessage('a'), userMessage('b', 2), userMessage('c', 3)]);
+
+ const first = await store.read('run-1', START, 2);
+ expect(first.records.map((r) => r.seq)).toEqual([1, 2]);
+ expect(first.cursor).toEqual({ afterSeq: 2 });
+
+ const next = await store.read('run-1', first.cursor);
+ expect(next.records.map((r) => r.seq)).toEqual([3]);
+
+ const end = await store.read('run-1', next.cursor);
+ expect(end.records).toEqual([]);
+ // An empty page must not rewind the cursor.
+ expect(end.cursor).toEqual({ afterSeq: 3 });
+ });
+
+ it('reports an empty page for an unknown run rather than throwing', async () => {
+ expect((await store.read('nope')).records).toEqual([]);
+ });
+});
+
+describe('cursorAfter', () => {
+ it('keeps the previous position when nothing was read', () => {
+ expect(cursorAfter([], { afterSeq: 7 })).toEqual({ afterSeq: 7 });
+ });
+});
+
+describe('readAll', () => {
+ it('pages to the end of the run', async () => {
+ const entries = Array.from({ length: 25 }, (_, i) => userMessage(`m${String(i)}`, i + 1));
+ await store.append('run-1', entries);
+ const records = await readAll(store, 'run-1', START, 10);
+ expect(records.map((r) => r.seq)).toEqual(entries.map((_, i) => i + 1));
+ });
+});
+
+describe('follow', () => {
+ const immediateSleep = async (): Promise => {};
+
+ it('yields batches as they arrive and stops on a terminal batch', async () => {
+ await store.append('run-1', [userMessage('a')]);
+
+ const batches: number[][] = [];
+ const iteration = (async () => {
+ for await (const batch of follow(store, 'run-1', {
+ sleep: immediateSleep,
+ isTerminal: (records) => records.some((r) => r.entry.type === 'run_finished')
+ })) {
+ batches.push(batch.map((r) => r.seq));
+ if (batches.length === 1) {
+ await store.append('run-1', [assistantText('b', 2)]);
+ } else if (batches.length === 2) {
+ await store.append('run-1', [
+ { type: 'run_finished', id: 'ffffffff', parentId: null, timestamp: '2026-01-01T00:00:09.000Z' }
+ ]);
+ }
+ }
+ })();
+
+ await iteration;
+ expect(batches).toEqual([[1], [2], [3]]);
+ });
+
+ it('stops when the caller aborts', async () => {
+ const controller = new AbortController();
+ const batches: number[][] = [];
+ const iteration = (async () => {
+ for await (const batch of follow(store, 'run-1', { sleep: immediateSleep, signal: controller.signal })) {
+ batches.push(batch.map((r) => r.seq));
+ controller.abort();
+ }
+ })();
+
+ await store.append('run-1', [userMessage('a')]);
+ await iteration;
+ expect(batches.length).toBeLessThanOrEqual(1);
+ });
+
+ it('races a push wakeup against the poll delay', async () => {
+ let wakeups = 0;
+ const iteration = (async () => {
+ for await (const batch of follow(store, 'run-1', {
+ sleep: immediateSleep,
+ waitForChange: async () => {
+ wakeups += 1;
+ if (wakeups === 1) await store.append('run-1', [userMessage('a')]);
+ },
+ isTerminal: () => true
+ })) {
+ expect(batch).toHaveLength(1);
+ }
+ })();
+
+ await iteration;
+ expect(wakeups).toBeGreaterThan(0);
+ expect(store.snapshot('run-1')).toHaveLength(1);
+ });
+});
diff --git a/agentic/run-log/jest.config.js b/agentic/run-log/jest.config.js
new file mode 100644
index 000000000..8a26efd6d
--- /dev/null
+++ b/agentic/run-log/jest.config.js
@@ -0,0 +1,21 @@
+/** @type {import('ts-jest').JestConfigWithTsJest} */
+module.exports = {
+ preset: 'ts-jest',
+ testEnvironment: 'node',
+ transform: {
+ '^.+\\.tsx?$': [
+ 'ts-jest',
+ {
+ babelConfig: false,
+ tsconfig: 'tsconfig.json',
+ },
+ ],
+ },
+ transformIgnorePatterns: [`/node_modules/*`],
+ testRegex: '(/__tests__/.*\\.(test|spec))\\.(jsx?|tsx?)$',
+ moduleFileExtensions: ['ts', 'tsx', 'js', 'jsx', 'json', 'node'],
+ modulePathIgnorePatterns: ['dist/*'],
+ moduleNameMapper: {
+ '^(\\.{1,2}/.*)\\.js$': '$1',
+ },
+};
diff --git a/agentic/run-log/package.json b/agentic/run-log/package.json
new file mode 100644
index 000000000..1a14d0cbb
--- /dev/null
+++ b/agentic/run-log/package.json
@@ -0,0 +1,38 @@
+{
+ "name": "@agentic-kit/run-log",
+ "version": "0.1.0",
+ "author": "Constructive ",
+ "description": "The append-only agent run log \u2014 pi session entries stored verbatim under a run/seq wrapper, with projections for transcript, usage, tool state and resumable sessions",
+ "main": "index.js",
+ "module": "esm/index.js",
+ "types": "index.d.ts",
+ "homepage": "https://github.com/constructive-io/constructive",
+ "license": "SEE LICENSE IN LICENSE",
+ "publishConfig": {
+ "access": "public",
+ "directory": "dist"
+ },
+ "repository": {
+ "type": "git",
+ "url": "https://github.com/constructive-io/constructive"
+ },
+ "bugs": {
+ "url": "https://github.com/constructive-io/constructive/issues"
+ },
+ "scripts": {
+ "clean": "makage clean",
+ "prepack": "npm run build",
+ "build": "makage build",
+ "build:dev": "makage build --dev",
+ "lint": "eslint . --fix",
+ "test": "jest",
+ "test:watch": "jest --watch"
+ },
+ "keywords": [
+ "agentic-kit",
+ "pi",
+ "coding-agent",
+ "run-log",
+ "constructive"
+ ]
+}
diff --git a/agentic/run-log/src/file-store.ts b/agentic/run-log/src/file-store.ts
new file mode 100644
index 000000000..4a01bf8de
--- /dev/null
+++ b/agentic/run-log/src/file-store.ts
@@ -0,0 +1,123 @@
+/**
+ * `@agentic-kit/run-log/file-store` — the node-only JSONL store.
+ *
+ * A separate entry point because the package's main entry is imported by
+ * browsers, Electron renderers and Next client components; a `node:fs` import
+ * there breaks those bundles. Same split, same reason, as `12factor-env/dotenv`.
+ *
+ * The file holds one wrapped record per line, so a run is recoverable with
+ * `tail -f` and a partially-written last line is detectable rather than silently
+ * dropped. This is the store a local run uses before (or without) a database,
+ * and the one tests use when they want the log to survive a process restart.
+ */
+
+import { appendFileSync, existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs';
+import { dirname } from 'node:path';
+
+import type { PiSessionEntry } from './pi-entry';
+import { projectSession, type SessionProjectionOptions } from './projectors/session';
+import {
+ assertRunEventRecord,
+ idempotencyKey,
+ type RunEventRecord,
+ SUPPORTED_PI_SESSION_VERSION,
+ wrapEntry
+} from './record';
+import {
+ type AppendOptions,
+ cursorAfter,
+ type RunLogCursor,
+ type RunLogPage,
+ type RunLogStore,
+ START
+} from './store';
+
+export interface FileRunLogStoreOptions {
+ /** Absolute path of the log file. Parent directories are created. */
+ path: string;
+}
+
+export class FileRunLogStore implements RunLogStore {
+ private readonly path: string;
+
+ constructor(options: FileRunLogStoreOptions) {
+ this.path = options.path;
+ }
+
+ private load(): RunEventRecord[] {
+ if (!existsSync(this.path)) return [];
+ const contents = readFileSync(this.path, 'utf8');
+ const records: RunEventRecord[] = [];
+ const lines = contents.split('\n');
+ for (let i = 0; i < lines.length; i += 1) {
+ const line = lines[i].trim();
+ if (line.length === 0) continue;
+ let parsed: unknown;
+ try {
+ parsed = JSON.parse(line);
+ } catch (error) {
+ throw new Error(
+ `run log ${this.path} line ${String(i + 1)} is not valid JSON: ${
+ error instanceof Error ? error.message : String(error)
+ }`
+ );
+ }
+ records.push(assertRunEventRecord(parsed));
+ }
+ return records;
+ }
+
+ async append(
+ runId: string,
+ entries: readonly PiSessionEntry[],
+ options: AppendOptions = {}
+ ): Promise {
+ const existing = this.load().filter((record) => record.runId === runId);
+ const seen = new Set(existing.map((record) => idempotencyKey(runId, record.entry)));
+ const written: RunEventRecord[] = [];
+
+ for (const entry of entries) {
+ const key = idempotencyKey(runId, entry);
+ if (seen.has(key)) continue;
+ written.push(
+ wrapEntry({
+ runId,
+ seq: existing.length + written.length + 1,
+ entry,
+ ...(options.recordedAt ? { recordedAt: options.recordedAt } : {}),
+ piSessionVersion: options.piSessionVersion ?? SUPPORTED_PI_SESSION_VERSION
+ })
+ );
+ seen.add(key);
+ }
+
+ if (written.length === 0) return written;
+
+ mkdirSync(dirname(this.path), { recursive: true });
+ if (!existsSync(this.path)) writeFileSync(this.path, '', { mode: 0o600 });
+ appendFileSync(this.path, written.map((record) => JSON.stringify(record)).join('\n') + '\n');
+ return written;
+ }
+
+ async read(runId: string, cursor: RunLogCursor = START, limit?: number): Promise {
+ const after = this.load().filter((record) => record.runId === runId && record.seq > cursor.afterSeq);
+ const records = typeof limit === 'number' ? after.slice(0, limit) : after;
+ return { records, cursor: cursorAfter(records, cursor) };
+ }
+}
+
+/**
+ * Write the run's pi session file so `SessionManager.open` can resume it. This
+ * is the cloud→local (and local→local restart) resume path: project, write,
+ * hand the path to pi.
+ */
+export function writeSessionFile(
+ path: string,
+ records: readonly RunEventRecord[],
+ options: SessionProjectionOptions = {}
+): string {
+ const projection = projectSession(records, options);
+ mkdirSync(dirname(path), { recursive: true });
+ writeFileSync(path, projection.jsonl, { mode: 0o600 });
+ return path;
+}
diff --git a/agentic/run-log/src/follow.ts b/agentic/run-log/src/follow.ts
new file mode 100644
index 000000000..3dbce6082
--- /dev/null
+++ b/agentic/run-log/src/follow.ts
@@ -0,0 +1,97 @@
+/**
+ * Following a run log.
+ *
+ * Every surface — a desktop chat pane, a web execution view, a CLI tail, the
+ * resume path — is the same reader: hold a cursor, ask for what came after it.
+ * That is what makes local and cloud runs indistinguishable to a UI, so this is
+ * the only reader loop in the system.
+ *
+ * A push wakeup (LISTEN/NOTIFY, IPC, websocket) is an optimisation, not a
+ * requirement: `waitForChange` short-circuits the delay when it resolves, and
+ * without one the loop degrades to polling at `pollIntervalMs`.
+ */
+
+import type { RunEventRecord } from './record';
+import { cursorAfter, type RunLogCursor, type RunLogReadStore, START } from './store';
+
+export interface FollowOptions {
+ cursor?: RunLogCursor;
+ /** Polling delay when no wakeup arrives. */
+ pollIntervalMs?: number;
+ /** Records per read. */
+ limit?: number;
+ /** Resolves when the run may have new records; races the poll delay. */
+ waitForChange?: (signal?: AbortSignal) => Promise;
+ /** Stop following once this returns true for the batch just yielded. */
+ isTerminal?: (records: readonly RunEventRecord[]) => boolean;
+ signal?: AbortSignal;
+ /** Injectable for tests; defaults to `setTimeout`. */
+ sleep?: (ms: number, signal?: AbortSignal) => Promise;
+}
+
+const defaultSleep = (ms: number, signal?: AbortSignal): Promise =>
+ new Promise((resolve) => {
+ const timer = setTimeout(resolve, ms);
+ signal?.addEventListener(
+ 'abort',
+ () => {
+ clearTimeout(timer);
+ resolve();
+ },
+ { once: true }
+ );
+ });
+
+/**
+ * Yield every batch of new records until the run reaches a terminal state or
+ * the caller aborts. Batches are yielded as read, so a caller can project
+ * incrementally rather than re-projecting the whole run per frame.
+ */
+export async function* follow(
+ store: RunLogReadStore,
+ runId: string,
+ options: FollowOptions = {}
+): AsyncGenerator {
+ const pollIntervalMs = options.pollIntervalMs ?? 500;
+ const sleep = options.sleep ?? defaultSleep;
+ let cursor = options.cursor ?? START;
+
+ while (!options.signal?.aborted) {
+ const page = await store.read(runId, cursor, options.limit);
+ if (page.records.length > 0) {
+ cursor = cursorAfter(page.records, cursor);
+ yield page.records;
+ if (options.isTerminal?.(page.records)) return;
+ // Drain before waiting: a burst of tool output should not be paced by the
+ // poll interval.
+ continue;
+ }
+
+ if (options.waitForChange) {
+ await Promise.race([
+ options.waitForChange(options.signal),
+ sleep(pollIntervalMs, options.signal)
+ ]);
+ } else {
+ await sleep(pollIntervalMs, options.signal);
+ }
+ }
+}
+
+/** Read a run to its current end in one pass. */
+export async function readAll(
+ store: RunLogReadStore,
+ runId: string,
+ cursor: RunLogCursor = START,
+ pageLimit = 500
+): Promise {
+ const records: RunEventRecord[] = [];
+ let position = cursor;
+ for (;;) {
+ const page = await store.read(runId, position, pageLimit);
+ if (page.records.length === 0) return records;
+ records.push(...page.records);
+ position = cursorAfter(page.records, position);
+ if (page.records.length < pageLimit) return records;
+ }
+}
diff --git a/agentic/run-log/src/index.ts b/agentic/run-log/src/index.ts
new file mode 100644
index 000000000..cb51aa22d
--- /dev/null
+++ b/agentic/run-log/src/index.ts
@@ -0,0 +1,106 @@
+/**
+ * `@agentic-kit/run-log` — the append-only run log: one ordered record of what
+ * an agent run did, wherever it ran.
+ *
+ * Browser-safe on purpose: renderers import the projectors, so nothing here may
+ * reach for a node builtin. The filesystem store lives behind
+ * `@agentic-kit/run-log/file-store`.
+ */
+
+export {
+ follow,
+ type FollowOptions,
+ readAll
+} from './follow';
+export {
+ assertPiSessionEntry,
+ contentText,
+ isAssistantMessage,
+ isPiBranchSummaryEntry,
+ isPiCompactionEntry,
+ isPiMessageEntry,
+ isPiSessionHeader,
+ isToolResultMessage,
+ type PiAssistantMessage,
+ type PiBashExecutionMessage,
+ type PiBranchSummaryEntry,
+ type PiCompactionEntry,
+ type PiContent,
+ type PiCustomMessage,
+ type PiEntryBase,
+ type PiImageContent,
+ type PiMessage,
+ type PiMessageEntry,
+ type PiOtherEntry,
+ type PiSessionEntry,
+ type PiSessionHeader,
+ type PiSummaryMessage,
+ type PiTextContent,
+ type PiThinkingContent,
+ type PiToolCallContent,
+ type PiToolResultMessage,
+ type PiUsage,
+ type PiUsageCost,
+ type PiUserMessage,
+ toolCalls
+} from './pi-entry';
+export {
+ type BashPart,
+ type Conversation,
+ type ConversationPart,
+ type CustomPart,
+ projectParts,
+ type SummaryPart,
+ type TextPart,
+ type ThinkingPart,
+ type ToolPart,
+ type ToolStatus,
+ type UnknownPart
+} from './projectors/parts';
+export {
+ parseSessionJsonl,
+ projectSession,
+ type SessionProjection,
+ type SessionProjectionOptions
+} from './projectors/session';
+export {
+ APPROVAL_REQUEST_TYPE,
+ APPROVAL_RESOLUTION_TYPE,
+ type ApprovalRequestInput,
+ approvalRequestMessage,
+ type ApprovalResolutionInput,
+ approvalResolutionMessage,
+ type ApprovalState,
+ projectToolState,
+ type ToolCallState,
+ type ToolCallStatus,
+ type ToolStateProjection
+} from './projectors/tool-state';
+export {
+ modelKey,
+ type ModelUsage,
+ projectUsage,
+ type RunUsage,
+ type UsageTotals
+} from './projectors/usage';
+export {
+ assertOrdered,
+ assertRunEventRecord,
+ idempotencyKey,
+ RUN_LOG_WRAPPER_VERSION,
+ type RunEventRecord,
+ SUPPORTED_PI_SESSION_VERSION,
+ wrapEntry,
+ type WrapEntryOptions
+} from './record';
+export {
+ type AppendOptions,
+ cursorAfter,
+ MemoryRunLogStore,
+ type RunLogAppendStore,
+ type RunLogCursor,
+ type RunLogPage,
+ type RunLogReadStore,
+ type RunLogStore,
+ START
+} from './store';
diff --git a/agentic/run-log/src/pi-entry.ts b/agentic/run-log/src/pi-entry.ts
new file mode 100644
index 000000000..8e3f90086
--- /dev/null
+++ b/agentic/run-log/src/pi-entry.ts
@@ -0,0 +1,238 @@
+/**
+ * The structural subset of pi's session entries this package reads.
+ *
+ * Deliberately structural, not imported from pi: a run log stores pi entries
+ * *verbatim*, so the types here describe what the projectors read rather than
+ * re-declaring pi's format. Every interface keeps an index signature so an
+ * entry produced by a newer pi still parses, and unknown `type` values are
+ * carried through untouched instead of being dropped.
+ *
+ * Reference: `@earendil-works/pi-coding-agent` `docs/session-format.md`
+ * (session file version 3).
+ */
+
+export interface PiUsageCost {
+ input?: number;
+ output?: number;
+ cacheRead?: number;
+ cacheWrite?: number;
+ total?: number;
+}
+
+export interface PiUsage {
+ input?: number;
+ output?: number;
+ cacheRead?: number;
+ cacheWrite?: number;
+ totalTokens?: number;
+ cost?: PiUsageCost;
+ [key: string]: unknown;
+}
+
+export interface PiTextContent {
+ type: 'text';
+ text: string;
+}
+
+export interface PiThinkingContent {
+ type: 'thinking';
+ thinking: string;
+}
+
+export interface PiImageContent {
+ type: 'image';
+ data: string;
+ mimeType: string;
+}
+
+export interface PiToolCallContent {
+ type: 'toolCall';
+ id: string;
+ name: string;
+ arguments?: Record;
+}
+
+export type PiContent = PiTextContent | PiThinkingContent | PiImageContent | PiToolCallContent;
+
+export interface PiUserMessage {
+ role: 'user';
+ content: string | PiContent[];
+ timestamp?: number;
+ [key: string]: unknown;
+}
+
+export interface PiAssistantMessage {
+ role: 'assistant';
+ content: PiContent[];
+ api?: string;
+ provider?: string;
+ model?: string;
+ usage?: PiUsage;
+ stopReason?: 'stop' | 'length' | 'toolUse' | 'error' | 'aborted';
+ errorMessage?: string;
+ timestamp?: number;
+ [key: string]: unknown;
+}
+
+export interface PiToolResultMessage {
+ role: 'toolResult';
+ toolCallId: string;
+ toolName: string;
+ content: PiContent[];
+ details?: unknown;
+ usage?: PiUsage;
+ isError?: boolean;
+ timestamp?: number;
+ [key: string]: unknown;
+}
+
+export interface PiBashExecutionMessage {
+ role: 'bashExecution';
+ command: string;
+ output: string;
+ exitCode?: number;
+ cancelled?: boolean;
+ truncated?: boolean;
+ timestamp?: number;
+ [key: string]: unknown;
+}
+
+export interface PiCustomMessage {
+ role: 'custom';
+ customType: string;
+ content: string | PiContent[];
+ display?: boolean;
+ details?: unknown;
+ timestamp?: number;
+ [key: string]: unknown;
+}
+
+export interface PiSummaryMessage {
+ role: 'branchSummary' | 'compactionSummary';
+ summary: string;
+ timestamp?: number;
+ [key: string]: unknown;
+}
+
+export type PiMessage =
+ | PiUserMessage
+ | PiAssistantMessage
+ | PiToolResultMessage
+ | PiBashExecutionMessage
+ | PiCustomMessage
+ | PiSummaryMessage;
+
+export interface PiSessionHeader {
+ type: 'session';
+ version: number;
+ id: string;
+ timestamp: string;
+ cwd?: string;
+ parentSession?: string;
+ [key: string]: unknown;
+}
+
+export interface PiEntryBase {
+ id: string;
+ parentId: string | null;
+ timestamp: string;
+}
+
+export interface PiMessageEntry extends PiEntryBase {
+ type: 'message';
+ message: PiMessage;
+ [key: string]: unknown;
+}
+
+export interface PiCompactionEntry extends PiEntryBase {
+ type: 'compaction';
+ summary: string;
+ tokensBefore?: number;
+ usage?: PiUsage;
+ [key: string]: unknown;
+}
+
+export interface PiBranchSummaryEntry extends PiEntryBase {
+ type: 'branch_summary';
+ summary: string;
+ fromId: string;
+ usage?: PiUsage;
+ [key: string]: unknown;
+}
+
+export interface PiOtherEntry extends PiEntryBase {
+ type: string;
+ [key: string]: unknown;
+}
+
+export type PiSessionEntry =
+ | PiSessionHeader
+ | PiMessageEntry
+ | PiCompactionEntry
+ | PiBranchSummaryEntry
+ | PiOtherEntry;
+
+const isRecord = (value: unknown): value is Record =>
+ typeof value === 'object' && value !== null && !Array.isArray(value);
+
+export const isPiSessionHeader = (entry: PiSessionEntry): entry is PiSessionHeader =>
+ entry.type === 'session';
+
+export const isPiMessageEntry = (entry: PiSessionEntry): entry is PiMessageEntry =>
+ entry.type === 'message' && isRecord((entry as PiMessageEntry).message);
+
+export const isPiCompactionEntry = (entry: PiSessionEntry): entry is PiCompactionEntry =>
+ entry.type === 'compaction';
+
+export const isPiBranchSummaryEntry = (entry: PiSessionEntry): entry is PiBranchSummaryEntry =>
+ entry.type === 'branch_summary';
+
+export const isAssistantMessage = (message: PiMessage): message is PiAssistantMessage =>
+ message.role === 'assistant';
+
+export const isToolResultMessage = (message: PiMessage): message is PiToolResultMessage =>
+ message.role === 'toolResult';
+
+/**
+ * Narrow an untrusted value (a database JSONB column, an HTTP body) to a pi
+ * entry. Throws rather than returning null: a log row that cannot be read is a
+ * corrupted log, never an empty one.
+ */
+export function assertPiSessionEntry(value: unknown): PiSessionEntry {
+ if (!isRecord(value)) {
+ throw new TypeError(`pi session entry must be an object, received ${typeof value}`);
+ }
+ if (typeof value.type !== 'string' || value.type.length === 0) {
+ throw new TypeError('pi session entry must carry a non-empty string `type`');
+ }
+ if (value.type !== 'session') {
+ if (typeof value.id !== 'string' || value.id.length === 0) {
+ throw new TypeError(`pi ${value.type} entry must carry a non-empty string \`id\``);
+ }
+ if (typeof value.timestamp !== 'string') {
+ throw new TypeError(`pi ${value.type} entry must carry an ISO string \`timestamp\``);
+ }
+ if (!('parentId' in value)) {
+ throw new TypeError(`pi ${value.type} entry must carry \`parentId\` (null for the first entry)`);
+ }
+ }
+ return value as PiSessionEntry;
+}
+
+/** The text of a message's content, whether it is a string or a block array. */
+export function contentText(content: string | PiContent[] | undefined): string {
+ if (typeof content === 'string') return content;
+ if (!Array.isArray(content)) return '';
+ return content
+ .filter((block): block is PiTextContent => isRecord(block) && block.type === 'text')
+ .map((block) => block.text)
+ .join('');
+}
+
+/** Tool calls requested by an assistant message, in order. */
+export function toolCalls(message: PiAssistantMessage): PiToolCallContent[] {
+ if (!Array.isArray(message.content)) return [];
+ return message.content.filter(
+ (block): block is PiToolCallContent => isRecord(block) && block.type === 'toolCall'
+ );
+}
diff --git a/agentic/run-log/src/projectors/parts.ts b/agentic/run-log/src/projectors/parts.ts
new file mode 100644
index 000000000..64593054f
--- /dev/null
+++ b/agentic/run-log/src/projectors/parts.ts
@@ -0,0 +1,234 @@
+/**
+ * Renderable projection: run log records → an ordered list of parts a UI draws.
+ *
+ * This is the projection that replaces per-host transcript encodings. A tool
+ * call and its later result collapse into one part, so a renderer never has to
+ * correlate two messages itself, and an unknown entry type becomes an
+ * `unknown` part rather than disappearing — a log written by a newer pi still
+ * renders, minus the detail this version understands.
+ */
+
+import {
+ contentText,
+ isAssistantMessage,
+ isPiBranchSummaryEntry,
+ isPiCompactionEntry,
+ isPiMessageEntry,
+ isPiSessionHeader,
+ isToolResultMessage,
+ type PiSessionEntry,
+ toolCalls
+} from '../pi-entry';
+import type { RunEventRecord } from '../record';
+
+export type ToolStatus = 'requested' | 'completed' | 'failed';
+
+export interface PartBase {
+ /** Sequence of the record that introduced the part — a stable React key. */
+ seq: number;
+ entryId?: string;
+}
+
+export interface TextPart extends PartBase {
+ kind: 'text';
+ role: 'user' | 'assistant';
+ text: string;
+ model?: string;
+ provider?: string;
+}
+
+export interface ThinkingPart extends PartBase {
+ kind: 'thinking';
+ text: string;
+}
+
+export interface ToolPart extends PartBase {
+ kind: 'tool';
+ toolCallId: string;
+ name: string;
+ arguments: Record;
+ status: ToolStatus;
+ /** Text of the tool result, once one has been logged. */
+ output?: string;
+ details?: unknown;
+ /** Sequence of the record that settled the call. */
+ settledSeq?: number;
+}
+
+export interface BashPart extends PartBase {
+ kind: 'bash';
+ command: string;
+ output: string;
+ exitCode?: number;
+}
+
+export interface CustomPart extends PartBase {
+ kind: 'custom';
+ customType: string;
+ text: string;
+ display: boolean;
+ details?: unknown;
+}
+
+export interface SummaryPart extends PartBase {
+ kind: 'summary';
+ reason: 'compaction' | 'branch';
+ summary: string;
+}
+
+export interface UnknownPart extends PartBase {
+ kind: 'unknown';
+ entryType: string;
+ entry: PiSessionEntry;
+}
+
+export type ConversationPart =
+ | TextPart
+ | ThinkingPart
+ | ToolPart
+ | BashPart
+ | CustomPart
+ | SummaryPart
+ | UnknownPart;
+
+export interface Conversation {
+ parts: ConversationPart[];
+ /** From the session header, when the log carries one. */
+ sessionId?: string;
+ cwd?: string;
+}
+
+/**
+ * Project records into a conversation. Pure and total: the same records always
+ * produce the same parts, whichever host wrote them.
+ */
+export function projectParts(records: readonly RunEventRecord[]): Conversation {
+ const parts: ConversationPart[] = [];
+ const toolsByCallId = new Map();
+ let sessionId: string | undefined;
+ let cwd: string | undefined;
+
+ for (const record of records) {
+ const { entry, seq } = record;
+
+ if (isPiSessionHeader(entry)) {
+ sessionId = entry.id;
+ if (typeof entry.cwd === 'string') cwd = entry.cwd;
+ continue;
+ }
+
+ if (isPiCompactionEntry(entry)) {
+ parts.push({ kind: 'summary', reason: 'compaction', summary: entry.summary, seq, entryId: entry.id });
+ continue;
+ }
+
+ if (isPiBranchSummaryEntry(entry)) {
+ parts.push({ kind: 'summary', reason: 'branch', summary: entry.summary, seq, entryId: entry.id });
+ continue;
+ }
+
+ if (!isPiMessageEntry(entry)) {
+ parts.push({ kind: 'unknown', entryType: entry.type, entry, seq, entryId: (entry as { id?: string }).id });
+ continue;
+ }
+
+ const message = entry.message;
+ const base = { seq, entryId: entry.id };
+
+ if (message.role === 'user') {
+ parts.push({ kind: 'text', role: 'user', text: contentText(message.content), ...base });
+ continue;
+ }
+
+ if (isAssistantMessage(message)) {
+ for (const block of Array.isArray(message.content) ? message.content : []) {
+ if (block.type === 'text' && block.text.length > 0) {
+ parts.push({
+ kind: 'text',
+ role: 'assistant',
+ text: block.text,
+ ...(message.model ? { model: message.model } : {}),
+ ...(message.provider ? { provider: message.provider } : {}),
+ ...base
+ });
+ } else if (block.type === 'thinking') {
+ parts.push({ kind: 'thinking', text: block.thinking, ...base });
+ }
+ }
+ for (const call of toolCalls(message)) {
+ const part: ToolPart = {
+ kind: 'tool',
+ toolCallId: call.id,
+ name: call.name,
+ arguments: call.arguments ?? {},
+ status: 'requested',
+ ...base
+ };
+ toolsByCallId.set(call.id, part);
+ parts.push(part);
+ }
+ continue;
+ }
+
+ if (isToolResultMessage(message)) {
+ const existing = toolsByCallId.get(message.toolCallId);
+ const output = contentText(message.content);
+ const status: ToolStatus = message.isError ? 'failed' : 'completed';
+ if (existing) {
+ existing.status = status;
+ existing.output = output;
+ existing.settledSeq = seq;
+ if (message.details !== undefined) existing.details = message.details;
+ } else {
+ // A result whose call is not in this window (paged read, forked branch).
+ parts.push({
+ kind: 'tool',
+ toolCallId: message.toolCallId,
+ name: message.toolName,
+ arguments: {},
+ status,
+ output,
+ settledSeq: seq,
+ ...base
+ });
+ }
+ continue;
+ }
+
+ if (message.role === 'bashExecution') {
+ parts.push({
+ kind: 'bash',
+ command: message.command,
+ output: message.output,
+ ...(typeof message.exitCode === 'number' ? { exitCode: message.exitCode } : {}),
+ ...base
+ });
+ continue;
+ }
+
+ if (message.role === 'custom') {
+ parts.push({
+ kind: 'custom',
+ customType: message.customType,
+ text: contentText(message.content),
+ display: message.display !== false,
+ ...(message.details !== undefined ? { details: message.details } : {}),
+ ...base
+ });
+ continue;
+ }
+
+ parts.push({
+ kind: 'summary',
+ reason: message.role === 'branchSummary' ? 'branch' : 'compaction',
+ summary: (message as { summary?: string }).summary ?? '',
+ ...base
+ });
+ }
+
+ return {
+ parts,
+ ...(sessionId ? { sessionId } : {}),
+ ...(cwd ? { cwd } : {})
+ };
+}
diff --git a/agentic/run-log/src/projectors/session.ts b/agentic/run-log/src/projectors/session.ts
new file mode 100644
index 000000000..abd53677a
--- /dev/null
+++ b/agentic/run-log/src/projectors/session.ts
@@ -0,0 +1,94 @@
+/**
+ * Session projection: run log records → a pi session file.
+ *
+ * This is what makes a run resumable anywhere. The log is the source of truth;
+ * a `.jsonl` session is a derived artifact, so a run that started in the cloud
+ * can be continued locally (and vice versa) by projecting the log back into the
+ * only format pi's `SessionManager` reads.
+ *
+ * Returns a string rather than writing a file: the projection is pure, and the
+ * node-side write lives in `@agentic-kit/run-log/file-store`.
+ */
+
+import { isPiSessionHeader, type PiSessionEntry } from '../pi-entry';
+import { assertOrdered, type RunEventRecord, SUPPORTED_PI_SESSION_VERSION } from '../record';
+
+export interface SessionProjectionOptions {
+ /** Used when the log carries no session header (a run logged headerless). */
+ sessionId?: string;
+ cwd?: string;
+ timestamp?: string;
+}
+
+export interface SessionProjection {
+ /** The full session file contents, newline-terminated. */
+ jsonl: string;
+ /** Entries in file order, header first. */
+ entries: PiSessionEntry[];
+ piSessionVersion: number;
+}
+
+/**
+ * Project records into a pi session file. Throws when the records cannot form a
+ * loadable session — an unresumable session must fail at projection time, not
+ * when pi later reads a truncated tree.
+ */
+export function projectSession(
+ records: readonly RunEventRecord[],
+ options: SessionProjectionOptions = {}
+): SessionProjection {
+ assertOrdered(records);
+
+ const versions = new Set(records.map((record) => record.piSessionVersion));
+ if (versions.size > 1) {
+ throw new Error(
+ `run log mixes pi session versions (${Array.from(versions).sort().join(', ')}); migrate the older entries before projecting a session`
+ );
+ }
+ const piSessionVersion = records[0]?.piSessionVersion ?? SUPPORTED_PI_SESSION_VERSION;
+
+ const entries = records.map((record) => record.entry);
+ const headerIndex = entries.findIndex(isPiSessionHeader);
+ if (headerIndex > 0) {
+ throw new Error(
+ `run log carries a session header at position ${String(headerIndex)}; a pi session file requires it first`
+ );
+ }
+
+ const body = headerIndex === 0 ? entries.slice(1) : entries;
+ const header: PiSessionEntry =
+ headerIndex === 0
+ ? entries[0]
+ : {
+ type: 'session',
+ version: piSessionVersion,
+ id: options.sessionId ?? records[0]?.runId ?? 'run-log',
+ timestamp: options.timestamp ?? records[0]?.recordedAt ?? new Date().toISOString(),
+ ...(options.cwd ? { cwd: options.cwd } : {})
+ };
+
+ const ordered = [header, ...body];
+ return {
+ entries: ordered,
+ piSessionVersion,
+ jsonl: ordered.map((entry) => JSON.stringify(entry)).join('\n') + '\n'
+ };
+}
+
+/** Parse a pi session file into entries — the inverse, for importing a session. */
+export function parseSessionJsonl(jsonl: string): PiSessionEntry[] {
+ const entries: PiSessionEntry[] = [];
+ const lines = jsonl.split('\n');
+ for (let i = 0; i < lines.length; i += 1) {
+ const line = lines[i].trim();
+ if (line.length === 0) continue;
+ try {
+ entries.push(JSON.parse(line) as PiSessionEntry);
+ } catch (error) {
+ throw new Error(
+ `pi session line ${String(i + 1)} is not valid JSON: ${error instanceof Error ? error.message : String(error)}`
+ );
+ }
+ }
+ return entries;
+}
diff --git a/agentic/run-log/src/projectors/tool-state.ts b/agentic/run-log/src/projectors/tool-state.ts
new file mode 100644
index 000000000..35204543b
--- /dev/null
+++ b/agentic/run-log/src/projectors/tool-state.ts
@@ -0,0 +1,192 @@
+/**
+ * Tool and approval projection: run log records → the state a UI needs to know
+ * what is running and what is waiting on a human.
+ *
+ * Approvals ride in the log as pi `custom` messages rather than a side channel,
+ * because "the run is blocked on you" is part of the run's history: a surface
+ * that reconnects hours later must be able to see a pending request without
+ * having been present when it was raised, and both placements then behave the
+ * same — the cloud already treats the conversation as the approval UI.
+ */
+
+import { contentText, isAssistantMessage, isPiMessageEntry, isToolResultMessage, toolCalls } from '../pi-entry';
+import type { RunEventRecord } from '../record';
+
+/** `customType` of an approval request written by the gate extension. */
+export const APPROVAL_REQUEST_TYPE = 'constructive.approval.request';
+/** `customType` of the human's answer to a request. */
+export const APPROVAL_RESOLUTION_TYPE = 'constructive.approval.resolution';
+
+export type ToolCallStatus = 'requested' | 'awaiting-approval' | 'rejected' | 'running' | 'completed' | 'failed';
+
+export interface ToolCallState {
+ toolCallId: string;
+ name: string;
+ arguments: Record;
+ status: ToolCallStatus;
+ requestedSeq: number;
+ settledSeq?: number;
+ output?: string;
+ approval?: ApprovalState;
+}
+
+export interface ApprovalState {
+ toolCallId: string;
+ requestedSeq: number;
+ prompt: string;
+ resolvedSeq?: number;
+ decision?: 'approved' | 'rejected';
+ reason?: string;
+ actorId?: string;
+}
+
+export interface ToolStateProjection {
+ tools: Record;
+ /** In log order — the oldest unanswered request first. */
+ pendingApprovals: ApprovalState[];
+}
+
+interface ApprovalDetails {
+ toolCallId?: unknown;
+ decision?: unknown;
+ reason?: unknown;
+ actorId?: unknown;
+}
+
+const details = (value: unknown): ApprovalDetails =>
+ typeof value === 'object' && value !== null ? (value as ApprovalDetails) : {};
+
+export function projectToolState(records: readonly RunEventRecord[]): ToolStateProjection {
+ const tools: Record = {};
+ const approvals = new Map();
+
+ for (const { entry, seq } of records) {
+ if (!isPiMessageEntry(entry)) continue;
+ const message = entry.message;
+
+ if (isAssistantMessage(message)) {
+ for (const call of toolCalls(message)) {
+ tools[call.id] = {
+ toolCallId: call.id,
+ name: call.name,
+ arguments: call.arguments ?? {},
+ status: 'requested',
+ requestedSeq: seq
+ };
+ }
+ continue;
+ }
+
+ if (isToolResultMessage(message)) {
+ const state = tools[message.toolCallId];
+ const settled: Partial = {
+ status: message.isError ? 'failed' : 'completed',
+ settledSeq: seq,
+ output: contentText(message.content)
+ };
+ tools[message.toolCallId] = state
+ ? { ...state, ...settled }
+ : {
+ toolCallId: message.toolCallId,
+ name: message.toolName,
+ arguments: {},
+ requestedSeq: seq,
+ status: settled.status as ToolCallStatus,
+ settledSeq: seq,
+ output: settled.output as string
+ };
+ continue;
+ }
+
+ if (message.role !== 'custom') continue;
+
+ if (message.customType === APPROVAL_REQUEST_TYPE) {
+ const info = details(message.details);
+ const toolCallId = typeof info.toolCallId === 'string' ? info.toolCallId : null;
+ if (!toolCallId) {
+ throw new Error(
+ `approval request at seq ${String(seq)} carries no toolCallId; the run log cannot attribute it`
+ );
+ }
+ const approval: ApprovalState = {
+ toolCallId,
+ requestedSeq: seq,
+ prompt: contentText(message.content)
+ };
+ approvals.set(toolCallId, approval);
+ const state = tools[toolCallId];
+ if (state) tools[toolCallId] = { ...state, status: 'awaiting-approval', approval };
+ continue;
+ }
+
+ if (message.customType === APPROVAL_RESOLUTION_TYPE) {
+ const info = details(message.details);
+ const toolCallId = typeof info.toolCallId === 'string' ? info.toolCallId : null;
+ if (!toolCallId) {
+ throw new Error(
+ `approval resolution at seq ${String(seq)} carries no toolCallId; the run log cannot attribute it`
+ );
+ }
+ const approved = info.decision === 'approved';
+ const existing = approvals.get(toolCallId);
+ const approval: ApprovalState = {
+ ...(existing ?? { toolCallId, requestedSeq: seq, prompt: '' }),
+ resolvedSeq: seq,
+ decision: approved ? 'approved' : 'rejected',
+ ...(typeof info.reason === 'string' ? { reason: info.reason } : {}),
+ ...(typeof info.actorId === 'string' ? { actorId: info.actorId } : {})
+ };
+ approvals.set(toolCallId, approval);
+ const state = tools[toolCallId];
+ if (state && state.status === 'awaiting-approval') {
+ tools[toolCallId] = { ...state, status: approved ? 'running' : 'rejected', approval };
+ } else if (state) {
+ tools[toolCallId] = { ...state, approval };
+ }
+ }
+ }
+
+ const pendingApprovals = Array.from(approvals.values())
+ .filter((approval) => approval.resolvedSeq === undefined)
+ .sort((a, b) => a.requestedSeq - b.requestedSeq);
+
+ return { tools, pendingApprovals };
+}
+
+/** The parts of an approval request an extension needs to write one. */
+export interface ApprovalRequestInput {
+ toolCallId: string;
+ prompt: string;
+}
+
+export interface ApprovalResolutionInput {
+ toolCallId: string;
+ decision: 'approved' | 'rejected';
+ reason?: string;
+ actorId?: string;
+}
+
+/** Build the pi `custom` message an approval request is carried in. */
+export const approvalRequestMessage = (input: ApprovalRequestInput) => ({
+ role: 'custom' as const,
+ customType: APPROVAL_REQUEST_TYPE,
+ content: input.prompt,
+ display: true,
+ details: { toolCallId: input.toolCallId },
+ timestamp: Date.now()
+});
+
+/** Build the pi `custom` message a human's answer is carried in. */
+export const approvalResolutionMessage = (input: ApprovalResolutionInput) => ({
+ role: 'custom' as const,
+ customType: APPROVAL_RESOLUTION_TYPE,
+ content: input.reason ?? input.decision,
+ display: true,
+ details: {
+ toolCallId: input.toolCallId,
+ decision: input.decision,
+ ...(input.reason ? { reason: input.reason } : {}),
+ ...(input.actorId ? { actorId: input.actorId } : {})
+ },
+ timestamp: Date.now()
+});
diff --git a/agentic/run-log/src/projectors/usage.ts b/agentic/run-log/src/projectors/usage.ts
new file mode 100644
index 000000000..9e150d3ad
--- /dev/null
+++ b/agentic/run-log/src/projectors/usage.ts
@@ -0,0 +1,113 @@
+/**
+ * Usage projection: run log records → token and cost totals.
+ *
+ * Every usage-bearing pi entry counts, not just assistant messages: a tool that
+ * performed nested LLM work reports `usage` on its result, and compaction and
+ * branch summaries are model calls the run paid for. Missing any of those makes
+ * a run look cheaper than it was, which is exactly the kind of drift metering
+ * exists to prevent.
+ *
+ * This projection describes what a run *observed*. It is reconciliation input,
+ * never the billing authority — a gateway-observed record is (see
+ * `@agentic-kit/pi-ext-metered-model`).
+ */
+
+import {
+ isAssistantMessage,
+ isPiBranchSummaryEntry,
+ isPiCompactionEntry,
+ isPiMessageEntry,
+ isToolResultMessage,
+ type PiUsage
+} from '../pi-entry';
+import type { RunEventRecord } from '../record';
+
+export interface UsageTotals {
+ input: number;
+ output: number;
+ cacheRead: number;
+ cacheWrite: number;
+ totalTokens: number;
+ cost: number;
+ /** Number of usage-bearing entries folded into these totals. */
+ calls: number;
+}
+
+export interface ModelUsage extends UsageTotals {
+ provider: string;
+ model: string;
+}
+
+export interface RunUsage extends UsageTotals {
+ /** Per provider+model breakdown, keyed `provider/model`. */
+ byModel: Record;
+}
+
+const empty = (): UsageTotals => ({
+ input: 0,
+ output: 0,
+ cacheRead: 0,
+ cacheWrite: 0,
+ totalTokens: 0,
+ cost: 0,
+ calls: 0
+});
+
+const num = (value: unknown): number => (typeof value === 'number' && Number.isFinite(value) ? value : 0);
+
+function fold(target: UsageTotals, usage: PiUsage): void {
+ const input = num(usage.input);
+ const output = num(usage.output);
+ const cacheRead = num(usage.cacheRead);
+ const cacheWrite = num(usage.cacheWrite);
+ target.input += input;
+ target.output += output;
+ target.cacheRead += cacheRead;
+ target.cacheWrite += cacheWrite;
+ // Providers that omit a total still have one: the parts they did report.
+ target.totalTokens += usage.totalTokens === undefined
+ ? input + output + cacheRead + cacheWrite
+ : num(usage.totalTokens);
+ target.cost += num(usage.cost?.total);
+ target.calls += 1;
+}
+
+export const modelKey = (provider: string, model: string): string => `${provider}/${model}`;
+
+/** Fold every usage-bearing entry in the records into run totals. */
+export function projectUsage(records: readonly RunEventRecord[]): RunUsage {
+ const totals: RunUsage = { ...empty(), byModel: {} };
+
+ const add = (usage: PiUsage | undefined, provider = 'unknown', model = 'unknown'): void => {
+ if (!usage) return;
+ fold(totals, usage);
+ const key = modelKey(provider, model);
+ const bucket = totals.byModel[key] ?? { ...empty(), provider, model };
+ fold(bucket, usage);
+ totals.byModel[key] = bucket;
+ };
+
+ let lastProvider = 'unknown';
+ let lastModel = 'unknown';
+
+ for (const { entry } of records) {
+ if (isPiMessageEntry(entry)) {
+ const message = entry.message;
+ if (isAssistantMessage(message)) {
+ lastProvider = message.provider ?? lastProvider;
+ lastModel = message.model ?? lastModel;
+ add(message.usage, message.provider ?? 'unknown', message.model ?? 'unknown');
+ } else if (isToolResultMessage(message)) {
+ // Nested model work inside a tool: attributed to the run's current model,
+ // which is the model that requested the tool.
+ add(message.usage, lastProvider, lastModel);
+ }
+ continue;
+ }
+ if (isPiCompactionEntry(entry) || isPiBranchSummaryEntry(entry)) {
+ add(entry.usage, lastProvider, lastModel);
+ }
+ }
+
+ return totals;
+}
diff --git a/agentic/run-log/src/record.ts b/agentic/run-log/src/record.ts
new file mode 100644
index 000000000..938437285
--- /dev/null
+++ b/agentic/run-log/src/record.ts
@@ -0,0 +1,115 @@
+/**
+ * The run log record: four platform-owned fields around a verbatim pi entry.
+ *
+ * The wrapper exists to give an entry identity (which run) and order (which
+ * position) across hosts — nothing else. It deliberately does not re-encode pi's
+ * semantics, because pi already versions and migrates its own session format;
+ * `piSessionVersion` records which version an entry was written under so a
+ * reader can hand old entries to pi's migrations instead of guessing.
+ */
+
+import { assertPiSessionEntry, type PiSessionEntry } from './pi-entry';
+
+/** Version of the wrapper itself — bumped only if these four fields change. */
+export const RUN_LOG_WRAPPER_VERSION = 1;
+
+/** The pi session format version this package projects without migration. */
+export const SUPPORTED_PI_SESSION_VERSION = 3;
+
+export interface RunEventRecord {
+ /** The run this entry belongs to. */
+ runId: string;
+ /** 1-based position within the run. Gapless and strictly increasing. */
+ seq: number;
+ /** When the platform durably recorded the entry (ISO 8601). */
+ recordedAt: string;
+ /** pi's session format version at write time. */
+ piSessionVersion: number;
+ /** The pi session entry, byte-for-byte as pi produced it. */
+ entry: PiSessionEntry;
+}
+
+export interface WrapEntryOptions {
+ runId: string;
+ seq: number;
+ entry: PiSessionEntry;
+ recordedAt?: string;
+ piSessionVersion?: number;
+}
+
+export function wrapEntry(options: WrapEntryOptions): RunEventRecord {
+ if (!options.runId) throw new TypeError('a run log record needs a runId');
+ if (!Number.isInteger(options.seq) || options.seq < 1) {
+ throw new TypeError(`run log seq must be a positive integer, received ${String(options.seq)}`);
+ }
+ return {
+ runId: options.runId,
+ seq: options.seq,
+ recordedAt: options.recordedAt ?? new Date().toISOString(),
+ piSessionVersion: options.piSessionVersion ?? SUPPORTED_PI_SESSION_VERSION,
+ entry: assertPiSessionEntry(options.entry)
+ };
+}
+
+/**
+ * Narrow an untrusted record (database row, HTTP body). Throws on anything
+ * unreadable — a log that cannot be parsed must fail loudly, never silently
+ * render as an empty conversation.
+ */
+export function assertRunEventRecord(value: unknown): RunEventRecord {
+ if (typeof value !== 'object' || value === null || Array.isArray(value)) {
+ throw new TypeError(`run log record must be an object, received ${typeof value}`);
+ }
+ const record = value as Record;
+ if (typeof record.runId !== 'string' || record.runId.length === 0) {
+ throw new TypeError('run log record must carry a non-empty runId');
+ }
+ if (!Number.isInteger(record.seq) || (record.seq as number) < 1) {
+ throw new TypeError(`run log record ${record.runId} has an invalid seq: ${String(record.seq)}`);
+ }
+ if (typeof record.recordedAt !== 'string') {
+ throw new TypeError(`run log record ${record.runId}#${String(record.seq)} must carry recordedAt`);
+ }
+ if (!Number.isInteger(record.piSessionVersion)) {
+ throw new TypeError(
+ `run log record ${record.runId}#${String(record.seq)} must carry an integer piSessionVersion`
+ );
+ }
+ return {
+ runId: record.runId,
+ seq: record.seq as number,
+ recordedAt: record.recordedAt,
+ piSessionVersion: record.piSessionVersion as number,
+ entry: assertPiSessionEntry(record.entry)
+ };
+}
+
+/**
+ * The de-duplication key for an append. pi entry ids are unique within a
+ * session, so a retried append (a Job restart re-emitting its tail, a
+ * reconnecting writer) is recognised rather than duplicated.
+ */
+export function idempotencyKey(runId: string, entry: PiSessionEntry): string {
+ const id = typeof (entry as { id?: unknown }).id === 'string' ? (entry as { id: string }).id : null;
+ if (id) return `${runId}:${entry.type}:${id}`;
+ // Session headers carry no tree id; there is exactly one per session file.
+ return `${runId}:${entry.type}:${String((entry as { id?: string }).id ?? 'header')}`;
+}
+
+/** Records must be contiguous and in order before anything projects them. */
+export function assertOrdered(records: readonly RunEventRecord[]): void {
+ for (let i = 1; i < records.length; i += 1) {
+ const previous = records[i - 1];
+ const current = records[i];
+ if (current.runId !== previous.runId) {
+ throw new Error(
+ `run log records mix runs: ${previous.runId} then ${current.runId} at index ${String(i)}`
+ );
+ }
+ if (current.seq <= previous.seq) {
+ throw new Error(
+ `run log records out of order in ${current.runId}: seq ${String(previous.seq)} followed by ${String(current.seq)}`
+ );
+ }
+ }
+}
diff --git a/agentic/run-log/src/store.ts b/agentic/run-log/src/store.ts
new file mode 100644
index 000000000..d95d72f76
--- /dev/null
+++ b/agentic/run-log/src/store.ts
@@ -0,0 +1,103 @@
+/**
+ * The storage contract. A run log is append-only and read by cursor, so the
+ * interfaces are deliberately two: a writer (the agent host) and a reader
+ * (every UI surface, the resume path, the usage rollup). Concrete stores live
+ * with their storage — Postgres in constructive-db, JSONL in `./file-store`,
+ * memory here for tests and for a run that has not been persisted yet.
+ */
+
+import type { PiSessionEntry } from './pi-entry';
+import {
+ assertOrdered,
+ idempotencyKey,
+ type RunEventRecord,
+ SUPPORTED_PI_SESSION_VERSION,
+ wrapEntry
+} from './record';
+
+/** Read position: `afterSeq` is exclusive, so `0` means "from the start". */
+export interface RunLogCursor {
+ afterSeq: number;
+}
+
+export const START: RunLogCursor = { afterSeq: 0 };
+
+export const cursorAfter = (records: readonly RunEventRecord[], from: RunLogCursor = START): RunLogCursor =>
+ records.length === 0 ? from : { afterSeq: records[records.length - 1].seq };
+
+export interface RunLogPage {
+ records: RunEventRecord[];
+ cursor: RunLogCursor;
+}
+
+export interface AppendOptions {
+ /** pi session format version the entries were produced under. */
+ piSessionVersion?: number;
+ recordedAt?: string;
+}
+
+export interface RunLogAppendStore {
+ /**
+ * Append entries to a run, returning the records actually written. Entries
+ * already present (matched by pi entry id) are skipped, which makes an append
+ * safe to retry.
+ */
+ append(runId: string, entries: readonly PiSessionEntry[], options?: AppendOptions): Promise;
+}
+
+export interface RunLogReadStore {
+ read(runId: string, cursor?: RunLogCursor, limit?: number): Promise;
+}
+
+export type RunLogStore = RunLogAppendStore & RunLogReadStore;
+
+/** In-memory store: the reference implementation and the test double. */
+export class MemoryRunLogStore implements RunLogStore {
+ private readonly runs = new Map();
+ private readonly seen = new Map>();
+
+ async append(
+ runId: string,
+ entries: readonly PiSessionEntry[],
+ options: AppendOptions = {}
+ ): Promise {
+ const records = this.runs.get(runId) ?? [];
+ const seen = this.seen.get(runId) ?? new Set();
+ const written: RunEventRecord[] = [];
+
+ for (const entry of entries) {
+ const key = idempotencyKey(runId, entry);
+ if (seen.has(key)) continue;
+ const record = wrapEntry({
+ runId,
+ seq: records.length + written.length + 1,
+ entry,
+ ...(options.recordedAt ? { recordedAt: options.recordedAt } : {}),
+ piSessionVersion: options.piSessionVersion ?? SUPPORTED_PI_SESSION_VERSION
+ });
+ written.push(record);
+ seen.add(key);
+ }
+
+ this.runs.set(runId, records.concat(written));
+ this.seen.set(runId, seen);
+ return written;
+ }
+
+ async read(runId: string, cursor: RunLogCursor = START, limit?: number): Promise {
+ const all = this.runs.get(runId) ?? [];
+ const after = all.filter((record) => record.seq > cursor.afterSeq);
+ const records = typeof limit === 'number' ? after.slice(0, limit) : after;
+ assertOrdered(records);
+ return { records, cursor: cursorAfter(records, cursor) };
+ }
+
+ /** Test/debug helper: every record of a run, ignoring cursors. */
+ snapshot(runId: string): RunEventRecord[] {
+ return (this.runs.get(runId) ?? []).slice();
+ }
+
+ runIds(): string[] {
+ return Array.from(this.runs.keys());
+ }
+}
diff --git a/agentic/run-log/tsconfig.esm.json b/agentic/run-log/tsconfig.esm.json
new file mode 100644
index 000000000..624ab17cf
--- /dev/null
+++ b/agentic/run-log/tsconfig.esm.json
@@ -0,0 +1,7 @@
+{
+ "extends": "./tsconfig.json",
+ "compilerOptions": {
+ "module": "es2022",
+ "outDir": "dist/esm"
+ }
+}
diff --git a/agentic/run-log/tsconfig.json b/agentic/run-log/tsconfig.json
new file mode 100644
index 000000000..df063b5ee
--- /dev/null
+++ b/agentic/run-log/tsconfig.json
@@ -0,0 +1,8 @@
+{
+ "extends": "../../tsconfig.json",
+ "compilerOptions": {
+ "outDir": "dist",
+ "rootDir": "src"
+ },
+ "include": ["src/**/*.ts"]
+}
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 7bdb4c757..8f9cee276 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -255,6 +255,17 @@ importers:
agentic/protocol:
publishDirectory: dist
+ agentic/pi-ext-run-log:
+ dependencies:
+ '@agentic-kit/run-log':
+ specifier: workspace:^
+ version: link:../run-log/dist
+ devDependencies:
+ '@earendil-works/pi-coding-agent':
+ specifier: 0.79.6
+ version: 0.79.6(ws@8.20.1)(zod@4.4.3)
+ publishDirectory: dist
+
agentic/react:
dependencies:
'@agentic-kit/agent':
@@ -287,6 +298,9 @@ importers:
version: 19.2.5(react@19.2.5)
publishDirectory: dist
+ agentic/run-log:
+ publishDirectory: dist
+
examples/codegen-integration:
dependencies:
'@0no-co/graphql.web':