@studnicky/process-kit

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

Install

pnpm add @studnicky/process-kit

Usage

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

import type { EffectHandlerInterface, FsmStepInterface } 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 type { JobEffectEntity } from './entities/JobEffectEntity.js';
import type { JobEventEntity } from './entities/JobEventEntity.js';
import type { JobStateEntity } from './entities/JobStateEntity.js';

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. ---

class JobProcess extends StateMachine<JobStateEntity.Type, JobEventEntity.Type, JobEffectEntity.Type> {
  readonly #stateWaiters = new Map<JobStateEntity.Type['variant'], Set<() => void>>();
  currentState: JobStateEntity.Type = { 'variant': 'idle' };

  static make(): JobProcess { return new JobProcess(); }

  getInitialState(): JobStateEntity.Type { return { 'variant': 'idle' }; }

  reduce(
    state: JobStateEntity.Type,
    event: JobEventEntity.Type
  ): FsmStepInterface<JobStateEntity.Type, JobEffectEntity.Type> {
    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
    });
  }

  waitForState(variant: JobStateEntity.Type['variant']): Promise<void> {
    if (this.currentState.variant === variant) {
      const result = Promise.resolve();
      return result;
    }

    return new Promise<void>((resolve) => {
      const waiters = this.#stateWaiters.get(variant) ?? new Set<() => void>();
      waiters.add(resolve);
      this.#stateWaiters.set(variant, waiters);
    });
  }

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

  protected override onEnterState(state: JobStateEntity.Type): void {
    this.currentState = state;
    const waiters = this.#stateWaiters.get(state.variant);
    if (waiters === undefined) {
      return;
    }
    this.#stateWaiters.delete(state.variant);
    for (const resolve of waiters) {
      resolve();
    }
  }
}

// `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(): {
    readonly 'kit': ProcessKit<JobStateEntity.Type, JobEventEntity.Type, JobEffectEntity.Type>;
    readonly 'machine': JobProcess;
  } {
    const handler: EffectHandlerInterface<JobEffectEntity.Type, JobEventEntity.Type> = (effect, dispatch) => {
      if (effect.variant === 'requestAck') {
        dispatch({ 'type': 'acknowledge' });
        return;
      }
      kit.scheduleDispatch(counter.nowMs() + effect.delayMs, { 'type': 'advance' });
    };

    const machine = JobProcess.make();
    const kit = ProcessKit.create<JobStateEntity.Type, JobEventEntity.Type, JobEffectEntity.Type>({
      'handler': handler,
      'machine': machine,
      'scheduler': scheduler
    });

    return { 'kit': kit, 'machine': machine };
  }
}

// Cancellation composed via Signal: an AbortSignal drives a 'cancel'
// event into the composed ProcessKit's public dispatch().
class CancellationWiring {
  static wireCancellation(
    kit: ProcessKit<JobStateEntity.Type, JobEventEntity.Type, JobEffectEntity.Type>,
    abortSignal: AbortSignal
  ): Promise<JobStateEntity.Type> {
    return new Promise<JobStateEntity.Type>((resolve, reject) => {
      abortSignal.addEventListener('abort', () => {
        kit.dispatch({ 'type': 'cancel' }).then(resolve, reject);
      }, { 'once': true });
    });
  }
}

Try it

Loading example…

The output shows Job A completing after a same-cycle self-acknowledgment followed by a VirtualScheduler-driven advance, and Job B's final state landing on cancelled after an AbortSignal fires — even though its pending scheduled advance is still cancelled out by ProcessKit#stop().

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
handler	EffectHandlerInterface<TEffect, TEvent>	undefined — no effects handled; configure through ProcessKit.create({ machine, handler })
scheduler	SchedulerProviderInterface (RealTimeScheduler/VirtualScheduler)	RealTimeScheduler.create()

ProcessKit exposes no collaborator getters. Callers retain their machine and optional scheduler references when they need those primitives' lifecycle APIs. The interpreter is owned internally and receives the singular handler through ProcessKit.create({ machine, handler, scheduler? }).

Import ProcessKit from @studnicky/process-kit and ProcessKitConfigInterface from @studnicky/process-kit/interfaces.

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 substrate's scope boundary of its 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, outside substrate's scope.
3. Checkpoint/resume creep — stop()/teardown must stay in-memory only; do not add a save/resume pair backed by a store.

When this composition tips into orchestration

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, not a hand-rolled registry or chain of ProcessKit instances.

Documentation

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

Interfaces

@studnicky/process-kit/interfaces exports process-kit configuration contracts.

import type { ProcessKitConfigInterface } from '@studnicky/process-kit/interfaces';

Exports

Symbol	Purpose	Import path
ProcessKit	Combines an FSM, effect interpreter, and scheduler for one process.	@studnicky/process-kit

Source on GitHub