diff --git a/packages/engine/src/engine.test.ts b/packages/engine/src/engine.test.ts index ebc1e5f..a43a391 100644 --- a/packages/engine/src/engine.test.ts +++ b/packages/engine/src/engine.test.ts @@ -1,40 +1,21 @@ import { describe, expect, it } from 'vitest'; -import { Engine, type CommandHost, type Message } from './engine.js'; - -/** A minimal command host that runs a `move` and emits `move:done`. */ -function moveHost(): CommandHost & { moves: Message[] } { - const handlers = new Map void>(); - const moves: Message[] = []; - return { - moves, - on(type, handler) { - handlers.set(type, handler); - return () => handlers.delete(type); - }, - dispatch(msg) { - if (msg.type === 'move') { - moves.push(msg); - // The command's completion is itself a message. - this.dispatch({ type: 'move:done', data: msg.data }); - } else if (msg.type === 'move:done') { - handlers.get('move:done')?.(msg); - } - }, - }; -} +import { Engine, type Message } from './engine.js'; describe('Engine', () => { - it('dispatches a message to the command host on tick', () => { - const host = moveHost(); - const engine = new Engine(host); + it('dispatches a message to handlers registered for its type', () => { + const engine = new Engine(); + const seen: Message[] = []; + engine.on('move', (m) => seen.push(m)); engine.enqueue({ type: 'move', data: { part: 'a', to: '/grid/5/5' } }); engine.tick(); - expect(host.moves).toHaveLength(1); + expect(seen).toHaveLength(1); + expect(seen[0]!.data).toEqual({ part: 'a', to: '/grid/5/5' }); }); it('a trigger reacts to a message and emits on the next tick', () => { - const host = moveHost(); - const engine = new Engine(host); + const engine = new Engine(); + const seen: Message[] = []; + engine.on('move', (m) => seen.push(m)); engine.registerTrigger({ type: 'tap', id: 'draw', @@ -44,14 +25,48 @@ describe('Engine', () => { engine.enqueue({ type: 'tap', data: { part: 'carcassonne:tile#a', trigger: 'draw' } }); engine.tick(); // tap processed; move emitted to the next tick - expect(host.moves).toHaveLength(0); + expect(seen).toHaveLength(0); engine.tick(); // move runs - expect(host.moves).toHaveLength(1); + expect(seen).toHaveLength(1); + }); + + it('runCommand runs the command and emits its result message', async () => { + const engine = new Engine(); + const results: Message[] = []; + engine.on('move:done', (m) => results.push(m)); + engine.on('move:error', (m) => results.push(m)); + + engine.runCommand('move', async ({ args }) => { + expect(args).toEqual({ part: 'a', to: '/grid/5/5' }); + return 'moved'; + }); + + engine.enqueue({ type: 'move', data: { part: 'a', to: '/grid/5/5' } }); + engine.tick(); // command runs; result emitted to the next tick + await Promise.resolve(); // flush the command's async resolution + engine.tick(); // result processed + expect(results).toEqual([{ type: 'move:done', data: 'moved' }]); + }); + + it('runCommand emits :error when the command throws', async () => { + const engine = new Engine(); + const results: Message[] = []; + engine.on('move:error', (m) => results.push(m)); + + engine.runCommand('move', async () => { + throw new Error('bad path'); + }); + + engine.enqueue({ type: 'move', data: {} }); + engine.tick(); + await Promise.resolve(); + await Promise.resolve(); // rejection hops through .then before .catch + engine.tick(); + expect(results).toEqual([{ type: 'move:error', error: new Error('bad path') }]); }); it('an orchestrator awaits a matching message and resumes on tick', async () => { - const host = moveHost(); - const engine = new Engine(host); + const engine = new Engine(); const log: string[] = []; const done = engine.runOrchestrator(async (ctx) => { @@ -71,6 +86,27 @@ describe('Engine', () => { expect(log).toEqual(['start', 'tap:carcassonne:tile#a']); // The move emitted by the orchestrator runs on the next tick. engine.tick(); - expect(host.moves).toHaveLength(1); + }); + + it('an orchestrator wait rejects when its signal aborts', async () => { + const engine = new Engine(); + const controller = new AbortController(); + const log: string[] = []; + + const done = engine.runOrchestrator( + async (ctx) => { + try { + await ctx.wait((m) => m.type === 'tap'); + log.push('resolved'); + } catch { + log.push('aborted'); + } + }, + controller.signal, + ); + + controller.abort(); + await done; + expect(log).toEqual(['aborted']); }); }); \ No newline at end of file diff --git a/packages/engine/src/engine.ts b/packages/engine/src/engine.ts index f0480bc..78cfa53 100644 --- a/packages/engine/src/engine.ts +++ b/packages/engine/src/engine.ts @@ -1,53 +1,104 @@ /** - * The engine: the message bus that ties the queue, triggers, orchestrators, - * and the command host together. + * The engine: the message bus that ties the queue, triggers, and handlers + * together. * * The engine is pure — no r3f, no React, no store. It defines the contract; - * `@tts/tabletop` implements the `CommandHost` with the built-in commands that - * mutate the tabletop store and drive the render layer. The engine never - * imports tabletop. + * `@tts/tabletop` registers the built-in commands (`move`, `focus`, `caption`, + * ...) that mutate the tabletop store and drive the render layer. The engine + * never imports tabletop. */ -import { MessageQueue, type Message } from './message.js'; +import { MessageQueue, type CommandResult, type Message, type MessageHandler } from './message.js'; import { TriggerRegistry, type Trigger } from './trigger.js'; -import { runOrchestrator, type Orchestrator } from './orchestrator.js'; +import { runOrchestrator } from './orchestrator.js'; +import { type Command, type Orchestrator, type RunContext } from './run.js'; /** - * The command host: how a command handler registers with the bus. `@tts/tabletop` - * implements this with the built-in commands (`move`, `focus`, `caption`, ...). - * The engine dispatches each message to the host's registered handler for its - * type; the host runs the command (the single mutation path) and emits the - * `type:done` result message. - */ -export interface CommandHost { - /** Register a handler for a message type. Returns an unsubscribe. */ - on(type: string, handler: (msg: Message) => void): () => void; - /** Dispatch a message to the registered handler for its type, if any. */ - dispatch(msg: Message): void; -} - -/** - * The engine. `enqueue` adds a message; `tick` drains the queue and processes - * it through the command host and triggers. Orchestrators run against the same - * queue and suspend on `wait` until a matching message is processed. + * The engine. `enqueue` adds a message; `tick` drains the queue and dispatches + * each message to the handlers registered for its `type`, then to matching + * triggers. Orchestrators run against the same queue and suspend on `wait` + * until a matching message is processed. */ export class Engine { private queue = new MessageQueue(); private triggers = new TriggerRegistry(); - private host: CommandHost; + private handlers = new Map(); - constructor(host: CommandHost) { - this.host = host; - // Every message goes to the command host first (it runs the command), then - // to triggers (they react to the message, including the `type:done` the - // host emits). Emissions from either land on the next tick. + constructor() { + // Every message goes to the handlers for its type (they run commands), then + // to triggers (they react to the message, including the `:done` a command + // emits). Emissions from either land on the next tick. this.queue.on((msg) => { - this.host.dispatch(msg); + for (const handler of this.handlers.get(msg.type) ?? []) handler(msg); for (const t of this.triggers.match(msg)) { for (const emit of t.emit) this.queue.enqueue(emit); } }); } + /** Register a handler for a message type. Returns an unsubscribe. */ + on(type: string, handler: MessageHandler): () => void { + const list = this.handlers.get(type) ?? []; + list.push(handler); + this.handlers.set(type, list); + return () => { + const cur = this.handlers.get(type); + if (!cur) return; + const next = cur.filter((h) => h !== handler); + if (next.length) this.handlers.set(type, next); + else this.handlers.delete(type); + }; + } + + /** + * Register a command. The engine builds a `RunContext` from the message, + * runs the command, and emits its result — `:done` on resolve, `:cancel` on + * abort, `:error` on throw. + */ + runCommand( + type: Name, + command: Command, + ): () => void { + return this.on(type, (msg) => { + const controller = new AbortController(); + const ctx: RunContext = { + signal: controller.signal, + emit: (m) => this.queue.enqueue(m), + wait: (pred) => + new Promise((resolve, reject) => { + if (controller.signal.aborted) { + reject(new Error('aborted')); + return; + } + const off = this.on('*', (m) => { + if (pred(m)) { + off(); + resolve(m); + } + }); + controller.signal.addEventListener('abort', () => { + off(); + reject(new Error('aborted')); + }, { once: true }); + }), + enableTrigger: (type, id) => this.triggers.enable(type, id), + disableTrigger: (type, id) => this.triggers.disable(type, id), + }; + const args = msg.data as Args; + command({ ...ctx, args }) + .then((result) => this.queue.enqueue({ type: `${type}:done`, data: result } as CommandResult)) + .catch((err: unknown) => { + if (controller.signal.aborted) { + this.queue.enqueue({ type: `${type}:cancel` } as CommandResult); + } else { + this.queue.enqueue({ + type: `${type}:error`, + error: err instanceof Error ? err : new Error(String(err)), + } as CommandResult); + } + }); + }); + } + enqueue(msg: Message): void { this.queue.enqueue(msg); } @@ -74,19 +125,20 @@ export class Engine { } /** Run an orchestrator against this engine's queue. */ - runOrchestrator(o: Orchestrator): Promise { + runOrchestrator(o: Orchestrator, signal?: AbortSignal): Promise { return runOrchestrator( o, (msg) => this.queue.enqueue(msg), (handler) => this.queue.on(handler), (type, id) => this.triggers.enable(type, id), (type, id) => this.triggers.disable(type, id), + signal, ); } } -export type { Message, MessageHandler } from './message.js'; +export type { Message, MessageHandler, CommandMessage, CommandResult } from './message.js'; export type { Trigger } from './trigger.js'; export { triggerMatches, TriggerRegistry } from './trigger.js'; -export type { Orchestrator, OrchestratorContext } from './orchestrator.js'; +export type { Orchestrator, RunContext, Command } from './run.js'; export { runOrchestrator } from './orchestrator.js'; \ No newline at end of file diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 99dd286..e4a6b41 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -1,4 +1,11 @@ -export { MessageQueue, type Message, type MessageHandler } from './message.js'; +export { + MessageQueue, + type Message, + type MessageHandler, + type CommandMessage, + type CommandResult, +} from './message.js'; export { TriggerRegistry, triggerMatches, type Trigger } from './trigger.js'; -export { runOrchestrator, type Orchestrator, type OrchestratorContext } from './orchestrator.js'; -export { Engine, type CommandHost } from './engine.js'; \ No newline at end of file +export { runOrchestrator } from './orchestrator.js'; +export type { Orchestrator, RunContext, Command } from './run.js'; +export { Engine } from './engine.js'; \ No newline at end of file diff --git a/packages/engine/src/message.ts b/packages/engine/src/message.ts index 35f058c..9a50356 100644 --- a/packages/engine/src/message.ts +++ b/packages/engine/src/message.ts @@ -2,10 +2,13 @@ * The message model and the queue that serializes it. * * A message is both an event (something happened) and an intent (something - * should happen). It is identified by `type` (and optionally `id`), matching - * the format's `type#id` convention. A `move` message both runs the move - * command and is observable as an event; the command's completion is itself a - * message (`move:done`), which is what triggers match and orchestrators await. + * should happen). It is identified by `type`, matching the format's `type#id` + * convention. A `move` message both runs the move command and is observable as + * an event; the command's completion is itself a message (`move:done`), which + * is what triggers match and orchestrators await. + * + * Messages are a discriminated union on `type`. The engine defines the generic + * shapes; the host's concrete union extends them with its own command types. * * Messages are not processed inline. They are enqueued and handled on the next * `tick()`. This kills reentrancy (a handler cannot cause unbounded @@ -13,12 +16,34 @@ * deterministic frame. The engine is pure — it has no render loop — so the * host calls `tick()` (a `useFrame` in `@tts/tabletop`, manually in tests). */ + +/** A command message names a command to run. Its handler is the command. */ +export interface CommandMessage { + type: Name; + data: Args; +} + +/** + * A command's result, discriminated on the type suffix. `:done` on resolve, + * `:cancel` on abort (superseded, skipped, surface disabled), `:error` on + * throw. A trigger matching `move:done` does not fire on a cancel. + */ +export type CommandResult = + | { type: `${Name}:done`; data: R } + | { type: `${Name}:cancel` } + | { type: `${Name}:error`; error: Error }; + +/** + * The base message type. The engine is host-agnostic, so this is a permissive + * structural type; the host defines a concrete discriminated union on `type` + * that extends it with its own command and interaction messages. + */ export interface Message { type: string; /** Command-specific payload. */ data?: unknown; - /** Optional identity, for matching and dedup. */ - id?: string; + /** Present on `:error` result messages. */ + error?: Error; } /** A handler consumes a message and may emit new ones. */ diff --git a/packages/engine/src/orchestrator.ts b/packages/engine/src/orchestrator.ts index 5a59980..97008f1 100644 --- a/packages/engine/src/orchestrator.ts +++ b/packages/engine/src/orchestrator.ts @@ -3,7 +3,8 @@ * * An orchestrator is the code counterpart to a trigger: an async function that * emits messages and awaits matching ones. It is a proper TS module, declared - * per folder as `main.ts`, unique per folder like `package.yaml`. + * per folder as `main.ts`, unique per folder like `package.yaml`, exported as a + * default async function. * * An orchestrator is a long-running command: it awaits events instead of * resolving immediately, so it inherits the run-context machinery (supersede @@ -11,25 +12,12 @@ * — declaration is data, activation is code. */ import type { Message } from './message.js'; - -/** The context an orchestrator runs against. */ -export interface OrchestratorContext { - /** Emit a message onto the queue. */ - emit(msg: Message): void; - /** Await the next message matching `pred`. */ - wait(pred: (msg: Message) => boolean): Promise; - /** Enable/disable a trigger by `type#id`. */ - enableTrigger(type: string, id?: string): void; - disableTrigger(type: string, id?: string): void; -} - -/** An orchestrator: an async function that emits and awaits messages. */ -export type Orchestrator = (ctx: OrchestratorContext) => Promise; +import type { Orchestrator, RunContext } from './run.js'; /** * Run an orchestrator against a queue. `emit` enqueues; `wait` suspends until - * a matching message is processed during a `tick()`. Returns a promise that - * resolves when the orchestrator completes. + * a matching message is processed during a `tick()`, rejecting on abort. + * Returns a promise that resolves when the orchestrator completes. */ export function runOrchestrator( orchestrator: Orchestrator, @@ -37,8 +25,18 @@ export function runOrchestrator( on: (handler: (msg: Message) => void) => () => void, enableTrigger: (type: string, id?: string) => void, disableTrigger: (type: string, id?: string) => void, + signal?: AbortSignal, ): Promise { - const pending: Array<{ pred: (msg: Message) => boolean; resolve: (m: Message) => void }> = []; + const controller = new AbortController(); + const pending: Array<{ pred: (msg: Message) => boolean; resolve: (m: Message) => void; reject: (e: Error) => void }> = []; + const abort = () => { + controller.abort(); + // Reject every pending wait so the orchestrator unwinds on cancel. + for (const p of pending.splice(0)) p.reject(new Error('aborted')); + }; + if (signal?.aborted) abort(); + else signal?.addEventListener('abort', abort, { once: true }); + const unsubscribe = on((msg) => { for (let i = 0; i < pending.length; i++) { const p = pending[i]!; @@ -50,15 +48,23 @@ export function runOrchestrator( } }); - const ctx: OrchestratorContext = { + const ctx: RunContext = { + signal: controller.signal, emit, wait: (pred) => - new Promise((resolve) => { - pending.push({ pred, resolve }); + new Promise((resolve, reject) => { + if (controller.signal.aborted) { + reject(new Error('aborted')); + return; + } + pending.push({ pred, resolve, reject }); }), enableTrigger, disableTrigger, }; - return orchestrator(ctx).finally(unsubscribe); + return orchestrator(ctx).finally(() => { + unsubscribe(); + controller.abort(); + }); } \ No newline at end of file diff --git a/packages/engine/src/run.ts b/packages/engine/src/run.ts new file mode 100644 index 0000000..b04d451 --- /dev/null +++ b/packages/engine/src/run.ts @@ -0,0 +1,30 @@ +/** + * The run context and the unified command/orchestrator shape. + * + * A command and an orchestrator are the same thing: an async function taking a + * `RunContext`. A command returns a result and is awaited by the engine, which + * emits `:done`/`:cancel`/`:error`; an orchestrator returns `void` and is never + * awaited by a parent. Cancellation is an `AbortSignal` — a superseded command + * or disabled surface aborts it, `wait` rejects on abort, and the `:cancel` + * result is emitted. Errors are thrown: a command that throws emits `:error`. + */ +import type { Message } from './message.js'; + +/** The handle every handler runs against. */ +export interface RunContext { + /** Cancellation: superseded, skipped, surface disabled. */ + signal: AbortSignal; + /** Emit a message onto the queue. */ + emit(msg: Message): void; + /** Await the next message matching `pred`. Rejects on abort. */ + wait(pred: (m: Message) => boolean): Promise; + /** Enable/disable a trigger by `type#id`. */ + enableTrigger(type: string, id?: string): void; + disableTrigger(type: string, id?: string): void; +} + +/** A command: an async function taking the context plus its args. */ +export type Command = (ctx: RunContext & { args: Args }) => Promise; + +/** An orchestrator: an async function that emits and awaits messages. */ +export type Orchestrator = (ctx: RunContext) => Promise; \ No newline at end of file