@studnicky/process-kit

Reducer-with-effects process pattern composing @studnicky/fsm, @studnicky/scheduler, and @studnicky/signal.

Install

pnpm add @studnicky/process-kit

Usage

ProcessKit wraps a caller-supplied StateMachine subclass with an internally-built EffectInterpreter, a SchedulerProviderType (real-time by default, or a VirtualScheduler for deterministic tests), and a Signal for cancellation composition. machine is the only required field — ProcessKit never invents a reducer, only wires one to its supporting primitives:

import type { FsmStepType } from '@studnicky/fsm';

import { VirtualTimeCounter } from '@studnicky/clock';
import { StateMachine, TransitionRejectedError } from '@studnicky/fsm';
import { VirtualScheduler } from '@studnicky/scheduler';
import { Signal } from '@studnicky/signal';
import assert from 'node:assert/strict';

import { ProcessKit } from '../src/index.js';

// --- Domain: a job that starts, self-acknowledges in the same cycle, waits for a scheduled
// advance, then settles. reduce() stays a pure function of (state, event) throughout. ---

type JobState =
  | { readonly 'variant': 'acknowledged' }
  | { readonly 'variant': 'cancelled' }
  | { readonly 'variant': 'completed' }
  | { readonly 'variant': 'idle' }
  | { readonly 'variant': 'waiting' };

type JobEvent =
  | { readonly 'type': 'acknowledge' }
  | { readonly 'type': 'advance' }
  | { readonly 'type': 'cancel' }
  | { readonly 'type': 'start' };

type JobEffect =
  | { readonly 'delayMs': number; readonly 'variant': 'scheduleAdvance' }
  | { readonly 'variant': 'requestAck' };

class JobProcess extends StateMachine<JobState, JobEvent, JobEffect> {
  static make(): JobProcess { return new JobProcess(); }

  getInitialState(): JobState { return { 'variant': 'idle' }; }

  reduce(state: JobState, event: JobEvent): FsmStepType<JobState, JobEffect> {
    if (state.variant === 'idle' && event.type === 'start') {
      return { 'effects': [{ 'variant': 'requestAck' }], 'state': { 'variant': 'waiting' } };
    }
    if (state.variant === 'waiting' && event.type === 'acknowledge') {
      return { 'effects': [{ 'delayMs': 50, 'variant': 'scheduleAdvance' }], 'state': { 'variant': 'acknowledged' } };
    }
    if (state.variant === 'acknowledged' && event.type === 'advance') {
      return { 'effects': [], 'state': { 'variant': 'completed' } };
    }
    if ((state.variant === 'waiting' || state.variant === 'acknowledged') && event.type === 'cancel') {
      return { 'effects': [], 'state': { 'variant': 'cancelled' } };
    }
    throw new TransitionRejectedError({
      'eventType': event.type,
      'reason': `no transition defined for state '${state.variant}'`,
      'stateVariant': state.variant
    });
  }

  // Once settled, further transitions are rejected outright — reduce() is never called.
  protected override isTerminated(state: JobState): boolean {
    return state.variant === 'completed' || state.variant === 'cancelled';
  }
}

// `VirtualScheduler` gives this example a deterministic, fast clock — no real timers.
const counter = VirtualTimeCounter.create({ 'startMs': 0 });
const scheduler = VirtualScheduler.create({ 'counter': counter });

class Kit {
  static make(): ProcessKit<JobState, JobEvent, JobEffect> {
    const kitRef = { 'current': null as unknown as ProcessKit<JobState, JobEvent, JobEffect> };

    const requestAckFn = (_effect: JobEffect, dispatch: (event: JobEvent) => void): void => {
      dispatch({ 'type': 'acknowledge' });
    };

    const scheduleAdvanceFn = (effect: JobEffect): void => {
      kitRef.current.scheduleDispatch(counter.nowMs() + (effect as Extract<JobEffect, { 'variant': 'scheduleAdvance' }>).delayMs, { 'type': 'advance' });
    };

    kitRef.current = ProcessKit.create<JobState, JobEvent, JobEffect>({
      'handlers': {
        'requestAck': requestAckFn,
        'scheduleAdvance': scheduleAdvanceFn
      },
      'machine': JobProcess.make(),
      'scheduler': scheduler
    });

    return kitRef.current;
  }
}

// Cancellation composed via the now-instantiable Signal: an AbortSignal drives a 'cancel'
// event into the composed ProcessKit's public dispatch().
class CancellationWiring {
  static wireCancellation(kit: ProcessKit<JobState, JobEvent, JobEffect>, abortSignal: AbortSignal): void {
    abortSignal.addEventListener('abort', () => {
      void kit.dispatch({ 'type': 'cancel' });
    }, { 'once': true });
  }
}

Transparency contract

ProcessKit introduces no hook of its own — every observable stage is already covered by the primitive it delegates to:

Config key	Accepts	Default
machine	StateMachine subclass instance	required — no default
handlers	EffectHandlerMapType<TEffect, TEvent>	undefined — no effects handled
scheduler	SchedulerProviderType (RealTimeScheduler/VirtualScheduler)	RealTimeScheduler.create()
signal	Signal instance	Signal.create()

Getter	Returns
getMachine()	The composed StateMachine instance
getInterpreter()	The composed EffectInterpreter instance
getScheduler()	The composed SchedulerProviderType instance
getSignal()	The composed Signal instance

Every getter returns the exact instance passed to create()/builder() — never a copy or wrapper. A caller who subclassed StateMachine for its 6 lifecycle hooks keeps full access to those hooks; EffectInterpreter's 9 hooks and the scheduler's own hooks remain reachable through getInterpreter()/getScheduler().

dispatch() vs. the effect-handler dispatch capability

EffectInterpreter's effect handlers receive their own (effect, dispatch) => void capability, whose dispatch(event) enqueues an event at the front of the mailbox and is only ever processed within the same drain cycle that invoked the handler. ProcessKit#dispatch(event) is the public, external entry point and always goes through the interpreter's real send(). ProcessKit#scheduleDispatch(atMs, event) schedules a callback that fires well after any drain cycle has ended, so it correctly calls dispatch()/send(), never the effect-handler capability — see the example above, where the scheduleAdvance effect's handler calls kit.scheduleDispatch(...) rather than the dispatch parameter it was given.

Orchestration-boundary risk flags

ProcessKit sits nearest the Dagonizer boundary of substrate's pattern kits. Three boundaries are enforced by convention, not by a runtime guard:

1. scheduleDispatch chaining — do not nest scheduleDispatch calls that branch on the resulting state to schedule the next step; that is hand-rolling a workflow scheduler. Let a single StateMachine own sequencing as ordinary transitions.
2. Multi-instance registries — do not build a registry/lookup of many named ProcessKit instances dispatched into by name; that is node-placement, which belongs to Dagonizer.
3. Checkpoint/resume creep — stop()/teardown must stay in-memory only; do not add a save/resume pair backed by a store.

See Composition Anti-Patterns and Substrate vs. Dagonizer Boundary for the full rationale.

When to stop using this and move to Dagonizer

ProcessKit drives exactly one process (one machine, one interpreter, one scheduler) through in-memory transitions. It has no concept of a node, a graph, or a dependency between multiple processes. Once a workflow needs to coordinate the outcome of one process to decide whether or how to run another — branching, fan-out across dependent processes, checkpoint/resume, or cross-process retry budgets — that is workflow orchestration and belongs in Dagonizer, not in a hand-rolled registry or chain of ProcessKit instances.

Documentation

Full reference: https://studnicky.github.io/substrate/packages/process-kit

Source on GitHub