feat(pi-ext-run-log): @agentic-kit/pi-ext-run-log — mirror pi session entries into the run log

.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

agentic/pi-ext-run-log/README.md

@@ -0,0 +1,48 @@
+<p align="center" width="100%">
+  <img height="250" src="https://raw.githubusercontent.com/constructive-io/constructive/refs/heads/main/assets/outline-logo.svg" />
+</p>
+
+# @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
+```

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<void>;
+  registered: string[];
+  entries: unknown[];
+}
+
+const fakePi = (): FakePi => {
+  const handlers = new Map<string, (event: unknown, ctx: unknown) => Promise<void> | void>();
+  const entries: unknown[] = [];
+  const sessionManager = { getHeader: () => header, getEntries: () => entries };
+  const api = {
+    on: (event: string, handler: (event: unknown, ctx: unknown) => Promise<void> | 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([]);
+  });
+});

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<string, unknown> = { 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);
+  });
+});

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',
+  },
+};