feat(dashboard/server): per-execution event journal + events API

Each `command_executions` row gets one JSONL file at
`.ocr/data/events/<execution_id>.jsonl`. command-runner appends one
StreamEvent per line; the dashboard's `GET /api/commands/:id/events`
route reads the file back for live-reload rehydration and history
replay. The journal is also load-bearing for the resume-recovery
primitive: when relational state misses a `vendor_session_id` capture,
the SessionCaptureService walks the JSONL and backfills.

Append-only, atomic-rename-friendly, malformed-line tolerant. The
file is read-only on the recovery path — writers go through the SQL
primitives in `@open-code-review/cli/db`; readers may consult JSONL
but never write it as authoritative state.

Co-Authored-By: claude-flow <ruv@ruv.net>

Author: spencermarx

packages/dashboard/src/server/routes/commands.ts

@@ -6,6 +6,7 @@ import { Router } from 'express'
 import type { Database } from 'sql.js'
 import { getCommandHistory } from '../db.js'
 import { getActiveCommands } from '../socket/command-runner.js'
+import { readEventJournal } from '../services/event-journal.js'
 
 type CommandDefinition = {
   name: string
@@ -44,7 +45,7 @@ const AVAILABLE_COMMANDS: CommandDefinition[] = [
   },
 ]
 
-export function createCommandsRouter(db: Database): Router {
+export function createCommandsRouter(db: Database, ocrDir: string): Router {
   const router = Router()
 
   // GET /api/commands — List available commands with descriptions
@@ -80,5 +81,32 @@ export function createCommandsRouter(db: Database): Router {
     }
   })
 
+  // GET /api/commands/:id/events — Replay the per-execution event stream.
+  //
+  // Returns the contents of `.ocr/data/events/<id>.jsonl` parsed back into
+  // a StreamEvent[]. Used by the client for two paths:
+  //   1. Rehydration when a tab reloads mid-run — the live socket
+  //      subscription only sees events from now on; this fills in the gap.
+  //   2. History replay — expanding a completed command in the history
+  //      list lazy-fetches its events to render the timeline.
+  //
+  // Returns an empty array (not 404) when no journal exists. Non-AI
+  // commands and rows that predate the events feature have no journal —
+  // the client treats empty as "use the legacy raw output instead."
+  router.get('/:id/events', (req, res) => {
+    const id = parseInt(req.params['id'] ?? '', 10)
+    if (!Number.isFinite(id) || id <= 0) {
+      res.status(400).json({ error: 'Invalid execution id' })
+      return
+    }
+    try {
+      const events = readEventJournal(ocrDir, id)
+      res.json({ execution_id: id, events })
+    } catch (err) {
+      console.error(`Failed to read events for execution ${id}:`, err)
+      res.status(500).json({ error: 'Failed to read event journal' })
+    }
+  })
+
   return router
 }

packages/dashboard/src/server/services/__tests__/event-journal.test.ts

@@ -0,0 +1,93 @@
+/**
+ * Event journal — round-trip + edge-case tests for the JSONL persistence
+ * helper that backs `command:event` rehydration.
+ */
+
+import { afterEach, beforeEach, describe, expect, it } from 'vitest'
+import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import {
+  EventJournalAppender,
+  eventJournalPath,
+  readEventJournal,
+} from '../event-journal.js'
+import type { StreamEvent } from '../ai-cli/types.js'
+
+let workspace: string
+let ocrDir: string
+
+beforeEach(() => {
+  workspace = mkdtempSync(join(tmpdir(), 'ocr-events-'))
+  ocrDir = join(workspace, '.ocr')
+})
+
+afterEach(() => {
+  rmSync(workspace, { recursive: true, force: true })
+})
+
+function makeEvent(seq: number, overrides: Partial<StreamEvent> = {}): StreamEvent {
+  return {
+    type: 'text_delta',
+    text: `chunk ${seq}`,
+    executionId: 1,
+    agentId: 'orchestrator',
+    timestamp: new Date(2026, 0, 1, 0, 0, seq).toISOString(),
+    seq,
+    ...overrides,
+  } as StreamEvent
+}
+
+describe('event-journal', () => {
+  it('appends each event as one JSON line and reads them back in order', async () => {
+    const appender = new EventJournalAppender(ocrDir, 1)
+    appender.append(makeEvent(1))
+    appender.append(makeEvent(2, { type: 'message', text: 'final', executionId: 1 } as never))
+    await appender.close()
+
+    const path = eventJournalPath(ocrDir, 1)
+    const raw = readFileSync(path, 'utf-8')
+    const lines = raw.trim().split('\n')
+    expect(lines).toHaveLength(2)
+
+    const events = readEventJournal(ocrDir, 1)
+    expect(events).toHaveLength(2)
+    expect(events[0]?.seq).toBe(1)
+    expect(events[1]?.seq).toBe(2)
+  })
+
+  it('returns an empty array when no journal exists', () => {
+    expect(readEventJournal(ocrDir, 999)).toEqual([])
+  })
+
+  it('skips malformed lines rather than throwing', async () => {
+    // Initialize directory by appending one valid event, then close.
+    const appender = new EventJournalAppender(ocrDir, 7)
+    appender.append(makeEvent(1))
+    await appender.close()
+    // Inject a malformed line at the end of the file.
+    const path = eventJournalPath(ocrDir, 7)
+    const original = readFileSync(path, 'utf-8')
+    writeFileSync(path, original + '{this is not json}\n', 'utf-8')
+
+    const events = readEventJournal(ocrDir, 7)
+    expect(events).toHaveLength(1)
+    expect(events[0]?.seq).toBe(1)
+  })
+
+  it('append after close is a no-op rather than throwing', () => {
+    const appender = new EventJournalAppender(ocrDir, 11)
+    appender.close()
+    expect(() => appender.append(makeEvent(1))).not.toThrow()
+  })
+
+  it('lazily creates the events directory on first appender', async () => {
+    // The appender's constructor should have created the directory; the
+    // path is what we care about.
+    const appender = new EventJournalAppender(ocrDir, 42)
+    appender.append(makeEvent(1))
+    await appender.close()
+    const path = eventJournalPath(ocrDir, 42)
+    expect(readFileSync(path, 'utf-8').length).toBeGreaterThan(0)
+  })
+})

