Dagonizer
Dagonizer

Migrating to Batch ​

What It Is ​

This is the upgrade guide for node code written around an older single-item mental model. Current Dagonizer nodes are batch-native: a node consumes Batch<TState> and returns RoutedBatchType<TOutput>.

Most migrations are mechanical. Keep the business logic, move the item loop inside execute(batch, context), partition items by output, and return one routed batch map.

How It Works ​

Wrap former per-item logic in a local loop over batch items, collect each item into the output batch it routes to, and return one RoutedBatchType. Scatter, reservoirs, gather strategies, and direct node tests all use this same contract.

Every node now lands on the same base contract: MonadicNode.execute(batch, context). Single-item behavior is still supported; it is represented as a batch with one item.

Diagrams, Examples, and Outputs ​

Migration is about node implementation shape, so this page uses runnable code snippets rather than a new graph. The topology does not need to change just because a node becomes batch-native.

Use these pages together:

  • Plural-native execution explains batches, partitioned routing, and the work-set scheduler.
  • Scatter Extensions demonstrates batch-native nodes, direct node calls, gather strategy folds, and reservoir configuration.
  • The Cartographer shows production-shaped per-item routing inside batch-native nodes.
  • Reference: Nodes documents the placement and execution policy surface.

What It Lets You Do ​

Use when ​

Use this guide when upgrading node implementations from single-state execution to Dagonizer's batch-native contract. The target shape is one node method that accepts a Batch<TState>, partitions items by output, and remains directly testable.

Code Samples ​

The snippets below are the migration points you edit in real code: node execution, gather strategies, direct node calls, and tests.

Per-item nodes → local batch loop ​

A node that processed one state and returned one output is a per-item node. Extend MonadicNode and keep the item loop inside execute(batch, context):

The Cartographer's RouteGeoNode is the runnable version of this pattern. It loops over the batch locally, partitions items into has-geo and needs-geo, and returns a RoutedBatchType to the engine:

ts
export class RouteGeoNode extends MonadicNode<CartographerState, 'has-geo' | 'needs-geo'> {
  readonly '@id' = 'urn:noocodec:node:route-geo';
  readonly 'name' = 'route-geo';
  readonly 'outputs' = ['has-geo', 'needs-geo'] as const;

  override get outputSchema(): Record<'has-geo' | 'needs-geo', SchemaObjectType> {
    return {
      'has-geo':   { 'type': 'object' },
      'needs-geo': { 'type': 'object' },
    };
  }

  override async execute(
    batch: Batch<CartographerState>,
    _context: NodeContextType,
  ): Promise<RoutedBatchType<'has-geo' | 'needs-geo', CartographerState>> {
    const acc = new Map<'has-geo' | 'needs-geo', ItemType<CartographerState>[]>();

    for (const item of batch) {
      const result = this.routeItem(item.state);
      for (const error of result.errors) {
        item.state.collectError(error);
      }
      const bucket = acc.get(result.output);
      if (bucket === undefined) {
        acc.set(result.output, [item]);
      } else {
        bucket.push(item);
      }
    }

    const routed = new Map<'has-geo' | 'needs-geo', Batch<CartographerState>>();
    for (const [output, items] of acc) {
      routed.set(output, Batch.from(items));
    }
    return routed;
  }

  private routeItem(state: CartographerState): NodeOutputType<'has-geo' | 'needs-geo'> {
    const geo = state.canonical.geo;
    // A source's pre-resolved geo only lets us skip the lookup when it actually
    // resolved a location — an 'UNK'/'Unmapped' placeholder (e.g. a ping whose
    // coords were out of range at the source) is NOT resolved, so it must run
    // the lookup path where validate-coords can reject the bad coords.
    const hasResolvedGeo =
      geo !== undefined &&
      geo.country.length > 0 &&
      geo.country !== 'UNK' &&
      geo.region.length > 0 &&
      geo.region !== 'Unmapped';

    if (hasResolvedGeo) {
      state.routing = { ...state.routing, 'geoLookupSkipped': true, 'geoLookupRun': false };
      return NodeOutput.create('has-geo');
    }
    state.routing = { ...state.routing, 'geoLookupRun': true, 'geoLookupSkipped': false };
    return NodeOutput.create('needs-geo');
  }
}

What changed:

  • implements NodeInterface → extends MonadicNode.
  • async execute(state, ctx) → async execute(batch, ctx) with a local loop over batch.
  • Drop the timeout boilerplate that just sets the default — MonadicNode supplies Timeout.none(). Keep it (with override) only when you set a real value, e.g. override readonly timeout = Timeout.ofMs(5000).
  • validate() / destroy() defaults are inherited; add override if you provide your own.

Batch-native nodes ​

A node that wants to process the whole batch in one call — to hit a shared LRU cache across items, vectorize, or fan out / partition — extends MonadicNode (the root) and implements execute(batch) directly:

