From eeea45bf3ecb203124fca2bd8f405dc2b80a88ad Mon Sep 17 00:00:00 2001 From: Devin Ivy Date: Sun, 19 Jul 2026 16:34:26 -0400 Subject: [PATCH] docs: runtime foundations implementation plan --- .../plans/2026-07-19-runtime-foundations.md | 930 ++++++++++++++++++ 1 file changed, 930 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-19-runtime-foundations.md diff --git a/docs/superpowers/plans/2026-07-19-runtime-foundations.md b/docs/superpowers/plans/2026-07-19-runtime-foundations.md new file mode 100644 index 0000000..a8f4c3c --- /dev/null +++ b/docs/superpowers/plans/2026-07-19-runtime-foundations.md @@ -0,0 +1,930 @@ +# Runtime Foundations Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Land the runtime-independent foundations from `docs/superpowers/specs/2026-07-19-runtime-design.md`: the `define()` primitive, the reshaped `WorkerHandle` (add `index`, shared active counts, drop `thread`), the new `Balancer` interface with `initialState()`, `leastBusy` as default, and generalizing dispatch from `Worker` to any port-like endpoint. + +**Architecture:** Each piece is standalone and useful without the global runtime (which is a follow-up plan: registerRuntime, ambient handle, broker mesh, peer dispatch, dedicated-worker removal). `define()` mirrors `mo()`: module-scope registration, `${url}#${index}` IDs, resolution-by-import on the receiving thread, task-arg-like wire form. Active counts move from a main-thread `Map` to a padded `Int32Array` on a `SharedArrayBuffer` so any thread can read/update them later. Balancer state moves from closures to an `initialState()` value that can contain shared memory. + +**Tech Stack:** Node 24 (native TS, `using`), `node:test`, `node:worker_threads`, pnpm. Run `nvm use 24` before any test command. Tests import from `'moroutine'` (package self-reference to `src/index.ts`). + +**This is a semver-major.** Breaking: default balancer round-robin → `leastBusy()`; `Balancer.select` gains a third `state` arg (plain closure balancers remain assignable); `WorkerHandle.thread` removed, `WorkerHandle.index` added. + +--- + +### Task 1: `define()` core + +`define()` tags an object with deterministic module identity (same ID scheme as `mo()`), registers it in a define-registry, and returns the same reference. Values must be objects/functions (primitives serialize fine on their own and cannot be branded). + +**Files:** +- Create: `src/define.ts` +- Modify: `src/index.ts` +- Create: `test/define.test.ts` +- Create: `test/fixtures/define.ts` + +- [ ] **Step 1: Write the failing unit tests** + +Create `test/fixtures/define.ts`: + +```ts +import { mo, define } from 'moroutine'; + +export const limits = define(import.meta, { maxBatch: 64, retry: 3 }); + +// A function inside a defined value cannot cross threads by serialization — +// if a worker can call it, resolution happened by module evaluation. +export const helpers = define(import.meta, { double: (x: number) => x * 2 }); + +export const readLimits = mo(import.meta, (l: { maxBatch: number; retry: number }): number => { + return l.maxBatch + l.retry; +}); + +export const useHelpers = mo(import.meta, (h: { double: (x: number) => number }): number => { + return h.double(21); +}); + +export function lateDefine(): void { + define(import.meta, { late: true }); +} +``` + +Create `test/define.test.ts`: + +```ts +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { define } from 'moroutine'; +import { limits } from './fixtures/define.ts'; + +describe('define()', () => { + it('returns the same object reference', () => { + assert.equal(limits.maxBatch, 64); + assert.equal(limits.retry, 3); + }); + + it('throws on primitive values', () => { + assert.throws(() => define(import.meta, 42 as any), TypeError); + assert.throws(() => define(import.meta, 'nope' as any), TypeError); + assert.throws(() => define(import.meta, null as any), TypeError); + }); +}); +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `node --test test/define.test.ts` +Expected: FAIL — `'moroutine'` has no export named `define`. + +- [ ] **Step 3: Implement `src/define.ts`** + +```ts +import { isModuleFrozen } from './registry.ts'; + +const DEFINE_BRAND = Symbol.for('moroutine.define'); + +const counters = new Map(); + +/** Registry of defined values by id, populated by module evaluation on each thread. */ +export const defineRegistry = new Map(); + +/** + * Tags a value with deterministic module identity so it can cross threads by + * reference and be resolved on the receiving thread by evaluating its module — + * the same resolution model as task-args, but for plain values. + * + * Must be called at module scope in a side-effect-free module. The value is + * produced by evaluating the module on each thread, so non-shared mutable state + * inside it is per-thread; cross-thread state must use shared memory. + * + * @param importMeta - The `import.meta` of the calling module. + * @param value - The value to define. Must be an object or function. + * @returns The same value, branded with its module identity. + */ +export function define(importMeta: ImportMeta, value: T): T { + const url = importMeta.url; + + if (isModuleFrozen(url)) { + throw new Error( + `Cannot call define() for ${url} after a task or defined value from this module has been dispatched. ` + + 'All define() calls must happen at module load time.', + ); + } + + if (value === null || (typeof value !== 'object' && typeof value !== 'function')) { + throw new TypeError('define() requires an object or function value'); + } + + const index = counters.get(url) ?? 0; + counters.set(url, index + 1); + const id = `${url}#${index}`; + + Object.defineProperty(value, DEFINE_BRAND, { value: id }); + defineRegistry.set(id, value); + return value; +} + +/** Returns the define id of a branded value, or undefined. */ +export function getDefineId(value: unknown): string | undefined { + if (value === null || (typeof value !== 'object' && typeof value !== 'function')) return undefined; + return (value as Record)[DEFINE_BRAND]; +} +``` + +Note: `define()` and `mo()` keep separate counters and registries, so their IDs may collide textually — that is fine because the wire forms are distinct and lookups hit different registries. They share the same module-freeze guard. + +- [ ] **Step 4: Export from `src/index.ts`** + +Add after the `mo` exports at the top: + +```ts +export { define } from './define.ts'; +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `node --test test/define.test.ts` +Expected: PASS (2 tests). + +- [ ] **Step 6: Commit** + +```bash +git add src/define.ts src/index.ts test/define.test.ts test/fixtures/define.ts +git commit -m "feat: define() module-identified values" +``` + +--- + +### Task 2: `define()` values as task args (cross-thread resolution) + +Defined values passed as moroutine args travel as `{ __define__: id }` and resolve on the worker by importing the module and reading the define-registry — mirroring how task-args resolve by execution. Sending a define ref freezes its module (ID determinism), same as dispatching a task does. + +**Files:** +- Modify: `src/execute.ts` (prepareArg) +- Modify: `src/worker-entry.ts` (needsAsyncResolve, resolveArg, sync deserialize paths) +- Modify: `test/define.test.ts` + +- [ ] **Step 1: Write the failing integration tests** + +Append to `test/define.test.ts`: + +```ts +import { workers } from 'moroutine'; +import { helpers, readLimits, useHelpers, lateDefine } from './fixtures/define.ts'; + +describe('define() as task args', () => { + it('resolves a defined value on a pool worker', async () => { + using run = workers(1); + const result = await run(readLimits(limits)); + assert.equal(result, 67); + }); + + it('resolves by module evaluation, not serialization (functions cross)', async () => { + using run = workers(1); + const result = await run(useHelpers(helpers)); + assert.equal(result, 42); + }); + + it('resolves on a dedicated worker via bare await', async () => { + const result = await useHelpers(helpers); + assert.equal(result, 42); + }); + + it('resolves repeatedly on the same worker (cached path)', async () => { + using run = workers(1); + const a = await run(readLimits(limits)); + const b = await run(readLimits(limits)); + assert.equal(a, 67); + assert.equal(b, 67); + }); + + it('freezes the module once a defined value has been dispatched', async () => { + using run = workers(1); + await run(readLimits(limits)); + assert.throws(lateDefine, { + message: /Cannot call define\(\).*after a task or defined value from this module has been dispatched/, + }); + }); +}); +``` + +Merge the imports with the existing ones at the top of the file (single import statement per module — `limits` is already imported). + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `node --test test/define.test.ts` +Expected: The new tests FAIL. Without ref-detection the branded object is structured-cloned: `readLimits` may pass incidentally, but `useHelpers` FAILS (functions cannot be cloned — DataCloneError), and the freeze test FAILS (no freeze on dispatch). + +- [ ] **Step 3: Send refs in `src/execute.ts`** + +Add to the imports: + +```ts +import { getDefineId } from './define.ts'; +``` + +In `prepareArg`, add the define check as the FIRST branch (a defined value is user-branded — it must win over structural detection like async-generator sniffing): + +```ts +function prepareArg(arg: unknown): unknown { + // Defined values cross by reference; the worker resolves via module import. + const defineId = getDefineId(arg); + if (defineId !== undefined) { + freezeModule(defineId.slice(0, defineId.lastIndexOf('#'))); + return { __define__: defineId }; + } + // ... existing branches unchanged (AbortSignal, AsyncGenerator, AsyncIterableTask, Channel, PromiseLikeTask, serializeArg) +``` + +`freezeModule` is already imported in `execute.ts`. + +- [ ] **Step 4: Resolve refs in `src/worker-entry.ts`** + +Add to the imports: + +```ts +import { defineRegistry } from './define.ts'; +``` + +Add alongside `isTaskArg`: + +```ts +function isDefinedRef(arg: unknown): arg is { __define__: string } { + return typeof arg === 'object' && arg !== null && '__define__' in arg; +} +``` + +Update `needsAsyncResolve` — a define ref only needs async resolution until its module has been imported: + +```ts +function needsAsyncResolve(arg: unknown): boolean { + if (isDefinedRef(arg)) return !defineRegistry.has(arg.__define__); + return arg instanceof MessagePort || isSerializedStreamHandle(arg) || isTaskArg(arg); +} +``` + +Add a sync-path wrapper and use it EVERYWHERE the sync loops currently call `deserializeArg(...)` on a task argument (the sync prologue in `invokeWithArgs`, the async tail in `invokeWithArgs`, the arg loop in `resolveArg`'s task-arg branch, and the arg loop in `handleStreamTask`): + +```ts +function deserializeLocal(arg: unknown): unknown { + if (isDefinedRef(arg)) return defineRegistry.get(arg.__define__); + return deserializeArg(arg); +} +``` + +For example, the sync prologue in `invokeWithArgs` becomes: + +```ts + resolved[i] = deserializeLocal(arg); +``` + +and the async tail / other loops become: + +```ts + resolved[i] = needsAsyncResolve(arg) ? await resolveArg(arg) : deserializeLocal(arg); +``` + +(Do NOT change the stream-value path in `portToAsyncIterable` — define refs only appear in task args.) + +Add the async branch at the top of `resolveArg`: + +```ts +async function resolveArg(arg: unknown): Promise { + if (isDefinedRef(arg)) { + const id = arg.__define__; + if (!defineRegistry.has(id)) { + const url = id.slice(0, id.lastIndexOf('#')); + await import(url); + if (!defineRegistry.has(id)) throw new Error(`Defined value not found: ${id}`); + } + return defineRegistry.get(id); + } + // ... existing branches unchanged +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `node --test test/define.test.ts` +Expected: PASS (all 7 tests). + +- [ ] **Step 6: Run the full suite for regressions** + +Run: `pnpm test` +Expected: PASS across the board (no existing arg path is affected — define refs are a new wire shape). + +- [ ] **Step 7: Commit** + +```bash +git add src/execute.ts src/worker-entry.ts test/define.test.ts test/fixtures/define.ts +git commit -m "feat: defined values as task args, resolved by module import" +``` + +--- + +### Task 3: WorkerHandle reshape — `index`, shared active counts, drop `thread` + +Active counts move from a main-thread `Map` to a SharedArrayBuffer-backed `Int32Array` with 64-byte stride (false-sharing padding), so future peer threads can read and update them. `WorkerHandle` gains `index` and loses `thread` (its only consumers were two test assertions; the raw `Worker` can be exposed on a main-only surface later if a need appears). + +**Files:** +- Create: `src/active-counts.ts` +- Create: `test/active-counts.test.ts` +- Modify: `src/runner.ts:43-51` (WorkerHandle interface) +- Modify: `src/worker-pool.ts` (counts, handle construction) +- Modify: `test/load-balancing.test.ts:39-47` (thread test → index test) +- Modify: `test/balancers.test.ts:6-8` (mockHandle) + +- [ ] **Step 1: Write the failing unit test for ActiveCounts** + +Create `test/active-counts.test.ts` (imports relatively — `ActiveCounts` is internal, not a package export): + +```ts +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { ActiveCounts } from '../src/active-counts.ts'; + +describe('ActiveCounts', () => { + it('starts at zero and tracks inc/dec per index', () => { + const counts = new ActiveCounts(4); + assert.equal(counts.get(0), 0); + counts.inc(0); + counts.inc(0); + counts.inc(3); + assert.equal(counts.get(0), 2); + assert.equal(counts.get(3), 1); + counts.dec(0); + assert.equal(counts.get(0), 1); + }); + + it('is backed by a SharedArrayBuffer and reconstructs over it', () => { + const counts = new ActiveCounts(2); + counts.inc(1); + const other = new ActiveCounts(counts.buffer); + assert.equal(other.get(1), 1); + other.inc(1); + assert.equal(counts.get(1), 2); + }); + + it('pads slots to 64-byte stride', () => { + const counts = new ActiveCounts(2); + assert.equal(counts.buffer.byteLength, 2 * 64); + }); +}); +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `node --test test/active-counts.test.ts` +Expected: FAIL — module not found. + +- [ ] **Step 3: Implement `src/active-counts.ts`** + +```ts +/** Ints per slot: 64-byte stride pads each counter to its own cache line, + * avoiding false sharing between adjacent workers' counters. */ +const STRIDE = 16; + +/** + * Per-worker in-flight task counters on a SharedArrayBuffer, so any thread + * can read (`leastBusy`) or update (dispatching thread) them. + */ +export class ActiveCounts { + readonly #view: Int32Array; + + constructor(sizeOrBuffer: number | SharedArrayBuffer) { + this.#view = new Int32Array( + typeof sizeOrBuffer === 'number' ? new SharedArrayBuffer(sizeOrBuffer * STRIDE * 4) : sizeOrBuffer, + ); + } + + get buffer(): SharedArrayBuffer { + return this.#view.buffer; + } + + inc(index: number): void { + Atomics.add(this.#view, index * STRIDE, 1); + } + + dec(index: number): void { + Atomics.sub(this.#view, index * STRIDE, 1); + } + + get(index: number): number { + return Atomics.load(this.#view, index * STRIDE); + } +} +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `node --test test/active-counts.test.ts` +Expected: PASS (3 tests). + +- [ ] **Step 5: Update the failing handle tests** + +In `test/load-balancing.test.ts`, replace the `exposes thread on WorkerHandle` test (lines 39–47) with: + +```ts + it('exposes index on WorkerHandle', () => { + const run = workers(2); + try { + assert.equal(run.workers[0].index, 0); + assert.equal(run.workers[1].index, 1); + } finally { + run[Symbol.dispose](); + } + }); +``` + +In `test/balancers.test.ts`, update `mockHandle`: + +```ts +function mockHandle(activeCount: number, index = 0): WorkerHandle { + return { activeCount, index, exec: {} as any }; +} +``` + +- [ ] **Step 6: Run tests to verify they fail** + +Run: `node --test test/load-balancing.test.ts test/balancers.test.ts` +Expected: load-balancing FAILS (`index` is undefined); balancers may pass (type-only change) — that is fine. + +- [ ] **Step 7: Reshape the interface and pool** + +In `src/runner.ts`, replace the `WorkerHandle` interface (lines 43–51): + +```ts +/** A handle to a specific worker in a pool. Uniform across threads. */ +export interface WorkerHandle { + /** Dispatches a task pinned to this worker. Returns `AsyncIterable` for streaming tasks, `Promise` otherwise. */ + exec(task: Task, opts?: ChannelOptions): RunResult; + /** Position of this worker in the pool. Stable across threads; the portable identity used by `assign()`. */ + readonly index: number; + /** Number of currently in-flight tasks on this worker. */ + readonly activeCount: number; +} +``` + +Remove the now-unused `import type { Worker } from 'node:worker_threads';` at the top of `runner.ts`. + +In `src/worker-pool.ts`: + +1. Add import: `import { ActiveCounts } from './active-counts.ts';` +2. Replace `const activeCounts = new Map();` with `const counts = new ActiveCounts(size);` +3. Change `trackValue` and `trackStream` to take an index instead of a handle: + +```ts + async function trackValue(index: number, promise: Promise): Promise { + inflight.add(promise); + counts.inc(index); + try { + return await promise; + } catch (err) { + if (err instanceof Error) { + const Ctor = err.constructor as ErrorConstructor; + throw new Ctor(err.message, { cause: err }); + } + throw new Error(String(err), { cause: err }); + } finally { + inflight.delete(promise); + counts.dec(index); + } + } + + function trackStream(index: number, done: Promise): void { + inflight.add(done); + counts.inc(index); + done.then(() => { + inflight.delete(done); + counts.dec(index); + }); + } +``` + +4. `resolveWorkerAndHandle` returns the index too: + +```ts + function resolveWorkerAndHandle(task: Task): { worker: Worker; handle: WorkerHandle; idx: number } { + if (task.worker != null) { + const idx = workerHandles.indexOf(task.worker); + if (idx !== -1) return { worker: pool[idx], handle: workerHandles[idx], idx }; + } + const handle = balancer.select(workerHandles, task); + const idx = workerHandles.indexOf(handle); + return { worker: pool[idx], handle, idx }; + } +``` + +5. `dispatch` uses the index: + +```ts + function dispatch(task: Task): Promise { + if (disposed) return Promise.reject(new Error('Worker pool is disposed')); + const { worker, idx } = resolveWorkerAndHandle(task); + return trackValue(idx, execute(worker, task.id, task.args)); + } +``` + +6. `makeWorkerHandle` drops `thread`, gains `index`, reads counts: + +```ts + function makeWorkerHandle(worker: Worker, idx: number): WorkerHandle { + return { + exec(task: Task, channelOpts?: ChannelOptions): any { + if (task instanceof AsyncIterableTask) { + if (disposed) throw new Error('Worker pool is disposed'); + const { iterable, done } = dispatchStream(worker, task.id, task.args, channelOpts); + trackStream(idx, done); + return iterable; + } + if (disposed) return Promise.reject(new Error('Worker pool is disposed')); + return trackValue(idx, execute(worker, task.id, task.args)); + }, + get index() { + return idx; + }, + get activeCount() { + return counts.get(idx); + }, + }; + } +``` + +7. In the streaming branch of the `run` callable, use the returned `idx` with `trackStream(idx, done)` (replacing `trackStream(handle, done)`), and destructure `idx` from `resolveWorkerAndHandle`. + +- [ ] **Step 8: Run the full suite** + +Run: `pnpm test` +Expected: PASS. `test/worker-handle.test.ts` and `test/serve/**` exercise handles heavily — they must be green. + +Run: `pnpm types` +Expected: clean — this catches any lingering `.thread` reference. + +- [ ] **Step 9: Commit** + +```bash +git add src/active-counts.ts src/runner.ts src/worker-pool.ts test/active-counts.test.ts test/load-balancing.test.ts test/balancers.test.ts +git commit -m "feat!: WorkerHandle index + shared active counts, drop thread" +``` + +--- + +### Task 4: Balancer `initialState()` + shared-state `roundRobin` + `leastBusy` default + +`Balancer` gains an optional `initialState()` returning concrete state (may contain shared memory); `select` receives it as a third argument. `roundRobin` moves its cursor from a closure to a `Uint32Atomic` so it will be correct when scheduling later happens on multiple threads. `leastBusy` (which reads shared active counts via handles and needs no own state) becomes the default everywhere. + +**Files:** +- Modify: `src/runner.ts:25-41` (Balancer, WorkerOptions doc) +- Modify: `src/balancers.ts` +- Modify: `src/worker-pool.ts:33,101` (default + state wiring) +- Modify: `test/balancers.test.ts` +- Modify: `test/load-balancing.test.ts:8-16` + +- [ ] **Step 1: Update the balancer tests to the new contract (failing first)** + +In `test/balancers.test.ts`, replace the `roundRobin()` describe block: + +```ts +describe('roundRobin()', () => { + it('cycles through workers in order', () => { + const b = roundRobin(); + const state = b.initialState!(); + const handles = [mockHandle(0, 0), mockHandle(0, 1), mockHandle(0, 2)]; + const task = { id: 'test', args: [], uid: 0 } as any; + assert.equal(b.select(handles, task, state), handles[0]); + assert.equal(b.select(handles, task, state), handles[1]); + assert.equal(b.select(handles, task, state), handles[2]); + assert.equal(b.select(handles, task, state), handles[0]); + }); + + it('state is shared-memory backed (two instances over the same state agree)', () => { + const b1 = roundRobin(); + const state = b1.initialState!(); + const b2 = roundRobin(); + const handles = [mockHandle(0, 0), mockHandle(0, 1)]; + const task = { id: 'test', args: [], uid: 0 } as any; + assert.equal(b1.select(handles, task, state), handles[0]); + assert.equal(b2.select(handles, task, state), handles[1]); + }); +}); +``` + +Update the `leastBusy()` describe block's select calls to pass the (undefined) state explicitly: + +```ts + assert.equal(b.select(handles, task, undefined), handles[1]); +``` + +(and likewise in the tie-break test). + +In `test/load-balancing.test.ts`, rename the first test: + +```ts + it('defaults to least-busy', async () => { + const run = workers(2); + try { + const result = await run(identity(42)); + assert.equal(result, 42); + } finally { + run[Symbol.dispose](); + } + }); +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `node --test test/balancers.test.ts` +Expected: FAIL — `b.initialState` is not a function. + +- [ ] **Step 3: Update the `Balancer` interface in `src/runner.ts`** + +Replace the `Balancer` interface (lines 25–33): + +```ts +/** A load balancing strategy for choosing which worker runs a task. */ +export interface Balancer { + /** Produces the balancer's state once at pool/runtime boot. May contain + * shared memory (which crosses threads by reference); plain values ride + * along by serialization. Closure state is safe only where scheduling is + * single-threaded (a `workers()` pool); shared, declared state works everywhere. */ + initialState?(): S; + /** Choose a worker for the given task. Called synchronously on every dispatch. */ + select(workers: readonly WorkerHandle[], task: Task, state: S): WorkerHandle; + /** Optional cleanup on sync dispose. */ + [Symbol.dispose]?(): void; + /** Optional cleanup on async dispose. */ + [Symbol.asyncDispose]?(): Promise; +} +``` + +Update `WorkerOptions`: + +```ts +/** Options for configuring a worker pool. */ +export interface WorkerOptions { + /** Maximum time in ms to wait for in-flight tasks during async dispose. If exceeded, workers are force-terminated. */ + shutdownTimeout?: number; + /** Load balancing strategy. Defaults to least-busy. */ + balance?: Balancer; +} +``` + +- [ ] **Step 4: Rework `src/balancers.ts`** + +```ts +import { uint32atomic } from './shared/descriptors.ts'; +import type { Uint32Atomic } from './shared/uint32-atomic.ts'; +import type { Balancer, WorkerHandle } from './runner.ts'; + +/** + * Creates a round-robin balancer that cycles through workers in order. + * The cursor lives in shared memory so all scheduling threads advance the + * same sequence. (The uint32 cursor wraps at 2^32; a wrap may skip one + * position, which round-robin semantics tolerate.) + * @returns A fresh Balancer instance. + */ +export function roundRobin(): Balancer<{ cursor: Uint32Atomic }> { + return { + initialState: () => ({ cursor: uint32atomic() }), + select(workers: readonly WorkerHandle[], _task, { cursor }): WorkerHandle { + return workers[cursor.add(1) % workers.length]; + }, + }; +} + +/** + * Creates a least-busy balancer that picks the worker with the lowest activeCount. + * Ties are broken by index (first wins). Reads the pool's shared active counts + * via the handles, so it needs no state of its own. + * @returns A fresh Balancer instance. + */ +export function leastBusy(): Balancer { + return { + select(workers: readonly WorkerHandle[]): WorkerHandle { + let best = workers[0]; + for (let i = 1; i < workers.length; i++) { + if (workers[i].activeCount < best.activeCount) { + best = workers[i]; + } + } + return best; + }, + }; +} +``` + +(`cursor.add(1)` returns the previous value, matching the old `next++` order: the first pick is index 0.) + +- [ ] **Step 5: Wire state through `src/worker-pool.ts`** + +Change the import and default (line 6 and 33): + +```ts +import { leastBusy } from './balancers.ts'; +``` + +```ts + const balancer: Balancer = opts?.balance ?? leastBusy(); + const balancerState: unknown = balancer.initialState?.(); +``` + +Change the select call in `resolveWorkerAndHandle`: + +```ts + const handle = balancer.select(workerHandles, task, balancerState); +``` + +- [ ] **Step 6: Run tests to verify they pass** + +Run: `node --test test/balancers.test.ts test/load-balancing.test.ts` +Expected: PASS. + +- [ ] **Step 7: Run the full suite** + +Run: `pnpm test` +Expected: PASS. Watch specifically for tests that implicitly assumed round-robin dispatch order (e.g. anything asserting which worker got a task in `test/pool.test.ts`, `test/serve/`); with `leastBusy` ties break to worker 0, so sequential awaited dispatches all land on worker 0. If a test fails for this reason, pin the old behavior in that test by passing `{ balance: roundRobin() }` explicitly — do not weaken assertions. + +Run: `pnpm types` +Expected: clean. + +- [ ] **Step 8: Commit** + +```bash +git add src/runner.ts src/balancers.ts src/worker-pool.ts test/balancers.test.ts test/load-balancing.test.ts +git commit -m "feat!: Balancer initialState(), shared-state roundRobin, leastBusy default" +``` + +--- + +### Task 5: Generalize dispatch to port-like endpoints + +Pure refactor, no behavior change: `setupWorker`/`execute`/`dispatchStream` accept any endpoint with `postMessage`/`on` (both `Worker` and `MessagePort` qualify) instead of `Worker`. This is the groundwork the follow-up runtime plan builds peer dispatch on. + +**Files:** +- Modify: `src/execute.ts:1,20,79,99-104` +- Modify: `src/dedicated-runner.ts` (type only, if needed) + +- [ ] **Step 1: Introduce the Endpoint type and generalize signatures** + +In `src/execute.ts`, replace the Worker-typed imports and signatures: + +```ts +import { MessageChannel } from 'node:worker_threads'; +import type { MessagePort, Transferable } from 'node:worker_threads'; +``` + +Add near the top: + +```ts +/** Anything dispatch can talk over: a Worker from the main thread, or a + * MessagePort between peers. Structurally satisfied by both. */ +export interface Endpoint { + postMessage(value: unknown, transferList?: readonly Transferable[]): void; + on(event: 'message', listener: (value: any) => void): unknown; +} +``` + +Change the three signatures (bodies unchanged): + +```ts +export function setupWorker(endpoint: Endpoint): void { +``` + +```ts +export function execute(endpoint: Endpoint, id: string, args: unknown[]): Promise { +``` + +```ts +export function dispatchStream( + endpoint: Endpoint, + id: string, + args: unknown[], + opts?: ChannelOptions, +): StreamDispatch { +``` + +Rename the parameter uses inside from `worker` to `endpoint` (`endpoint.postMessage(...)`, `endpoint.on(...)`). + +- [ ] **Step 2: Verify types and behavior** + +Run: `pnpm types` +Expected: clean — `Worker` satisfies `Endpoint` structurally, so `worker-pool.ts` and `dedicated-runner.ts` compile unchanged. If `postMessage` overloads clash, loosen the `transferList` parameter to `readonly any[]` to match Node's signature. + +Run: `pnpm test` +Expected: PASS — no behavior change. + +- [ ] **Step 3: Commit** + +```bash +git add src/execute.ts +git commit -m "refactor: generalize dispatch from Worker to port-like Endpoint" +``` + +--- + +### Task 6: Docs and changeset + +**Files:** +- Modify: `README.md` (Core API, Load Balancing sections) +- Create: `.changeset/runtime-foundations.md` + +- [ ] **Step 1: README — document `define()`** + +Add to the Core API section of `README.md`, after the `workers()` subsection: + +````markdown +### `define(import.meta, value)` + +Tags an object with module identity so it crosses threads **by reference**: workers resolve +it by evaluating its module, like moroutine functions themselves. Useful for config objects, +lookup tables, and values containing non-serializable members (functions). + +```ts +// config.ts +import { define } from 'moroutine'; + +export const limits = define(import.meta, { maxBatch: 64, retry: 3 }); +``` + +Pass it to any moroutine — the reference crosses, and the worker resolves its own copy from +the module: + +```ts +await run(processBatch(items, limits)); +``` + +Since the value is produced by evaluating the module on each thread, non-shared mutable +state inside it is per-thread. Cross-thread state must use shared memory types. +```` + +- [ ] **Step 2: README — update the Load Balancing section** + +- Change "The pool uses round-robin scheduling by default" to "The pool uses least-busy scheduling by default. Pass a `balance` option to change the strategy:" and swap the example to `{ balance: roundRobin() }`. +- Update the built-ins list: `leastBusy()` first, marked `(default)`; note `roundRobin()` keeps its cursor in shared memory. +- Update the custom `Balancer` example to the new signature and mention `initialState()`: + +````markdown +Custom balancers implement the `Balancer` interface. Stateless strategies just implement +`select`: + +```ts +import type { Balancer } from 'moroutine'; + +const random: Balancer = { + select(workers, task) { + return workers[Math.floor(Math.random() * workers.length)]; + }, +}; +``` + +Stateful strategies produce their state once via `initialState()`, and `select` receives it +as a third argument. Put cross-thread state (counters, cursors) in shared memory types: + +```ts +import { uint32atomic } from 'moroutine'; +import type { Balancer, Uint32Atomic } from 'moroutine'; + +const everyOther: Balancer<{ cursor: Uint32Atomic }> = { + initialState: () => ({ cursor: uint32atomic() }), + select(workers, task, { cursor }) { + return workers[(cursor.add(1) * 2) % workers.length]; + }, +}; +``` +```` + +- Update the `WorkerHandle` sentence: "Each `WorkerHandle` exposes `index` (its position in the pool) and `activeCount` (in-flight tasks) for building custom strategies." (removing the `thread` mention). + +- [ ] **Step 3: Create the changeset** + +Create `.changeset/runtime-foundations.md`: + +```markdown +--- +'moroutine': major +--- + +Runtime foundations: + +- New `define()`: module-identified values that cross threads by reference and resolve per + thread via module evaluation; supported as task args. +- **Breaking:** default balancer is now `leastBusy()` (was round-robin). +- **Breaking:** `Balancer` gains optional `initialState()` and `select` receives the state + as a third argument. Plain closure balancers remain valid for `workers()` pools. +- **Breaking:** `WorkerHandle.thread` removed; `WorkerHandle.index` added. Active counts + are SharedArrayBuffer-backed. +``` + +- [ ] **Step 4: Verify and commit** + +Run: `pnpm test && pnpm types && pnpm lint` +Expected: all clean. If prettier complains, run `pnpm lint:fix`. + +```bash +git add README.md .changeset/runtime-foundations.md +git commit -m "docs: define(), balancer state, leastBusy default" +``` + +--- + +## Deferred to the follow-up runtime plan + +Everything requiring the global runtime: `registerRuntime()`, the internal default runtime module, the ambient `runtime` handle, boot handshake (`workerData` with define ID + counts buffer + serialized balancer state), lazy brokered peer mesh, peer dispatch + same-worker shortcut, bare-await retargeting, dedicated-runner removal, unref-based teardown, `runtime.shutdown()`, and their tests. Write that plan after this one lands. -- knotmirror 0.0.1