diff --git a/.changeset/ai-autopilot-generic-surface.md b/.changeset/ai-autopilot-generic-surface.md new file mode 100644 index 0000000..f0138e0 --- /dev/null +++ b/.changeset/ai-autopilot-generic-surface.md @@ -0,0 +1,5 @@ +--- +"@gemstack/ai-autopilot": minor +--- + +Generalize the surface stream to be event-type generic. `EventStream`, `AutopilotHandle`, and `launchAutopilot` 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. diff --git a/packages/ai-autopilot/src/surface/events.test.ts b/packages/ai-autopilot/src/surface/events.test.ts index 30399ea..b45afa1 100644 --- a/packages/ai-autopilot/src/surface/events.test.ts +++ b/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() + 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[] = [] diff --git a/packages/ai-autopilot/src/surface/events.ts b/packages/ai-autopilot/src/surface/events.ts index 0bd86a9..ac0188a 100644 --- a/packages/ai-autopilot/src/surface/events.ts +++ b/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()`). */ -export class EventStream { - private readonly buffer: SupervisorEvent[] = [] +export class EventStream { + 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 { + [Symbol.asyncIterator](): AsyncIterableIterator { let index = 0 const stream = this return { [Symbol.asyncIterator]() { return this }, - next(): Promise> { + next(): Promise> { if (index < stream.buffer.length) { return Promise.resolve({ value: stream.buffer[index++]!, done: false }) } diff --git a/packages/ai-autopilot/src/surface/launch.test.ts b/packages/ai-autopilot/src/surface/launch.test.ts index e9825cb..f456970 100644 --- a/packages/ai-autopilot/src/surface/launch.test.ts +++ b/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(onEvent => { + emit = onEvent + return new Promise(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()) diff --git a/packages/ai-autopilot/src/surface/launch.ts b/packages/ai-autopilot/src/surface/launch.ts index 43cabf0..d27e330 100644 --- a/packages/ai-autopilot/src/surface/launch.ts +++ b/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`. */ -export interface AutopilotHandle { +export interface AutopilotHandle { /** 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 + stream(): AsyncIterableIterator /** Resolves with the run's result, or rejects if the run threw. */ - result(): Promise + result(): Promise } /** 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, +export function launchAutopilot( + start: (onEvent: (event: E) => void) => Promise, opts: LaunchOptions = {}, -): AutopilotHandle { - const stream = new EventStream() +): AutopilotHandle { + const stream = new EventStream() const id = opts.id ?? `autopilot-${++counter}` let state: AutopilotStatus = 'running'