feat(ai-autopilot): generalize surface stream to be event-type generic (#120)

.changeset/ai-autopilot-generic-surface.md

@@ -0,0 +1,5 @@
+---
+"@gemstack/ai-autopilot": minor
+---
+
+Generalize the surface stream to be event-type generic. `EventStream<E>`, `AutopilotHandle<E, R>`, and `launchAutopilot<E, R>` now take the event (and result) type as parameters, defaulting to the supervisor's `SupervisorEvent` / `SupervisorRun` so every existing supervisor surface is unchanged. This lets bootstrap (and any future surface) stream its own narration events and return its own result over the same replayable, detached transport. Closes #120.

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

@@ -83,6 +83,28 @@ describe('formatEvent', () => {
   }
 })
 
+describe('EventStream generic element type', () => {
+  interface BootEvent {
+    type: 'narrate' | 'decision'
+    message: string
+  }
+
+  it('carries a custom event type through push, history, and iteration', async () => {
+    const s = new EventStream<BootEvent>()
+    s.push({ type: 'narrate', message: 'picking the stack' })
+    s.push({ type: 'decision', message: 'Vike + universal-orm' })
+    s.close()
+
+    assert.deepEqual(
+      s.history().map(e => e.message),
+      ['picking the stack', 'Vike + universal-orm'],
+    )
+    const seen: BootEvent[] = []
+    for await (const e of s) seen.push(e)
+    assert.deepEqual(seen.map(e => e.type), ['narrate', 'decision'])
+  })
+})
+
 describe('terminalSink', () => {
   it('writes a formatted line per event to the provided writer', () => {
     const lines: string[] = []

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

@@ -1,31 +1,35 @@
 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.
+ * A replayable, multi-consumer stream of events. 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.
+ *
+ * The element type `E` defaults to {@link SupervisorEvent} so existing supervisor
+ * surfaces are unchanged; bootstrap and any future surface pass their own event
+ * type (`new EventStream<BootstrapEvent>()`).
  */
-export class EventStream {
-  private readonly buffer: SupervisorEvent[] = []
+export class EventStream<E = SupervisorEvent> {
+  private readonly buffer: E[] = []
   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 => {
+  /** Append an event. Wire this in as an `onEvent` sink. Ignored once closed. */
+  readonly push = (event: E): 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 {
+  get sink(): (event: E) => void {
     return this.push
   }
 
   /** Events buffered so far, from `fromOffset` (default 0) — Flue-style tail replay. */
-  history(fromOffset = 0): SupervisorEvent[] {
+  history(fromOffset = 0): E[] {
     return this.buffer.slice(fromOffset)
   }
 
@@ -52,14 +56,14 @@ export class EventStream {
    * Independent iterators each keep their own cursor, so late consumers still
    * see the full history.
    */
-  [Symbol.asyncIterator](): AsyncIterableIterator<SupervisorEvent> {
+  [Symbol.asyncIterator](): AsyncIterableIterator<E> {
     let index = 0
     const stream = this
     return {
       [Symbol.asyncIterator]() {
         return this
       },
-      next(): Promise<IteratorResult<SupervisorEvent>> {
+      next(): Promise<IteratorResult<E>> {
         if (index < stream.buffer.length) {
           return Promise.resolve({ value: stream.buffer[index++]!, done: false })
         }

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

@@ -69,6 +69,36 @@ describe('launchAutopilot', () => {
     assert.deepEqual(seen, [])
   })
 
+  it('carries a custom event and result type (bootstrap-shaped surface)', async () => {
+    interface BootEvent {
+      type: string
+      message: string
+    }
+    interface BootResult {
+      app: string
+      blockers: string[]
+    }
+
+    let emit!: (e: BootEvent) => void
+    let finish!: (r: BootResult) => void
+    const handle = launchAutopilot<BootEvent, BootResult>(onEvent => {
+      emit = onEvent
+      return new Promise<BootResult>(resolve => {
+        finish = resolve
+      })
+    })
+
+    emit({ type: 'narrate', message: 'scaffolding' })
+    assert.deepEqual(
+      handle.events().map(e => e.message),
+      ['scaffolding'],
+    )
+    finish({ app: 'shop', blockers: [] })
+    const result = await handle.result()
+    assert.equal(result.app, 'shop')
+    assert.deepEqual(result.blockers, [])
+  })
+
   it('honors an explicit id and generates unique ids otherwise', async () => {
     const a = launchAutopilot(async () => fakeRun(), { id: 'my-run' })
     const b = launchAutopilot(async () => fakeRun())

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

@@ -11,18 +11,22 @@ export type AutopilotStatus = 'running' | 'done' | 'error'
  * {@link result}. The same handle backs an in-page surface (iterate `stream()`
  * and push each event over SSE) and a background process (persist the handle,
  * let a client poll `status()` + `events(offset)`).
+ *
+ * The type params default to the supervisor's `E`/`R` so supervisor surfaces are
+ * unchanged; bootstrap emits its own narration events and returns its own result,
+ * so it launches with `AutopilotHandle<BootstrapEvent, BootstrapResult>`.
  */
-export interface AutopilotHandle {
+export interface AutopilotHandle<E = SupervisorEvent, R = SupervisorRun> {
   /** Stable id for this run. */
   readonly id: string
   /** Current lifecycle state. */
   status(): AutopilotStatus
   /** Events emitted so far, from `fromOffset` (default 0). */
-  events(fromOffset?: number): SupervisorEvent[]
+  events(fromOffset?: number): E[]
   /** A fresh async iterator over events: replays history, then live, then ends. */
-  stream(): AsyncIterableIterator<SupervisorEvent>
+  stream(): AsyncIterableIterator<E>
   /** Resolves with the run's result, or rejects if the run threw. */
-  result(): Promise<SupervisorRun>
+  result(): Promise<R>
 }
 
 /** Options for {@link launchAutopilot}. */
@@ -40,11 +44,11 @@ let counter = 0
  * Keeping `start` caller-provided means this surface knows nothing about how the
  * Supervisor is built; it only owns the event stream and lifecycle.
  */
-export function launchAutopilot(
-  start: (onEvent: (event: SupervisorEvent) => void) => Promise<SupervisorRun>,
+export function launchAutopilot<E = SupervisorEvent, R = SupervisorRun>(
+  start: (onEvent: (event: E) => void) => Promise<R>,
   opts: LaunchOptions = {},
-): AutopilotHandle {
-  const stream = new EventStream()
+): AutopilotHandle<E, R> {
+  const stream = new EventStream<E>()
   const id = opts.id ?? `autopilot-${++counter}`
   let state: AutopilotStatus = 'running'