feat(ai-autopilot): surfaces (terminal / in-page / background)

.changeset/ai-autopilot-surfaces.md

@@ -0,0 +1,5 @@
+---
+"@gemstack/ai-autopilot": minor
+---
+
+Add surfaces: run the same autopilot in the terminal, an in-page UI, or a background process, all over the Supervisor's `onEvent` stream. `terminalSink()` prints events inline (`formatEvent()` renders one event as a line); `EventStream` is a replayable multi-consumer transport with offset/tail replay (borrowing Flue's Durable-Streams `tail=N`); `launchAutopilot(start)` runs a Supervisor detached and returns an `AutopilotHandle` (`status()`, `events(offset)`, live async `stream()`, `result()`) that backs both the background and in-page (SSE) surfaces. Verified end-to-end against a real Supervisor. Closes the surfaces child (#100) of the ai-autopilot epic (#97).

packages/ai-autopilot/README.md

@@ -106,6 +106,45 @@ a `prefix` to avoid name collisions.
 To implement a real runner, satisfy the `Runner` interface: `boot()` returns a
 `RunnerSession` with an `fs`, `exec()`, an optional `preview()`, and `dispose()`.
 
+## Surfaces — terminal, in-page, background
+
+The Supervisor already emits progress via `onEvent`. **Surfaces** run the same
+autopilot in three places by adapting that event stream:
+
+**Terminal** — print each event inline:
+
+```ts
+import { Supervisor, terminalSink } from '@gemstack/ai-autopilot'
+
+const supervisor = new Supervisor({ ...opts, onEvent: terminalSink() })
+await supervisor.run(task)
+// ▶ plan: 2 subtask(s) for "…"
+//   → s1: …
+//   ✓ s1
+//   ▶ synthesize: 2 result(s)
+```
+
+**Background** — launch detached and get a handle; nothing blocks:
+
+```ts
+import { launchAutopilot } from '@gemstack/ai-autopilot'
+
+const run = launchAutopilot(onEvent => new Supervisor({ ...opts, onEvent }).run(task))
+run.status()        // 'running' → 'done' | 'error'
+run.events(offset)  // replay history from an offset (Flue-style tail=N)
+const result = await run.result()
+```
+
+**In-page** — the same handle exposes a live async stream to push over SSE:
+
+```ts
+for await (const event of run.stream()) sendToClient(event)  // replays history, then live, then ends
+```
+
+`EventStream` is the underlying replayable, multi-consumer transport; a late
+subscriber still sees the full history. Use `formatEvent(event)` to render an
+event as a line yourself.
+
 ## Guardrails
 
 - **`concurrency`** (optional, default 4) — max workers in flight; positive integer.

packages/ai-autopilot/src/index.ts

@@ -26,6 +26,13 @@
  *
  * - {@link FakeRunner} — in-memory runner for tests
  * - {@link runnerTools} — expose a booted session to an agent as sandbox tools
+ *
+ * Surfaces run the same autopilot in the terminal, an in-page UI, or a
+ * background process — all over the Supervisor's `onEvent` stream.
+ *
+ * - {@link terminalSink} — print events inline (terminal surface)
+ * - {@link EventStream} — replayable multi-consumer event transport
+ * - {@link launchAutopilot} — a detached background run handle
  */
 export { Supervisor } from './supervisor.js'
 export { agentPlanner, type AgentPlannerOptions } from './planner.js'
@@ -65,6 +72,16 @@ export {
   type RecordedExec,
   type RunnerToolsOptions,
 } from './runner/index.js'
+export {
+  EventStream,
+  formatEvent,
+  terminalSink,
+  launchAutopilot,
+  type TerminalSinkOptions,
+  type AutopilotHandle,
+  type AutopilotStatus,
+  type LaunchOptions,
+} from './surface/index.js'
 export type {
   Subtask,
   PlannedSubtask,

packages/ai-autopilot/src/surface/events.test.ts

@@ -0,0 +1,96 @@
+import { describe, it } from 'node:test'
+import assert from 'node:assert/strict'
+import { EventStream, formatEvent, terminalSink } from './events.js'
+import type { SupervisorEvent, PlannedSubtask, SubtaskResult } from '../types.js'
+
+const sub: PlannedSubtask = { id: 'subtask-1', description: 'do a thing' }
+const okResult: SubtaskResult = {
+  subtask: sub,
+  text: 'done',
+  ok: true,
+  usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 },
+}
+
+describe('EventStream history / tail replay', () => {
+  it('buffers events and replays from an offset', () => {
+    const s = new EventStream()
+    s.push({ type: 'plan', task: 't', subtasks: [sub] })
+    s.push({ type: 'dispatch-start', subtask: sub })
+    assert.equal(s.length, 2)
+    assert.equal(s.history().length, 2)
+    assert.deepEqual(
+      s.history(1).map(e => e.type),
+      ['dispatch-start'],
+    )
+  })
+
+  it('ignores pushes after close', () => {
+    const s = new EventStream()
+    s.close()
+    s.push({ type: 'plan', task: 't', subtasks: [] })
+    assert.equal(s.length, 0)
+    assert.equal(s.isClosed, true)
+  })
+})
+
+describe('EventStream async iteration', () => {
+  it('replays buffered events then ends once closed', async () => {
+    const s = new EventStream()
+    s.push({ type: 'plan', task: 't', subtasks: [sub] })
+    s.push({ type: 'synthesize', results: [okResult] })
+    s.close()
+    const seen: string[] = []
+    for await (const e of s) seen.push(e.type)
+    assert.deepEqual(seen, ['plan', 'synthesize'])
+  })
+
+  it('delivers events pushed after iteration starts, and to multiple iterators', async () => {
+    const s = new EventStream()
+    const collect = async () => {
+      const out: string[] = []
+      for await (const e of s) out.push(e.type)
+      return out
+    }
+    const a = collect()
+    const b = collect()
+    // let the iterators reach their waiting state, then push + close
+    await Promise.resolve()
+    s.push({ type: 'plan', task: 't', subtasks: [] })
+    s.push({ type: 'dispatch-result', result: okResult })
+    s.close()
+    assert.deepEqual(await a, ['plan', 'dispatch-result'])
+    assert.deepEqual(await b, ['plan', 'dispatch-result'])
+  })
+})
+
+describe('formatEvent', () => {
+  const cases: Array<[SupervisorEvent, RegExp]> = [
+    [{ type: 'plan', task: 'brief', subtasks: [sub] }, /plan: 1 subtask.*brief/],
+    [{ type: 'plan-trimmed', kept: 2, dropped: 3, reason: 'maxSubtasks' }, /trimmed: kept 2, dropped 3/],
+    [{ type: 'dispatch-start', subtask: sub }, /→ subtask-1: do a thing/],
+    [{ type: 'dispatch-result', result: okResult }, /✓ subtask-1/],
+    [
+      { type: 'dispatch-result', result: { ...okResult, ok: false, text: '', error: new Error('boom') } },
+      /✗ subtask-1 \(boom\)/,
+    ],
+    [{ type: 'budget-exceeded', spentTokens: 210, limitTokens: 200, skipped: 2 }, /budget exceeded: 210\/200.*2 skipped/],
+    [{ type: 'synthesize', results: [okResult] }, /synthesize: 1 result/],
+  ]
+  for (const [event, re] of cases) {
+    it(`renders a ${event.type} event`, () => {
+      assert.match(formatEvent(event), re)
+    })
+  }
+})
+
+describe('terminalSink', () => {
+  it('writes a formatted line per event to the provided writer', () => {
+    const lines: string[] = []
+    const sink = terminalSink({ write: l => lines.push(l) })
+    sink({ type: 'plan', task: 't', subtasks: [sub] })
+    sink({ type: 'dispatch-result', result: okResult })
+    assert.equal(lines.length, 2)
+    assert.match(lines[0]!, /plan: 1 subtask/)
+    assert.match(lines[1]!, /✓ subtask-1/)
+  })
+})

packages/ai-autopilot/src/surface/events.ts

@@ -0,0 +1,121 @@
+import type { SupervisorEvent } from '../types.js'
+
+/**
+ * A replayable, multi-consumer stream of {@link SupervisorEvent}s. It is the
+ * transport shared by the in-page and background surfaces: buffer every event,
+ * hand out live async iterators, and replay history from an offset (borrowing
+ * Flue's Durable-Streams `tail=N`). The terminal surface uses {@link terminalSink}
+ * directly and does not need this.
+ */
+export class EventStream {
+  private readonly buffer: SupervisorEvent[] = []
+  private readonly waiters: Array<() => void> = []
+  private closed = false
+
+  /** Append an event. Wire this in as a Supervisor `onEvent`. Ignored once closed. */
+  readonly push = (event: SupervisorEvent): void => {
+    if (this.closed) return
+    this.buffer.push(event)
+    for (const wake of this.waiters.splice(0)) wake()
+  }
+
+  /** Alias for {@link push}, reads well at the `onEvent:` call site. */
+  get sink(): (event: SupervisorEvent) => void {
+    return this.push
+  }
+
+  /** Events buffered so far, from `fromOffset` (default 0) — Flue-style tail replay. */
+  history(fromOffset = 0): SupervisorEvent[] {
+    return this.buffer.slice(fromOffset)
+  }
+
+  /** Number of events buffered. */
+  get length(): number {
+    return this.buffer.length
+  }
+
+  /** True once {@link close} has run. */
+  get isClosed(): boolean {
+    return this.closed
+  }
+
+  /** End the stream: live iterators drain their backlog, then finish. Idempotent. */
+  close(): void {
+    if (this.closed) return
+    this.closed = true
+    for (const wake of this.waiters.splice(0)) wake()
+  }
+
+  /**
+   * A fresh async iterator that replays every buffered event, then yields new
+   * ones as they arrive, and finishes once the stream is closed and drained.
+   * Independent iterators each keep their own cursor, so late consumers still
+   * see the full history.
+   */
+  [Symbol.asyncIterator](): AsyncIterableIterator<SupervisorEvent> {
+    let index = 0
+    const stream = this
+    return {
+      [Symbol.asyncIterator]() {
+        return this
+      },
+      next(): Promise<IteratorResult<SupervisorEvent>> {
+        if (index < stream.buffer.length) {
+          return Promise.resolve({ value: stream.buffer[index++]!, done: false })
+        }
+        if (stream.closed) return Promise.resolve({ value: undefined, done: true })
+        return new Promise(resolve => {
+          stream.waiters.push(() => {
+            if (index < stream.buffer.length) resolve({ value: stream.buffer[index++]!, done: false })
+            else resolve({ value: undefined, done: true })
+          })
+        })
+      },
+    }
+  }
+}
+
+function errorText(error: unknown): string {
+  if (error instanceof Error) return error.message
+  return String(error)
+}
+
+/** Render a {@link SupervisorEvent} as a single human-readable line. */
+export function formatEvent(event: SupervisorEvent): string {
+  switch (event.type) {
+    case 'plan':
+      return `▶ plan: ${event.subtasks.length} subtask(s) for "${event.task}"`
+    case 'plan-trimmed':
+      return `  plan trimmed: kept ${event.kept}, dropped ${event.dropped} (${event.reason})`
+    case 'dispatch-start':
+      return `  → ${event.subtask.id}: ${event.subtask.description}`
+    case 'dispatch-result':
+      return event.result.ok
+        ? `  ✓ ${event.result.subtask.id}`
+        : `  ✗ ${event.result.subtask.id} (${errorText(event.result.error)})`
+    case 'budget-exceeded':
+      return `  ! budget exceeded: ${event.spentTokens}/${event.limitTokens} tokens, ${event.skipped} skipped`
+    case 'synthesize':
+      return `▶ synthesize: ${event.results.length} result(s)`
+  }
+}
+
+/** Options for {@link terminalSink}. */
+export interface TerminalSinkOptions {
+  /** Where to write each formatted line. Default: `process.stdout` (with a newline). */
+  write?: (line: string) => void
+}
+
+/**
+ * The terminal surface: an `onEvent` sink that prints each event as a formatted
+ * line. Pass it as a Supervisor `onEvent` and run inline.
+ *
+ * ```ts
+ * const supervisor = new Supervisor({ ...opts, onEvent: terminalSink() })
+ * await supervisor.run(task)
+ * ```
+ */
+export function terminalSink(opts: TerminalSinkOptions = {}): (event: SupervisorEvent) => void {
+  const write = opts.write ?? ((line: string) => process.stdout.write(line + '\n'))
+  return event => write(formatEvent(event))
+}

packages/ai-autopilot/src/surface/index.ts

@@ -0,0 +1,21 @@
+/**
+ * Surfaces — run the same autopilot in the terminal, in an in-page UI, or as a
+ * background process. All three consume the Supervisor's `onEvent` stream; they
+ * differ only in how events are rendered and whether the run blocks.
+ *
+ * - {@link terminalSink} / {@link formatEvent} — the terminal surface (print inline)
+ * - {@link EventStream} — replayable multi-consumer transport (in-page / background)
+ * - {@link launchAutopilot} — the background surface: a detached {@link AutopilotHandle}
+ */
+export {
+  EventStream,
+  formatEvent,
+  terminalSink,
+  type TerminalSinkOptions,
+} from './events.js'
+export {
+  launchAutopilot,
+  type AutopilotHandle,
+  type AutopilotStatus,
+  type LaunchOptions,
+} from './launch.js'