feat(sdk): flows replay verb — journal time-travel (#309)

packages/sdk/src/cli.ts

@@ -16,6 +16,7 @@ import {
   type RunReport,
 } from './cli/run.js';
 import { runDirectFlow } from './cli/direct-run.js';
+import { parseReplayArgs, replayJournal, type ReplayArgs } from './cli/replay.js';
 import { runCloudCli } from './cli/cloud-run.js';
 import { isAuthoredFlowPath } from './direct-input.js';
 import { runHnMonitor } from './cli/hn-monitor.js';
@@ -35,6 +36,7 @@ export interface CliIo {
 
 type CliExitCode = 0 | 1 | 2 | 3;
 type ParsedArgs =
+  | ReplayArgs
   | { command: 'cloud-run'; value: string; json: boolean; wait: boolean }
   | { command: 'check'; json: boolean; value: string }
   | { command: 'run'; localAgent: boolean; dataDir: string; input: string | undefined; json: boolean; spawn: boolean; noObserverLink: boolean; value: string }
@@ -54,6 +56,7 @@ const USAGE = [
   'flows run [--json] [--no-spawn] [--no-observer-link] [--data-dir <dir>] [--local-agent] <flow.ts> --input <inline-json-or-file>',
   'flows tick start --schedule-id <id> --interval-ms <ms> [--epoch-ms <ms>] [--max-catch-up <n>] [--poll-interval-ms <ms>] [--data-dir <dir>] <spec.json>',
   'flows resume [--json] [--no-spawn] [--no-observer-link] [--data-dir <dir>] <run-id>',
+  'flows replay [--json] [--data-dir <dir>] <run-id> [--at <step-id>]',
   'flows observer [--data-dir <dir>]',
   'flows hn-monitor start [--data-dir <dir>] [--poll-interval-ms <n>] <spec.json>',
 ].join('\n');
@@ -90,6 +93,7 @@ export async function runCli(
   }
 
   if (parsed.command === 'cloud-run') return runCloudCli(parsed, io);
+  if (parsed.command === 'replay') return replayJournal(parsed, io);
 
   if (parsed.command === 'check') {
     // Deliberately daemon-free (kernel/DAEMON-LIFECYCLE.md §4). `checkFlow` is
@@ -356,6 +360,7 @@ function emitWait(
 
 function parseArgs(args: readonly string[]): ParsedArgs | undefined {
   const command = args[0];
+  if (command === 'replay') return parseReplayArgs(args.slice(1));
   if (command === 'hn-monitor') return parseHnMonitorArgs(args.slice(1));
   if (command === 'tick') return parseTickArgs(args.slice(1));
   if (command === 'observer') return parseObserverArgs(args.slice(1));

packages/sdk/src/cli/replay.ts

@@ -0,0 +1,72 @@
+import type { CliIo } from '../cli.js';
+import { canonicalize } from '../canonical.js';
+import { JournalReadError, walkJournal } from '../journal-client.js';
+
+export interface ReplayArgs {
+  command: 'replay';
+  value: string;
+  json: boolean;
+  dataDir: string;
+  at?: string;
+}
+
+export function parseReplayArgs(args: readonly string[]): ReplayArgs | undefined {
+  let json = false;
+  let dataDir: string | undefined;
+  let at: string | undefined;
+  const positionals: string[] = [];
+  for (let index = 0; index < args.length; index += 1) {
+    const argument = args[index]!;
+    if (argument === '--json') {
+      if (json) return undefined;
+      json = true;
+    } else if (argument === '--data-dir' || argument === '--at') {
+      const value = args[++index];
+      if (value === undefined || value.length === 0 || value.startsWith('-')) return undefined;
+      if (argument === '--data-dir') {
+        if (dataDir !== undefined) return undefined;
+        dataDir = value;
+      } else {
+        if (at !== undefined) return undefined;
+        at = value;
+      }
+    } else if (argument.startsWith('-')) {
+      return undefined;
+    } else {
+      positionals.push(argument);
+    }
+  }
+  if (positionals.length !== 1) return undefined;
+  return { command: 'replay', value: positionals[0]!, json, dataDir: dataDir ?? '.relayflowd', at };
+}
+
+export async function replayJournal(args: ReplayArgs, io: CliIo): Promise<0 | 1 | 2> {
+  let emitted = false;
+  try {
+    for await (const event of walkJournal(args.value, args.dataDir, { at: args.at })) {
+      const payload = event.payload !== null && typeof event.payload === 'object' && !Array.isArray(event.payload)
+        ? event.payload as Record<string, unknown> : {};
+      io.stdout(args.json ? canonicalize({
+        step_id: event.step_id,
+        kind: event.entry_type,
+        event,
+        verification: payload['verification'] ?? null,
+        spend: payload['budget'] ?? payload['budget_total'] ?? null,
+      })
+        : `${event.seq} ${event.at_ms} ${event.entry_type}`
+          + (event.step_id === null ? '' : ` step=${JSON.stringify(event.step_id)}`)
+          + (event.attempt === null ? '' : ` attempt=${event.attempt}`)
+          + ` ${canonicalize(event.payload)}`);
+      emitted = true;
+    }
+    return 0;
+  } catch (error) {
+    const diagnostic = {
+      severity: 'refusal',
+      kind: error instanceof JournalReadError ? error.code : 'journal_read_failed',
+      message: error instanceof Error ? error.message : String(error),
+    };
+    io.stderr(`${emitted ? 'FAILED' : 'REFUSED'} [${diagnostic.kind}] ${diagnostic.message}`);
+    return emitted ? 1 : 2;
+  }
+}

packages/sdk/src/journal-client.ts

@@ -10,6 +10,7 @@
 // loopback double in tests.
 
 import { EventEmitter } from 'node:events';
+export { walkJournal, JournalReadError, type JournalEvent, type JournalReadFailure } from './journal-reader.js';
 import { randomUUID } from 'node:crypto';
 import { createConnection, type Socket } from 'node:net';
 import type { VerbContract, EventSubmitParams } from './protocol.js';

packages/sdk/src/journal-offset.ts

@@ -0,0 +1,60 @@
+import { readFileSync } from 'node:fs';
+
+/** Locate a table row's cell in a SQLite snapshot for read-error diagnostics. */
+export function journalRecordOffset(path: string, rootPage: number, seq: number): string {
+  const main = readFileSync(path);
+  const encodedSize = main.readUInt16BE(16);
+  const pageSize = encodedSize === 1 ? 65536 : encodedSize;
+  const pages = new Map<number, { bytes: Buffer; offset: number; file: string }>();
+  let wal: Buffer | undefined;
+  try { wal = readFileSync(`${path}-wal`); } catch (error) {
+    if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
+  }
+  if (wal !== undefined) {
+    const frameSize = pageSize + 24;
+    let lastCommit = 0;
+    for (let frame = 32; frame + frameSize <= wal.length; frame += frameSize) {
+      // RESTART checkpoints reuse the WAL without truncating its old tail.
+      // Frames from that prior generation are not part of SQLite's view.
+      if (!wal.subarray(frame + 8, frame + 16).equals(wal.subarray(16, 24))) break;
+      if (wal.readUInt32BE(frame + 4) !== 0) lastCommit = frame;
+    }
+    for (let frame = 32; frame <= lastCommit; frame += frameSize) {
+      pages.set(wal.readUInt32BE(frame), { bytes: wal, offset: frame + 24, file: 'WAL' });
+    }
+  }
+  const visited = new Set<number>();
+  let page = rootPage;
+  while (!visited.has(page)) {
+    visited.add(page);
+    const source = pages.get(page) ?? { bytes: main, offset: (page - 1) * pageSize, file: 'journal' };
+    const header = source.offset + (page === 1 ? 100 : 0);
+    const kind = source.bytes[header];
+    if (kind !== 5 && kind !== 13) break;
+    const count = source.bytes.readUInt16BE(header + 3);
+    let next = kind === 5 ? source.bytes.readUInt32BE(header + 8) : 0;
+    for (let index = 0; index < count; index += 1) {
+      const cell = source.offset + source.bytes.readUInt16BE(header + (kind === 5 ? 12 : 8) + 2 * index);
+      const rowIdOffset = kind === 5 ? cell + 4 : varint(source.bytes, cell).next;
+      const rowId = varint(source.bytes, rowIdOffset).value;
+      if (kind === 13 && rowId === BigInt(seq)) return `${source.file} byte offset ${cell}`;
+      if (kind === 5 && BigInt(seq) <= rowId) {
+        next = source.bytes.readUInt32BE(cell);
+        break;
+      }
+    }
+    if (next === 0) break;
+    page = next;
+  }
+  throw new Error('Cannot locate journal record cell.');
+}
+
+function varint(bytes: Buffer, offset: number): { value: bigint; next: number } {
+  let value = 0n;
+  for (let index = 0; index < 9; index += 1) {
+    const byte = bytes.readUInt8(offset++);
+    value = (value << (index === 8 ? 8n : 7n)) | BigInt(index === 8 ? byte : byte & 0x7f);
+    if (byte < 0x80 || index === 8) return { value, next: offset };
+  }
+  throw new Error('Invalid SQLite varint.');
+}

packages/sdk/src/journal-reader.ts

@@ -0,0 +1,165 @@
+import { constants } from 'node:fs';
+import { copyFile, mkdtemp, open, rm, stat } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { createRequire } from 'node:module';
+import { join, resolve } from 'node:path';
+import { journalRecordOffset } from './journal-offset.js';
+
+// Journal version 1's closed vocabulary (relayflowd-core/src/entry.rs).
+const ENTRY_TYPES = new Set([
+  'run.spawned', 'run.cancel.requested', 'event.received', 'subscription.registered',
+  'subscription.matched', 'subscription.stale', 'step.routed', 'step.attempt.started',
+  'step.completed', 'wait.event', 'wait.human', 'sleep.until', 'wait.completed',
+  'stream.appended', 'memory.injected', 'channel.appended', 'channel.delivered',
+  'channel.acknowledged', 'effect.recorded', 'effect.confirmed', 'epoch.summary',
+  'segment.closed', 'run.completed',
+]);
+
+/** The persisted envelope from relayflowd-core/src/entry.rs. */
+export interface JournalEvent {
+  seq: number;
+  segment_id: number;
+  entry_type: string;
+  run_id: string;
+  step_id: string | null;
+  attempt: number | null;
+  at_ms: number;
+  payload: unknown;
+}
+
+export type JournalReadFailure =
+  | 'invalid_run_id' | 'run_not_found' | 'step_not_found' | 'journal_read_failed';
+
+export class JournalReadError extends Error {
+  constructor(readonly code: JournalReadFailure, message: string) {
+    super(message);
+    this.name = 'JournalReadError';
+  }
+}
+
+async function fingerprint(path: string): Promise<string | undefined> {
+  try {
+    const info = await stat(path, { bigint: true });
+    if (!info.isFile()) throw new Error(`Journal path is not a regular file: ${path}`);
+    return [info.dev, info.ino, info.size, info.mtimeNs, info.ctimeNs].join(':');
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === 'ENOENT') return undefined;
+    throw error;
+  }
+}
+
+/**
+ * Walk a stable on-disk journal in sequence order, across every segment.
+ * `at` includes the step's terminal completion, or its last journaled event
+ * if it has not terminated. Retries are included.
+ *
+ * SQLite read-only connections can still create/change WAL shared-memory files.
+ * Copy the database and its WAL into private scratch space before opening it,
+ * and reject concurrent source changes instead of returning a torn snapshot.
+ * Only scratch files are written; the run, registry and daemon are untouched.
+ */
+export async function* walkJournal(
+  runId: string,
+  dataDir: string,
+  options: { at?: string } = {},
+): AsyncIterable<JournalEvent> {
+  // Run ids are path components, never paths. Permit imported run names as
+  // well as the ULIDs produced by the kernel.
+  if (!/^[A-Za-z0-9][A-Za-z0-9_-]*$/.test(runId)) {
+    throw new JournalReadError('invalid_run_id', 'Run id must contain only letters, digits, underscores or hyphens, starting with a letter or digit.');
+  }
+  const source = join(resolve(dataDir), 'runs', `${runId}.sqlite3`);
+  let scratch: string | undefined;
+  let database: import('node:sqlite').DatabaseSync | undefined;
+  try {
+    const before = await Promise.all([fingerprint(source), fingerprint(`${source}-wal`)]);
+    if (before[0] === undefined) {
+      throw new JournalReadError('run_not_found', `Run "${runId}" does not exist in "${dataDir}".`);
+    }
+    scratch = await mkdtemp(join(tmpdir(), 'flows-replay-'));
+    const snapshot = join(scratch, 'journal.sqlite3');
+    await copyFile(source, snapshot, constants.COPYFILE_EXCL);
+    if (before[1] !== undefined) await copyFile(`${source}-wal`, `${snapshot}-wal`, constants.COPYFILE_EXCL);
+    const after = await Promise.all([fingerprint(source), fingerprint(`${source}-wal`)]);
+    if (before.some((value, index) => value !== after[index])) {
+      throw new Error('Journal changed while taking the replay snapshot; retry replay.');
+    }
+    const file = await open(snapshot, 'r');
+    try {
+      const header = Buffer.alloc(16);
+      const { bytesRead } = await file.read(header, 0, 16, 0);
+      const expected = Buffer.from('SQLite format 3\0');
+      for (let offset = 0; offset < expected.length; offset += 1) {
+        if (offset >= bytesRead || header[offset] !== expected[offset]) {
+          throw new Error(`Invalid SQLite header at journal byte offset ${offset}.`);
+        }
+      }
+    } finally { await file.close(); }
+
+    // Lazy loading keeps all existing CLI verbs independent of node:sqlite.
+    const { DatabaseSync } = createRequire(import.meta.url)('node:sqlite') as typeof import('node:sqlite');
+    database = new DatabaseSync(snapshot, { readOnly: true });
+    const metadata = database.prepare("SELECT value FROM meta WHERE key = 'run_id'").get();
+    if (metadata?.['value'] !== runId) throw new Error('Journal run id does not match the requested run.');
+    const version = database.prepare("SELECT value FROM meta WHERE key = 'journal_version'").get();
+    if (version?.['value'] !== '1') throw new Error('Unsupported journal version.');
+    const integrity = database.prepare('PRAGMA quick_check').all();
+    if (integrity.length !== 1 || integrity[0]?.['quick_check'] !== 'ok') throw new Error('Journal integrity check failed.');
+
+    let through: number | undefined;
+    if (options.at !== undefined) {
+      const row = database.prepare(`
+        SELECT COALESCE(MAX(CASE WHEN entry_type = 'step.completed'
+          AND json_valid(payload) AND json_extract(payload, '$.disposition') = 'step_done'
+          THEN seq END), MAX(seq)) AS seq FROM entries WHERE step_id = ?
+      `).get(options.at);
+      if (row?.['seq'] === null || row === undefined) {
+        throw new JournalReadError('step_not_found', `Step "${options.at}" is not journaled in run "${runId}".`);
+      }
+      through = integer(row['seq'], 'seq');
+    }
+    const rows = database.prepare(
+      'SELECT seq, segment_id, entry_type, step_id, attempt, at_ms, payload FROM entries'
+        + (through === undefined ? '' : ' WHERE seq <= ?') + ' ORDER BY seq',
+    ).iterate(...(through === undefined ? [] : [through]));
+    for (const row of rows) {
+      try {
+        if (typeof row['entry_type'] !== 'string' || !ENTRY_TYPES.has(row['entry_type'])
+          || typeof row['payload'] !== 'string'
+          || (row['step_id'] !== null && typeof row['step_id'] !== 'string')) {
+          throw new Error('Invalid journal entry or unknown record type.');
+        }
+        yield {
+          seq: integer(row['seq'], 'seq'),
+          segment_id: integer(row['segment_id'], 'segment_id'),
+          entry_type: row['entry_type'],
+          run_id: runId,
+          step_id: row['step_id'],
+          attempt: row['attempt'] === null ? null : integer(row['attempt'], 'attempt'),
+          at_ms: integer(row['at_ms'], 'at_ms'),
+          payload: JSON.parse(row['payload']),
+        };
+      } catch (error) {
+        const root = database.prepare("SELECT rootpage FROM sqlite_schema WHERE name = 'entries'").get();
+        const location = journalRecordOffset(snapshot, integer(root?.['rootpage'], 'rootpage'), integer(row['seq'], 'seq'));
+        throw new Error(`Entry seq ${row['seq']} at ${location}: ${error instanceof Error ? error.message : String(error)}`);
+      }
+    }
+  } catch (error) {
+    if (error instanceof JournalReadError) throw error;
+    throw new JournalReadError('journal_read_failed', `Cannot read journal for run "${runId}": ${error instanceof Error ? error.message : String(error)}`);
+  } finally {
+    try {
+      database?.close();
+    } finally {
+      if (scratch !== undefined) await rm(scratch, { recursive: true, force: true });
+    }
+  }
+}
+
+function integer(value: unknown, field: string): number {
+  if (typeof value !== 'number' || !Number.isSafeInteger(value)) {
+    throw new Error(`Invalid journal ${field}: expected a safe integer.`);
+  }
+  return value;
+}