packages/dashboard/src/server/services/event-journal.ts

@@ -0,0 +1,130 @@
+/**
+ * Event journal — JSONL persistence for live command streams.
+ *
+ * Each command_executions row gets one journal file at
+ * `.ocr/data/events/<execution_id>.jsonl`. The command-runner appends one
+ * `StreamEvent` per JSON line as the AI CLI emits them; the dashboard's
+ * `GET /api/commands/:id/events` route reads the file back for rehydration
+ * (page reload mid-run) and history-replay.
+ *
+ * Why JSONL on disk rather than a sqlite table:
+ *   1. Append-only writes avoid the sql.js merge-before-write rename dance
+ *      under high event throughput
+ *   2. The format is trivially `tail -f`-able for humans debugging a run
+ *   3. Event volume per execution is bounded but non-trivial (hundreds to
+ *      low-thousands per active review) — keeping it out of the DB keeps
+ *      the in-memory sql.js DB small
+ *   4. No schema migration needed if the event union evolves
+ *
+ * Writes are best-effort and intentionally non-blocking — if the journal
+ * write fails, the live socket emit still happens, and the user just loses
+ * the ability to replay/reload-rehydrate that one event. The command itself
+ * does NOT fail because of a journal error.
+ */
+
+import {
+  createWriteStream,
+  existsSync,
+  mkdirSync,
+  readFileSync,
+  type WriteStream,
+} from 'node:fs'
+import { join } from 'node:path'
+import type { StreamEvent } from './ai-cli/types.js'
+
+/**
+ * Resolves the directory where event journals live for a given workspace.
+ * Lazily creates the directory so first-run installs work without setup.
+ */
+export function eventsDir(ocrDir: string): string {
+  const dir = join(ocrDir, 'data', 'events')
+  if (!existsSync(dir)) {
+    mkdirSync(dir, { recursive: true })
+  }
+  return dir
+}
+
+/**
+ * Resolves the journal file path for a single execution.
+ * The file may or may not exist yet — appendEvent creates it on first write.
+ */
+export function eventJournalPath(ocrDir: string, executionId: number): string {
+  return join(eventsDir(ocrDir), `${executionId}.jsonl`)
+}
+
+/**
+ * Per-execution append handle. Keeps a write stream open for the lifetime
+ * of the execution so we don't pay the open/close cost on every event.
+ *
+ * Call `close()` when the execution finishes. Idempotent.
+ */
+export class EventJournalAppender {
+  private stream: WriteStream | null
+  readonly path: string
+
+  constructor(ocrDir: string, executionId: number) {
+    this.path = eventJournalPath(ocrDir, executionId)
+    // 'a' = append, creates if missing
+    this.stream = createWriteStream(this.path, { flags: 'a' })
+    // Errors on the stream are logged but don't crash the runner — this is
+    // a best-effort journal, not a load-bearing path.
+    this.stream.on('error', (err) => {
+      console.error(`[event-journal] write error for ${this.path}:`, err)
+      this.stream = null
+    })
+  }
+
+  append(event: StreamEvent): void {
+    if (!this.stream) return
+    this.stream.write(JSON.stringify(event) + '\n')
+  }
+
+  /**
+   * Close the underlying write stream. Returns a promise that resolves
+   * once the OS has flushed all pending writes, so callers that need
+   * to read the file back synchronously (tests, the events route on
+   * a just-finished execution) can await this.
+   *
+   * Idempotent — calling close after the stream is already closed is
+   * a no-op that resolves immediately.
+   */
+  close(): Promise<void> {
+    if (!this.stream) return Promise.resolve()
+    const stream = this.stream
+    this.stream = null
+    return new Promise<void>((resolve) => {
+      stream.end(() => resolve())
+    })
+  }
+}
+
+/**
+ * Reads all events for a given execution. Returns an empty array when no
+ * journal exists yet (pre-AI command, journal write failed, or execution
+ * predates the event-stream feature).
+ *
+ * The events are returned in write order. Malformed lines are skipped with
+ * a warning rather than throwing — partial recovery is more useful than
+ * an all-or-nothing failure for a debug surface.
+ */
+export function readEventJournal(ocrDir: string, executionId: number): StreamEvent[] {
+  const path = eventJournalPath(ocrDir, executionId)
+  if (!existsSync(path)) return []
+  let raw: string
+  try {
+    raw = readFileSync(path, 'utf-8')
+  } catch {
+    return []
+  }
+  const events: StreamEvent[] = []
+  const lines = raw.split('\n')
+  for (const line of lines) {
+    if (!line.trim()) continue
+    try {
+      events.push(JSON.parse(line) as StreamEvent)
+    } catch (err) {
+      console.warn(`[event-journal] malformed line in ${path}:`, err)
+    }
+  }
+  return events
+}