ts
// BatchEnrichNode: processes the whole batch in one call.
// Extends MonadicNode (the root); the batch arrives as Batch<RankingState>.
// Use this pattern when you need access to all items at once (e.g., to hit a
// shared cache, vectorize, or partition by a property across the whole batch).
export class BatchEnrichNode extends MonadicNode<RankingState, 'enriched'> {
  readonly name = 'batch-enrich';
  readonly '@id' = 'urn:noocodec:node:batch-enrich';
  readonly outputs = ['enriched'] as const;
  override get outputSchema(): Record<'enriched', SchemaObjectType> {
    return { 'enriched': { 'type': 'object' } };
  }

  async execute(
    batch: Batch<RankingState>,
    _ctx: NodeContextType,
  ): Promise<RoutedBatchType<'enriched', RankingState>> {
    // Operate on the whole batch at once: normalise every item's score.
    const items = batch.items();
    const max = Math.max(...items.map((item) => item.state.candidate.score), 1);
    for (const item of items) {
      item.state.candidate = {
        ...item.state.candidate,
        score: item.state.candidate.score / max,
      };
    }
    return RoutedBatch.create('enriched', batch);
  }
}

The authoring direction ​

MonadicNode is the node base (the monad — execute(batch, context)). If you maintained a custom node base, point it at MonadicNode and keep any per-item loop local to that custom base or to each concrete node.

MonadicNode is exported from the root (@studnicky/dagonizer) and ./core; it is also re-exported from ./patterns for co-import with the pattern surface.

Gather strategies → one fold ​

The gather contract is now a single fold — initial → reduce → finalize. The seed / apply / applyIncremental methods and IncrementalGatherStrategy are removed:

ts
/**
 * TopNGatherStrategy: collects each clone's `candidate` field, keeps the
 * top-N by score, and writes the result to `topCandidates` on the parent.
 * Registered under 'top-n'. Reference from a GatherNode after scatter to
 * collect and rank clone outputs in one pass.
 */
interface ScoredItem {
  readonly score: number;
}

class TopNGatherStrategy extends GatherStrategy {
  readonly name = 'top-n';
  readonly '@id' = 'urn:noocodec:node:top-n';

  override reduce(
    _config: GatherConfigType,
    _batch: Parameters<GatherStrategy['reduce']>[1],
    _state: NodeStateInterface,
    _accessor: StateAccessorInterface,
  ): void {
    // accumulate nothing per-clone — finalize handles all records
  }

  override async finalize(
    config: GatherConfigType,
    execution: GatherExecutionType<NodeStateInterface>,
  ): Promise<void> {
    const target = config.target ?? 'topCandidates';
    const n = 3;
    const all = execution.records.map((r) =>
      execution.accessor.get(r.cloneState, 'candidate'),
    ).filter((c): c is ScoredItem => typeof c === 'object' && c !== null && 'score' in c && typeof c.score === 'number');
    const sorted = [...all].sort((a, b) => b.score - a.score).slice(0, n);
    execution.accessor.set(execution.state, target, sorted);
  }
}

GatherStrategies.register(new TopNGatherStrategy());
// GatherStrategies.resolve('top-n') now works in any scatter placement.

"Incremental" is a reduce over a batch of 1; "all-at-once" is a reduce over a batch of N. The built-in gather config (map / append / partition / collect / discard / custom) is unchanged.

Removed symbols ​

RemovedReplacement
PluralNodeInterface, the PLURAL brandone contract — NodeInterface.execute(batch)
ParallelNodeScatterNode (scatter + gather is the one fan-out)
IncrementalGatherStrategy, GatherStrategy.apply / applyIncrementalGatherStrategy.reduce (one fold)
GatherStrategy.seed()GatherStrategy.initial()

Calling a node directly ​

Tests or nodes that invoked another node's execute(state) must pass a batch and read the routed result:

ts
// Direct node invocation: create a Batch from a single state and call execute().
// Returns a RoutedBatchType; check routed.has('success') to inspect results.
// Used in tests for per-node isolation without a full dispatcher.
export class ScoreNodeRunner {
  static async score(item: string): Promise<boolean> {
    const state = new RankingState();
    state.candidate = { title: item, score: 0 };
    const node = new ScoreNode();
    const ctx: NodeContextType = {
      signal: new AbortController().signal,
      dagName: RESERVOIR_DEMO_DAG_IRI,
      nodeName: 'score',
      validateOutputs: false,
      outputSchemaValidator: null,
    };
    const routed = await node.execute(Batch.of(state), ctx);
    return routed.has('success');
  }
}

Details for Nerds ​

Checklist ​

  1. Per-item nodes: extends MonadicNode, execute(state, ctx) → execute(batch, ctx) with local routing.
  2. Batch-native nodes: extends MonadicNode, implement execute(batch, context).
  3. Custom gather strategies: rewrite to initial / reduce / finalize.
  4. Replace removed symbols.
  5. Update direct execute call sites to execute(Batch.of(state), ctx).
  6. npm run typecheck && npm run lint && npm run test.

Watched over by the Order of Dagon.

MIT License, © 2026 Andrew Studnicky